fix(app-server): report dynamic tool schema errors

codex-rs/app-server/README.md

@@ -1410,6 +1410,8 @@ Dynamic tool identifiers follow the same constraints as Responses function tools
 
 Each dynamic tool may set `deferLoading`. When omitted, it defaults to `false`. Set it to `true` to keep the tool registered and callable by runtime features such as `code_mode`, while excluding it from the model-facing tool list sent on ordinary turns. When `tool_search` is available, deferred dynamic tools are searchable and can be exposed by a matching search result.
 
+If the model backend rejects a dynamic tool schema during turn startup, app-server emits the normal `error` notification and stores the failure on the turn returned by `thread/read`. When the backend error names a dynamic tool, `TurnError.data.reason` is `dynamicToolInputSchema` and includes the tool namespace/name, `schemaPath`, backend, underlying backend message, and remediation text.
+
 When a dynamic tool is invoked during a turn, the server sends an `item/tool/call` JSON-RPC request to the client:
 
 ```json

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

@@ -910,11 +910,8 @@ pub(crate) async fn apply_bespoke_event_handling(
                 return;
             }
 
-            let turn_error = TurnError {
-                message: ev.message,
-                codex_error_info: ev.codex_error_info.map(V2CodexErrorInfo::from),
-                additional_details: None,
-            };
+            let dynamic_tools = conversation.dynamic_tools().await;
+            let turn_error = TurnError::from_core_error_event(&ev, Some(dynamic_tools.as_slice()));
             handle_error(conversation_id, turn_error.clone(), &thread_state).await;
             outgoing
                 .send_server_notification(ServerNotification::Error(ErrorNotification {
@@ -932,6 +929,7 @@ pub(crate) async fn apply_bespoke_event_handling(
                 message: ev.message,
                 codex_error_info: ev.codex_error_info.map(V2CodexErrorInfo::from),
                 additional_details: ev.additional_details,
+                data: None,
             };
             outgoing
                 .send_server_notification(ServerNotification::Error(ErrorNotification {
@@ -3178,6 +3176,7 @@ mod tests {
                 message: "boom".to_string(),
                 codex_error_info: Some(V2CodexErrorInfo::InternalServerError),
                 additional_details: None,
+                data: None,
             },
             &thread_state,
         )
@@ -3190,6 +3189,7 @@ mod tests {
                 message: "boom".to_string(),
                 codex_error_info: Some(V2CodexErrorInfo::InternalServerError),
                 additional_details: None,
+                data: None,
             })
         );
         Ok(())
@@ -3350,6 +3350,7 @@ mod tests {
                 message: "oops".to_string(),
                 codex_error_info: None,
                 additional_details: None,
+                data: None,
             },
             &thread_state,
         )
@@ -3400,6 +3401,7 @@ mod tests {
                 message: "bad".to_string(),
                 codex_error_info: Some(V2CodexErrorInfo::Other),
                 additional_details: None,
+                data: None,
             },
             &thread_state,
         )
@@ -3435,6 +3437,7 @@ mod tests {
                         message: "bad".to_string(),
                         codex_error_info: Some(V2CodexErrorInfo::Other),
                         additional_details: None,
+                        data: None,
                     })
                 );
                 assert_eq!(n.turn.completed_at, Some(TEST_TURN_COMPLETED_AT));
@@ -3645,6 +3648,7 @@ mod tests {
                 message: "a1".to_string(),
                 codex_error_info: Some(V2CodexErrorInfo::BadRequest),
                 additional_details: None,
+                data: None,
             },
             &thread_state,
         )
@@ -3666,6 +3670,7 @@ mod tests {
                 message: "b1".to_string(),
                 codex_error_info: None,
                 additional_details: None,
+                data: None,
             },
             &thread_state,
         )
@@ -3702,6 +3707,7 @@ mod tests {
                         message: "a1".to_string(),
                         codex_error_info: Some(V2CodexErrorInfo::BadRequest),
                         additional_details: None,
+                        data: None,
                     })
                 );
             }
@@ -3720,6 +3726,7 @@ mod tests {
                         message: "b1".to_string(),
                         codex_error_info: None,
                         additional_details: None,
+                        data: None,
                     })
                 );
             }

codex-rs/app-server/src/request_processors/thread_processor.rs

@@ -2086,7 +2086,14 @@ impl ThreadRequestProcessor {
                 .load_history(/*include_archived*/ true)
                 .await
                 .map_err(|err| thread_read_history_load_error(thread_id, err))?;
-            thread.turns = build_api_turns_from_rollout_items(&history.items);
+            let live_turn = {
+                let thread_state = self.thread_state_manager.thread_state(thread_id).await;
+                let state = thread_state.lock().await;
+                state
+                    .active_turn_snapshot()
+                    .or_else(|| state.last_terminal_turn_snapshot())
+            };
+            populate_thread_turns_from_history(thread, &history.items, live_turn.as_ref());
         }
 
         Ok(())

codex-rs/app-server/src/request_processors/thread_processor_tests.rs

@@ -133,6 +133,39 @@ mod thread_processor_behavior_tests {
         validate_dynamic_tools(&tools).expect("valid schema");
     }
 
+    #[test]
+    fn validate_dynamic_tools_accepts_root_any_of_input_schema() {
+        let tools = vec![ApiDynamicToolSpec {
+            namespace: Some("repro".to_string()),
+            name: "create".to_string(),
+            description: "test".to_string(),
+            input_schema: json!({
+                "anyOf": [
+                    {
+                        "type": "object",
+                        "properties": {
+                            "kind": { "type": "string", "enum": ["wikiEdit"] },
+                            "slug": { "type": "string" }
+                        },
+                        "required": ["kind", "slug"],
+                        "additionalProperties": false
+                    },
+                    {
+                        "type": "object",
+                        "properties": {
+                            "kind": { "type": "string", "enum": ["chatMessage"] },
+                            "channel": { "type": "string" }
+                        },
+                        "required": ["kind", "channel"],
+                        "additionalProperties": false
+                    }
+                ]
+            }),
+            defer_loading: false,
+        }];
+        validate_dynamic_tools(&tools).expect("valid schema");
+    }
+
     #[test]
     fn validate_dynamic_tools_accepts_same_name_in_different_namespaces() {
         let tools = vec![

codex-rs/app-server/src/request_processors/turn_processor.rs

@@ -787,6 +787,7 @@ impl TurnRequestProcessor {
                                 turn_kind: turn_kind.into(),
                             }),
                             additional_details: None,
+                            data: None,
                         };
                         let data = match serde_json::to_value(error) {
                             Ok(data) => Some(data),

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

@@ -6,6 +6,7 @@ use codex_app_server_protocol::ThreadHistoryBuilder;
 use codex_app_server_protocol::ThreadSettings;
 use codex_app_server_protocol::Turn;
 use codex_app_server_protocol::TurnError;
+use codex_app_server_protocol::TurnStatus;
 use codex_core::CodexThread;
 use codex_core::ThreadConfigSnapshot;
 use codex_file_watcher::WatchRegistration;
@@ -74,6 +75,7 @@ pub(crate) struct ThreadState {
     pub(crate) pending_rollbacks: Option<ConnectionRequestId>,
     pub(crate) turn_summary: TurnSummary,
     pub(crate) last_terminal_turn_id: Option<String>,
+    pub(crate) last_terminal_turn: Option<Turn>,
     pub(crate) cancel_tx: Option<oneshot::Sender<()>>,
     pub(crate) experimental_raw_events: bool,
     pub(crate) listener_generation: u64,
@@ -117,6 +119,8 @@ impl ThreadState {
         }
         self.listener_command_tx = None;
         self.current_turn_history.reset();
+        self.last_terminal_turn_id = None;
+        self.last_terminal_turn = None;
         self.listener_thread = None;
         self.watch_registration = WatchRegistration::default();
     }
@@ -135,15 +139,30 @@ impl ThreadState {
         self.current_turn_history.active_turn_snapshot()
     }
 
+    pub(crate) fn last_terminal_turn_snapshot(&self) -> Option<Turn> {
+        self.last_terminal_turn.clone()
+    }
+
     pub(crate) fn track_current_turn_event(&mut self, event_turn_id: &str, event: &EventMsg) {
         if let EventMsg::TurnStarted(payload) = event {
             self.turn_summary.started_at = payload.started_at;
+            self.last_terminal_turn_id = None;
+            self.last_terminal_turn = None;
         }
         self.current_turn_history.handle_event(event);
         if matches!(event, EventMsg::TurnAborted(_) | EventMsg::TurnComplete(_))
             && !self.current_turn_history.has_active_turn()
         {
             self.last_terminal_turn_id = Some(event_turn_id.to_string());
+            let mut terminal_turn = self.current_turn_history.active_turn_snapshot();
+            if let Some(turn) = terminal_turn.as_mut()
+                && let Some(error) = self.turn_summary.last_error.clone()
+            {
+                turn.status = TurnStatus::Failed;
+                turn.error = Some(error);
+            }
+            self.last_terminal_turn = terminal_turn
+                .filter(|turn| turn.error.is_some() || matches!(turn.status, TurnStatus::Failed));
             self.current_turn_history.reset();
         }
     }

codex-rs/app-server/tests/suite/v2/dynamic_tools.rs

@@ -9,6 +9,7 @@ use codex_app_server_protocol::DynamicToolCallParams;
 use codex_app_server_protocol::DynamicToolCallResponse;
 use codex_app_server_protocol::DynamicToolCallStatus;
 use codex_app_server_protocol::DynamicToolSpec;
+use codex_app_server_protocol::ErrorNotification;
 use codex_app_server_protocol::ItemCompletedNotification;
 use codex_app_server_protocol::ItemStartedNotification;
 use codex_app_server_protocol::JSONRPCNotification;
@@ -16,10 +17,13 @@ use codex_app_server_protocol::JSONRPCResponse;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::ServerRequest;
 use codex_app_server_protocol::ThreadItem;
+use codex_app_server_protocol::ThreadReadParams;
+use codex_app_server_protocol::ThreadReadResponse;
 use codex_app_server_protocol::ThreadStartParams;
 use codex_app_server_protocol::ThreadStartResponse;
 use codex_app_server_protocol::TurnStartParams;
 use codex_app_server_protocol::TurnStartResponse;
+use codex_app_server_protocol::TurnStatus;
 use codex_app_server_protocol::UserInput as V2UserInput;
 use codex_protocol::models::DEFAULT_IMAGE_DETAIL;
 use codex_protocol::models::FunctionCallOutputBody;
@@ -34,6 +38,7 @@ use std::time::Duration;
 use tempfile::TempDir;
 use tokio::time::timeout;
 use wiremock::MockServer;
+use wiremock::ResponseTemplate;
 
 // macOS and Windows Bazel CI can spend tens of seconds starting app-server
 // subprocesses or processing test RPCs under load.
@@ -279,6 +284,156 @@ async fn thread_start_rejects_dynamic_tools_not_supported_by_responses() -> Resu
     Ok(())
 }
 
+#[tokio::test]
+async fn backend_dynamic_tool_schema_error_is_reported_to_clients() -> Result<()> {
+    let server = MockServer::start().await;
+    let backend_error = "invalid_function_parameters ... Invalid schema for function 'create': schema must be a JSON Schema of 'type: \"object\"', got 'type: \"None\"";
+    let _response_mock = responses::mount_response_once(
+        &server,
+        ResponseTemplate::new(400).set_body_json(json!({
+            "error": {
+                "message": backend_error,
+                "type": "invalid_request_error",
+                "param": "tools",
+                "code": "invalid_function_parameters"
+            }
+        })),
+    )
+    .await;
+
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri())?;
+
+    let mut mcp = McpProcess::new(codex_home.path()).await?;
+    timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
+
+    let dynamic_tool = DynamicToolSpec {
+        namespace: Some("repro".to_string()),
+        name: "create".to_string(),
+        description: "Repro tool with a top-level anyOf schema.".to_string(),
+        input_schema: json!({
+            "anyOf": [
+                {
+                    "type": "object",
+                    "properties": {
+                        "kind": { "type": "string", "enum": ["wikiEdit"] },
+                        "slug": { "type": "string", "minLength": 1 }
+                    },
+                    "required": ["kind", "slug"],
+                    "additionalProperties": false
+                },
+                {
+                    "type": "object",
+                    "properties": {
+                        "kind": { "type": "string", "enum": ["chatMessage"] },
+                        "channel": { "type": "string", "minLength": 1 }
+                    },
+                    "required": ["kind", "channel"],
+                    "additionalProperties": false
+                }
+            ]
+        }),
+        defer_loading: false,
+    };
+
+    let thread_req = mcp
+        .send_thread_start_request(ThreadStartParams {
+            dynamic_tools: Some(vec![dynamic_tool]),
+            ..Default::default()
+        })
+        .await?;
+    let thread_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(thread_req)),
+    )
+    .await??;
+    let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(thread_resp)?;
+
+    let turn_req = mcp
+        .send_turn_start_request(TurnStartParams {
+            thread_id: thread.id.clone(),
+            input: vec![V2UserInput::Text {
+                text: "Say hello.".to_string(),
+                text_elements: Vec::new(),
+            }],
+            ..Default::default()
+        })
+        .await?;
+    let turn_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(turn_req)),
+    )
+    .await??;
+    let turn_start: TurnStartResponse = to_response(turn_resp)?;
+
+    let error_notification = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("error"),
+    )
+    .await??;
+    let error_notification: ErrorNotification =
+        serde_json::from_value(error_notification.params.context("error params")?)?;
+    let underlying_error = error_notification.error.message.clone();
+    let expected_error_data = expected_dynamic_tool_schema_error_data(backend_error);
+
+    assert!(underlying_error.contains("Invalid schema for function 'create'"));
+    assert_eq!(
+        error_notification.error.data,
+        Some(expected_error_data.clone())
+    );
+
+    timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_notification_message("turn/completed"),
+    )
+    .await??;
+
+    let read_req = mcp
+        .send_thread_read_request(ThreadReadParams {
+            thread_id: thread.id.clone(),
+            include_turns: true,
+        })
+        .await?;
+    let read_resp: JSONRPCResponse = timeout(
+        DEFAULT_READ_TIMEOUT,
+        mcp.read_stream_until_response_message(RequestId::Integer(read_req)),
+    )
+    .await??;
+    let read: ThreadReadResponse = to_response(read_resp)?;
+    let failed_turn = read
+        .thread
+        .turns
+        .iter()
+        .find(|turn| turn.id == turn_start.turn.id)
+        .context("expected failed turn in thread/read")?;
+    assert_eq!(failed_turn.status, TurnStatus::Failed);
+    assert_eq!(
+        failed_turn
+            .error
+            .as_ref()
+            .and_then(|error| error.data.clone()),
+        Some(expected_error_data)
+    );
+
+    Ok(())
+}
+
+fn expected_dynamic_tool_schema_error_data(underlying_error: &str) -> Value {
+    json!({
+        "reason": "dynamicToolInputSchema",
+        "backend": "Responses API",
+        "tool": {
+            "index": 0,
+            "namespace": "repro",
+            "name": "create",
+            "qualifiedName": "repro.create"
+        },
+        "schemaPath": "dynamicTools[0].inputSchema.anyOf",
+        "underlyingError": underlying_error,
+        "remediation": "Adjust the dynamic tool inputSchema to the JSON Schema subset accepted by the model backend. For discriminated unions, wrap the union in a top-level object property or flatten it into one object schema."
+    })
+}
+
 /// Exercises the full dynamic tool call path (server request, client response, model output).
 #[tokio::test]
 async fn dynamic_tool_call_round_trip_sends_text_content_items_to_model() -> Result<()> {

codex-rs/app-server/tests/suite/v2/safety_check_downgrade.rs

@@ -155,6 +155,7 @@ async fn cyber_policy_response_emits_typed_error_notification_v2() -> Result<()>
                 message: CYBER_POLICY_MESSAGE.to_string(),
                 codex_error_info: Some(CodexErrorInfo::CyberPolicy),
                 additional_details: None,
+                data: None,
             },
             will_retry: false,
             thread_id: thread.id,