From 9036b0a85324c2f4bf44aa9c350044228c5253bb Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Mon, 6 Jul 2026 14:36:25 -0700 Subject: [PATCH] feat(core): emit canonical dynamic tool call items --- .../app-server/src/bespoke_event_handling.rs | 158 +++++++++++++----- codex-rs/core/src/tools/handlers/dynamic.rs | 74 ++++---- 2 files changed, 157 insertions(+), 75 deletions(-) diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index 4bd1d6e108e1..863aeac79bf8 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/codex-rs/app-server/src/bespoke_event_handling.rs @@ -22,7 +22,6 @@ use codex_app_server_protocol::CommandExecutionSource; use codex_app_server_protocol::CommandExecutionStatus; use codex_app_server_protocol::DeprecationNoticeNotification; use codex_app_server_protocol::DynamicToolCallParams; -use codex_app_server_protocol::DynamicToolCallStatus; use codex_app_server_protocol::ErrorNotification; use codex_app_server_protocol::ExecPolicyAmendment as V2ExecPolicyAmendment; use codex_app_server_protocol::FileChangeApprovalDecision; @@ -823,52 +822,16 @@ pub(crate) async fn apply_bespoke_event_handling( on_request_permissions_response(pending_response, conversation, thread_state).await; }); } - EventMsg::DynamicToolCallRequest(request) => { - let call_id = request.call_id; - let turn_id = request.turn_id; - let namespace = request.namespace; - let tool = request.tool; - let arguments = request.arguments; - let item = ThreadItem::DynamicToolCall { - id: call_id.clone(), - namespace: namespace.clone(), - tool: tool.clone(), - arguments: arguments.clone(), - status: DynamicToolCallStatus::InProgress, - content_items: None, - success: None, - duration_ms: None, - }; - let notification = ItemStartedNotification { - thread_id: conversation_id.to_string(), - turn_id: turn_id.clone(), - started_at_ms: request.started_at_ms, - item, - }; - outgoing - .send_server_notification(ServerNotification::ItemStarted(notification)) - .await; - let params = DynamicToolCallParams { - thread_id: conversation_id.to_string(), - turn_id: turn_id.clone(), - call_id: call_id.clone(), - namespace, - tool: tool.clone(), - arguments: arguments.clone(), - }; - let (_pending_request_id, rx) = outgoing - .send_request(ServerRequestPayload::DynamicToolCall(params)) - .await; - tokio::spawn(async move { - crate::dynamic_tools::on_call_response(call_id, rx, conversation).await; - }); + EventMsg::DynamicToolCallRequest(_) | EventMsg::DynamicToolCallResponse(_) => { + // Deprecated dynamic-tool events are still fanned out for raw-event and rollout + // compatibility consumers. App-server v2 receives the canonical DynamicToolCall + // item lifecycle and dispatches client requests from canonical starts instead. } EventMsg::McpToolCallBegin(_) | EventMsg::McpToolCallEnd(_) => { // Deprecated MCP tool-call events are still fanned out for legacy clients. // App-server v2 receives the canonical TurnItem::McpToolCall lifecycle instead. } - msg @ (EventMsg::DynamicToolCallResponse(_) - | EventMsg::CollabAgentSpawnBegin(_) + msg @ (EventMsg::CollabAgentSpawnBegin(_) | EventMsg::CollabAgentSpawnEnd(_) | EventMsg::CollabAgentInteractionBegin(_) | EventMsg::CollabAgentInteractionEnd(_) @@ -1040,6 +1003,17 @@ pub(crate) async fn apply_bespoke_event_handling( .insert(item.id.clone()), _ => true, }; + let dynamic_tool_call_params = match &event.item { + CoreTurnItem::DynamicToolCall(item) => Some(DynamicToolCallParams { + thread_id: conversation_id.to_string(), + turn_id: event.turn_id.clone(), + call_id: item.id.clone(), + namespace: item.namespace.clone(), + tool: item.tool.clone(), + arguments: item.arguments.clone(), + }), + _ => None, + }; if should_emit { let notification = item_event_to_server_notification( EventMsg::ItemStarted(event), @@ -1048,6 +1022,15 @@ pub(crate) async fn apply_bespoke_event_handling( ); outgoing.send_server_notification(notification).await; } + if let Some(params) = dynamic_tool_call_params { + let call_id = params.call_id.clone(); + let (_pending_request_id, rx) = outgoing + .send_request(ServerRequestPayload::DynamicToolCall(params)) + .await; + tokio::spawn(async move { + crate::dynamic_tools::on_call_response(call_id, rx, conversation).await; + }); + } } EventMsg::ItemCompleted(event) => { apply_canonical_item_completed_side_effects(&thread_state, &event.item).await; @@ -2185,10 +2168,14 @@ mod tests { use codex_app_server_protocol::AutoReviewDecisionSource; use codex_app_server_protocol::GuardianApprovalReviewStatus; use codex_app_server_protocol::JSONRPCErrorError; + use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::TurnPlanStepStatus; use codex_login::CodexAuth; use codex_protocol::AgentPath; + use codex_protocol::items::DynamicToolCallItem; + use codex_protocol::items::DynamicToolCallStatus as CoreDynamicToolCallStatus; use codex_protocol::items::HookPromptFragment; + use codex_protocol::items::TurnItem as CoreTurnItem; use codex_protocol::items::build_hook_prompt_message; use codex_protocol::models::FileSystemPermissions as CoreFileSystemPermissions; use codex_protocol::models::NetworkPermissions as CoreNetworkPermissions; @@ -2205,6 +2192,7 @@ mod tests { use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::GuardianAssessmentEvent; use codex_protocol::protocol::GuardianAssessmentStatus; + use codex_protocol::protocol::ItemStartedEvent; use codex_protocol::protocol::RateLimitSnapshot; use codex_protocol::protocol::RateLimitWindow; use codex_protocol::protocol::RolloutItem; @@ -3482,6 +3470,92 @@ mod tests { Ok(()) } + #[tokio::test] + async fn canonical_dynamic_tool_start_emits_item_and_requests_client() -> Result<()> { + let codex_home = TempDir::new()?; + let config = load_default_config_for_test(&codex_home).await; + let thread_manager = Arc::new( + codex_core::test_support::thread_manager_with_models_provider_and_home( + CodexAuth::create_dummy_chatgpt_auth_for_testing(), + config.model_provider.clone(), + config.codex_home.to_path_buf(), + Arc::new(codex_exec_server::EnvironmentManager::default_for_tests()), + ), + ); + let codex_core::NewThread { + thread_id: conversation_id, + thread: conversation, + .. + } = thread_manager.start_thread(config).await?; + let (tx, mut rx) = mpsc::channel(CHANNEL_CAPACITY); + let outgoing = Arc::new(OutgoingMessageSender::new( + tx, + codex_analytics::AnalyticsEventsClient::disabled(), + )); + let outgoing = ThreadScopedOutgoingMessageSender::new( + outgoing, + vec![ConnectionId(1)], + conversation_id, + ); + + apply_bespoke_event_handling( + Event { + id: "turn-1".to_string(), + msg: EventMsg::ItemStarted(ItemStartedEvent { + thread_id: conversation_id, + turn_id: "turn-1".to_string(), + item: CoreTurnItem::DynamicToolCall(DynamicToolCallItem { + id: "dynamic-1".to_string(), + namespace: Some("apps".to_string()), + tool: "lookup".to_string(), + arguments: json!({"id": "123"}), + status: CoreDynamicToolCallStatus::InProgress, + content_items: None, + success: None, + error: None, + duration: None, + }), + started_at_ms: 42, + }), + }, + conversation_id, + conversation, + thread_manager, + outgoing, + new_thread_state(), + ThreadWatchManager::new(), + Arc::new(tokio::sync::Semaphore::new(/*permits*/ 1)), + "test-provider".to_string(), + ) + .await; + + let item_started = recv_broadcast_message(&mut rx).await?; + let OutgoingMessage::AppServerNotification(ServerNotification::ItemStarted(payload)) = + item_started + else { + bail!("unexpected message: {item_started:?}"); + }; + assert_eq!(payload.item.id(), "dynamic-1"); + + let request = recv_broadcast_message(&mut rx).await?; + let OutgoingMessage::Request(ServerRequest::DynamicToolCall { params, .. }) = request + else { + bail!("unexpected message: {request:?}"); + }; + assert_eq!( + params, + DynamicToolCallParams { + thread_id: conversation_id.to_string(), + turn_id: "turn-1".to_string(), + call_id: "dynamic-1".to_string(), + namespace: Some("apps".to_string()), + tool: "lookup".to_string(), + arguments: json!({"id": "123"}), + } + ); + Ok(()) + } + #[tokio::test] async fn test_handle_turn_complete_emits_completed_without_error() -> Result<()> { let conversation_id = ThreadId::new(); diff --git a/codex-rs/core/src/tools/handlers/dynamic.rs b/codex-rs/core/src/tools/handlers/dynamic.rs index 28af3434ff12..4b8fa037974e 100644 --- a/codex-rs/core/src/tools/handlers/dynamic.rs +++ b/codex-rs/core/src/tools/handlers/dynamic.rs @@ -9,14 +9,13 @@ use crate::tools::handlers::parse_arguments; use crate::tools::registry::CoreToolRuntime; use crate::tools::registry::ToolExecutor; use crate::tools::registry::ToolExposure; -use crate::turn_timing::now_unix_timestamp_ms; -use codex_protocol::dynamic_tools::DynamicToolCallRequest; use codex_protocol::dynamic_tools::DynamicToolFunctionSpec; use codex_protocol::dynamic_tools::DynamicToolNamespaceSpec; use codex_protocol::dynamic_tools::DynamicToolResponse; +use codex_protocol::items::DynamicToolCallItem; +use codex_protocol::items::DynamicToolCallStatus; +use codex_protocol::items::TurnItem; use codex_protocol::models::FunctionCallOutputContentItem; -use codex_protocol::protocol::DynamicToolCallResponseEvent; -use codex_protocol::protocol::EventMsg; use codex_tools::ResponsesApiNamespace; use codex_tools::ResponsesApiNamespaceTool; use codex_tools::ToolName; @@ -178,7 +177,6 @@ async fn request_dynamic_tool( ) -> Option { let namespace = tool_name.namespace; let tool = tool_name.name; - let turn_id = turn_context.sub_id.clone(); let (tx_response, rx_response) = oneshot::channel(); let event_id = call_id.clone(); let prev_entry = { @@ -196,45 +194,55 @@ async fn request_dynamic_tool( } let started_at = Instant::now(); - let started_at_ms = now_unix_timestamp_ms(); - let event = EventMsg::DynamicToolCallRequest(DynamicToolCallRequest { - call_id: call_id.clone(), - turn_id: turn_id.clone(), - started_at_ms, - namespace: namespace.clone(), - tool: tool.clone(), - arguments: arguments.clone(), - }); - session.send_event(turn_context, event).await; + session + .emit_turn_item_started( + turn_context, + &TurnItem::DynamicToolCall(DynamicToolCallItem { + id: call_id.clone(), + namespace: namespace.clone(), + tool: tool.clone(), + arguments: arguments.clone(), + status: DynamicToolCallStatus::InProgress, + content_items: None, + success: None, + error: None, + duration: None, + }), + ) + .await; let response = rx_response.await.ok(); - let response_event = match &response { - Some(response) => EventMsg::DynamicToolCallResponse(DynamicToolCallResponseEvent { - call_id, - turn_id, - completed_at_ms: now_unix_timestamp_ms(), + let item = match &response { + Some(response) => DynamicToolCallItem { + id: call_id, namespace, tool, arguments, - content_items: response.content_items.clone(), - success: response.success, + status: if response.success { + DynamicToolCallStatus::Completed + } else { + DynamicToolCallStatus::Failed + }, + content_items: Some(response.content_items.clone()), + success: Some(response.success), error: None, - duration: started_at.elapsed(), - }), - None => EventMsg::DynamicToolCallResponse(DynamicToolCallResponseEvent { - call_id, - turn_id, - completed_at_ms: now_unix_timestamp_ms(), + duration: Some(started_at.elapsed()), + }, + None => DynamicToolCallItem { + id: call_id, namespace, tool, arguments, - content_items: Vec::new(), - success: false, + status: DynamicToolCallStatus::Failed, + content_items: Some(Vec::new()), + success: Some(false), error: Some("dynamic tool call was cancelled before receiving a response".to_string()), - duration: started_at.elapsed(), - }), + duration: Some(started_at.elapsed()), + }, }; - session.send_event(turn_context, response_event).await; + session + .emit_turn_item_completed(turn_context, TurnItem::DynamicToolCall(item)) + .await; response }