diff --git a/codex-rs/app-server-protocol/schema/json/ServerNotification.json b/codex-rs/app-server-protocol/schema/json/ServerNotification.json index 90899cb152..489d2cfaaf 100644 --- a/codex-rs/app-server-protocol/schema/json/ServerNotification.json +++ b/codex-rs/app-server-protocol/schema/json/ServerNotification.json @@ -4830,6 +4830,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json index dda285476d..c532665190 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json @@ -18259,6 +18259,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json index 3a23ba3a69..a80599d356 100644 --- a/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json +++ b/codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.v2.schemas.json @@ -16083,6 +16083,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ErrorNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ErrorNotification.json index fd55d08764..6f31beb813 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ErrorNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ErrorNotification.json @@ -164,6 +164,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json index 877eeb1278..4d587543e2 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ReviewStartResponse.json @@ -1390,6 +1390,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json index 8c45f57d38..0deadad717 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json @@ -1950,6 +1950,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json index a9a186228b..2a5fdda1dd 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json @@ -1765,6 +1765,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json index a2a4834cb3..8bebcdf368 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadMetadataUpdateResponse.json @@ -1765,6 +1765,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json index 5d76f9b181..c2ca377790 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json @@ -1765,6 +1765,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json index 8055ff58fc..d9e542f34d 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json @@ -1950,6 +1950,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json index d0e25a1eaa..7af419c3bb 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json @@ -1765,6 +1765,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json index d456ccebc4..4dd55d350c 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json @@ -1950,6 +1950,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json index dd7de34b82..8b8c9ecfb0 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json @@ -1765,6 +1765,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json b/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json index 5929510520..c07d2f49d6 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json @@ -1765,6 +1765,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json index ddaab863f7..1f0800afa6 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnCompletedNotification.json @@ -1390,6 +1390,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json b/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json index a2836f7f03..30af2dd4af 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnStartResponse.json @@ -1390,6 +1390,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json b/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json index 8c57890a3b..f6ee8cf652 100644 --- a/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json +++ b/codex-rs/app-server-protocol/schema/json/v2/TurnStartedNotification.json @@ -1390,6 +1390,9 @@ } ] }, + "data": { + "default": null + }, "message": { "type": "string" } diff --git a/codex-rs/app-server-protocol/schema/typescript/v2/TurnError.ts b/codex-rs/app-server-protocol/schema/typescript/v2/TurnError.ts index 765a8e050b..e067369922 100644 --- a/codex-rs/app-server-protocol/schema/typescript/v2/TurnError.ts +++ b/codex-rs/app-server-protocol/schema/typescript/v2/TurnError.ts @@ -1,6 +1,7 @@ // GENERATED CODE! DO NOT MODIFY BY HAND! // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { JsonValue } from "../serde_json/JsonValue"; import type { CodexErrorInfo } from "./CodexErrorInfo"; -export type TurnError = { message: string, codexErrorInfo: CodexErrorInfo | null, additionalDetails: string | null, }; +export type TurnError = { message: string, codexErrorInfo: CodexErrorInfo | null, additionalDetails: string | null, data: JsonValue | null, }; diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index 0fbb93f397..d9e07c48b4 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -21,6 +21,7 @@ use crate::protocol::v2::TurnItemsView; use crate::protocol::v2::TurnStatus; use crate::protocol::v2::UserInput; use crate::protocol::v2::WebSearchAction; +use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::items::parse_hook_prompt_message; use codex_protocol::models::MessagePhase; use codex_protocol::protocol::AgentReasoningEvent; @@ -46,9 +47,11 @@ use codex_protocol::protocol::PatchApplyBeginEvent; use codex_protocol::protocol::PatchApplyEndEvent; use codex_protocol::protocol::ReviewOutputEvent; use codex_protocol::protocol::RolloutItem; +use codex_protocol::protocol::SessionMetaLine; use codex_protocol::protocol::ThreadRolledBackEvent; use codex_protocol::protocol::TurnAbortedEvent; use codex_protocol::protocol::TurnCompleteEvent; +use codex_protocol::protocol::TurnContextItem; use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::UserMessageEvent; use codex_protocol::protocol::ViewImageToolCallEvent; @@ -86,6 +89,8 @@ pub fn build_turns_from_rollout_items(items: &[RolloutItem]) -> Vec { pub struct ThreadHistoryBuilder { turns: Vec, current_turn: Option, + dynamic_tools: Vec, + pending_context_turn_id: Option, next_item_index: i64, current_rollout_index: usize, next_rollout_index: usize, @@ -102,6 +107,8 @@ impl ThreadHistoryBuilder { Self { turns: Vec::new(), current_turn: None, + dynamic_tools: Vec::new(), + pending_context_turn_id: None, next_item_index: 1, current_rollout_index: 0, next_rollout_index: 0, @@ -232,10 +239,19 @@ impl ThreadHistoryBuilder { RolloutItem::EventMsg(event) => self.handle_event(event), RolloutItem::Compacted(payload) => self.handle_compacted(payload), RolloutItem::ResponseItem(item) => self.handle_response_item(item), - RolloutItem::TurnContext(_) | RolloutItem::SessionMeta(_) => {} + RolloutItem::TurnContext(payload) => self.handle_turn_context(payload), + RolloutItem::SessionMeta(payload) => self.handle_session_meta(payload), } } + fn handle_session_meta(&mut self, payload: &SessionMetaLine) { + self.dynamic_tools = payload.meta.dynamic_tools.clone().unwrap_or_default(); + } + + fn handle_turn_context(&mut self, payload: &TurnContextItem) { + self.pending_context_turn_id.clone_from(&payload.turn_id); + } + fn handle_response_item(&mut self, item: &codex_protocol::models::ResponseItem) { let codex_protocol::models::ResponseItem::Message { role, content, id, .. @@ -880,15 +896,23 @@ impl ThreadHistoryBuilder { if !payload.affects_turn_status() { return; } + if self.current_turn.is_none() + && let Some(turn_id) = self.pending_context_turn_id.take() + { + self.current_turn = Some( + self.new_turn(Some(turn_id)) + .with_status(TurnStatus::Failed) + .opened_explicitly(), + ); + } let Some(turn) = self.current_turn.as_mut() else { return; }; turn.status = TurnStatus::Failed; - turn.error = Some(V2TurnError { - message: payload.message.clone(), - codex_error_info: payload.codex_error_info.clone().map(Into::into), - additional_details: None, - }); + turn.error = Some(V2TurnError::from_core_error_event( + payload, + Some(&self.dynamic_tools), + )); } fn handle_turn_aborted(&mut self, payload: &TurnAbortedEvent) { @@ -919,6 +943,7 @@ impl ThreadHistoryBuilder { } fn handle_turn_started(&mut self, payload: &TurnStartedEvent) { + self.pending_context_turn_id = None; self.finish_current_turn(); self.current_turn = Some( self.new_turn(Some(payload.turn_id.clone())) @@ -929,6 +954,7 @@ impl ThreadHistoryBuilder { } fn handle_turn_complete(&mut self, payload: &TurnCompleteEvent) { + self.pending_context_turn_id = None; let mark_completed = |turn: &mut PendingTurn| { if matches!(turn.status, TurnStatus::Completed | TurnStatus::InProgress) { turn.status = TurnStatus::Completed; @@ -993,7 +1019,11 @@ impl ThreadHistoryBuilder { fn finish_current_turn(&mut self) { if let Some(turn) = self.current_turn.take() { - if turn.items.is_empty() && !turn.opened_explicitly && !turn.saw_compaction { + if turn.items.is_empty() + && turn.error.is_none() + && !turn.opened_explicitly + && !turn.saw_compaction + { return; } self.turns.push(Turn::from(turn)); @@ -3093,6 +3123,85 @@ mod tests { ); } + #[test] + fn error_after_turn_context_creates_failed_turn_with_dynamic_tool_schema_data() { + let error_message = "invalid_function_parameters ... Invalid schema for function 'create': schema must be a JSON Schema of 'type: \"object\"', got 'type: \"None\""; + let items = vec![ + RolloutItem::SessionMeta(SessionMetaLine { + meta: codex_protocol::protocol::SessionMeta { + dynamic_tools: Some(vec![DynamicToolSpec { + namespace: Some("repro".to_string()), + name: "create".to_string(), + description: "test".to_string(), + input_schema: serde_json::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, + }]), + ..Default::default() + }, + git: None, + }), + RolloutItem::TurnContext(turn_context_item("turn-a")), + RolloutItem::EventMsg(EventMsg::Error(ErrorEvent { + message: error_message.to_string(), + codex_error_info: Some(CodexErrorInfo::BadRequest), + })), + ]; + + let turns = build_turns_from_rollout_items(&items); + + assert_eq!( + turns, + vec![Turn { + id: "turn-a".to_string(), + items: Vec::new(), + items_view: TurnItemsView::Full, + status: TurnStatus::Failed, + error: Some(TurnError { + message: error_message.to_string(), + codex_error_info: Some(crate::protocol::v2::CodexErrorInfo::BadRequest), + additional_details: None, + data: Some(serde_json::json!({ + "reason": "dynamicToolInputSchema", + "backend": "Responses API", + "tool": { + "index": 0, + "namespace": "repro", + "name": "create", + "qualifiedName": "repro.create" + }, + "schemaPath": "dynamicTools[0].inputSchema.anyOf", + "underlyingError": error_message, + "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." + })), + }), + started_at: None, + completed_at: None, + duration_ms: None, + }] + ); + } + #[test] fn error_then_turn_complete_preserves_failed_status() { let events = vec![ @@ -3142,6 +3251,7 @@ mod tests { } ), additional_details: None, + data: None, }) ); } @@ -3229,4 +3339,24 @@ mod tests { assert_eq!(turns.len(), 1); assert!(turns[0].items.is_empty()); } + + fn turn_context_item(turn_id: &str) -> TurnContextItem { + TurnContextItem { + turn_id: Some(turn_id.to_string()), + cwd: PathBuf::from("/tmp"), + current_date: None, + timezone: None, + approval_policy: codex_protocol::protocol::AskForApproval::OnRequest, + sandbox_policy: codex_protocol::protocol::SandboxPolicy::DangerFullAccess, + permission_profile: None, + network: None, + file_system_sandbox_policy: None, + model: "gpt-5".to_string(), + personality: None, + collaboration_mode: None, + realtime_active: None, + effort: None, + summary: codex_protocol::config_types::ReasoningSummary::Auto, + } + } } diff --git a/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs index f0c518adf8..a098482ec6 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/thread_data.rs @@ -2,6 +2,8 @@ use super::CodexErrorInfo; use super::ThreadItem; use super::ThreadStatus; use super::TurnStatus; +use codex_protocol::dynamic_tools::DynamicToolSpec as CoreDynamicToolSpec; +use codex_protocol::protocol::ErrorEvent as CoreErrorEvent; use codex_protocol::protocol::SessionSource as CoreSessionSource; use codex_protocol::protocol::SubAgentSource as CoreSubAgentSource; use codex_protocol::protocol::ThreadSource as CoreThreadSource; @@ -9,6 +11,8 @@ use codex_utils_absolute_path::AbsolutePathBuf; use schemars::JsonSchema; use serde::Deserialize; use serde::Serialize; +use serde_json::Value as JsonValue; +use serde_json::json; use std::path::PathBuf; use thiserror::Error; use ts_rs::TS; @@ -193,4 +197,117 @@ pub struct TurnError { pub codex_error_info: Option, #[serde(default)] pub additional_details: Option, + #[serde(default)] + pub data: Option, +} + +impl TurnError { + pub fn from_core_error_event( + event: &CoreErrorEvent, + dynamic_tools: Option<&[CoreDynamicToolSpec]>, + ) -> Self { + Self { + message: event.message.clone(), + codex_error_info: event.codex_error_info.clone().map(Into::into), + additional_details: None, + data: dynamic_tool_input_schema_error_data( + &event.message, + dynamic_tools.unwrap_or(&[]), + ), + } + } +} + +fn dynamic_tool_input_schema_error_data( + message: &str, + dynamic_tools: &[CoreDynamicToolSpec], +) -> Option { + let underlying_error = underlying_backend_error_message(message); + let backend_function_name = backend_function_name_from_schema_error(&underlying_error)?; + let matching_tools = dynamic_tools + .iter() + .enumerate() + .filter(|(_, tool)| { + backend_function_name == tool.name + || backend_function_name == qualified_dynamic_tool_name(tool) + }) + .collect::>(); + + let [(index, tool)] = matching_tools.as_slice() else { + return None; + }; + let qualified_name = qualified_dynamic_tool_name(tool); + let schema_path = dynamic_tool_schema_path(*index, &tool.input_schema); + + Some(json!({ + "reason": "dynamicToolInputSchema", + "backend": "Responses API", + "tool": { + "index": index, + "namespace": tool.namespace, + "name": tool.name, + "qualifiedName": qualified_name, + }, + "schemaPath": schema_path, + "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.", + })) +} + +fn underlying_backend_error_message(message: &str) -> String { + let Ok(value) = serde_json::from_str::(message) else { + return message.to_string(); + }; + value + .get("error") + .and_then(|error| error.get("message")) + .and_then(JsonValue::as_str) + .map(str::trim) + .filter(|message| !message.is_empty()) + .unwrap_or(message) + .to_string() +} + +fn backend_function_name_from_schema_error(message: &str) -> Option<&str> { + if !message.contains("Invalid schema for function") + && !message.contains("invalid_function_parameters") + { + return None; + } + + for (prefix, suffix) in [ + ("function '", "'"), + ("function \"", "\""), + ("function `", "`"), + ] { + let Some((_, rest)) = message.split_once(prefix) else { + continue; + }; + let Some((name, _)) = rest.split_once(suffix) else { + continue; + }; + if !name.is_empty() { + return Some(name); + } + } + + None +} + +fn dynamic_tool_schema_path(index: usize, input_schema: &JsonValue) -> String { + let suffix = if input_schema.get("anyOf").is_some() { + ".anyOf" + } else if input_schema.get("oneOf").is_some() { + ".oneOf" + } else { + "" + }; + format!("dynamicTools[{index}].inputSchema{suffix}") +} + +fn qualified_dynamic_tool_name(tool: &CoreDynamicToolSpec) -> String { + match tool.namespace.as_deref() { + Some(namespace) if !namespace.is_empty() => format!("{namespace}.{}", tool.name), + Some(_) | None => tool.name.clone(), + } } diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 2ceffc86fe..17bbf365bd 100644 --- a/codex-rs/app-server/README.md +++ b/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 diff --git a/codex-rs/app-server/src/bespoke_event_handling.rs b/codex-rs/app-server/src/bespoke_event_handling.rs index efe6c69a32..1a4aa35e6f 100644 --- a/codex-rs/app-server/src/bespoke_event_handling.rs +++ b/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, }) ); } diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index 6b6e1729a5..acaf4ddb6f 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/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(()) diff --git a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs b/codex-rs/app-server/src/request_processors/thread_processor_tests.rs index f4593ab7c4..f3528a1df3 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor_tests.rs +++ b/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![ diff --git a/codex-rs/app-server/src/request_processors/turn_processor.rs b/codex-rs/app-server/src/request_processors/turn_processor.rs index ff2825be67..8702736154 100644 --- a/codex-rs/app-server/src/request_processors/turn_processor.rs +++ b/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), diff --git a/codex-rs/app-server/src/thread_state.rs b/codex-rs/app-server/src/thread_state.rs index 7f864265b2..ceb0876dc3 100644 --- a/codex-rs/app-server/src/thread_state.rs +++ b/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, pub(crate) turn_summary: TurnSummary, pub(crate) last_terminal_turn_id: Option, + pub(crate) last_terminal_turn: Option, pub(crate) cancel_tx: Option>, 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 { + 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(); } } diff --git a/codex-rs/app-server/tests/suite/v2/dynamic_tools.rs b/codex-rs/app-server/tests/suite/v2/dynamic_tools.rs index b357e139db..51a706fcfc 100644 --- a/codex-rs/app-server/tests/suite/v2/dynamic_tools.rs +++ b/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::(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<()> { diff --git a/codex-rs/app-server/tests/suite/v2/safety_check_downgrade.rs b/codex-rs/app-server/tests/suite/v2/safety_check_downgrade.rs index 7c1619e38a..634f94123b 100644 --- a/codex-rs/app-server/tests/suite/v2/safety_check_downgrade.rs +++ b/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, diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index 1b40387c3f..34661dbe8e 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -498,6 +498,10 @@ impl CodexThread { self.codex.thread_config_snapshot().await } + pub async fn dynamic_tools(&self) -> Vec { + self.codex.dynamic_tools().await + } + pub async fn config(&self) -> Arc { self.codex.session.get_config().await } diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index c73b43a838..8f160f6bb4 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -782,6 +782,11 @@ impl Codex { state.session_configuration.environments.clone() } + pub(crate) async fn dynamic_tools(&self) -> Vec { + let state = self.session.state.lock().await; + state.session_configuration.dynamic_tools.clone() + } + pub(crate) fn state_db(&self) -> Option { self.session.state_db() }