From 722636539721d08f719284e6bda0c384d22d056f Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Mon, 27 Oct 2025 14:05:35 -0700 Subject: [PATCH 1/3] Centralize truncation in conversation history (#5652) move the truncation logic to conversation history to use on any tool output. This will help us in avoiding edge cases while truncating the tool calls and mcp calls. --- codex-rs/core/src/codex.rs | 93 ------- codex-rs/core/src/conversation_history.rs | 318 +++++++++++++++++++++- codex-rs/core/src/tools/events.rs | 13 +- codex-rs/core/src/tools/mod.rs | 258 +----------------- codex-rs/core/tests/suite/mod.rs | 1 + codex-rs/core/tests/suite/truncation.rs | 270 ++++++++++++++++++ 6 files changed, 588 insertions(+), 365 deletions(-) create mode 100644 codex-rs/core/tests/suite/truncation.rs diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index f811211c4f..47b651e650 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -2366,10 +2366,6 @@ mod tests { use crate::state::TaskKind; use crate::tasks::SessionTask; use crate::tasks::SessionTaskContext; - use crate::tools::MODEL_FORMAT_HEAD_LINES; - use crate::tools::MODEL_FORMAT_MAX_BYTES; - use crate::tools::MODEL_FORMAT_MAX_LINES; - use crate::tools::MODEL_FORMAT_TAIL_LINES; use crate::tools::ToolRouter; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolOutput; @@ -2456,95 +2452,6 @@ mod tests { assert_eq!(expected, got); } - #[test] - fn model_truncation_head_tail_by_lines() { - // Build 400 short lines so line-count limit, not byte budget, triggers truncation - let lines: Vec = (1..=400).map(|i| format!("line{i}")).collect(); - let full = lines.join("\n"); - - let exec = ExecToolCallOutput { - exit_code: 0, - stdout: StreamOutput::new(String::new()), - stderr: StreamOutput::new(String::new()), - aggregated_output: StreamOutput::new(full), - duration: StdDuration::from_secs(1), - timed_out: false, - }; - - let out = format_exec_output_str(&exec); - - // Strip truncation header if present for subsequent assertions - let body = out - .strip_prefix("Total output lines: ") - .and_then(|rest| rest.split_once("\n\n").map(|x| x.1)) - .unwrap_or(out.as_str()); - - // Expect elision marker with correct counts - let omitted = 400 - MODEL_FORMAT_MAX_LINES; // 144 - let marker = format!("\n[... omitted {omitted} of 400 lines ...]\n\n"); - assert!(out.contains(&marker), "missing marker: {out}"); - - // Validate head and tail - let parts: Vec<&str> = body.split(&marker).collect(); - assert_eq!(parts.len(), 2, "expected one marker split"); - let head = parts[0]; - let tail = parts[1]; - - let expected_head: String = (1..=MODEL_FORMAT_HEAD_LINES) - .map(|i| format!("line{i}")) - .collect::>() - .join("\n"); - assert!(head.starts_with(&expected_head), "head mismatch"); - - let expected_tail: String = ((400 - MODEL_FORMAT_TAIL_LINES + 1)..=400) - .map(|i| format!("line{i}")) - .collect::>() - .join("\n"); - assert!(tail.ends_with(&expected_tail), "tail mismatch"); - } - - #[test] - fn model_truncation_respects_byte_budget() { - // Construct a large output (about 100kB) so byte budget dominates - let big_line = "x".repeat(100); - let full = std::iter::repeat_n(big_line, 1000) - .collect::>() - .join("\n"); - - let exec = ExecToolCallOutput { - exit_code: 0, - stdout: StreamOutput::new(String::new()), - stderr: StreamOutput::new(String::new()), - aggregated_output: StreamOutput::new(full.clone()), - duration: StdDuration::from_secs(1), - timed_out: false, - }; - - let out = format_exec_output_str(&exec); - // Keep strict budget on the truncated body (excluding header) - let body = out - .strip_prefix("Total output lines: ") - .and_then(|rest| rest.split_once("\n\n").map(|x| x.1)) - .unwrap_or(out.as_str()); - assert!(body.len() <= MODEL_FORMAT_MAX_BYTES, "exceeds byte budget"); - assert!(out.contains("omitted"), "should contain elision marker"); - - // Ensure head and tail are drawn from the original - assert!(full.starts_with(body.chars().take(8).collect::().as_str())); - assert!( - full.ends_with( - body.chars() - .rev() - .take(8) - .collect::() - .chars() - .rev() - .collect::() - .as_str() - ) - ); - } - #[test] fn includes_timed_out_message() { let exec = ExecToolCallOutput { diff --git a/codex-rs/core/src/conversation_history.rs b/codex-rs/core/src/conversation_history.rs index 08fa8cebed..f230a7979b 100644 --- a/codex-rs/core/src/conversation_history.rs +++ b/codex-rs/core/src/conversation_history.rs @@ -2,9 +2,18 @@ use codex_protocol::models::FunctionCallOutputPayload; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TokenUsageInfo; +use codex_utils_string::take_bytes_at_char_boundary; +use codex_utils_string::take_last_bytes_at_char_boundary; use std::ops::Deref; use tracing::error; +// Model-formatting limits: clients get full streams; only content sent to the model is truncated. +pub(crate) const MODEL_FORMAT_MAX_BYTES: usize = 10 * 1024; // 10 KiB +pub(crate) const MODEL_FORMAT_MAX_LINES: usize = 256; // lines +pub(crate) const MODEL_FORMAT_HEAD_LINES: usize = MODEL_FORMAT_MAX_LINES / 2; +pub(crate) const MODEL_FORMAT_TAIL_LINES: usize = MODEL_FORMAT_MAX_LINES - MODEL_FORMAT_HEAD_LINES; // 128 +pub(crate) const MODEL_FORMAT_HEAD_BYTES: usize = MODEL_FORMAT_MAX_BYTES / 2; + /// Transcript of conversation history #[derive(Debug, Clone, Default)] pub(crate) struct ConversationHistory { @@ -47,7 +56,8 @@ impl ConversationHistory { continue; } - self.items.push(item.clone()); + let processed = Self::process_item(&item); + self.items.push(processed); } } @@ -68,6 +78,22 @@ impl ConversationHistory { } } + pub(crate) fn replace(&mut self, items: Vec) { + self.items = items; + } + + pub(crate) fn update_token_info( + &mut self, + usage: &TokenUsage, + model_context_window: Option, + ) { + self.token_info = TokenUsageInfo::new_or_append( + &self.token_info, + &Some(usage.clone()), + model_context_window, + ); + } + /// This function enforces a couple of invariants on the in-memory history: /// 1. every call (function/custom) has a corresponding output entry /// 2. every output has a corresponding call entry @@ -253,10 +279,6 @@ impl ConversationHistory { } } - pub(crate) fn replace(&mut self, items: Vec) { - self.items = items; - } - /// Removes the corresponding paired item for the provided `item`, if any. /// /// Pairs: @@ -326,19 +348,108 @@ impl ConversationHistory { } } - pub(crate) fn update_token_info( - &mut self, - usage: &TokenUsage, - model_context_window: Option, - ) { - self.token_info = TokenUsageInfo::new_or_append( - &self.token_info, - &Some(usage.clone()), - model_context_window, - ); + fn process_item(item: &ResponseItem) -> ResponseItem { + match item { + ResponseItem::FunctionCallOutput { call_id, output } => { + let truncated = format_output_for_model_body(output.content.as_str()); + ResponseItem::FunctionCallOutput { + call_id: call_id.clone(), + output: FunctionCallOutputPayload { + content: truncated, + success: output.success, + }, + } + } + ResponseItem::CustomToolCallOutput { call_id, output } => { + let truncated = format_output_for_model_body(output); + ResponseItem::CustomToolCallOutput { + call_id: call_id.clone(), + output: truncated, + } + } + ResponseItem::Message { .. } + | ResponseItem::Reasoning { .. } + | ResponseItem::LocalShellCall { .. } + | ResponseItem::FunctionCall { .. } + | ResponseItem::WebSearchCall { .. } + | ResponseItem::CustomToolCall { .. } + | ResponseItem::GhostSnapshot { .. } + | ResponseItem::Other => item.clone(), + } } } +pub(crate) fn format_output_for_model_body(content: &str) -> String { + // Head+tail truncation for the model: show the beginning and end with an elision. + // Clients still receive full streams; only this formatted summary is capped. + let total_lines = content.lines().count(); + if content.len() <= MODEL_FORMAT_MAX_BYTES && total_lines <= MODEL_FORMAT_MAX_LINES { + return content.to_string(); + } + let output = truncate_formatted_exec_output(content, total_lines); + format!("Total output lines: {total_lines}\n\n{output}") +} + +fn truncate_formatted_exec_output(content: &str, total_lines: usize) -> String { + let segments: Vec<&str> = content.split_inclusive('\n').collect(); + let head_take = MODEL_FORMAT_HEAD_LINES.min(segments.len()); + let tail_take = MODEL_FORMAT_TAIL_LINES.min(segments.len().saturating_sub(head_take)); + let omitted = segments.len().saturating_sub(head_take + tail_take); + + let head_slice_end: usize = segments + .iter() + .take(head_take) + .map(|segment| segment.len()) + .sum(); + let tail_slice_start: usize = if tail_take == 0 { + content.len() + } else { + content.len() + - segments + .iter() + .rev() + .take(tail_take) + .map(|segment| segment.len()) + .sum::() + }; + let head_slice = &content[..head_slice_end]; + let tail_slice = &content[tail_slice_start..]; + let truncated_by_bytes = content.len() > MODEL_FORMAT_MAX_BYTES; + // this is a bit wrong. We are counting metadata lines and not just shell output lines. + let marker = if omitted > 0 { + Some(format!( + "\n[... omitted {omitted} of {total_lines} lines ...]\n\n" + )) + } else if truncated_by_bytes { + Some(format!( + "\n[... output truncated to fit {MODEL_FORMAT_MAX_BYTES} bytes ...]\n\n" + )) + } else { + None + }; + + let marker_len = marker.as_ref().map_or(0, String::len); + let base_head_budget = MODEL_FORMAT_HEAD_BYTES.min(MODEL_FORMAT_MAX_BYTES); + let head_budget = base_head_budget.min(MODEL_FORMAT_MAX_BYTES.saturating_sub(marker_len)); + let head_part = take_bytes_at_char_boundary(head_slice, head_budget); + let mut result = String::with_capacity(MODEL_FORMAT_MAX_BYTES.min(content.len())); + + result.push_str(head_part); + if let Some(marker_text) = marker.as_ref() { + result.push_str(marker_text); + } + + let remaining = MODEL_FORMAT_MAX_BYTES.saturating_sub(result.len()); + if remaining == 0 { + return result; + } + + let tail_part = take_last_bytes_at_char_boundary(tail_slice, remaining); + result.push_str(tail_part); + + result +} + #[inline] fn error_or_panic(message: String) { if cfg!(debug_assertions) || env!("CARGO_PKG_VERSION").contains("alpha") { @@ -533,6 +644,183 @@ mod tests { assert_eq!(h.contents(), vec![]); } + #[test] + fn record_items_truncates_function_call_output_content() { + let mut history = ConversationHistory::new(); + let long_line = "a very long line to trigger truncation\n"; + let long_output = long_line.repeat(2_500); + let item = ResponseItem::FunctionCallOutput { + call_id: "call-100".to_string(), + output: FunctionCallOutputPayload { + content: long_output.clone(), + success: Some(true), + }, + }; + + history.record_items([&item]); + + assert_eq!(history.items.len(), 1); + match &history.items[0] { + ResponseItem::FunctionCallOutput { output, .. } => { + assert_ne!(output.content, long_output); + assert!( + output.content.starts_with("Total output lines:"), + "expected truncated summary, got {}", + output.content + ); + } + other => panic!("unexpected history item: {other:?}"), + } + } + + #[test] + fn record_items_truncates_custom_tool_call_output_content() { + let mut history = ConversationHistory::new(); + let line = "custom output that is very long\n"; + let long_output = line.repeat(2_500); + let item = ResponseItem::CustomToolCallOutput { + call_id: "tool-200".to_string(), + output: long_output.clone(), + }; + + history.record_items([&item]); + + assert_eq!(history.items.len(), 1); + match &history.items[0] { + ResponseItem::CustomToolCallOutput { output, .. } => { + assert_ne!(output, &long_output); + assert!( + output.starts_with("Total output lines:"), + "expected truncated summary, got {output}" + ); + } + other => panic!("unexpected history item: {other:?}"), + } + } + + // The following tests were adapted from tools::mod truncation tests to + // target the new truncation functions in conversation_history. + + use regex_lite::Regex; + + fn assert_truncated_message_matches(message: &str, line: &str, total_lines: usize) { + let pattern = truncated_message_pattern(line, total_lines); + let regex = Regex::new(&pattern).unwrap_or_else(|err| { + panic!("failed to compile regex {pattern}: {err}"); + }); + let captures = regex + .captures(message) + .unwrap_or_else(|| panic!("message failed to match pattern {pattern}: {message}")); + let body = captures + .name("body") + .expect("missing body capture") + .as_str(); + assert!( + body.len() <= MODEL_FORMAT_MAX_BYTES, + "body exceeds byte limit: {} bytes", + body.len() + ); + } + + fn truncated_message_pattern(line: &str, total_lines: usize) -> String { + let head_take = MODEL_FORMAT_HEAD_LINES.min(total_lines); + let tail_take = MODEL_FORMAT_TAIL_LINES.min(total_lines.saturating_sub(head_take)); + let omitted = total_lines.saturating_sub(head_take + tail_take); + let escaped_line = regex_lite::escape(line); + if omitted == 0 { + return format!( + r"(?s)^Total output lines: {total_lines}\n\n(?P{escaped_line}.*\n\[\.{{3}} output truncated to fit {MODEL_FORMAT_MAX_BYTES} bytes \.{{3}}]\n\n.*)$", + ); + } + format!( + r"(?s)^Total output lines: {total_lines}\n\n(?P{escaped_line}.*\n\[\.{{3}} omitted {omitted} of {total_lines} lines \.{{3}}]\n\n.*)$", + ) + } + + #[test] + fn format_exec_output_truncates_large_error() { + let line = "very long execution error line that should trigger truncation\n"; + let large_error = line.repeat(2_500); // way beyond both byte and line limits + + let truncated = format_output_for_model_body(&large_error); + + let total_lines = large_error.lines().count(); + assert_truncated_message_matches(&truncated, line, total_lines); + assert_ne!(truncated, large_error); + } + + #[test] + fn format_exec_output_marks_byte_truncation_without_omitted_lines() { + let long_line = "a".repeat(MODEL_FORMAT_MAX_BYTES + 50); + let truncated = format_output_for_model_body(&long_line); + + assert_ne!(truncated, long_line); + let marker_line = + format!("[... output truncated to fit {MODEL_FORMAT_MAX_BYTES} bytes ...]"); + assert!( + truncated.contains(&marker_line), + "missing byte truncation marker: {truncated}" + ); + assert!( + !truncated.contains("omitted"), + "line omission marker should not appear when no lines were dropped: {truncated}" + ); + } + + #[test] + fn format_exec_output_returns_original_when_within_limits() { + let content = "example output\n".repeat(10); + + assert_eq!(format_output_for_model_body(&content), content); + } + + #[test] + fn format_exec_output_reports_omitted_lines_and_keeps_head_and_tail() { + let total_lines = MODEL_FORMAT_MAX_LINES + 100; + let content: String = (0..total_lines) + .map(|idx| format!("line-{idx}\n")) + .collect(); + + let truncated = format_output_for_model_body(&content); + let omitted = total_lines - MODEL_FORMAT_MAX_LINES; + let expected_marker = format!("[... omitted {omitted} of {total_lines} lines ...]"); + + assert!( + truncated.contains(&expected_marker), + "missing omitted marker: {truncated}" + ); + assert!( + truncated.contains("line-0\n"), + "expected head line to remain: {truncated}" + ); + + let last_line = format!("line-{}\n", total_lines - 1); + assert!( + truncated.contains(&last_line), + "expected tail line to remain: {truncated}" + ); + } + + #[test] + fn format_exec_output_prefers_line_marker_when_both_limits_exceeded() { + let total_lines = MODEL_FORMAT_MAX_LINES + 42; + let long_line = "x".repeat(256); + let content: String = (0..total_lines) + .map(|idx| format!("line-{idx}-{long_line}\n")) + .collect(); + + let truncated = format_output_for_model_body(&content); + + assert!( + truncated.contains("[... omitted 42 of 298 lines ...]"), + "expected omitted marker when line count exceeds limit: {truncated}" + ); + assert!( + !truncated.contains("output truncated to fit"), + "line omission marker should take precedence over byte marker: {truncated}" + ); + } + //TODO(aibrahim): run CI in release mode. #[cfg(not(debug_assertions))] #[test] diff --git a/codex-rs/core/src/tools/events.rs b/codex-rs/core/src/tools/events.rs index 5578295547..cb267c8917 100644 --- a/codex-rs/core/src/tools/events.rs +++ b/codex-rs/core/src/tools/events.rs @@ -19,7 +19,6 @@ use std::path::Path; use std::path::PathBuf; use std::time::Duration; -use super::format_exec_output; use super::format_exec_output_str; #[derive(Clone, Copy)] @@ -146,7 +145,7 @@ impl ToolEmitter { (*message).to_string(), -1, Duration::ZERO, - format_exec_output(&message), + message.clone(), ) .await; } @@ -241,7 +240,7 @@ impl ToolEmitter { (*message).to_string(), -1, Duration::ZERO, - format_exec_output(&message), + message.clone(), ) .await; } @@ -277,7 +276,7 @@ impl ToolEmitter { } Err(ToolError::Codex(err)) => { let message = format!("execution error: {err:?}"); - let response = super::format_exec_output(&message); + let response = message.clone(); event = ToolEventStage::Failure(ToolEventFailure::Message(message)); Err(FunctionCallError::RespondToModel(response)) } @@ -289,9 +288,9 @@ impl ToolEmitter { } else { msg }; - let response = super::format_exec_output(&normalized); - event = ToolEventStage::Failure(ToolEventFailure::Message(normalized)); - Err(FunctionCallError::RespondToModel(response)) + let response = &normalized; + event = ToolEventStage::Failure(ToolEventFailure::Message(normalized.clone())); + Err(FunctionCallError::RespondToModel(response.clone())) } }; self.emit(ctx, event).await; diff --git a/codex-rs/core/src/tools/mod.rs b/codex-rs/core/src/tools/mod.rs index f22d064b51..f5ae4a12e1 100644 --- a/codex-rs/core/src/tools/mod.rs +++ b/codex-rs/core/src/tools/mod.rs @@ -9,19 +9,11 @@ pub mod runtimes; pub mod sandboxing; pub mod spec; +use crate::conversation_history::format_output_for_model_body; use crate::exec::ExecToolCallOutput; -use codex_utils_string::take_bytes_at_char_boundary; -use codex_utils_string::take_last_bytes_at_char_boundary; pub use router::ToolRouter; use serde::Serialize; -// Model-formatting limits: clients get full streams; only content sent to the model is truncated. -pub(crate) const MODEL_FORMAT_MAX_BYTES: usize = 10 * 1024; // 10 KiB -pub(crate) const MODEL_FORMAT_MAX_LINES: usize = 256; // lines -pub(crate) const MODEL_FORMAT_HEAD_LINES: usize = MODEL_FORMAT_MAX_LINES / 2; -pub(crate) const MODEL_FORMAT_TAIL_LINES: usize = MODEL_FORMAT_MAX_LINES - MODEL_FORMAT_HEAD_LINES; // 128 -pub(crate) const MODEL_FORMAT_HEAD_BYTES: usize = MODEL_FORMAT_MAX_BYTES / 2; - // Telemetry preview limits: keep log events smaller than model budgets. pub(crate) const TELEMETRY_PREVIEW_MAX_BYTES: usize = 2 * 1024; // 2 KiB pub(crate) const TELEMETRY_PREVIEW_MAX_LINES: usize = 64; // lines @@ -73,249 +65,15 @@ pub fn format_exec_output_str(exec_output: &ExecToolCallOutput) -> String { let content = aggregated_output.text.as_str(); - if exec_output.timed_out { - let prefixed = format!( + let body = if exec_output.timed_out { + format!( "command timed out after {} milliseconds\n{content}", exec_output.duration.as_millis() - ); - return format_exec_output(&prefixed); - } - - format_exec_output(content) -} - -pub(super) fn format_exec_output(content: &str) -> String { - // Head+tail truncation for the model: show the beginning and end with an elision. - // Clients still receive full streams; only this formatted summary is capped. - let total_lines = content.lines().count(); - if content.len() <= MODEL_FORMAT_MAX_BYTES && total_lines <= MODEL_FORMAT_MAX_LINES { - return content.to_string(); - } - let output = truncate_formatted_exec_output(content, total_lines); - format!("Total output lines: {total_lines}\n\n{output}") -} - -fn truncate_formatted_exec_output(content: &str, total_lines: usize) -> String { - let segments: Vec<&str> = content.split_inclusive('\n').collect(); - let head_take = MODEL_FORMAT_HEAD_LINES.min(segments.len()); - let tail_take = MODEL_FORMAT_TAIL_LINES.min(segments.len().saturating_sub(head_take)); - let omitted = segments.len().saturating_sub(head_take + tail_take); - - let head_slice_end: usize = segments - .iter() - .take(head_take) - .map(|segment| segment.len()) - .sum(); - let tail_slice_start: usize = if tail_take == 0 { - content.len() - } else { - content.len() - - segments - .iter() - .rev() - .take(tail_take) - .map(|segment| segment.len()) - .sum::() - }; - let head_slice = &content[..head_slice_end]; - let tail_slice = &content[tail_slice_start..]; - let truncated_by_bytes = content.len() > MODEL_FORMAT_MAX_BYTES; - let marker = if omitted > 0 { - Some(format!( - "\n[... omitted {omitted} of {total_lines} lines ...]\n\n" - )) - } else if truncated_by_bytes { - Some(format!( - "\n[... output truncated to fit {MODEL_FORMAT_MAX_BYTES} bytes ...]\n\n" - )) - } else { - None - }; - - let marker_len = marker.as_ref().map_or(0, String::len); - let base_head_budget = MODEL_FORMAT_HEAD_BYTES.min(MODEL_FORMAT_MAX_BYTES); - let head_budget = base_head_budget.min(MODEL_FORMAT_MAX_BYTES.saturating_sub(marker_len)); - let head_part = take_bytes_at_char_boundary(head_slice, head_budget); - let mut result = String::with_capacity(MODEL_FORMAT_MAX_BYTES.min(content.len())); - - result.push_str(head_part); - if let Some(marker_text) = marker.as_ref() { - result.push_str(marker_text); - } - - let remaining = MODEL_FORMAT_MAX_BYTES.saturating_sub(result.len()); - if remaining == 0 { - return result; - } - - let tail_part = take_last_bytes_at_char_boundary(tail_slice, remaining); - result.push_str(tail_part); - - result -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::function_tool::FunctionCallError; - use regex_lite::Regex; - - fn truncate_function_error(err: FunctionCallError) -> FunctionCallError { - match err { - FunctionCallError::RespondToModel(msg) => { - FunctionCallError::RespondToModel(format_exec_output(&msg)) - } - FunctionCallError::Denied(msg) => FunctionCallError::Denied(format_exec_output(&msg)), - FunctionCallError::Fatal(msg) => FunctionCallError::Fatal(format_exec_output(&msg)), - other => other, - } - } - - fn assert_truncated_message_matches(message: &str, line: &str, total_lines: usize) { - let pattern = truncated_message_pattern(line, total_lines); - let regex = Regex::new(&pattern).unwrap_or_else(|err| { - panic!("failed to compile regex {pattern}: {err}"); - }); - let captures = regex - .captures(message) - .unwrap_or_else(|| panic!("message failed to match pattern {pattern}: {message}")); - let body = captures - .name("body") - .expect("missing body capture") - .as_str(); - assert!( - body.len() <= MODEL_FORMAT_MAX_BYTES, - "body exceeds byte limit: {} bytes", - body.len() - ); - } - - fn truncated_message_pattern(line: &str, total_lines: usize) -> String { - let head_take = MODEL_FORMAT_HEAD_LINES.min(total_lines); - let tail_take = MODEL_FORMAT_TAIL_LINES.min(total_lines.saturating_sub(head_take)); - let omitted = total_lines.saturating_sub(head_take + tail_take); - let escaped_line = regex_lite::escape(line); - if omitted == 0 { - return format!( - r"(?s)^Total output lines: {total_lines}\n\n(?P{escaped_line}.*\n\[\.{{3}} output truncated to fit {MODEL_FORMAT_MAX_BYTES} bytes \.{{3}}]\n\n.*)$", - ); - } - format!( - r"(?s)^Total output lines: {total_lines}\n\n(?P{escaped_line}.*\n\[\.{{3}} omitted {omitted} of {total_lines} lines \.{{3}}]\n\n.*)$", ) - } + } else { + content.to_string() + }; - #[test] - fn truncate_formatted_exec_output_truncates_large_error() { - let line = "very long execution error line that should trigger truncation\n"; - let large_error = line.repeat(2_500); // way beyond both byte and line limits - - let truncated = format_exec_output(&large_error); - - let total_lines = large_error.lines().count(); - assert_truncated_message_matches(&truncated, line, total_lines); - assert_ne!(truncated, large_error); - } - - #[test] - fn truncate_function_error_trims_respond_to_model() { - let line = "respond-to-model error that should be truncated\n"; - let huge = line.repeat(3_000); - let total_lines = huge.lines().count(); - - let err = truncate_function_error(FunctionCallError::RespondToModel(huge)); - match err { - FunctionCallError::RespondToModel(message) => { - assert_truncated_message_matches(&message, line, total_lines); - } - other => panic!("unexpected error variant: {other:?}"), - } - } - - #[test] - fn truncate_function_error_trims_fatal() { - let line = "fatal error output that should be truncated\n"; - let huge = line.repeat(3_000); - let total_lines = huge.lines().count(); - - let err = truncate_function_error(FunctionCallError::Fatal(huge)); - match err { - FunctionCallError::Fatal(message) => { - assert_truncated_message_matches(&message, line, total_lines); - } - other => panic!("unexpected error variant: {other:?}"), - } - } - - #[test] - fn truncate_formatted_exec_output_marks_byte_truncation_without_omitted_lines() { - let long_line = "a".repeat(MODEL_FORMAT_MAX_BYTES + 50); - let truncated = format_exec_output(&long_line); - - assert_ne!(truncated, long_line); - let marker_line = - format!("[... output truncated to fit {MODEL_FORMAT_MAX_BYTES} bytes ...]"); - assert!( - truncated.contains(&marker_line), - "missing byte truncation marker: {truncated}" - ); - assert!( - !truncated.contains("omitted"), - "line omission marker should not appear when no lines were dropped: {truncated}" - ); - } - - #[test] - fn truncate_formatted_exec_output_returns_original_when_within_limits() { - let content = "example output\n".repeat(10); - - assert_eq!(format_exec_output(&content), content); - } - - #[test] - fn truncate_formatted_exec_output_reports_omitted_lines_and_keeps_head_and_tail() { - let total_lines = MODEL_FORMAT_MAX_LINES + 100; - let content: String = (0..total_lines) - .map(|idx| format!("line-{idx}\n")) - .collect(); - - let truncated = format_exec_output(&content); - let omitted = total_lines - MODEL_FORMAT_MAX_LINES; - let expected_marker = format!("[... omitted {omitted} of {total_lines} lines ...]"); - - assert!( - truncated.contains(&expected_marker), - "missing omitted marker: {truncated}" - ); - assert!( - truncated.contains("line-0\n"), - "expected head line to remain: {truncated}" - ); - - let last_line = format!("line-{}\n", total_lines - 1); - assert!( - truncated.contains(&last_line), - "expected tail line to remain: {truncated}" - ); - } - - #[test] - fn truncate_formatted_exec_output_prefers_line_marker_when_both_limits_exceeded() { - let total_lines = MODEL_FORMAT_MAX_LINES + 42; - let long_line = "x".repeat(256); - let content: String = (0..total_lines) - .map(|idx| format!("line-{idx}-{long_line}\n")) - .collect(); - - let truncated = format_exec_output(&content); - - assert!( - truncated.contains("[... omitted 42 of 298 lines ...]"), - "expected omitted marker when line count exceeds limit: {truncated}" - ); - assert!( - !truncated.contains("output truncated to fit"), - "line omission marker should take precedence over byte marker: {truncated}" - ); - } + // Truncate for model consumption before serialization. + format_output_for_model_body(&body) } diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index bfaef15a92..300ca146ec 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -33,6 +33,7 @@ mod stream_no_completed; mod tool_harness; mod tool_parallelism; mod tools; +mod truncation; mod unified_exec; mod user_notification; mod view_image; diff --git a/codex-rs/core/tests/suite/truncation.rs b/codex-rs/core/tests/suite/truncation.rs new file mode 100644 index 0000000000..74d90f1de7 --- /dev/null +++ b/codex-rs/core/tests/suite/truncation.rs @@ -0,0 +1,270 @@ +#![cfg(not(target_os = "windows"))] +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use anyhow::Context; +use anyhow::Result; +use codex_core::features::Feature; +use codex_core::model_family::find_family_for_model; +use codex_core::protocol::SandboxPolicy; +use core_test_support::assert_regex_match; +use core_test_support::responses; +use core_test_support::responses::ev_assistant_message; +use core_test_support::responses::ev_completed; +use core_test_support::responses::ev_function_call; +use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_sse_once_match; +use core_test_support::responses::mount_sse_sequence; +use core_test_support::responses::sse; +use core_test_support::responses::start_mock_server; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use escargot::CargoBuild; +use regex_lite::Regex; +use serde_json::Value; +use serde_json::json; +use wiremock::matchers::any; + +// Verifies byte-truncation formatting for function error output (RespondToModel errors) +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn truncate_function_error_trims_respond_to_model() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let mut builder = test_codex().with_config(|config| { + // Use the test model that wires function tools like grep_files + config.model = "test-gpt-5-codex".to_string(); + config.model_family = + find_family_for_model("test-gpt-5-codex").expect("model family for test model"); + }); + let test = builder.build(&server).await?; + + // Construct a very long, non-existent path to force a RespondToModel error with a large message + let long_path = "a".repeat(20_000); + let call_id = "grep-huge-error"; + let args = json!({ + "pattern": "alpha", + "path": long_path, + "limit": 10 + }); + let responses = vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_function_call(call_id, "grep_files", &serde_json::to_string(&args)?), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_assistant_message("msg-1", "done"), + ev_completed("resp-2"), + ]), + ]; + let mock = mount_sse_sequence(&server, responses).await; + + test.submit_turn_with_policy( + "trigger grep_files with long path to test truncation", + SandboxPolicy::DangerFullAccess, + ) + .await?; + + let output = mock + .function_call_output_text(call_id) + .context("function error output present")?; + + tracing::debug!(output = %output, "truncated function error output"); + + // Expect plaintext with byte-truncation marker and no omitted-lines marker + assert!( + serde_json::from_str::(&output).is_err(), + "expected error output to be plain text", + ); + let truncated_pattern = r#"(?s)^Total output lines: 1\s+.*\[\.\.\. output truncated to fit 10240 bytes \.\.\.\]\s*$"#; + assert_regex_match(truncated_pattern, &output); + assert!( + !output.contains("omitted"), + "line omission marker should not appear when no lines were dropped: {output}" + ); + + Ok(()) +} + +// Verifies that a standard tool call (shell) exceeding the model formatting +// limits is truncated before being sent back to the model. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn tool_call_output_exceeds_limit_truncated_for_model() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + + // Use a model that exposes the generic shell tool. + let mut builder = test_codex().with_config(|config| { + config.model = "gpt-5-codex".to_string(); + config.model_family = + find_family_for_model("gpt-5-codex").expect("gpt-5-codex is a model family"); + }); + let fixture = builder.build(&server).await?; + + let call_id = "shell-too-large"; + let args = serde_json::json!({ + "command": ["/bin/sh", "-c", "seq 1 400"], + "timeout_ms": 5_000, + }); + + // First response: model tells us to run the tool; second: complete the turn. + mount_sse_once_match( + &server, + any(), + sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_function_call(call_id, "shell", &serde_json::to_string(&args)?), + responses::ev_completed("resp-1"), + ]), + ) + .await; + let mock2 = mount_sse_once_match( + &server, + any(), + sse(vec![ + responses::ev_assistant_message("msg-1", "done"), + responses::ev_completed("resp-2"), + ]), + ) + .await; + + fixture + .submit_turn_with_policy("trigger big shell output", SandboxPolicy::DangerFullAccess) + .await?; + + // Inspect what we sent back to the model; it should contain a truncated + // function_call_output for the shell call. + let output = mock2 + .single_request() + .function_call_output_text(call_id) + .context("function_call_output present for shell call")?; + + // Expect plain text (not JSON) with truncation markers and line elision. + assert!( + serde_json::from_str::(&output).is_err(), + "expected truncated shell output to be plain text" + ); + let truncated_pattern = r#"(?s)^Exit code: 0 +Wall time: .* seconds +Total output lines: 400 +Output: +1 +2 +3 +4 +5 +6 +.* +\[\.{3} omitted 144 of 400 lines \.{3}\] + +.* +396 +397 +398 +399 +400 +$"#; + assert_regex_match(truncated_pattern, &output); + + Ok(()) +} + +// Verifies that an MCP tool call result exceeding the model formatting limits +// is truncated before being sent back to the model. +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn mcp_tool_call_output_exceeds_limit_truncated_for_model() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + + let call_id = "rmcp-truncated"; + let server_name = "rmcp"; + let tool_name = format!("mcp__{server_name}__echo"); + + // Build a very large message to exceed 10KiB once serialized. + let large_msg = "long-message-with-newlines-".repeat(600); + let args_json = serde_json::json!({ "message": large_msg }); + + mount_sse_once_match( + &server, + any(), + sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_function_call(call_id, &tool_name, &args_json.to_string()), + responses::ev_completed("resp-1"), + ]), + ) + .await; + let mock2 = mount_sse_once_match( + &server, + any(), + sse(vec![ + responses::ev_assistant_message("msg-1", "rmcp echo tool completed."), + responses::ev_completed("resp-2"), + ]), + ) + .await; + + // Compile the rmcp stdio test server and configure it. + let rmcp_test_server_bin = CargoBuild::new() + .package("codex-rmcp-client") + .bin("test_stdio_server") + .run()? + .path() + .to_string_lossy() + .into_owned(); + + let mut builder = test_codex().with_config(move |config| { + config.features.enable(Feature::RmcpClient); + config.mcp_servers.insert( + server_name.to_string(), + codex_core::config_types::McpServerConfig { + transport: codex_core::config_types::McpServerTransportConfig::Stdio { + command: rmcp_test_server_bin, + args: Vec::new(), + env: None, + env_vars: Vec::new(), + cwd: None, + }, + enabled: true, + startup_timeout_sec: Some(std::time::Duration::from_secs(10)), + tool_timeout_sec: None, + enabled_tools: None, + disabled_tools: None, + }, + ); + }); + let fixture = builder.build(&server).await?; + + fixture + .submit_turn_with_policy( + "call the rmcp echo tool with a very large message", + SandboxPolicy::ReadOnly, + ) + .await?; + + // The MCP tool call output is converted to a function_call_output for the model. + let output = mock2 + .single_request() + .function_call_output_text(call_id) + .context("function_call_output present for rmcp call")?; + + // Expect plain text with byte-based truncation marker. + assert!( + serde_json::from_str::(&output).is_err(), + "expected truncated MCP output to be plain text" + ); + assert!( + output.starts_with("Total output lines: 1\n\n{"), + "expected total line header and JSON head, got: {output}" + ); + let byte_marker = Regex::new(r"\[\.\.\. output truncated to fit 10240 bytes \.\.\.\]") + .expect("compile regex"); + assert!( + byte_marker.is_match(&output), + "expected byte truncation marker, got: {output}" + ); + + Ok(()) +} From 67a219ffc2c08eb0e7ac4176908964458a1fd2a4 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 27 Oct 2025 14:06:13 -0700 Subject: [PATCH 2/3] fix: move account struct to app-server-protocol and use camelCase (#5829) Makes sense to move this struct to `app-server-protocol/` since we want to serialize as camelCase, but we don't for structs defined in `protocol/` It was: ``` export type Account = { "type": "ApiKey", api_key: string, } | { "type": "chatgpt", email: string | null, plan_type: PlanType, }; ``` But we want: ``` export type Account = { "type": "apiKey", apiKey: string, } | { "type": "chatgpt", email: string | null, planType: PlanType, }; ``` --- codex-rs/app-server-protocol/src/protocol.rs | 47 +++++++++++++++++++- codex-rs/protocol/src/account.rs | 15 ------- 2 files changed, 46 insertions(+), 16 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol.rs b/codex-rs/app-server-protocol/src/protocol.rs index f164ce5601..65a233a376 100644 --- a/codex-rs/app-server-protocol/src/protocol.rs +++ b/codex-rs/app-server-protocol/src/protocol.rs @@ -5,7 +5,7 @@ use crate::JSONRPCNotification; use crate::JSONRPCRequest; use crate::RequestId; use codex_protocol::ConversationId; -use codex_protocol::account::Account; +use codex_protocol::account::PlanType; use codex_protocol::config_types::ForcedLoginMethod; use codex_protocol::config_types::ReasoningEffort; use codex_protocol::config_types::ReasoningSummary; @@ -236,6 +236,22 @@ client_request_definitions! { }, } +#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, JsonSchema, TS)] +#[serde(tag = "type", rename_all = "camelCase")] +#[ts(tag = "type")] +pub enum Account { + #[serde(rename = "apiKey", rename_all = "camelCase")] + #[ts(rename = "apiKey", rename_all = "camelCase")] + ApiKey { api_key: String }, + + #[serde(rename = "chatgpt", rename_all = "camelCase")] + #[ts(rename = "chatgpt", rename_all = "camelCase")] + ChatGpt { + email: Option, + plan_type: PlanType, + }, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] pub struct GetAccountResponse { @@ -1246,6 +1262,35 @@ mod tests { Ok(()) } + #[test] + fn account_serializes_fields_in_camel_case() -> Result<()> { + let api_key = Account::ApiKey { + api_key: "secret".to_string(), + }; + assert_eq!( + json!({ + "type": "apiKey", + "apiKey": "secret", + }), + serde_json::to_value(&api_key)?, + ); + + let chatgpt = Account::ChatGpt { + email: Some("user@example.com".to_string()), + plan_type: PlanType::Plus, + }; + assert_eq!( + json!({ + "type": "chatgpt", + "email": "user@example.com", + "planType": "plus", + }), + serde_json::to_value(&chatgpt)?, + ); + + Ok(()) + } + #[test] fn serialize_list_models() -> Result<()> { let request = ClientRequest::ListModels { diff --git a/codex-rs/protocol/src/account.rs b/codex-rs/protocol/src/account.rs index 1d63910c64..fb707c3a73 100644 --- a/codex-rs/protocol/src/account.rs +++ b/codex-rs/protocol/src/account.rs @@ -18,18 +18,3 @@ pub enum PlanType { #[serde(other)] Unknown, } - -#[derive(Debug, Clone, PartialEq, Deserialize, Serialize, JsonSchema, TS)] -#[serde(tag = "type")] -#[ts(tag = "type")] -pub enum Account { - ApiKey { - api_key: String, - }, - #[serde(rename = "chatgpt")] - #[ts(rename = "chatgpt")] - ChatGpt { - email: Option, - plan_type: PlanType, - }, -} From b0bdc04c309385a7529ec06fabdef0662274cb01 Mon Sep 17 00:00:00 2001 From: Gabriel Peal Date: Mon, 27 Oct 2025 14:55:57 -0700 Subject: [PATCH 3/3] [MCP] Render MCP tool call result images to the model (#5600) It's pretty amazing we have gotten here without the ability for the model to see image content from MCP tool calls. This PR builds off of 4391 and fixes #4819. I would like @KKcorps to get adequete credit here but I also want to get this fix in ASAP so I gave him a week to update it and haven't gotten a response so I'm going to take it across the finish line. This test highlights how absured the current situation is. I asked the model to read this image using the Chrome MCP image After this change, it correctly outputs: > Captured the page: image dhows a dark terminal-style UI labeled `OpenAI Codex (v0.0.0)` with prompt `model: gpt-5-codex medium` and working directory `/codex/codex-rs` (and more) Before this change, it said: > Took the full-page screenshot you asked for. It shows a long, horizontally repeating pattern of stylized people in orange, light-blue, and mustard clothing, holding hands in alternating poses against a white background. No text or other graphics-just rows of flat illustration stretching off to the right. Without this change, the Figma, Playwright, Chrome, and other visual MCP servers are pretty much entirely useless. I tested this change with the openai respones api as well as a third party completions api --- codex-rs/core/src/chat_completions.rs | 46 ++- codex-rs/core/src/codex.rs | 51 +-- codex-rs/core/src/conversation_history.rs | 26 +- codex-rs/core/src/mcp_tool_call.rs | 1 + codex-rs/core/src/response_processing.rs | 7 +- codex-rs/core/src/tools/context.rs | 19 +- .../core/src/tools/handlers/apply_patch.rs | 2 + .../core/src/tools/handlers/grep_files.rs | 2 + codex-rs/core/src/tools/handlers/list_dir.rs | 1 + codex-rs/core/src/tools/handlers/mcp.rs | 12 +- .../core/src/tools/handlers/mcp_resource.rs | 16 +- codex-rs/core/src/tools/handlers/plan.rs | 1 + codex-rs/core/src/tools/handlers/read_file.rs | 1 + codex-rs/core/src/tools/handlers/shell.rs | 3 + codex-rs/core/src/tools/handlers/test_sync.rs | 1 + .../core/src/tools/handlers/unified_exec.rs | 1 + .../core/src/tools/handlers/view_image.rs | 1 + codex-rs/core/src/tools/parallel.rs | 2 +- codex-rs/core/src/tools/router.rs | 1 + codex-rs/core/tests/suite/rmcp_client.rs | 352 ++++++++++++++++++ codex-rs/protocol/src/models.rs | 256 +++++++++++-- codex-rs/protocol/src/protocol.rs | 6 +- .../rmcp-client/src/bin/test_stdio_server.rs | 47 ++- codex-rs/tui/src/history_cell.rs | 2 +- 24 files changed, 749 insertions(+), 108 deletions(-) diff --git a/codex-rs/core/src/chat_completions.rs b/codex-rs/core/src/chat_completions.rs index 76cbdf9f0a..1fb2230d79 100644 --- a/codex-rs/core/src/chat_completions.rs +++ b/codex-rs/core/src/chat_completions.rs @@ -17,6 +17,7 @@ use crate::util::backoff; use bytes::Bytes; use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::models::ContentItem; +use codex_protocol::models::FunctionCallOutputContentItem; use codex_protocol::models::ReasoningItemContent; use codex_protocol::models::ResponseItem; use eventsource_stream::Eventsource; @@ -159,16 +160,26 @@ pub(crate) async fn stream_chat_completions( for (idx, item) in input.iter().enumerate() { match item { ResponseItem::Message { role, content, .. } => { + // Build content either as a plain string (typical for assistant text) + // or as an array of content items when images are present (user/tool multimodal). let mut text = String::new(); + let mut items: Vec = Vec::new(); + let mut saw_image = false; + for c in content { match c { ContentItem::InputText { text: t } | ContentItem::OutputText { text: t } => { text.push_str(t); + items.push(json!({"type":"text","text": t})); + } + ContentItem::InputImage { image_url } => { + saw_image = true; + items.push(json!({"type":"image_url","image_url": {"url": image_url}})); } - _ => {} } } + // Skip exact-duplicate assistant messages. if role == "assistant" { if let Some(prev) = &last_assistant_text @@ -179,7 +190,17 @@ pub(crate) async fn stream_chat_completions( last_assistant_text = Some(text.clone()); } - let mut msg = json!({"role": role, "content": text}); + // For assistant messages, always send a plain string for compatibility. + // For user messages, if an image is present, send an array of content items. + let content_value = if role == "assistant" { + json!(text) + } else if saw_image { + json!(items) + } else { + json!(text) + }; + + let mut msg = json!({"role": role, "content": content_value}); if role == "assistant" && let Some(reasoning) = reasoning_by_anchor_index.get(&idx) && let Some(obj) = msg.as_object_mut() @@ -238,10 +259,29 @@ pub(crate) async fn stream_chat_completions( messages.push(msg); } ResponseItem::FunctionCallOutput { call_id, output } => { + // Prefer structured content items when available (e.g., images) + // otherwise fall back to the legacy plain-string content. + let content_value = if let Some(items) = &output.content_items { + let mapped: Vec = items + .iter() + .map(|it| match it { + FunctionCallOutputContentItem::InputText { text } => { + json!({"type":"text","text": text}) + } + FunctionCallOutputContentItem::InputImage { image_url } => { + json!({"type":"image_url","image_url": {"url": image_url}}) + } + }) + .collect(); + json!(mapped) + } else { + json!(output.content) + }; + messages.push(json!({ "role": "tool", "tool_call_id": call_id, - "content": output.content, + "content": content_value, })); } ResponseItem::CustomToolCall { diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 47b651e650..1e33335f4b 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -2047,7 +2047,7 @@ async fn try_run_turn( call_id: String::new(), output: FunctionCallOutputPayload { content: msg.to_string(), - success: None, + ..Default::default() }, }; add_completed(ProcessedResponseItem { @@ -2061,7 +2061,7 @@ async fn try_run_turn( call_id: String::new(), output: FunctionCallOutputPayload { content: message, - success: None, + ..Default::default() }, }; add_completed(ProcessedResponseItem { @@ -2199,41 +2199,6 @@ pub(super) fn get_last_assistant_message_from_turn(responses: &[ResponseItem]) - } }) } -pub(crate) fn convert_call_tool_result_to_function_call_output_payload( - call_tool_result: &CallToolResult, -) -> FunctionCallOutputPayload { - let CallToolResult { - content, - is_error, - structured_content, - } = call_tool_result; - - // In terms of what to send back to the model, we prefer structured_content, - // if available, and fallback to content, otherwise. - let mut is_success = is_error != &Some(true); - let content = if let Some(structured_content) = structured_content - && structured_content != &serde_json::Value::Null - && let Ok(serialized_structured_content) = serde_json::to_string(&structured_content) - { - serialized_structured_content - } else { - match serde_json::to_string(&content) { - Ok(serialized_content) => serialized_content, - Err(err) => { - // If we could not serialize either content or structured_content to - // JSON, flag this as an error. - is_success = false; - err.to_string() - } - } - }; - - FunctionCallOutputPayload { - content, - success: Some(is_success), - } -} - /// Emits an ExitedReviewMode Event with optional ReviewOutput, /// and records a developer message with the review output. pub(crate) async fn exit_review_mode( @@ -2439,7 +2404,7 @@ mod tests { })), }; - let got = convert_call_tool_result_to_function_call_output_payload(&ctr); + let got = FunctionCallOutputPayload::from(&ctr); let expected = FunctionCallOutputPayload { content: serde_json::to_string(&json!({ "ok": true, @@ -2447,6 +2412,7 @@ mod tests { })) .unwrap(), success: Some(true), + ..Default::default() }; assert_eq!(expected, got); @@ -2479,11 +2445,12 @@ mod tests { structured_content: Some(serde_json::Value::Null), }; - let got = convert_call_tool_result_to_function_call_output_payload(&ctr); + let got = FunctionCallOutputPayload::from(&ctr); let expected = FunctionCallOutputPayload { content: serde_json::to_string(&vec![text_block("hello"), text_block("world")]) .unwrap(), success: Some(true), + ..Default::default() }; assert_eq!(expected, got); @@ -2497,10 +2464,11 @@ mod tests { structured_content: Some(json!({ "message": "bad" })), }; - let got = convert_call_tool_result_to_function_call_output_payload(&ctr); + let got = FunctionCallOutputPayload::from(&ctr); let expected = FunctionCallOutputPayload { content: serde_json::to_string(&json!({ "message": "bad" })).unwrap(), success: Some(false), + ..Default::default() }; assert_eq!(expected, got); @@ -2514,10 +2482,11 @@ mod tests { structured_content: None, }; - let got = convert_call_tool_result_to_function_call_output_payload(&ctr); + let got = FunctionCallOutputPayload::from(&ctr); let expected = FunctionCallOutputPayload { content: serde_json::to_string(&vec![text_block("alpha")]).unwrap(), success: Some(true), + ..Default::default() }; assert_eq!(expected, got); diff --git a/codex-rs/core/src/conversation_history.rs b/codex-rs/core/src/conversation_history.rs index f230a7979b..e9583ff0fb 100644 --- a/codex-rs/core/src/conversation_history.rs +++ b/codex-rs/core/src/conversation_history.rs @@ -136,7 +136,7 @@ impl ConversationHistory { call_id: call_id.clone(), output: FunctionCallOutputPayload { content: "aborted".to_string(), - success: None, + ..Default::default() }, }, )); @@ -183,7 +183,7 @@ impl ConversationHistory { call_id: call_id.clone(), output: FunctionCallOutputPayload { content: "aborted".to_string(), - success: None, + ..Default::default() }, }, )); @@ -565,7 +565,7 @@ mod tests { call_id: "call-1".to_string(), output: FunctionCallOutputPayload { content: "ok".to_string(), - success: None, + ..Default::default() }, }, ]; @@ -581,7 +581,7 @@ mod tests { call_id: "call-2".to_string(), output: FunctionCallOutputPayload { content: "ok".to_string(), - success: None, + ..Default::default() }, }, ResponseItem::FunctionCall { @@ -615,7 +615,7 @@ mod tests { call_id: "call-3".to_string(), output: FunctionCallOutputPayload { content: "ok".to_string(), - success: None, + ..Default::default() }, }, ]; @@ -848,7 +848,7 @@ mod tests { call_id: "call-x".to_string(), output: FunctionCallOutputPayload { content: "aborted".to_string(), - success: None, + ..Default::default() }, }, ] @@ -925,7 +925,7 @@ mod tests { call_id: "shell-1".to_string(), output: FunctionCallOutputPayload { content: "aborted".to_string(), - success: None, + ..Default::default() }, }, ] @@ -939,7 +939,7 @@ mod tests { call_id: "orphan-1".to_string(), output: FunctionCallOutputPayload { content: "ok".to_string(), - success: None, + ..Default::default() }, }]; let mut h = create_history_with_items(items); @@ -979,7 +979,7 @@ mod tests { call_id: "c2".to_string(), output: FunctionCallOutputPayload { content: "ok".to_string(), - success: None, + ..Default::default() }, }, // Will get an inserted custom tool output @@ -1021,7 +1021,7 @@ mod tests { call_id: "c1".to_string(), output: FunctionCallOutputPayload { content: "aborted".to_string(), - success: None, + ..Default::default() }, }, ResponseItem::CustomToolCall { @@ -1051,7 +1051,7 @@ mod tests { call_id: "s1".to_string(), output: FunctionCallOutputPayload { content: "aborted".to_string(), - success: None, + ..Default::default() }, }, ] @@ -1116,7 +1116,7 @@ mod tests { call_id: "orphan-1".to_string(), output: FunctionCallOutputPayload { content: "ok".to_string(), - success: None, + ..Default::default() }, }]; let mut h = create_history_with_items(items); @@ -1150,7 +1150,7 @@ mod tests { call_id: "c2".to_string(), output: FunctionCallOutputPayload { content: "ok".to_string(), - success: None, + ..Default::default() }, }, ResponseItem::CustomToolCall { diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index 09846b719a..5166410372 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -35,6 +35,7 @@ pub(crate) async fn handle_mcp_tool_call( output: FunctionCallOutputPayload { content: format!("err: {e}"), success: Some(false), + ..Default::default() }, }; } diff --git a/codex-rs/core/src/response_processing.rs b/codex-rs/core/src/response_processing.rs index e18fdd45c8..5c30c1126c 100644 --- a/codex-rs/core/src/response_processing.rs +++ b/codex-rs/core/src/response_processing.rs @@ -61,14 +61,11 @@ pub(crate) async fn process_items( ) => { items_to_record_in_conversation_history.push(item); let output = match result { - Ok(call_tool_result) => { - crate::codex::convert_call_tool_result_to_function_call_output_payload( - call_tool_result, - ) - } + Ok(call_tool_result) => FunctionCallOutputPayload::from(call_tool_result), Err(err) => FunctionCallOutputPayload { content: err.clone(), success: Some(false), + ..Default::default() }, }; items_to_record_in_conversation_history.push(ResponseItem::FunctionCallOutput { diff --git a/codex-rs/core/src/tools/context.rs b/codex-rs/core/src/tools/context.rs index 27d309dc25..d2e47f926f 100644 --- a/codex-rs/core/src/tools/context.rs +++ b/codex-rs/core/src/tools/context.rs @@ -5,6 +5,7 @@ use crate::tools::TELEMETRY_PREVIEW_MAX_LINES; use crate::tools::TELEMETRY_PREVIEW_TRUNCATION_NOTICE; use crate::turn_diff_tracker::TurnDiffTracker; use codex_otel::otel_event_manager::OtelEventManager; +use codex_protocol::models::FunctionCallOutputContentItem; use codex_protocol::models::FunctionCallOutputPayload; use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ShellToolCallParams; @@ -65,7 +66,10 @@ impl ToolPayload { #[derive(Clone)] pub enum ToolOutput { Function { + // Plain text representation of the tool output. content: String, + // Some tool calls such as MCP calls may return structured content that can get parsed into an array of polymorphic content items. + content_items: Option>, success: Option, }, Mcp { @@ -90,7 +94,11 @@ impl ToolOutput { pub fn into_response(self, call_id: &str, payload: &ToolPayload) -> ResponseInputItem { match self { - ToolOutput::Function { content, success } => { + ToolOutput::Function { + content, + content_items, + success, + } => { if matches!(payload, ToolPayload::Custom { .. }) { ResponseInputItem::CustomToolCallOutput { call_id: call_id.to_string(), @@ -99,7 +107,11 @@ impl ToolOutput { } else { ResponseInputItem::FunctionCallOutput { call_id: call_id.to_string(), - output: FunctionCallOutputPayload { content, success }, + output: FunctionCallOutputPayload { + content, + content_items, + success, + }, } } } @@ -163,6 +175,7 @@ mod tests { }; let response = ToolOutput::Function { content: "patched".to_string(), + content_items: None, success: Some(true), } .into_response("call-42", &payload); @@ -183,6 +196,7 @@ mod tests { }; let response = ToolOutput::Function { content: "ok".to_string(), + content_items: None, success: Some(true), } .into_response("fn-1", &payload); @@ -191,6 +205,7 @@ mod tests { ResponseInputItem::FunctionCallOutput { call_id, output } => { assert_eq!(call_id, "fn-1"); assert_eq!(output.content, "ok"); + assert!(output.content_items.is_none()); assert_eq!(output.success, Some(true)); } other => panic!("expected FunctionCallOutput, got {other:?}"), diff --git a/codex-rs/core/src/tools/handlers/apply_patch.rs b/codex-rs/core/src/tools/handlers/apply_patch.rs index 126b734242..1e82b9cf10 100644 --- a/codex-rs/core/src/tools/handlers/apply_patch.rs +++ b/codex-rs/core/src/tools/handlers/apply_patch.rs @@ -82,6 +82,7 @@ impl ToolHandler for ApplyPatchHandler { let content = item?; Ok(ToolOutput::Function { content, + content_items: None, success: Some(true), }) } @@ -126,6 +127,7 @@ impl ToolHandler for ApplyPatchHandler { let content = emitter.finish(event_ctx, out).await?; Ok(ToolOutput::Function { content, + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/grep_files.rs b/codex-rs/core/src/tools/handlers/grep_files.rs index de3cd3411c..5473f86935 100644 --- a/codex-rs/core/src/tools/handlers/grep_files.rs +++ b/codex-rs/core/src/tools/handlers/grep_files.rs @@ -90,11 +90,13 @@ impl ToolHandler for GrepFilesHandler { if search_results.is_empty() { Ok(ToolOutput::Function { content: "No matches found.".to_string(), + content_items: None, success: Some(false), }) } else { Ok(ToolOutput::Function { content: search_results.join("\n"), + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/list_dir.rs b/codex-rs/core/src/tools/handlers/list_dir.rs index bcea4a756d..1c08243f72 100644 --- a/codex-rs/core/src/tools/handlers/list_dir.rs +++ b/codex-rs/core/src/tools/handlers/list_dir.rs @@ -106,6 +106,7 @@ impl ToolHandler for ListDirHandler { output.extend(entries); Ok(ToolOutput::Function { content: output.join("\n"), + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/mcp.rs b/codex-rs/core/src/tools/handlers/mcp.rs index 4b2bf3b80e..9798fb8241 100644 --- a/codex-rs/core/src/tools/handlers/mcp.rs +++ b/codex-rs/core/src/tools/handlers/mcp.rs @@ -56,8 +56,16 @@ impl ToolHandler for McpHandler { Ok(ToolOutput::Mcp { result }) } codex_protocol::models::ResponseInputItem::FunctionCallOutput { output, .. } => { - let codex_protocol::models::FunctionCallOutputPayload { content, success } = output; - Ok(ToolOutput::Function { content, success }) + let codex_protocol::models::FunctionCallOutputPayload { + content, + content_items, + success, + } = output; + Ok(ToolOutput::Function { + content, + content_items, + success, + }) } _ => Err(FunctionCallError::RespondToModel( "mcp handler received unexpected response variant".to_string(), diff --git a/codex-rs/core/src/tools/handlers/mcp_resource.rs b/codex-rs/core/src/tools/handlers/mcp_resource.rs index be496f01ec..b601591ac1 100644 --- a/codex-rs/core/src/tools/handlers/mcp_resource.rs +++ b/codex-rs/core/src/tools/handlers/mcp_resource.rs @@ -297,7 +297,10 @@ async fn handle_list_resources( match payload_result { Ok(payload) => match serialize_function_output(payload) { Ok(output) => { - let ToolOutput::Function { content, success } = &output else { + let ToolOutput::Function { + content, success, .. + } = &output + else { unreachable!("MCP resource handler should return function output"); }; let duration = start.elapsed(); @@ -403,7 +406,10 @@ async fn handle_list_resource_templates( match payload_result { Ok(payload) => match serialize_function_output(payload) { Ok(output) => { - let ToolOutput::Function { content, success } = &output else { + let ToolOutput::Function { + content, success, .. + } = &output + else { unreachable!("MCP resource handler should return function output"); }; let duration = start.elapsed(); @@ -489,7 +495,10 @@ async fn handle_read_resource( match payload_result { Ok(payload) => match serialize_function_output(payload) { Ok(output) => { - let ToolOutput::Function { content, success } = &output else { + let ToolOutput::Function { + content, success, .. + } = &output + else { unreachable!("MCP resource handler should return function output"); }; let duration = start.elapsed(); @@ -618,6 +627,7 @@ where Ok(ToolOutput::Function { content, + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/plan.rs b/codex-rs/core/src/tools/handlers/plan.rs index ba8de6bef7..073319bf1c 100644 --- a/codex-rs/core/src/tools/handlers/plan.rs +++ b/codex-rs/core/src/tools/handlers/plan.rs @@ -88,6 +88,7 @@ impl ToolHandler for PlanHandler { Ok(ToolOutput::Function { content, + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/read_file.rs b/codex-rs/core/src/tools/handlers/read_file.rs index f9b6ae4dab..58b6ea6888 100644 --- a/codex-rs/core/src/tools/handlers/read_file.rs +++ b/codex-rs/core/src/tools/handlers/read_file.rs @@ -149,6 +149,7 @@ impl ToolHandler for ReadFileHandler { }; Ok(ToolOutput::Function { content: collected.join("\n"), + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/shell.rs b/codex-rs/core/src/tools/handlers/shell.rs index cab313077f..76650992b9 100644 --- a/codex-rs/core/src/tools/handlers/shell.rs +++ b/codex-rs/core/src/tools/handlers/shell.rs @@ -136,6 +136,7 @@ impl ShellHandler { let content = item?; return Ok(ToolOutput::Function { content, + content_items: None, success: Some(true), }); } @@ -179,6 +180,7 @@ impl ShellHandler { let content = emitter.finish(event_ctx, out).await?; return Ok(ToolOutput::Function { content, + content_items: None, success: Some(true), }); } @@ -226,6 +228,7 @@ impl ShellHandler { let content = emitter.finish(event_ctx, out).await?; Ok(ToolOutput::Function { content, + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/test_sync.rs b/codex-rs/core/src/tools/handlers/test_sync.rs index e340ab47f7..d217c1e8a6 100644 --- a/codex-rs/core/src/tools/handlers/test_sync.rs +++ b/codex-rs/core/src/tools/handlers/test_sync.rs @@ -95,6 +95,7 @@ impl ToolHandler for TestSyncHandler { Ok(ToolOutput::Function { content: "ok".to_string(), + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/unified_exec.rs b/codex-rs/core/src/tools/handlers/unified_exec.rs index 7d1102121e..32ace6c959 100644 --- a/codex-rs/core/src/tools/handlers/unified_exec.rs +++ b/codex-rs/core/src/tools/handlers/unified_exec.rs @@ -171,6 +171,7 @@ impl ToolHandler for UnifiedExecHandler { Ok(ToolOutput::Function { content, + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/handlers/view_image.rs b/codex-rs/core/src/tools/handlers/view_image.rs index b25642d803..6b308c0944 100644 --- a/codex-rs/core/src/tools/handlers/view_image.rs +++ b/codex-rs/core/src/tools/handlers/view_image.rs @@ -85,6 +85,7 @@ impl ToolHandler for ViewImageHandler { Ok(ToolOutput::Function { content: "attached local image path".to_string(), + content_items: None, success: Some(true), }) } diff --git a/codex-rs/core/src/tools/parallel.rs b/codex-rs/core/src/tools/parallel.rs index 449b8e6553..7340f5cc9c 100644 --- a/codex-rs/core/src/tools/parallel.rs +++ b/codex-rs/core/src/tools/parallel.rs @@ -105,7 +105,7 @@ impl ToolCallRuntime { call_id: call.call_id.clone(), output: FunctionCallOutputPayload { content: "aborted".to_string(), - success: None, + ..Default::default() }, }, } diff --git a/codex-rs/core/src/tools/router.rs b/codex-rs/core/src/tools/router.rs index 161997fb6c..19098aa80d 100644 --- a/codex-rs/core/src/tools/router.rs +++ b/codex-rs/core/src/tools/router.rs @@ -181,6 +181,7 @@ impl ToolRouter { output: codex_protocol::models::FunctionCallOutputPayload { content: message, success: Some(false), + ..Default::default() }, } } diff --git a/codex-rs/core/tests/suite/rmcp_client.rs b/codex-rs/core/tests/suite/rmcp_client.rs index 99c863b8f9..85698e9216 100644 --- a/codex-rs/core/tests/suite/rmcp_client.rs +++ b/codex-rs/core/tests/suite/rmcp_client.rs @@ -14,6 +14,8 @@ use codex_core::features::Feature; use codex_core::protocol::AskForApproval; use codex_core::protocol::EventMsg; +use codex_core::protocol::McpInvocation; +use codex_core::protocol::McpToolCallBeginEvent; use codex_core::protocol::Op; use codex_core::protocol::SandboxPolicy; use codex_protocol::config_types::ReasoningSummary; @@ -25,7 +27,9 @@ use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use core_test_support::wait_for_event_with_timeout; use escargot::CargoBuild; +use mcp_types::ContentBlock; use serde_json::Value; +use serde_json::json; use serial_test::serial; use tempfile::tempdir; use tokio::net::TcpStream; @@ -35,6 +39,8 @@ use tokio::time::Instant; use tokio::time::sleep; use wiremock::matchers::any; +static OPENAI_PNG: &str = "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAD0AAAA9CAYAAAAeYmHpAAAE6klEQVR4Aeyau44UVxCGx1fZsmRLlm3Zoe0XcGQ5cUiCCIgJeS9CHgAhMkISQnIuGQgJEkBcxLW+nqnZ6uqqc+nuWRC7q/P3qetf9e+MtOwyX25O4Nep6JPyop++0qev9HrfgZ+F6r2DuB/vHOrt/UIkqdDHYvujOW6fO7h/CNEI+a5jc+pBR8uy0jVFsziYu5HtfSUk+Io34q921hLNctFSX0gwww+S8wce8K1LfCU+cYW4888aov8NxqvQILUPPReLOrm6zyLxa4i+6VZuFbJo8d1MOHZm+7VUtB/aIvhPWc/3SWg49JcwFLlHxuXKjtyloo+YNhuW3VS+WPBuUEMvCFKjEDVgFBQHXrnazpqiSxNZCkQ1kYiozsbm9Oz7l4i2Il7vGccGNWAc3XosDrZe/9P3ZnMmzHNEQw4smf8RQ87XEAMsC7Az0Au+dgXerfH4+sHvEc0SYGic8WBBUGqFH2gN7yDrazy7m2pbRTeRmU3+MjZmr1h6LJgPbGy23SI6GlYT0brQ71IY8Us4PNQCm+zepSbaD2BY9xCaAsD9IIj/IzFmKMSdHHonwdZATbTnYREf6/VZGER98N9yCWIvXQwXDoDdhZJoT8jwLnJXDB9w4Sb3e6nK5ndzlkTLnP3JBu4LKkbrYrU69gCVceV0JvpyuW1xlsUVngzhwMetn/XamtTORF9IO5YnWNiyeF9zCAfqR3fUW+vZZKLtgP+ts8BmQRBREAdRDhH3o8QuRh/YucNFz2BEjxbRN6LGzphfKmvP6v6QhqIQyZ8XNJ0W0X83MR1PEcJBNO2KC2Z1TW/v244scp9FwRViZxIOBF0Lctk7ZVSavdLvRlV1hz/ysUi9sr8CIcB3nvWBwA93ykTz18eAYxQ6N/K2DkPA1lv3iXCwmDUT7YkjIby9siXueIJj9H+pzSqJ9oIuJWTUgSSt4WO7o/9GGg0viR4VinNRUDoIj34xoCd6pxD3aK3zfdbnx5v1J3ZNNEJsE0sBG7N27ReDrJc4sFxz7dI/ZAbOmmiKvHBitQXpAdR6+F7v+/ol/tOouUV01EeMZQF2BoQDn6dP4XNr+j9GZEtEK1/L8pFw7bd3a53tsTa7WD+054jOFmPg1XBKPQgnqFfmFcy32ZRvjmiIIQTYFvyDxQ8nH8WIwwGwlyDjDznnilYyFr6njrlZwsKkBpO59A7OwgdzPEWRm+G+oeb7IfyNuzjEEVLrOVxJsxvxwF8kmCM6I2QYmJunz4u4TrADpfl7mlbRTWQ7VmrBzh3+C9f6Grc3YoGN9dg/SXFthpRsT6vobfXRs2VBlgBHXVMLHjDNbIZv1sZ9+X3hB09cXdH1JKViyG0+W9bWZDa/r2f9zAFR71sTzGpMSWz2iI4YssWjWo3REy1MDGjdwe5e0dFSiAC1JakBvu4/CUS8Eh6dqHdU0Or0ioY3W5ClSqDXAy7/6SRfgw8vt4I+tbvvNtFT2kVDhY5+IGb1rCqYaXNF08vSALsXCPmt0kQNqJT1p5eI1mkIV/BxCY1z85lOzeFbPBQHURkkPTlwTYK9gTVE25l84IbFFN+YJDHjdpn0gq6mrHht0dkcjbM4UL9283O5p77GN+SPW/QwVB4IUYg7Or+Kp7naR6qktP98LNF2UxWo9yObPIT9KYg+hK4i56no4rfnM0qeyFf6AwAAAP//trwR3wAAAAZJREFUAwBZ0sR75itw5gAAAABJRU5ErkJggg=="; + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial(mcp_test_value)] async fn stdio_server_round_trip() -> anyhow::Result<()> { @@ -175,6 +181,352 @@ async fn stdio_server_round_trip() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial(mcp_test_value)] +async fn stdio_image_responses_round_trip() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + + let call_id = "img-1"; + let server_name = "rmcp"; + let tool_name = format!("mcp__{server_name}__image"); + + // First stream: model decides to call the image tool. + mount_sse_once_match( + &server, + any(), + responses::sse(vec![ + responses::ev_response_created("resp-1"), + responses::ev_function_call(call_id, &tool_name, "{}"), + responses::ev_completed("resp-1"), + ]), + ) + .await; + // Second stream: after tool execution, assistant emits a message and completes. + let final_mock = mount_sse_once_match( + &server, + any(), + responses::sse(vec![ + responses::ev_assistant_message("msg-1", "rmcp image tool completed successfully."), + responses::ev_completed("resp-2"), + ]), + ) + .await; + + // Build the stdio rmcp server and pass the image as data URL so it can construct ImageContent. + let rmcp_test_server_bin = CargoBuild::new() + .package("codex-rmcp-client") + .bin("test_stdio_server") + .run()? + .path() + .to_string_lossy() + .into_owned(); + + let fixture = test_codex() + .with_config(move |config| { + config.features.enable(Feature::RmcpClient); + config.mcp_servers.insert( + server_name.to_string(), + McpServerConfig { + transport: McpServerTransportConfig::Stdio { + command: rmcp_test_server_bin, + args: Vec::new(), + env: Some(HashMap::from([( + "MCP_TEST_IMAGE_DATA_URL".to_string(), + OPENAI_PNG.to_string(), + )])), + env_vars: Vec::new(), + cwd: None, + }, + enabled: true, + startup_timeout_sec: Some(Duration::from_secs(10)), + tool_timeout_sec: None, + enabled_tools: None, + disabled_tools: None, + }, + ); + }) + .build(&server) + .await?; + let session_model = fixture.session_configured.model.clone(); + + fixture + .codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "call the rmcp image tool".into(), + }], + final_output_json_schema: None, + cwd: fixture.cwd.path().to_path_buf(), + approval_policy: AskForApproval::Never, + sandbox_policy: SandboxPolicy::ReadOnly, + model: session_model, + effort: None, + summary: ReasoningSummary::Auto, + }) + .await?; + + // Wait for tool begin/end and final completion. + let begin_event = wait_for_event_with_timeout( + &fixture.codex, + |ev| matches!(ev, EventMsg::McpToolCallBegin(_)), + Duration::from_secs(10), + ) + .await; + let EventMsg::McpToolCallBegin(begin) = begin_event else { + unreachable!("begin"); + }; + assert_eq!( + begin, + McpToolCallBeginEvent { + call_id: call_id.to_string(), + invocation: McpInvocation { + server: server_name.to_string(), + tool: "image".to_string(), + arguments: Some(json!({})), + }, + }, + ); + + let end_event = wait_for_event(&fixture.codex, |ev| { + matches!(ev, EventMsg::McpToolCallEnd(_)) + }) + .await; + let EventMsg::McpToolCallEnd(end) = end_event else { + unreachable!("end"); + }; + assert_eq!(end.call_id, call_id); + assert_eq!( + end.invocation, + McpInvocation { + server: server_name.to_string(), + tool: "image".to_string(), + arguments: Some(json!({})), + } + ); + let result = end.result.expect("rmcp image tool should return success"); + assert_eq!(result.is_error, Some(false)); + assert_eq!(result.content.len(), 1); + let base64_only = OPENAI_PNG + .strip_prefix("data:image/png;base64,") + .expect("data url prefix"); + match &result.content[0] { + ContentBlock::ImageContent(img) => { + assert_eq!(img.mime_type, "image/png"); + assert_eq!(img.r#type, "image"); + assert_eq!(img.data, base64_only); + } + other => panic!("expected image content, got {other:?}"), + } + + wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; + + let output_item = final_mock.single_request().function_call_output(call_id); + assert_eq!( + output_item, + json!({ + "type": "function_call_output", + "call_id": call_id, + "output": [{ + "type": "input_image", + "image_url": OPENAI_PNG + }] + }) + ); + server.verify().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial(mcp_test_value)] +async fn stdio_image_completions_round_trip() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = responses::start_mock_server().await; + + let call_id = "img-cc-1"; + let server_name = "rmcp"; + let tool_name = format!("mcp__{server_name}__image"); + + let tool_call = json!({ + "choices": [ + { + "delta": { + "tool_calls": [ + { + "id": call_id, + "type": "function", + "function": {"name": tool_name, "arguments": "{}"} + } + ] + }, + "finish_reason": "tool_calls" + } + ] + }); + let sse_tool_call = format!( + "data: {}\n\ndata: [DONE]\n\n", + serde_json::to_string(&tool_call)? + ); + + let final_assistant = json!({ + "choices": [ + { + "delta": {"content": "rmcp image tool completed successfully."}, + "finish_reason": "stop" + } + ] + }); + let sse_final = format!( + "data: {}\n\ndata: [DONE]\n\n", + serde_json::to_string(&final_assistant)? + ); + + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; + struct ChatSeqResponder { + num_calls: AtomicUsize, + bodies: Vec, + } + impl wiremock::Respond for ChatSeqResponder { + fn respond(&self, _: &wiremock::Request) -> wiremock::ResponseTemplate { + let idx = self.num_calls.fetch_add(1, Ordering::SeqCst); + match self.bodies.get(idx) { + Some(body) => wiremock::ResponseTemplate::new(200) + .insert_header("content-type", "text/event-stream") + .set_body_string(body.clone()), + None => panic!("no chat completion response for index {idx}"), + } + } + } + + let chat_seq = ChatSeqResponder { + num_calls: AtomicUsize::new(0), + bodies: vec![sse_tool_call, sse_final], + }; + wiremock::Mock::given(wiremock::matchers::method("POST")) + .and(wiremock::matchers::path("/v1/chat/completions")) + .respond_with(chat_seq) + .expect(2) + .mount(&server) + .await; + + let rmcp_test_server_bin = CargoBuild::new() + .package("codex-rmcp-client") + .bin("test_stdio_server") + .run()? + .path() + .to_string_lossy() + .into_owned(); + + let fixture = test_codex() + .with_config(move |config| { + config.model_provider.wire_api = codex_core::WireApi::Chat; + config.features.enable(Feature::RmcpClient); + config.mcp_servers.insert( + server_name.to_string(), + McpServerConfig { + transport: McpServerTransportConfig::Stdio { + command: rmcp_test_server_bin, + args: Vec::new(), + env: Some(HashMap::from([( + "MCP_TEST_IMAGE_DATA_URL".to_string(), + OPENAI_PNG.to_string(), + )])), + env_vars: Vec::new(), + cwd: None, + }, + enabled: true, + startup_timeout_sec: Some(Duration::from_secs(10)), + tool_timeout_sec: None, + enabled_tools: None, + disabled_tools: None, + }, + ); + }) + .build(&server) + .await?; + let session_model = fixture.session_configured.model.clone(); + + fixture + .codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "call the rmcp image tool".into(), + }], + final_output_json_schema: None, + cwd: fixture.cwd.path().to_path_buf(), + approval_policy: AskForApproval::Never, + sandbox_policy: SandboxPolicy::ReadOnly, + model: session_model, + effort: None, + summary: ReasoningSummary::Auto, + }) + .await?; + + let begin_event = wait_for_event_with_timeout( + &fixture.codex, + |ev| matches!(ev, EventMsg::McpToolCallBegin(_)), + Duration::from_secs(10), + ) + .await; + let EventMsg::McpToolCallBegin(begin) = begin_event else { + unreachable!("begin"); + }; + assert_eq!( + begin, + McpToolCallBeginEvent { + call_id: call_id.to_string(), + invocation: McpInvocation { + server: server_name.to_string(), + tool: "image".to_string(), + arguments: Some(json!({})), + }, + }, + ); + + let end_event = wait_for_event(&fixture.codex, |ev| { + matches!(ev, EventMsg::McpToolCallEnd(_)) + }) + .await; + let EventMsg::McpToolCallEnd(end) = end_event else { + unreachable!("end"); + }; + assert!(end.result.as_ref().is_ok(), "tool call should succeed"); + + wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TaskComplete(_))).await; + + // Chat Completions assertion: the second POST should include a tool role message + // with an array `content` containing an item with the expected data URL. + let requests = server.received_requests().await.expect("requests captured"); + assert!(requests.len() >= 2, "expected two chat completion calls"); + let second = &requests[1]; + let body: Value = serde_json::from_slice(&second.body)?; + let messages = body + .get("messages") + .and_then(Value::as_array) + .cloned() + .expect("messages array"); + let tool_msg = messages + .iter() + .find(|m| { + m.get("role") == Some(&json!("tool")) && m.get("tool_call_id") == Some(&json!(call_id)) + }) + .cloned() + .expect("tool message present"); + assert_eq!( + tool_msg, + json!({ + "role": "tool", + "tool_call_id": call_id, + "content": [{"type": "image_url", "image_url": {"url": OPENAI_PNG}}] + }) + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[serial(mcp_test_value)] async fn stdio_server_propagates_whitelisted_env_vars() -> anyhow::Result<()> { diff --git a/codex-rs/protocol/src/models.rs b/codex-rs/protocol/src/models.rs index 4430a0998a..614a5ff2b0 100644 --- a/codex-rs/protocol/src/models.rs +++ b/codex-rs/protocol/src/models.rs @@ -3,6 +3,7 @@ use std::collections::HashMap; use base64::Engine; use codex_utils_image::load_and_resize_to_fit; use mcp_types::CallToolResult; +use mcp_types::ContentBlock; use serde::Deserialize; use serde::Deserializer; use serde::Serialize; @@ -83,9 +84,8 @@ pub enum ResponseItem { // NOTE: The input schema for `function_call_output` objects that clients send to the // OpenAI /v1/responses endpoint is NOT the same shape as the objects the server returns on the // SSE stream. When *sending* we must wrap the string output inside an object that includes a - // required `success` boolean. The upstream TypeScript CLI does this implicitly. To ensure we - // serialize exactly the expected shape we introduce a dedicated payload struct and flatten it - // here. + // required `success` boolean. To ensure we serialize exactly the expected shape we introduce + // a dedicated payload struct and flatten it here. FunctionCallOutput { call_id: String, output: FunctionCallOutputPayload, @@ -160,19 +160,17 @@ impl From for ResponseItem { ResponseInputItem::FunctionCallOutput { call_id, output } => { Self::FunctionCallOutput { call_id, output } } - ResponseInputItem::McpToolCallOutput { call_id, result } => Self::FunctionCallOutput { - call_id, - output: FunctionCallOutputPayload { - success: Some(result.is_ok()), - content: result.map_or_else( - |tool_call_err| format!("err: {tool_call_err:?}"), - |result| { - serde_json::to_string(&result) - .unwrap_or_else(|e| format!("JSON serialization error: {e}")) - }, - ), - }, - }, + ResponseInputItem::McpToolCallOutput { call_id, result } => { + let output = match result { + Ok(result) => FunctionCallOutputPayload::from(&result), + Err(tool_call_err) => FunctionCallOutputPayload { + content: format!("err: {tool_call_err:?}"), + success: Some(false), + ..Default::default() + }, + }; + Self::FunctionCallOutput { call_id, output } + } ResponseInputItem::CustomToolCallOutput { call_id, output } => { Self::CustomToolCallOutput { call_id, output } } @@ -290,31 +288,53 @@ pub struct ShellToolCallParams { pub justification: Option, } -#[derive(Debug, Clone, PartialEq, JsonSchema, TS)] +/// Responses API compatible content items that can be returned by a tool call. +/// This is a subset of ContentItem with the types we support as function call outputs. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema, TS)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum FunctionCallOutputContentItem { + // Do not rename, these are serialized and used directly in the responses API. + InputText { text: String }, + // Do not rename, these are serialized and used directly in the responses API. + InputImage { image_url: String }, +} + +/// The payload we send back to OpenAI when reporting a tool call result. +/// +/// `content` preserves the historical plain-string payload so downstream +/// integrations (tests, logging, etc.) can keep treating tool output as +/// `String`. When an MCP server returns richer data we additionally populate +/// `content_items` with the structured form that the Responses/Chat +/// Completions APIs understand. +#[derive(Debug, Default, Clone, PartialEq, JsonSchema, TS)] pub struct FunctionCallOutputPayload { pub content: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub content_items: Option>, // TODO(jif) drop this. pub success: Option, } +#[derive(Deserialize)] +#[serde(untagged)] +enum FunctionCallOutputPayloadSerde { + Text(String), + Items(Vec), +} + // The Responses API expects two *different* shapes depending on success vs failure: // • success → output is a plain string (no nested object) // • failure → output is an object { content, success:false } -// The upstream TypeScript CLI implements this by special‑casing the serialize path. -// We replicate that behavior with a manual Serialize impl. - impl Serialize for FunctionCallOutputPayload { fn serialize(&self, serializer: S) -> Result where S: Serializer, { - // The upstream TypeScript CLI always serializes `output` as a *plain string* regardless - // of whether the function call succeeded or failed. The boolean is purely informational - // for local bookkeeping and is NOT sent to the OpenAI endpoint. Sending the nested object - // form `{ content, success:false }` triggers the 400 we are still seeing. Mirror the JS CLI - // exactly: always emit a bare string. - - serializer.serialize_str(&self.content) + if let Some(items) = &self.content_items { + items.serialize(serializer) + } else { + serializer.serialize_str(&self.content) + } } } @@ -323,14 +343,106 @@ impl<'de> Deserialize<'de> for FunctionCallOutputPayload { where D: Deserializer<'de>, { - let s = String::deserialize(deserializer)?; - Ok(FunctionCallOutputPayload { - content: s, - success: None, - }) + match FunctionCallOutputPayloadSerde::deserialize(deserializer)? { + FunctionCallOutputPayloadSerde::Text(content) => Ok(FunctionCallOutputPayload { + content, + ..Default::default() + }), + FunctionCallOutputPayloadSerde::Items(items) => { + let content = serde_json::to_string(&items).map_err(serde::de::Error::custom)?; + Ok(FunctionCallOutputPayload { + content, + content_items: Some(items), + success: None, + }) + } + } } } +impl From<&CallToolResult> for FunctionCallOutputPayload { + fn from(call_tool_result: &CallToolResult) -> Self { + let CallToolResult { + content, + structured_content, + is_error, + } = call_tool_result; + + let is_success = is_error != &Some(true); + + if let Some(structured_content) = structured_content + && !structured_content.is_null() + { + match serde_json::to_string(structured_content) { + Ok(serialized_structured_content) => { + return FunctionCallOutputPayload { + content: serialized_structured_content, + success: Some(is_success), + ..Default::default() + }; + } + Err(err) => { + return FunctionCallOutputPayload { + content: err.to_string(), + success: Some(false), + ..Default::default() + }; + } + } + } + + let serialized_content = match serde_json::to_string(content) { + Ok(serialized_content) => serialized_content, + Err(err) => { + return FunctionCallOutputPayload { + content: err.to_string(), + success: Some(false), + ..Default::default() + }; + } + }; + + let content_items = convert_content_blocks_to_items(content); + + FunctionCallOutputPayload { + content: serialized_content, + content_items, + success: Some(is_success), + } + } +} + +fn convert_content_blocks_to_items( + blocks: &[ContentBlock], +) -> Option> { + let mut saw_image = false; + let mut items = Vec::with_capacity(blocks.len()); + + for block in blocks { + match block { + ContentBlock::TextContent(text) => { + items.push(FunctionCallOutputContentItem::InputText { + text: text.text.clone(), + }); + } + ContentBlock::ImageContent(image) => { + saw_image = true; + // Just in case the content doesn't include a data URL, add it. + let image_url = if image.data.starts_with("data:") { + image.data.clone() + } else { + format!("data:{};base64,{}", image.mime_type, image.data) + }; + items.push(FunctionCallOutputContentItem::InputImage { image_url }); + } + // TODO: render audio, resource, and embedded resource content to the model. + _ => return None, + } + } + + if saw_image { Some(items) } else { None } +} + // Implement Display so callers can treat the payload like a plain string when logging or doing // trivial substring checks in tests (existing tests call `.contains()` on the output). Display // returns the raw `content` field. @@ -354,6 +466,8 @@ impl std::ops::Deref for FunctionCallOutputPayload { mod tests { use super::*; use anyhow::Result; + use mcp_types::ImageContent; + use mcp_types::TextContent; use tempfile::tempdir; #[test] @@ -362,7 +476,7 @@ mod tests { call_id: "call1".into(), output: FunctionCallOutputPayload { content: "ok".into(), - success: None, + ..Default::default() }, }; @@ -381,6 +495,7 @@ mod tests { output: FunctionCallOutputPayload { content: "bad".into(), success: Some(false), + ..Default::default() }, }; @@ -391,6 +506,81 @@ mod tests { Ok(()) } + #[test] + fn serializes_image_outputs_as_array() -> Result<()> { + let call_tool_result = CallToolResult { + content: vec![ + ContentBlock::TextContent(TextContent { + annotations: None, + text: "caption".into(), + r#type: "text".into(), + }), + ContentBlock::ImageContent(ImageContent { + annotations: None, + data: "BASE64".into(), + mime_type: "image/png".into(), + r#type: "image".into(), + }), + ], + is_error: None, + structured_content: None, + }; + + let payload = FunctionCallOutputPayload::from(&call_tool_result); + assert_eq!(payload.success, Some(true)); + let items = payload.content_items.clone().expect("content items"); + assert_eq!( + items, + vec![ + FunctionCallOutputContentItem::InputText { + text: "caption".into(), + }, + FunctionCallOutputContentItem::InputImage { + image_url: "data:image/png;base64,BASE64".into(), + }, + ] + ); + + let item = ResponseInputItem::FunctionCallOutput { + call_id: "call1".into(), + output: payload, + }; + + let json = serde_json::to_string(&item)?; + let v: serde_json::Value = serde_json::from_str(&json)?; + + let output = v.get("output").expect("output field"); + assert!(output.is_array(), "expected array output"); + + Ok(()) + } + + #[test] + fn deserializes_array_payload_into_items() -> Result<()> { + let json = r#"[ + {"type": "input_text", "text": "note"}, + {"type": "input_image", "image_url": "data:image/png;base64,XYZ"} + ]"#; + + let payload: FunctionCallOutputPayload = serde_json::from_str(json)?; + + assert_eq!(payload.success, None); + let expected_items = vec![ + FunctionCallOutputContentItem::InputText { + text: "note".into(), + }, + FunctionCallOutputContentItem::InputImage { + image_url: "data:image/png;base64,XYZ".into(), + }, + ]; + assert_eq!(payload.content_items, Some(expected_items.clone())); + + let expected_content = serde_json::to_string(&expected_items)?; + assert_eq!(payload.content, expected_content); + + Ok(()) + } + #[test] fn deserialize_shell_tool_call_params() -> Result<()> { let json = r#"{ diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index bb14f3797b..a7f5241bb3 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -814,7 +814,7 @@ pub struct AgentReasoningDeltaEvent { pub delta: String, } -#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS, PartialEq)] pub struct McpInvocation { /// Name of the MCP server as defined in the config. pub server: String, @@ -824,14 +824,14 @@ pub struct McpInvocation { pub arguments: Option, } -#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS, PartialEq)] pub struct McpToolCallBeginEvent { /// Identifier so this can be paired with the McpToolCallEnd event. pub call_id: String, pub invocation: McpInvocation, } -#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] +#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS, PartialEq)] pub struct McpToolCallEndEvent { /// Identifier for the corresponding McpToolCallBegin that finished. pub call_id: String, diff --git a/codex-rs/rmcp-client/src/bin/test_stdio_server.rs b/codex-rs/rmcp-client/src/bin/test_stdio_server.rs index 44ae50f02f..aafba59324 100644 --- a/codex-rs/rmcp-client/src/bin/test_stdio_server.rs +++ b/codex-rs/rmcp-client/src/bin/test_stdio_server.rs @@ -40,7 +40,7 @@ pub fn stdio() -> (tokio::io::Stdin, tokio::io::Stdout) { } impl TestToolServer { fn new() -> Self { - let tools = vec![Self::echo_tool()]; + let tools = vec![Self::echo_tool(), Self::image_tool()]; let resources = vec![Self::memo_resource()]; let resource_templates = vec![Self::memo_template()]; Self { @@ -70,6 +70,22 @@ impl TestToolServer { ) } + fn image_tool() -> Tool { + #[expect(clippy::expect_used)] + let schema: JsonObject = serde_json::from_value(serde_json::json!({ + "type": "object", + "properties": {}, + "additionalProperties": false + })) + .expect("image tool schema should deserialize"); + + Tool::new( + Cow::Borrowed("image"), + Cow::Borrowed("Return a single image content block."), + Arc::new(schema), + ) + } + fn memo_resource() -> Resource { let raw = RawResource { uri: MEMO_URI.to_string(), @@ -214,6 +230,35 @@ impl ServerHandler for TestToolServer { meta: None, }) } + "image" => { + // Read a data URL (e.g. data:image/png;base64,AAA...) from env and convert to + // an MCP image content block. Tests set MCP_TEST_IMAGE_DATA_URL. + let data_url = std::env::var("MCP_TEST_IMAGE_DATA_URL").map_err(|_| { + McpError::invalid_params( + "missing MCP_TEST_IMAGE_DATA_URL env var for image tool", + None, + ) + })?; + + fn parse_data_url(url: &str) -> Option<(String, String)> { + let rest = url.strip_prefix("data:")?; + let (mime_and_opts, data) = rest.split_once(',')?; + let (mime, _opts) = + mime_and_opts.split_once(';').unwrap_or((mime_and_opts, "")); + Some((mime.to_string(), data.to_string())) + } + + let (mime_type, data_b64) = parse_data_url(&data_url).ok_or_else(|| { + McpError::invalid_params( + format!("invalid data URL for image tool: {data_url}"), + None, + ) + })?; + + Ok(CallToolResult::success(vec![rmcp::model::Content::image( + data_b64, mime_type, + )])) + } other => Err(McpError::invalid_params( format!("unknown tool: {other}"), None, diff --git a/codex-rs/tui/src/history_cell.rs b/codex-rs/tui/src/history_cell.rs index e630e00175..406a892561 100644 --- a/codex-rs/tui/src/history_cell.rs +++ b/codex-rs/tui/src/history_cell.rs @@ -465,7 +465,7 @@ struct CompletedMcpToolCallWithImageOutput { } impl HistoryCell for CompletedMcpToolCallWithImageOutput { fn display_lines(&self, _width: u16) -> Vec> { - vec!["tool result (image output omitted)".into()] + vec!["tool result (image output)".into()] } }