diff --git a/apis/src/openai/responses/agentic_loop/mod.rs b/apis/src/openai/responses/agentic_loop/mod.rs index e3346b658c..9f6ea3eec6 100644 --- a/apis/src/openai/responses/agentic_loop/mod.rs +++ b/apis/src/openai/responses/agentic_loop/mod.rs @@ -334,13 +334,13 @@ fn evaluate_loop_decision( ) -> Result { if state.tool_calls.is_empty() && state.web_search_calls.is_empty() { trace!("no tool calls, signaling done"); - finalize_response_body(state, body); + state.finalize_response_body(body); return set_done(ctx); } match check_exit_conditions(state, config) { Some(ExitReason::FinishReasonLength) => { ctx.set_metadata(META_STATUS, "incomplete"); - finalize_response_body(state, body); + state.finalize_response_body(body); set_action(ctx, ACTION_DONE)?; Ok(FilterAction::Continue) }, @@ -354,7 +354,7 @@ fn evaluate_loop_decision( state.iteration += 1; let (tc, wsc) = (state.tool_calls.len(), state.web_search_calls.len()); debug!(iteration = state.iteration, tc, wsc, "pending calls, signaling loop"); - finalize_response_body(state, body); + state.finalize_response_body(body); set_action(ctx, ACTION_LOOP)?; Ok(FilterAction::Continue) }, @@ -480,29 +480,6 @@ fn is_finish_reason_length(state: &ResponsesState) -> bool { // Helpers // ----------------------------------------------------------------------------- -/// Build the final response body from accumulated state. -/// -/// Replaces `response_object["output"]` with the full -/// `accumulated_output` (all rounds), stamps accumulated usage, -/// and serializes back to body bytes. -fn finalize_response_body(state: &ResponsesState, body: &mut Option) { - if !state.response_object.is_object() { - return; - } - let mut response = state.response_object.clone(); - if let Some(obj) = response.as_object_mut() { - if !state.accumulated_output.is_empty() { - obj.insert("output".to_owned(), Value::Array(state.accumulated_output.clone())); - } - if !state.usage.is_null() { - obj.insert("usage".to_owned(), state.usage.clone()); - } - } - if let Ok(serialized) = serde_json::to_vec(&response) { - *body = Some(Bytes::from(serialized)); - } -} - /// Shorthand: set `action = "done"` and return `Continue`. fn set_done(ctx: &mut HttpFilterContext<'_>) -> Result { set_action(ctx, ACTION_DONE)?; diff --git a/apis/src/openai/responses/mcp_dispatch/mod.rs b/apis/src/openai/responses/mcp_dispatch/mod.rs index 6597cae883..3390a6dcf4 100644 --- a/apis/src/openai/responses/mcp_dispatch/mod.rs +++ b/apis/src/openai/responses/mcp_dispatch/mod.rs @@ -117,6 +117,7 @@ impl McpDispatchFilter { /// Handle a tool call that requires approval. fn handle_approval_required( ctx: &mut HttpFilterContext<'_>, + body: &mut Option, pending: &PendingApproval, ) -> Result { debug!( @@ -139,6 +140,9 @@ impl McpDispatchFilter { }; state.accumulated_output.push(approval_event); + // Re-serialize the response body with the new mcp_approval_request event + state.finalize_response_body(body); + ctx.set_metadata("openai_mcp_dispatch.action".to_owned(), "done".to_owned()); set_action(ctx, ACTION_DONE)?; @@ -220,7 +224,7 @@ impl HttpFilter for McpDispatchFilter { fn on_response_body( &self, ctx: &mut HttpFilterContext<'_>, - _body: &mut Option, + body: &mut Option, end_of_stream: bool, ) -> Result { if !end_of_stream { @@ -238,7 +242,7 @@ impl HttpFilter for McpDispatchFilter { } if let Some(pending) = find_approval_required(&mcp_calls, &state.mcp_tool_map) { - return Self::handle_approval_required(ctx, &pending); + return Self::handle_approval_required(ctx, body, &pending); } ctx.set_metadata("openai_mcp_dispatch.action".to_owned(), "execute_mcp".to_owned()); diff --git a/apis/src/openai/responses/mcp_dispatch/tests.rs b/apis/src/openai/responses/mcp_dispatch/tests.rs index b26efee8f4..e78c23db2f 100644 --- a/apis/src/openai/responses/mcp_dispatch/tests.rs +++ b/apis/src/openai/responses/mcp_dispatch/tests.rs @@ -1033,6 +1033,41 @@ fn on_response_body_approval_emits_correct_arguments() { ); } +#[test] +fn on_response_body_approval_serializes_approval_request_into_body() { + let filter = make_dispatch_filter(); + let req = make_request(http::Method::POST, "/v1/responses"); + let mut ctx = make_filter_context(&req); + let state = ResponsesState { + mcp_tool_map: sample_tool_map(), + tool_calls: vec![json!({ + "name": "weather__get_weather", + "call_id": "c1", + "arguments": "{\"city\":\"Paris\"}" + })], + response_object: json!({ + "id": "resp_123", + "output": [] + }), + ..ResponsesState::default() + }; + ctx.extensions.insert(state); + + let mut body = Some(Bytes::from(r#"{"id":"resp_123","output":[]}"#)); + let result = filter.on_response_body(&mut ctx, &mut body, true).unwrap(); + assert!(matches!(result, FilterAction::Continue)); + + let bytes = body.expect("response body should be serialized with approval request"); + let response_json: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + let output = response_json["output"].as_array().expect("output should be an array"); + assert_eq!(output.len(), 1, "output array should contain 1 item"); + assert_eq!( + output[0]["type"], "mcp_approval_request", + "output item should be mcp_approval_request" + ); + assert_eq!(output[0]["id"], "c1"); +} + // ========================================================================= // on_request (HttpFilter trait) // ========================================================================= diff --git a/apis/src/openai/responses/state.rs b/apis/src/openai/responses/state.rs index 81e22d68a6..770ccd56c5 100644 --- a/apis/src/openai/responses/state.rs +++ b/apis/src/openai/responses/state.rs @@ -12,6 +12,9 @@ use std::collections::HashMap; +use bytes::Bytes; +use serde_json::Value; + /// Maximum citation file mappings retained during one response execution. pub(crate) const MAX_CITATION_FILES: usize = 1_024; @@ -34,19 +37,19 @@ pub(crate) struct ResponsesState { /// /// Preserves the full object from the request so filters can /// inspect both the strategy type and any parameters. - pub context_management: Option, + pub context_management: Option, /// Conversation scope for multi-turn state. /// /// Can be a string ID or an object with `id`. Controls which /// stored conversation this request belongs to. - pub conversation: Option, + pub conversation: Option, /// Public output retained across local file-search inference rounds. /// /// This is request-local continuation state. Private search context stays /// in [`Self::messages`] and is never exposed through Conversations. - pub file_search_output_items: Vec, + pub file_search_output_items: Vec, /// Additional fields to include in the response. /// @@ -66,7 +69,7 @@ pub(crate) struct ResponsesState { /// Preserved as-is so downstream filters can inspect what the /// client actually sent, independent of conversation history /// resolved by `rehydrate`. - pub input: Vec, + pub input: Vec, /// Current agentic loop iteration (0-indexed). Incremented by /// `openai_agentic_loop` at the start of each new inference round. @@ -84,7 +87,7 @@ pub(crate) struct ResponsesState { /// /// Built by `openai_mcp_tool_resolve` from `tools/list` responses. /// Consumed by `mcp_tool` (#27) for dispatch routing. - pub mcp_tool_map: HashMap<(String, String), serde_json::Value>, + pub mcp_tool_map: HashMap<(String, String), Value>, /// Resolved conversation history sent to the backend. /// @@ -94,7 +97,7 @@ pub(crate) struct ResponsesState { /// loops. `openai_responses_proxy` reads this as the authoritative /// conversation to send to the backend. Output-only metadata /// items must be omitted from this field. - pub messages: Vec, + pub messages: Vec, /// Whether tool calls may execute concurrently within an /// iteration. Defaults to `true` per the API spec. @@ -105,7 +108,7 @@ pub(crate) struct ResponsesState { /// This may include output-only metadata items omitted from /// [`Self::messages`] because it is not forwarded to backend /// inference. - pub persisted_messages: Vec, + pub persisted_messages: Vec, /// ID of a previous response to continue from. /// @@ -114,19 +117,19 @@ pub(crate) struct ResponsesState { pub previous_response_id: Option, /// MCP tool listings recovered from the previous response. - pub previous_tools: Vec, + pub previous_tools: Vec, /// Token usage reported by the previous response. - pub previous_usage: Option, + pub previous_usage: Option, /// Parsed request body as received from the client. - pub request_body: serde_json::Value, + pub request_body: Value, /// Whether provider-visible request fields require outbound serialization. pub request_body_rebuild: RequestBodyRebuild, /// The constructed response object for the current iteration. - pub response_object: serde_json::Value, + pub response_object: Value, /// Tool calls from the current inference response only. /// @@ -134,7 +137,7 @@ pub(crate) struct ResponsesState { /// before `stream_events` writes new ones. Without explicit /// clearing, stale tool calls from a previous iteration cause /// duplicate dispatch. - pub tool_calls: Vec, + pub tool_calls: Vec, /// Web search calls from the current inference response only. /// @@ -143,20 +146,20 @@ pub(crate) struct ResponsesState { /// items have a different shape (`action.query` instead of /// `name`/`arguments`) and must not trigger the /// one-function-call-per-round limit. - pub web_search_calls: Vec, + pub web_search_calls: Vec, /// Tool choice setting. Reset to `"auto"` by `openai_agentic_loop` /// after the first iteration; the original value from the /// request only applies to the first inference call. - pub tool_choice: serde_json::Value, + pub tool_choice: Value, /// Processed tool definitions from the request. - pub tools: Vec, + pub tools: Vec, /// Token usage accumulated across all iterations within the /// request. `stream_events` merges per-iteration usage into /// the running total. - pub usage: serde_json::Value, + pub usage: Value, /// Output items accumulated across all agentic loop iterations. /// @@ -164,7 +167,7 @@ pub(crate) struct ResponsesState { /// appended here so the final response contains the complete /// trace. `openai_agentic_loop` writes model items, `mcp_dispatch` /// writes `mcp_call` and `mcp_approval_request` items. - pub accumulated_output: Vec, + pub accumulated_output: Vec, } /// Whether the proxy can preserve the original request bytes. @@ -197,14 +200,14 @@ impl Default for ResponsesState { previous_response_id: None, previous_tools: Vec::new(), previous_usage: None, - request_body: serde_json::Value::Null, + request_body: Value::Null, request_body_rebuild: RequestBodyRebuild::PreserveOriginal, - response_object: serde_json::Value::Null, + response_object: Value::Null, tool_calls: Vec::new(), web_search_calls: Vec::new(), - tool_choice: serde_json::Value::String("auto".to_owned()), + tool_choice: Value::String("auto".to_owned()), tools: Vec::new(), - usage: serde_json::Value::Null, + usage: Value::Null, accumulated_output: Vec::new(), } } @@ -212,13 +215,13 @@ impl Default for ResponsesState { impl ResponsesState { /// Create initial state from a parsed request body. - pub(crate) fn from_request_body(body: serde_json::Value) -> Self { + pub(crate) fn from_request_body(body: Value) -> Self { let messages = normalize_input(&body); let persisted_messages = messages.clone(); let tool_choice = body .get("tool_choice") .cloned() - .unwrap_or_else(|| serde_json::Value::String("auto".to_owned())); + .unwrap_or_else(|| Value::String("auto".to_owned())); let tools = extract_array_field(&body, "tools"); Self { @@ -245,29 +248,29 @@ impl ResponsesState { } /// Borrow the public output owned by [`Self::response_object`]. - pub(crate) fn output_items(&self) -> &[serde_json::Value] { + pub(crate) fn output_items(&self) -> &[Value] { self.response_object .get("output") - .and_then(serde_json::Value::as_array) + .and_then(Value::as_array) .map(Vec::as_slice) .unwrap_or_default() } /// Mutably borrow public output, creating a valid array when absent. - pub(crate) fn output_items_mut(&mut self) -> &mut Vec { + pub(crate) fn output_items_mut(&mut self) -> &mut Vec { if !self.response_object.is_object() { - self.response_object = serde_json::Value::Object(serde_json::Map::new()); + self.response_object = Value::Object(serde_json::Map::new()); } - let serde_json::Value::Object(response) = &mut self.response_object else { + let Value::Object(response) = &mut self.response_object else { unreachable!("response_object was normalized to an object") }; let output = response .entry("output".to_owned()) - .or_insert_with(|| serde_json::Value::Array(Vec::new())); + .or_insert_with(|| Value::Array(Vec::new())); if !output.is_array() { - *output = serde_json::Value::Array(Vec::new()); + *output = Value::Array(Vec::new()); } - let serde_json::Value::Array(items) = output else { + let Value::Array(items) = output else { unreachable!("output was normalized to an array") }; items @@ -277,6 +280,28 @@ impl ResponsesState { pub(crate) fn request_body_requires_rebuild(&self) -> bool { self.request_body_rebuild == RequestBodyRebuild::Required } + + /// Build the final response body from accumulated state. + /// + /// Replaces `response_object["output"]` with the full `accumulated_output` + /// (all rounds), stamps accumulated usage, and serializes back to body bytes. + pub(crate) fn finalize_response_body(&self, body: &mut Option) { + if !self.response_object.is_object() { + return; + } + let mut response = self.response_object.clone(); + if let Some(obj) = response.as_object_mut() { + if !self.accumulated_output.is_empty() { + obj.insert("output".to_owned(), Value::Array(self.accumulated_output.clone())); + } + if !self.usage.is_null() { + obj.insert("usage".to_owned(), self.usage.clone()); + } + } + if let Ok(serialized) = serde_json::to_vec(&response) { + *body = Some(Bytes::from(serialized)); + } + } } /// Normalize the `input` field into a message array. @@ -284,11 +309,11 @@ impl ResponsesState { /// The Responses API `input` can be a string (single user message), /// a single item object, or an array of items. Normalizes all three /// forms to a `Vec`. -fn normalize_input(body: &serde_json::Value) -> Vec { +fn normalize_input(body: &Value) -> Vec { match body.get("input") { - Some(serde_json::Value::Array(arr)) => arr.clone(), - Some(input @ serde_json::Value::Object(_)) => vec![input.clone()], - Some(serde_json::Value::String(s)) => { + Some(Value::Array(arr)) => arr.clone(), + Some(input @ Value::Object(_)) => vec![input.clone()], + Some(Value::String(s)) => { vec![serde_json::json!({ "type": "message", "role": "user", @@ -300,36 +325,26 @@ fn normalize_input(body: &serde_json::Value) -> Vec { } /// Extract a JSON array field by name, defaulting to empty. -fn extract_array_field(body: &serde_json::Value, field: &str) -> Vec { - body.get(field) - .and_then(serde_json::Value::as_array) - .cloned() - .unwrap_or_default() +fn extract_array_field(body: &Value, field: &str) -> Vec { + body.get(field).and_then(Value::as_array).cloned().unwrap_or_default() } /// Extract a string field by name. -fn extract_string(body: &serde_json::Value, field: &str) -> Option { - body.get(field) - .and_then(serde_json::Value::as_str) - .map(ToOwned::to_owned) +fn extract_string(body: &Value, field: &str) -> Option { + body.get(field).and_then(Value::as_str).map(ToOwned::to_owned) } /// Extract an array of strings by name, defaulting to empty. -fn extract_string_array(body: &serde_json::Value, field: &str) -> Vec { +fn extract_string_array(body: &Value, field: &str) -> Vec { body.get(field) - .and_then(serde_json::Value::as_array) - .map(|arr| { - arr.iter() - .filter_map(serde_json::Value::as_str) - .map(ToOwned::to_owned) - .collect() - }) + .and_then(Value::as_array) + .map(|arr| arr.iter().filter_map(Value::as_str).map(ToOwned::to_owned).collect()) .unwrap_or_default() } /// Extract a `u32` field by name, logging when a value is present /// but not representable as `u32`. -fn extract_u32(body: &serde_json::Value, field: &str) -> Option { +fn extract_u32(body: &Value, field: &str) -> Option { let raw = body.get(field)?; let result = raw.as_u64().and_then(|v| u32::try_from(v).ok()); if result.is_none() { @@ -339,8 +354,8 @@ fn extract_u32(body: &serde_json::Value, field: &str) -> Option { } /// Extract a bool field by name, returning a default if absent. -fn extract_bool_or(body: &serde_json::Value, field: &str, default: bool) -> bool { - body.get(field).and_then(serde_json::Value::as_bool).unwrap_or(default) +fn extract_bool_or(body: &Value, field: &str, default: bool) -> bool { + body.get(field).and_then(Value::as_bool).unwrap_or(default) } // -----------------------------------------------------------------------------