feat(core): emit canonical dynamic tool call items (#31298)

This PR depends on [#31296](https://github.com/openai/codex/pull/31296)
for the canonical-to-legacy event mappings.

## Description

This PR makes dynamic tools emit canonical `TurnItem::DynamicToolCall`
lifecycle instead of `DynamicToolCallRequest` /
`DynamicToolCallResponse` directly.

App-server v2 now sends the client `DynamicToolCall` request from the
canonical item start. It ignores the mapped legacy request/response
events, so clients receive one item start and one tool request.

## Why

Dynamic tools are a separate migration boundary because their start
event also drives a client request. Keeping that routing change isolated
makes it easier to verify that the request still happens exactly once.

## What changed

- Emit in-progress and completed/failed dynamic tool items from the
dynamic tool handler.
- Move app-server client request dispatch onto canonical dynamic item
starts.
- Add focused app-server coverage for the canonical start notification
and client request.

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();

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<DynamicToolResponse> {
     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
 }