diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index fb4025b4a8..3484dfedef 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -122,6 +122,12 @@ pub(crate) async fn handle_mcp_tool_call( } let request_meta = build_mcp_tool_call_request_meta(turn_context.as_ref(), &server, metadata.as_ref()); + let connector_id = metadata + .as_ref() + .and_then(|metadata| metadata.connector_id.clone()); + let connector_name = metadata + .as_ref() + .and_then(|metadata| metadata.connector_name.clone()); let server_origin = sess .services .mcp_connection_manager @@ -163,6 +169,8 @@ pub(crate) async fn handle_mcp_tool_call( arguments_value: arguments_value.clone(), request_meta: request_meta.clone(), server_origin: server_origin.as_deref(), + connector_id: connector_id.clone(), + connector_name: connector_name.clone(), }, ) .await; @@ -248,6 +256,8 @@ pub(crate) async fn handle_mcp_tool_call( arguments_value: arguments_value.clone(), request_meta, server_origin: server_origin.as_deref(), + connector_id, + connector_name, }, ) .await; @@ -282,10 +292,14 @@ async fn execute_mcp_tool_call( let tool_call_span = crate::network_trace::mcp_tool_call_span( sess.as_ref(), turn_context.as_ref(), - request.server, - request.tool_name, - request.call_id, - request.server_origin, + crate::network_trace::McpToolCallTrace { + server_name: request.server, + tool_name: request.tool_name, + call_id: request.call_id, + server_origin: request.server_origin, + connector_id: request.connector_id.as_deref(), + connector_name: request.connector_name.as_deref(), + }, ); let result = async { sess.call_tool( @@ -319,6 +333,8 @@ struct McpToolCallRequest<'a> { arguments_value: Option, request_meta: Option, server_origin: Option<&'a str>, + connector_id: Option, + connector_name: Option, } async fn maybe_mark_thread_memory_mode_polluted(sess: &Session, turn_context: &TurnContext) { diff --git a/codex-rs/core/src/network_trace.rs b/codex-rs/core/src/network_trace.rs index 3fd08d7c1a..e243d1192c 100644 --- a/codex-rs/core/src/network_trace.rs +++ b/codex-rs/core/src/network_trace.rs @@ -96,15 +96,21 @@ pub(crate) fn responses_http_request_span( span } +pub(crate) struct McpToolCallTrace<'a> { + pub server_name: &'a str, + pub tool_name: &'a str, + pub call_id: &'a str, + pub server_origin: Option<&'a str>, + pub connector_id: Option<&'a str>, + pub connector_name: Option<&'a str>, +} + pub(crate) fn mcp_tool_call_span( session: &Session, turn_context: &TurnContext, - server_name: &str, - tool_name: &str, - call_id: &str, - server_origin: Option<&str>, + trace: McpToolCallTrace<'_>, ) -> Span { - let transport = match server_origin { + let transport = match trace.server_origin { Some("stdio") => "stdio", Some(_) => "streamable_http", None => "", @@ -114,11 +120,13 @@ pub(crate) fn mcp_tool_call_span( otel.kind = "client", rpc.system = "jsonrpc", rpc.method = "tools/call", - mcp.server.name = server_name, - mcp.server.origin = server_origin.unwrap_or(""), + mcp.server.name = trace.server_name, + mcp.server.origin = trace.server_origin.unwrap_or(""), mcp.transport = transport, - tool.name = tool_name, - tool.call_id = call_id, + mcp.connector.id = trace.connector_id.unwrap_or(""), + mcp.connector.name = trace.connector_name.unwrap_or(""), + tool.name = trace.tool_name, + tool.call_id = trace.call_id, conversation.id = Empty, session.id = Empty, turn.id = Empty, @@ -126,6 +134,6 @@ pub(crate) fn mcp_tool_call_span( server.port = Empty, ); CorrelationFields::from_turn_context(session, turn_context).record_on(&span); - record_server_fields(&span, server_origin); + record_server_fields(&span, trace.server_origin); span } diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index f4891d58c5..8e72d9efc1 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -92,6 +92,8 @@ mod model_visible_layout; mod models_cache_ttl; mod models_etag_responses; mod otel; +#[cfg(not(target_os = "windows"))] +mod otel_apps; mod pending_input; mod permissions_messages; mod personality; diff --git a/codex-rs/core/tests/suite/otel_apps.rs b/codex-rs/core/tests/suite/otel_apps.rs new file mode 100644 index 0000000000..75e45dac14 --- /dev/null +++ b/codex-rs/core/tests/suite/otel_apps.rs @@ -0,0 +1,104 @@ +#![allow(clippy::unwrap_used, clippy::expect_used)] + +use anyhow::Result; +use codex_core::CodexAuth; +use codex_features::Feature; +use codex_protocol::protocol::EventMsg; +use codex_protocol::protocol::Op; +use codex_protocol::user_input::UserInput; +use core_test_support::apps_test_server::AppsTestServer; +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; +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 core_test_support::wait_for_event; +use std::sync::Mutex; +use tracing::Level; +use tracing_subscriber::fmt::format::FmtSpan; +use tracing_test::internal::MockWriter; + +#[tokio::test(flavor = "current_thread")] +async fn codex_apps_mcp_span_records_connector_metadata() -> Result<()> { + skip_if_no_network!(Ok(())); + + let buffer: &'static Mutex> = Box::leak(Box::new(Mutex::new(Vec::new()))); + let subscriber = tracing_subscriber::fmt() + .with_level(true) + .with_ansi(false) + .with_max_level(Level::TRACE) + .with_span_events(FmtSpan::FULL) + .with_writer(MockWriter::new(buffer)) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + + let server = start_mock_server().await; + let apps_server = AppsTestServer::mount(&server).await?; + + mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-1"), + ev_function_call( + "calendar-call-1", + "mcp__codex_apps__calendar_create_event", + r#"{"title":"Lunch","starts_at":"2026-03-10T12:00:00Z"}"#, + ), + ev_completed("resp-1"), + ]), + ) + .await; + mount_sse_once( + &server, + sse(vec![ + ev_assistant_message("msg-1", "calendar tool completed successfully."), + ev_completed("resp-2"), + ]), + ) + .await; + + let fixture = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_config(move |config| { + config + .features + .enable(Feature::Apps) + .expect("test config should allow feature update"); + config.chatgpt_base_url = apps_server.chatgpt_base_url; + config.model = Some("gpt-5-codex".to_string()); + }) + .build(&server) + .await?; + + fixture + .codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "create a calendar event".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + }) + .await?; + + wait_for_event(&fixture.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let logs = String::from_utf8(buffer.lock().unwrap().clone()).unwrap(); + assert!( + logs.contains("mcp.tools.call{otel.kind=\"client\"") + && logs.contains("mcp.server.name=\"codex_apps\"") + && logs.contains("mcp.connector.id=\"calendar\"") + && logs.contains("mcp.connector.name=\"Calendar\"") + && logs.contains("tool.name=\"calendar_create_event\""), + "missing connector metadata on mcp.tools.call span\nlogs:\n{logs}" + ); + + Ok(()) +} diff --git a/codex-rs/core/tests/suite/rmcp_client.rs b/codex-rs/core/tests/suite/rmcp_client.rs index 39f8f9db3c..b7e1b84a44 100644 --- a/codex-rs/core/tests/suite/rmcp_client.rs +++ b/codex-rs/core/tests/suite/rmcp_client.rs @@ -48,16 +48,28 @@ use tokio::process::Child; use tokio::process::Command; use tokio::time::Instant; use tokio::time::sleep; -use tracing_test::traced_test; +use tracing::Level; +use tracing_subscriber::fmt::format::FmtSpan; +use tracing_test::internal::MockWriter; 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)] +#[tokio::test(flavor = "current_thread")] #[serial(mcp_test_value)] -#[traced_test] async fn stdio_server_round_trip() -> anyhow::Result<()> { skip_if_no_network!(Ok(())); + let buffer: &'static std::sync::Mutex> = + Box::leak(Box::new(std::sync::Mutex::new(Vec::new()))); + let subscriber = tracing_subscriber::fmt() + .with_level(true) + .with_ansi(false) + .with_max_level(Level::TRACE) + .with_span_events(FmtSpan::FULL) + .with_writer(MockWriter::new(buffer)) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + let server = responses::start_mock_server().await; let call_id = "call-123"; @@ -191,30 +203,19 @@ async fn stdio_server_round_trip() -> anyhow::Result<()> { wait_for_event(&fixture.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - logs_assert(|lines: &[&str]| { - lines - .iter() - .find(|line| { - line.contains("turn{otel.name=\"session_task.turn\"") - && line.contains("conversation.id=") - && line.contains("thread.id=") - && line.contains("turn.id=") - && line.contains("model=") - && line.contains("mcp.tools.call{") - && line.contains("otel.kind=\"client\"") - && line.contains("rpc.system=\"jsonrpc\"") - && line.contains("rpc.method=\"tools/call\"") - && line.contains("mcp.server.name=\"rmcp\"") - && line.contains("mcp.transport=\"stdio\"") - && line.contains("tool.name=\"echo\"") - && line.contains("tool.call_id=\"call-123\"") - && line.contains("turn.id=") - }) - .map(|_| Ok(())) - .unwrap_or_else(|| { - Err("missing mcp.tools.call span nested under session_task.turn".to_string()) - }) - }); + let logs = String::from_utf8(buffer.lock().unwrap().clone()).unwrap(); + assert!( + logs.contains("turn{otel.name=\"session_task.turn\"") + && logs.contains("mcp.tools.call{otel.kind=\"client\"") + && logs.contains("rpc.system=\"jsonrpc\"") + && logs.contains("rpc.method=\"tools/call\"") + && logs.contains("mcp.server.name=\"rmcp\"") + && logs.contains("mcp.transport=\"stdio\"") + && logs.contains("tool.name=\"echo\"") + && logs.contains("tool.call_id=\"call-123\"") + && logs.contains("turn.id="), + "missing mcp.tools.call span nested under session_task.turn\nlogs:\n{logs}" + ); server.verify().await;