diff --git a/crates/agentic-core/src/executor/engine.rs b/crates/agentic-core/src/executor/engine.rs index 13fc691..4dd8fb1 100644 --- a/crates/agentic-core/src/executor/engine.rs +++ b/crates/agentic-core/src/executor/engine.rs @@ -13,10 +13,10 @@ use tokio::sync::mpsc; use tracing::{debug, warn}; use super::gateway::{ - append_gateway_calls_to_new_input, append_output_items_to_input, append_tool_outputs, + LoopDecision, append_gateway_calls_to_new_input, append_output_items_to_input, append_tool_outputs, classify_round, execute_and_emit_output_calls, has_client_owned_calls, public_output_items, }; -use crate::executor::error::{ExecutorError, ExecutorResult}; +use crate::executor::error::ExecutorResult; use crate::executor::inference::DONE_MARKER; use crate::executor::persist::persist_if_needed; use crate::executor::rehydrate::rehydrate_conversation; @@ -24,7 +24,7 @@ use crate::executor::request::{ExecutionContext, RequestContext}; use crate::executor::upstream::{fetch_blocking_payload, fetch_stream_payload}; use crate::tool::ToolRegistry; use crate::types::io::{ResponseUsage, ToolChoice}; -use crate::types::request_response::{RequestPayload, ResponsePayload}; +use crate::types::request_response::{IncompleteDetails, RequestPayload, ResponsePayload}; use crate::utils::common::serialize_to_string; pub use crate::executor::inference::BoxStream; @@ -114,54 +114,83 @@ async fn run_until_gateway_tools_complete( let mut combined_output: Vec = Vec::new(); let mut combined_usage: Option = None; - for _ in 0..MAX_GATEWAY_TOOL_ROUNDS { + for round in 0..MAX_GATEWAY_TOOL_ROUNDS { let mut payload: ResponsePayload = if stream_upstream { fetch_stream_payload(&ctx, exec_ctx, auth, ®istry, stream_events).await? } else { fetch_blocking_payload(&ctx, exec_ctx, auth).await? }; registry.restore_final_payload_output(&mut payload.output); - accumulate_usage(&mut combined_usage, payload.usage); + accumulate_usage(&mut combined_usage, payload.usage.take()); let current_output = std::mem::take(&mut payload.output); - let has_client_owned_calls = has_client_owned_calls(¤t_output, ®istry); + let has_client_owned = has_client_owned_calls(¤t_output, ®istry); let gateway_results = execute_and_emit_output_calls(¤t_output, ®istry, combined_output.len(), stream_events).await?; let public_output = public_output_items(¤t_output, ®istry, &gateway_results); + combined_output.extend(public_output); - if has_client_owned_calls { - combined_output.extend(public_output); - append_gateway_calls_to_new_input(&mut ctx, ¤t_output, ®istry); - append_tool_outputs( - &mut ctx, - gateway_results.into_iter().map(|result| result.input_item).collect(), - ); - payload.output = combined_output; - payload.usage = combined_usage; - ctx.inject_ids(&mut payload); - return Ok((payload, ctx)); - } - - if gateway_results.is_empty() { - combined_output.extend(public_output); - payload.output = combined_output; - payload.usage = combined_usage; - ctx.inject_ids(&mut payload); - return Ok((payload, ctx)); + match classify_round(has_client_owned, &gateway_results, round, MAX_GATEWAY_TOOL_ROUNDS) { + // Client-owned calls (plain function or Codex namespace tools) are + // handed back to the caller. Gateway calls in the same turn are + // still recorded so the returned conversation is complete. + LoopDecision::RequiresClientAction => { + append_gateway_calls_to_new_input(&mut ctx, ¤t_output, ®istry); + append_tool_outputs( + &mut ctx, + gateway_results.into_iter().map(|result| result.input_item).collect(), + ); + finalize_loop(&mut payload, combined_output, combined_usage, &ctx); + return Ok((payload, ctx)); + } + // No gateway work remains — this turn is the final response. + LoopDecision::Done => { + finalize_loop(&mut payload, combined_output, combined_usage, &ctx); + return Ok((payload, ctx)); + } + // Budget exhausted while the model was still requesting gateway + // tools: surface the accumulated work as a partial + // `status: "incomplete"` response instead of failing the request. + // The final round's gateway calls and outputs are recorded so a + // continuation is not fed a dangling tool call. + LoopDecision::Incomplete(reason) => { + append_gateway_calls_to_new_input(&mut ctx, ¤t_output, ®istry); + append_tool_outputs( + &mut ctx, + gateway_results.into_iter().map(|result| result.input_item).collect(), + ); + finalize_loop(&mut payload, combined_output, combined_usage, &ctx); + "incomplete".clone_into(&mut payload.status); + payload.incomplete_details = Some(IncompleteDetails { reason: Some(reason) }); + return Ok((payload, ctx)); + } + // Gateway tools ran and rounds remain; feed outputs back and loop. + LoopDecision::Continue => { + ctx.enriched_request.tool_choice = Some(ToolChoice::Auto); + append_output_items_to_input(&mut ctx.enriched_request.input, ¤t_output); + append_gateway_calls_to_new_input(&mut ctx, ¤t_output, ®istry); + append_tool_outputs( + &mut ctx, + gateway_results.into_iter().map(|result| result.input_item).collect(), + ); + } } - - combined_output.extend(public_output); - ctx.enriched_request.tool_choice = Some(ToolChoice::Auto); - append_output_items_to_input(&mut ctx.enriched_request.input, ¤t_output); - append_gateway_calls_to_new_input(&mut ctx, ¤t_output, ®istry); - append_tool_outputs( - &mut ctx, - gateway_results.into_iter().map(|result| result.input_item).collect(), - ); } - Err(ExecutorError::InvalidRequest(format!( - "gateway tool execution exceeded {MAX_GATEWAY_TOOL_ROUNDS} rounds" - ))) + unreachable!("the final round returns Done, RequiresClientAction, or Incomplete"); +} + +/// Move accumulated output/usage onto the terminating round's payload and +/// inject the response/conversation IDs. The payload's `model`/`created_at`/ +/// `status` from the latest inference turn are preserved. +fn finalize_loop( + payload: &mut ResponsePayload, + combined_output: Vec, + combined_usage: Option, + ctx: &RequestContext, +) { + payload.output = combined_output; + payload.usage = combined_usage; + ctx.inject_ids(payload); } async fn run_blocking( diff --git a/crates/agentic-core/src/executor/gateway.rs b/crates/agentic-core/src/executor/gateway.rs index 61e5149..b13b9bd 100644 --- a/crates/agentic-core/src/executor/gateway.rs +++ b/crates/agentic-core/src/executor/gateway.rs @@ -1,16 +1,81 @@ +use std::time::Duration; + use futures::StreamExt; use futures::stream as futures_stream; use tokio::sync::mpsc; use crate::executor::error::{ExecutorError, ExecutorResult}; use crate::executor::request::RequestContext; -use crate::tool::{ToolError, ToolOutput, ToolRegistry, ToolType}; +use crate::tool::{GatewayDispatchResult, ToolError, ToolOutput, ToolRegistry, ToolType}; use crate::types::io::output::{FunctionToolCall, WebSearchCallStatus}; use crate::types::io::{InputItem, OutputItem, ResponsesInput}; use crate::utils::common::serialize_to_string; -const MAX_GATEWAY_TOOL_CALLS_PER_ROUND: usize = 8; -const MAX_GATEWAY_TOOL_CONCURRENCY: usize = 4; +/// Max gateway tool calls executing at once within a round. A sliding window: +/// as one call finishes, the next is admitted, so a round with N calls never +/// runs more than this many concurrently but still drains all N. Bounds +/// outbound fan-out without a hard per-round count cap. +/// +/// The call count is bounded upstream by the model's output size — there is no +/// unbounded in-memory materialisation from the model emitting arbitrarily many +/// tool calls. The window + per-call timeout bound outbound HTTP and latency. +const MAX_CONCURRENT_GATEWAY_CALLS: usize = 5; + +/// Per-call wall-clock budget. A tool exceeding this yields an error output fed +/// back to the model (never a whole-request failure). `Duration::ZERO` disables +/// the timeout — for providers that manage their own. +/// +/// Note: this bounds a single call, not the whole request. Worst-case request +/// latency scales with rounds and fan-out; an outer request-level deadline +/// would be the place to cap total time end-to-end. +const GATEWAY_TOOL_TIMEOUT: Duration = Duration::from_secs(60); + +/// Outcome of inspecting one inference turn's output, deciding whether the +/// gateway tool loop should run another round, stop, or surface a partial result. +/// +/// `#[non_exhaustive]` so downstream variants can be added without breaking +/// existing match arms. +#[derive(Debug)] +#[non_exhaustive] +pub(super) enum LoopDecision { + /// Gateway tools were dispatched this round; loop again with their outputs + /// appended to the conversation. + Continue, + /// No gateway work remains — the turn is final and the loop terminates. + Done, + /// One or more calls are client-owned (plain `function` or Codex + /// `namespace` tools); hand the turn back to the caller to execute. + RequiresClientAction, + /// The round cap was hit before the model stopped requesting tools. The + /// response is returned with `status: "incomplete"` rather than as an error. + Incomplete(String), +} + +/// Classify one turn's output into a [`LoopDecision`]. +/// +/// Order matters: client-owned calls take precedence (they must be handed back +/// even when gateway calls are also present in the same turn), then a +/// no-gateway-work turn is `Done`. Otherwise gateway tools ran — the loop would +/// continue, unless this was the last permitted round, in which case the budget +/// is exhausted and the turn is `Incomplete`. +/// +/// `round` is zero-based; `max_rounds` is the total budget. +pub(super) fn classify_round( + has_client_owned_calls: bool, + gateway_results: &[GatewayCallResult], + round: usize, + max_rounds: usize, +) -> LoopDecision { + if has_client_owned_calls { + LoopDecision::RequiresClientAction + } else if gateway_results.is_empty() { + LoopDecision::Done + } else if round + 1 >= max_rounds { + LoopDecision::Incomplete(format!("gateway tool execution exceeded {max_rounds} rounds")) + } else { + LoopDecision::Continue + } +} #[derive(Clone)] pub(super) struct GatewayCallResult { @@ -55,12 +120,52 @@ fn execution_error_output(call: &FunctionToolCall, message: &str) -> ExecutorRes } async fn execute_gateway_call(call: FunctionToolCall, registry: &ToolRegistry) -> ExecutorResult { - let Some(dispatch) = registry.dispatch(&call).await else { + execute_gateway_call_with_timeout(call, registry, GATEWAY_TOOL_TIMEOUT).await +} + +async fn execute_gateway_call_with_timeout( + call: FunctionToolCall, + registry: &ToolRegistry, + timeout: Duration, +) -> ExecutorResult { + // Resolve the tool type up front so a timeout (which yields no dispatch + // result) can still shape the correct public output. + let Some(tool_type) = registry.lookup(&call.name).map(|entry| entry.tool_type) else { return Err(ExecutorError::InvalidRequest(format!( "gateway tool '{}' was not dispatchable", call.name ))); }; + + // Per-call timeout: a hung tool becomes an error output fed back to the + // model, never a whole-request failure. `Duration::ZERO` opts out. + let dispatched = if timeout.is_zero() { + registry.dispatch(&call).await + } else { + match tokio::time::timeout(timeout, registry.dispatch(&call)).await { + Ok(dispatched) => dispatched, + Err(_elapsed) => Some(GatewayDispatchResult { + tool_type, + output: Err(ToolError::Execution(format!( + "gateway tool '{}' timed out after {timeout:?}", + call.name + ))), + }), + } + }; + + // An entry exists (the call was filtered to gateway-owned) but carries no + // handler — this server was built without that tool's executor. Treat it + // like the timeout path: surface an error output fed back to the model + // rather than failing the whole request, keeping the "never a + // whole-request failure" contract total. + let dispatch = dispatched.unwrap_or_else(|| GatewayDispatchResult { + tool_type, + output: Err(ToolError::Execution(format!( + "gateway tool '{}' has no registered handler", + call.name + ))), + }); let (output, status) = match dispatch.output { Ok(output) => (output, WebSearchCallStatus::Completed), Err(ToolError::Execution(message) | ToolError::Config(message)) => { @@ -97,20 +202,18 @@ pub(super) async fn execute_output_calls( ) -> ExecutorResult> { let calls = function_calls(output_items); let gateway_calls = registry.gateway_owned(&calls); - if gateway_calls.len() > MAX_GATEWAY_TOOL_CALLS_PER_ROUND { - return Err(ExecutorError::InvalidRequest(format!( - "gateway tool call limit exceeded: got {}, max {MAX_GATEWAY_TOOL_CALLS_PER_ROUND} per round", - gateway_calls.len() - ))); - } + // Execute all gateway calls concurrently with a sliding window of + // `MAX_CONCURRENT_GATEWAY_CALLS`: `buffered` admits the next call as soon as + // one finishes, so arbitrary fan-out drains safely without a hard count cap. + // Each call is individually timeout-bounded in `execute_gateway_call`. futures_stream::iter( gateway_calls .into_iter() .cloned() .map(|call| execute_gateway_call(call, registry)), ) - .buffered(MAX_GATEWAY_TOOL_CONCURRENCY) + .buffered(MAX_CONCURRENT_GATEWAY_CALLS) .collect::>() .await .into_iter() @@ -298,3 +401,192 @@ pub(super) fn append_gateway_calls_to_new_input( is_gateway_owned_call(call, registry).then(|| InputItem::FunctionCall(call.clone())) })); } + +#[cfg(test)] +mod tests { + use super::{GatewayCallResult, LoopDecision, classify_round}; + use crate::types::io::InputItem; + use crate::types::io::output::FunctionToolCall; + + const MAX: usize = 10; + + fn dummy_result() -> GatewayCallResult { + let call = FunctionToolCall { + id: "id".to_owned(), + call_id: "call".to_owned(), + name: "web_search".to_owned(), + arguments: "{}".to_owned(), + status: crate::types::event::MessageStatus::Completed, + namespace: None, + }; + GatewayCallResult { + call, + input_item: InputItem::FunctionCallOutput( + crate::tool::ToolOutput { + call_id: "call".to_owned(), + output: "{}".to_owned(), + } + .into(), + ), + public_output: None, + } + } + + #[test] + fn client_owned_calls_take_precedence_over_gateway_results() { + // Even with gateway results present in the same turn, a client-owned call + // must hand control back to the caller. + let decision = classify_round(true, &[dummy_result()], 0, MAX); + assert!(matches!(decision, LoopDecision::RequiresClientAction)); + } + + #[test] + fn no_gateway_work_is_done() { + let decision = classify_round(false, &[], 0, MAX); + assert!(matches!(decision, LoopDecision::Done)); + } + + #[test] + fn gateway_results_with_budget_remaining_continue() { + let decision = classify_round(false, &[dummy_result()], 0, MAX); + assert!(matches!(decision, LoopDecision::Continue)); + } + + #[test] + fn gateway_results_on_final_round_are_incomplete() { + // round is zero-based: round 9 is the 10th and last permitted round. + let decision = classify_round(false, &[dummy_result()], MAX - 1, MAX); + match decision { + LoopDecision::Incomplete(reason) => assert!(reason.contains("exceeded 10 rounds")), + other => panic!("expected Incomplete, got {other:?}"), + } + } + + #[test] + fn incomplete_only_fires_when_gateway_work_remains() { + // On the final round with no gateway work, the turn is still Done — the + // cap only matters when the model is still requesting tools. + let decision = classify_round(false, &[], MAX - 1, MAX); + assert!(matches!(decision, LoopDecision::Done)); + } + + // --- Per-call timeout --------------------------------------------------- + + use std::pin::Pin; + use std::sync::Arc; + + use serde_json::Value; + + use super::execute_gateway_call_with_timeout; + use crate::tool::{GatewayExecutor, ToolError, ToolHandler, ToolOutput, ToolRegistry, ToolType}; + use crate::types::io::OutputItem; + use crate::types::io::tools::FunctionTool; + use crate::types::tools::ResponsesTool; + + /// A gateway executor that sleeps ~50ms — comfortably longer than the tiny + /// timeout the test injects, forcing the timeout path without a paused clock. + struct SlowExecutor; + + impl ToolHandler for SlowExecutor { + fn tool_type(&self) -> ToolType { + ToolType::WebSearch + } + fn validate(&self, _param: &Value) -> Result<(), ToolError> { + Ok(()) + } + fn normalize(&self, _param: &Value) -> Vec { + Vec::new() + } + } + + impl GatewayExecutor for SlowExecutor { + fn execute( + &self, + call_id: &str, + _tool_name: &str, + _arguments: &str, + _config: &Value, + ) -> Pin> + Send + '_>> { + let call_id = call_id.to_owned(); + Box::pin(async move { + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + Ok(ToolOutput { + call_id, + output: "unreachable".to_owned(), + }) + }) + } + } + + fn web_search_call(call_id: &str) -> FunctionToolCall { + FunctionToolCall { + id: format!("fc_{call_id}"), + call_id: call_id.to_owned(), + name: "web_search".to_owned(), + arguments: "{}".to_owned(), + status: crate::types::event::MessageStatus::Completed, + namespace: None, + } + } + + #[tokio::test] + async fn hung_gateway_call_times_out_into_error_output() { + let web_search: ResponsesTool = + serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).expect("web_search tool param"); + let registry = ToolRegistry::build_with_handlers(&[web_search], |tool_type| match tool_type { + ToolType::WebSearch => Some(Arc::new(SlowExecutor) as Arc), + _ => None, + }) + .expect("registry builds"); + + // 1ms budget vs a 50ms tool → the timeout fires. Must return (not hang): + // the stuck call becomes an error output the loop can feed back. + let result = execute_gateway_call_with_timeout( + web_search_call("call_hang"), + ®istry, + std::time::Duration::from_millis(1), + ) + .await + .expect("timeout is isolated as an error output, not a dispatch failure"); + + assert_eq!(result.call.call_id, "call_hang"); + // A failed web_search still yields a public web_search_call item. + assert!(matches!(result.public_output, Some(OutputItem::WebSearchCall(_)))); + // The fed-back tool output is an error JSON mentioning the timeout. + let InputItem::FunctionCallOutput(msg) = &result.input_item else { + panic!("expected a function_call_output"); + }; + let body = serde_json::to_string(msg).expect("serialize output"); + assert!( + body.contains("timed out"), + "error output should mention the timeout: {body}" + ); + } + + #[tokio::test] + async fn gateway_call_without_registered_handler_becomes_error_output() { + // Declare web_search but build the registry with NO executor for it — + // the entry exists and is gateway-owned, so the call is not filtered + // out, but `dispatch` yields `None`. This must surface an error output, + // not fail the whole request. + let web_search: ResponsesTool = + serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).expect("web_search tool param"); + let registry = ToolRegistry::build_with_handlers(&[web_search], |_tool_type| None).expect("registry builds"); + + let result = + execute_gateway_call_with_timeout(web_search_call("call_no_handler"), ®istry, std::time::Duration::ZERO) + .await + .expect("a missing handler is isolated as an error output, not a dispatch failure"); + + assert_eq!(result.call.call_id, "call_no_handler"); + assert!(matches!(result.public_output, Some(OutputItem::WebSearchCall(_)))); + let InputItem::FunctionCallOutput(msg) = &result.input_item else { + panic!("expected a function_call_output"); + }; + let body = serde_json::to_string(msg).expect("serialize output"); + assert!( + body.contains("no registered handler"), + "error output should mention the missing handler: {body}" + ); + } +} diff --git a/crates/agentic-core/tests/cassettes/tool_calls/multi_turn/gateway_web_search_then_codex_namespace.yaml b/crates/agentic-core/tests/cassettes/tool_calls/multi_turn/gateway_web_search_then_codex_namespace.yaml new file mode 100644 index 0000000..e25820b --- /dev/null +++ b/crates/agentic-core/tests/cassettes/tool_calls/multi_turn/gateway_web_search_then_codex_namespace.yaml @@ -0,0 +1,65 @@ +# Composed multi-turn cassette for the gateway tool loop integration test. +# +# Turn 0 body is a recorded-shape gateway `web_search` function call (the tool +# the gateway executes server-side). Turn 1 body is the flat, model-visible +# Codex namespace call `agentic_ns__mcp__agentic_fixture__add_numbers` (the same +# flattened form as cassettes/codex/tools/direct_vllm_flat_namespace_tool.json), +# which the loop restores to `{namespace, name}` and hands back to the client. +# +# Driven by dispatch_loop_cassette_test.rs:: +# openai_gateway_web_search_then_codex_namespace_across_turns +turns: + - filename: t0 + request: + body: + input: search then add + method: POST + path: /v1/responses + response: + status_code: 200 + body: + id: resp_ws_0 + object: response + created_at: 0 + model: test-model + status: completed + output: + - id: fc_ws + type: function_call + call_id: call_ws + name: web_search + arguments: '{"query":"rust async"}' + status: completed + usage: null + incomplete_details: null + error: null + previous_response_id: null + conversation_id: null + instructions: null + - filename: t1 + request: + body: + input: continue + method: POST + path: /v1/responses + response: + status_code: 200 + body: + id: resp_ns_1 + object: response + created_at: 0 + model: test-model + status: completed + output: + - id: fc_ns + type: function_call + call_id: call_ns + name: agentic_ns__mcp__agentic_fixture__add_numbers + arguments: '{"numbers":[8,0]}' + status: completed + usage: null + incomplete_details: null + error: null + previous_response_id: resp_ws_0 + conversation_id: null + instructions: null diff --git a/crates/agentic-core/tests/dispatch_loop_cassette_test.rs b/crates/agentic-core/tests/dispatch_loop_cassette_test.rs new file mode 100644 index 0000000..4a99ffb --- /dev/null +++ b/crates/agentic-core/tests/dispatch_loop_cassette_test.rs @@ -0,0 +1,337 @@ +//! Cassette-driven integration tests for the reworked gateway tool loop +//! (`run_until_gateway_tools_complete` via `ExecuteRequest::run`). +//! +//! These drive real recorded `OpenAI` Responses wire bodies through the loop to +//! confirm the `LoopDecision` classification behaves end-to-end: +//! * client-owned (function) tool calls terminate the loop in one model call +//! and are handed back to the caller (`RequiresClientAction`); +//! * parallel function calls in a single turn are preserved on output; +//! * a gateway-owned call drives another round and comes back `completed`. +//! +//! The synthetic-payload edge cases (usage accumulation, fanout cap, error +//! feedback, the round-cap `incomplete` path) live in `web_search_tool_test.rs`; +//! this file focuses on coverage against recorded `OpenAI` wire bodies. + +use std::sync::Arc; + +use agentic_core::executor::{ConversationHandler, ExecuteRequest, ExecutionContext, ResponseHandler}; +use agentic_core::storage::{ConversationStore, ResponseStore}; +use agentic_core::tool::WebSearchHandler; +use agentic_core::types::io::{OutputItem, ResponsesInput}; +use agentic_core::types::request_response::RequestPayload; +use agentic_core::types::tools::ResponsesTool; +use either::Either; + +mod support; + +const MULTI_TURN_DIR: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/cassettes/tool_calls/multi_turn"); + +/// Load turn N (0-based) response body from a multi-turn cassette as a mock +/// JSON response for the LLM `MockServer`. +/// +/// Uses a permissive local parser (input typed as `Value`) rather than the +/// shared strict `Cassette` — these `OpenAI` cassettes carry array-valued `input` +/// on later turns, which the shared string-typed loader rejects. We only need +/// the response body, so the request shape is irrelevant here. +fn cassette_turn(filename: &str, turn: usize) -> support::MockResponse { + use serde::Deserialize; + + #[derive(Deserialize)] + struct Doc { + turns: Vec, + } + #[derive(Deserialize)] + struct DocTurn { + response: DocResponse, + } + #[derive(Deserialize)] + struct DocResponse { + #[serde(default)] + body: Option, + #[serde(default)] + sse: Option>, + } + + let path = format!("{MULTI_TURN_DIR}/{filename}"); + let text = std::fs::read_to_string(&path).unwrap_or_else(|e| panic!("read {path}: {e}")); + let doc: Doc = serde_yaml::from_str(&text).unwrap_or_else(|e| panic!("parse {path}: {e}")); + let resp = doc + .turns + .into_iter() + .nth(turn) + .unwrap_or_else(|| panic!("cassette {filename} missing turn {turn}")) + .response; + + if let Some(body) = resp.body { + support::MockResponse::Json(serde_json::to_string(&body).expect("cassette body serializes")) + } else if let Some(sse) = resp.sse { + support::MockResponse::Sse(sse.join("")) + } else { + panic!("cassette {filename} turn {turn} has neither body nor sse"); + } +} + +async fn build_exec_ctx(llm_url: &str) -> Arc { + let pool = support::setup_pool().await; + let conv_handler = ConversationHandler::new(ConversationStore::new(Arc::clone(&pool))); + let resp_handler = ResponseHandler::new(ResponseStore::new(pool)); + let client = Arc::new(reqwest::Client::new()); + Arc::new(ExecutionContext::new( + conv_handler, + resp_handler, + client, + llm_url.to_owned(), + )) +} + +/// Same as `build_exec_ctx` but registers a `WebSearchHandler` gateway executor +/// backed by the given You.com mock URL. +async fn build_exec_ctx_with_web_search(llm_url: &str, you_url: &str) -> Arc { + let pool = support::setup_pool().await; + let conv_handler = ConversationHandler::new(ConversationStore::new(Arc::clone(&pool))); + let resp_handler = ResponseHandler::new(ResponseStore::new(pool)); + let client = Arc::new(reqwest::Client::new()); + Arc::new( + ExecutionContext::new(conv_handler, resp_handler, Arc::clone(&client), llm_url.to_owned()) + .with_gateway_executor(Arc::new(WebSearchHandler::with_api_key( + client, + "secret-you-key".to_owned(), + you_url, + ))), + ) +} + +fn request(text: &str, tools: Option>) -> RequestPayload { + RequestPayload { + model: "test-model".to_owned(), + input: ResponsesInput::Text(text.to_owned()), + instructions: None, + previous_response_id: None, + conversation_id: None, + tools, + tool_choice: None, + stream: false, + store: true, + include: None, + temperature: None, + top_p: None, + max_output_tokens: Some(1024), + truncation: None, + metadata: None, + } +} + +fn function_call_names(output: &[OutputItem]) -> Vec<&str> { + output + .iter() + .filter_map(|item| match item { + OutputItem::FunctionCall(fc) => Some(fc.name.as_str()), + _ => None, + }) + .collect() +} + +/// `OpenAI` cassette, turn 1 emits a single client-owned `get_job_status` +/// function call. With no gateway executor registered, the loop must classify +/// this as `RequiresClientAction`: exactly one model call, the call handed back +/// on `output`, `status: "completed"`. +#[tokio::test] +async fn openai_single_client_owned_call_terminates_in_one_round() { + let llm = support::MockServer::start_deque(vec![cassette_turn("openai_responses_tool_calls_3turn.yaml", 0)]).await; + let exec_ctx = build_exec_ctx(llm.url()).await; + + let result = ExecuteRequest::new(request("check job status", None), exec_ctx) + .run() + .await + .expect("execute should succeed"); + let Either::Left(response) = result else { + panic!("non-streaming request should return a payload"); + }; + + // Client-owned function call: no gateway execution, exactly one LLM call. + assert_eq!( + llm.request_bodies().await.len(), + 1, + "client-owned tools take one model call" + ); + assert_eq!(response.status, "completed"); + assert_eq!( + function_call_names(&response.output), + vec!["get_job_status"], + "the client-owned call must be handed back on output" + ); +} + +/// `OpenAI` cassette, turn 1 emits two parallel client-owned function calls +/// (`get_job_status` + `web_search`). With no gateway executor, both are handed +/// back unexecuted in a single round. +#[tokio::test] +async fn openai_parallel_client_owned_calls_preserved_on_output() { + let llm = + support::MockServer::start_deque(vec![cassette_turn("openai_responses_tool_calls_parallel.yaml", 0)]).await; + let exec_ctx = build_exec_ctx(llm.url()).await; + + let result = ExecuteRequest::new(request("check status and search", None), exec_ctx) + .run() + .await + .expect("execute should succeed"); + let Either::Left(response) = result else { + panic!("non-streaming request should return a payload"); + }; + + assert_eq!(llm.request_bodies().await.len(), 1); + let names = function_call_names(&response.output); + assert_eq!(names.len(), 2, "both parallel calls preserved, got {names:?}"); + assert!( + names.contains(&"get_job_status"), + "expected get_job_status in {names:?}" + ); + assert!(names.contains(&"web_search"), "expected web_search in {names:?}"); +} + +/// When `web_search` is declared as a gateway tool AND a `WebSearchHandler` is +/// registered, the same first-turn call is gateway-owned: the loop executes it, +/// feeds the output back, and the recorded second turn returns text. The +/// `get_job_status` call in the parallel cassette stays client-owned, so the +/// turn is `RequiresClientAction` and terminates after one model call — the +/// gateway `web_search` output is still recorded alongside it. +#[tokio::test] +async fn openai_mixed_gateway_and_client_owned_hands_back_after_gateway_exec() { + let (you_url, mut captured_you, _you_handle) = spawn_mock_you().await; + let llm = + support::MockServer::start_deque(vec![cassette_turn("openai_responses_tool_calls_parallel.yaml", 0)]).await; + let exec_ctx = build_exec_ctx_with_web_search(llm.url(), &you_url).await; + let web_search: ResponsesTool = serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).unwrap(); + + let result = ExecuteRequest::new(request("status and search", Some(vec![web_search])), exec_ctx) + .run() + .await + .expect("execute should succeed"); + let Either::Left(response) = result else { + panic!("non-streaming request should return a payload"); + }; + + // web_search is gateway-owned → executed against You.com mock. + let search = captured_you.recv().await.expect("web_search should hit You.com"); + assert!(search.body.get("query").is_some(), "web_search executed with a query"); + + // get_job_status is still client-owned → RequiresClientAction, one model call. + assert_eq!(llm.request_bodies().await.len(), 1); + assert!( + function_call_names(&response.output).contains(&"get_job_status"), + "client-owned call handed back alongside the executed gateway call" + ); +} + +/// A gateway `web_search` turn followed by a Codex `namespace` turn, driven +/// through the loop as one conversation (per @maralbahari's review ask on #83). +/// +/// Round 0: the model emits a gateway-owned `web_search` call → the loop +/// executes it against the You.com mock and continues. Round 1: the model emits +/// the flat, model-visible namespace call +/// `agentic_ns__mcp__agentic_fixture__add_numbers` → the loop restores it to +/// `{namespace: "mcp__agentic_fixture", name: "add_numbers"}`, classifies it as +/// client-owned, and hands the turn back (`RequiresClientAction`). Two model +/// calls total; the namespace call is returned restored, never flattened. +#[tokio::test] +async fn openai_gateway_web_search_then_codex_namespace_across_turns() { + let (you_url, mut captured_you, _you_handle) = spawn_mock_you().await; + let llm = support::MockServer::start_deque(vec![ + cassette_turn("gateway_web_search_then_codex_namespace.yaml", 0), + cassette_turn("gateway_web_search_then_codex_namespace.yaml", 1), + ]) + .await; + let exec_ctx = build_exec_ctx_with_web_search(llm.url(), &you_url).await; + + let web_search: ResponsesTool = serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).unwrap(); + let codex_namespace: ResponsesTool = serde_json::from_value(serde_json::json!({ + "type": "namespace", + "name": "mcp__agentic_fixture", + "tools": [{"type": "function", "name": "add_numbers", "parameters": {"type": "object"}}] + })) + .unwrap(); + + let result = ExecuteRequest::new( + request("search then add", Some(vec![web_search, codex_namespace])), + exec_ctx, + ) + .run() + .await + .expect("execute should succeed"); + let Either::Left(response) = result else { + panic!("non-streaming request should return a payload"); + }; + + // Round 0 executed the gateway web_search against the You.com mock, then + // round 1 emitted the namespace call — two model calls in one conversation. + let search = captured_you.recv().await.expect("web_search should hit You.com"); + assert!(search.body.get("query").is_some(), "web_search executed with a query"); + assert_eq!( + llm.request_bodies().await.len(), + 2, + "gateway round + client-action round" + ); + + // The namespace call is handed back restored to {namespace, name}, never + // the flat model-visible name. + let output = serde_json::to_value(&response.output).unwrap(); + let items = output.as_array().unwrap(); + let ns_call = items + .iter() + .find(|it| it["type"] == "function_call" && it["call_id"] == "call_ns") + .expect("namespace call handed back to the client"); + assert_eq!(ns_call["name"], "add_numbers", "flat name restored to member name"); + assert_eq!(ns_call["namespace"], "mcp__agentic_fixture", "namespace restored"); + assert!( + !items + .iter() + .any(|it| it["name"] == "agentic_ns__mcp__agentic_fixture__add_numbers"), + "flat namespaced name must not leak to the client" + ); + // The gateway web_search surfaced as a public web_search_call, not a raw fn. + assert!( + items.iter().any(|it| it["type"] == "web_search_call"), + "web_search resolved into a public web_search_call: {output:#}" + ); +} + +// --- You.com mock (mirrors web_search_tool_test.rs) ------------------------- + +struct CapturedSearch { + body: serde_json::Value, +} + +async fn spawn_mock_you() -> ( + String, + tokio::sync::mpsc::Receiver, + tokio::task::JoinHandle<()>, +) { + use axum::extract::State; + use axum::routing::post; + use axum::{Json, Router}; + use tokio::net::TcpListener; + use tokio::sync::mpsc; + + let (tx, rx) = mpsc::channel(16); + let app = Router::new() + .route( + "/v1/search", + post( + move |State(tx): State>, Json(body): Json| async move { + tx.send(CapturedSearch { body }).await.unwrap(); + ( + axum::http::StatusCode::OK, + Json(serde_json::json!({ + "results": {"web": [{"url": "https://example.com", "title": "R"}], "news": []}, + "metadata": {"query": "q"} + })), + ) + }, + ), + ) + .with_state(tx); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let handle = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + (format!("http://{addr}"), rx, handle) +} diff --git a/crates/agentic-core/tests/web_search_tool_test.rs b/crates/agentic-core/tests/web_search_tool_test.rs index 4bec51b..92813f5 100644 --- a/crates/agentic-core/tests/web_search_tool_test.rs +++ b/crates/agentic-core/tests/web_search_tool_test.rs @@ -1089,7 +1089,7 @@ async fn execute_feeds_web_search_execution_errors_back_to_model() { } #[tokio::test] -async fn execute_errors_after_max_gateway_tool_rounds() { +async fn execute_returns_incomplete_after_max_gateway_tool_rounds() { let (you_url, mut captured_you, _you_handle) = spawn_mock_you().await; let llm_responses = std::iter::repeat_with(web_search_function_call_response) .take(10) @@ -1115,11 +1115,19 @@ async fn execute_errors_after_max_gateway_tool_rounds() { metadata: None, }; - let result = ExecuteRequest::new(payload, exec_ctx).run().await; - assert!( - result - .err() - .is_some_and(|err| err.to_string().contains("gateway tool execution exceeded 10 rounds")) + // Budget exhausted while the model keeps requesting tools → the response is + // surfaced as status: "incomplete" (Responses API semantics), not an error. + let result = ExecuteRequest::new(payload, exec_ctx) + .run() + .await + .expect("hitting the round cap returns an incomplete response, not an error"); + let Either::Left(response) = result else { + panic!("non-streaming request should return a payload"); + }; + assert_eq!(response.status, "incomplete"); + assert_eq!( + response.incomplete_details.and_then(|d| d.reason).as_deref(), + Some("gateway tool execution exceeded 10 rounds") ); for _ in 0..10 { captured_you.recv().await.expect("mock You.com should receive request"); @@ -1182,9 +1190,16 @@ async fn execute_feeds_invalid_web_search_arguments_back_to_model() { } #[tokio::test] -async fn execute_errors_when_gateway_tool_call_fanout_exceeds_limit() { +async fn execute_runs_large_gateway_fanout_without_hard_cap() { + // A single turn requesting 9 gateway calls (more than the concurrency + // window of 5) must all execute — the sliding window drains them safely, + // with no hard per-round count cap that fails the request. let (you_url, mut captured_you, _you_handle) = spawn_mock_you().await; - let llm = support::MockServer::start_deque(vec![many_web_search_function_call_response(9)]).await; + let llm = support::MockServer::start_deque(vec![ + many_web_search_function_call_response(9), + support::text_response("done with all searches"), + ]) + .await; let exec_ctx = build_exec_ctx(llm.url(), you_url).await; let web_search: ResponsesTool = serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).unwrap(); let payload = RequestPayload { @@ -1205,17 +1220,24 @@ async fn execute_errors_when_gateway_tool_call_fanout_exceeds_limit() { metadata: None, }; - let result = ExecuteRequest::new(payload, exec_ctx).run().await; - assert!( - result - .err() - .is_some_and(|err| err.to_string().contains("gateway tool call limit exceeded")) - ); - assert!( - captured_you.try_recv().is_err(), - "fanout limit should fail before paid searches" + let result = ExecuteRequest::new(payload, exec_ctx) + .run() + .await + .expect("large fan-out executes; no hard cap error"); + let Either::Left(response) = result else { + panic!("non-streaming request should return a payload"); + }; + assert_eq!(response.status, "completed"); + + let mut searches = 0; + while captured_you.try_recv().is_ok() { + searches += 1; + } + assert_eq!( + searches, 9, + "all 9 gateway calls execute despite the concurrency window of 5" ); - assert_eq!(llm.request_bodies().await.len(), 1); + assert_eq!(llm.request_bodies().await.len(), 2); } #[tokio::test] @@ -1275,3 +1297,182 @@ async fn stream_error_events_escape_error_messages() { let error = parsed["error"]["message"].as_str().unwrap(); assert!(error.contains(r#"bad \"quoted\" upstream response"#), "{error}"); } + +fn web_search_function_call_response_with_id(call_id: &str) -> support::MockResponse { + support::MockResponse::Json( + serde_json::json!({ + "id": format!("resp_{call_id}"), + "object": "response", + "created_at": 0, + "model": "test-model", + "status": "completed", + "output": [{ + "id": format!("fc_{call_id}"), + "type": "function_call", + "call_id": call_id, + "name": "web_search", + "arguments": "{\"query\":\"rust async\",\"count\":2}", + "status": "completed" + }], + "usage": null, + "incomplete_details": null, + "error": null, + "previous_response_id": null, + "conversation_id": null, + "instructions": null + }) + .to_string(), + ) +} + +#[tokio::test] +async fn incomplete_turn_persists_a_consistent_conversation_for_continuation() { + // 10 web_search rounds -> incomplete, then one text turn for the continuation. + // Each round emits a UNIQUE call_id (call_r0..call_r9) so the FINAL round's + // call is individually detectable in the persisted conversation. + let (you_url, _captured_you, _you_handle) = spawn_mock_you().await; + let mut llm_responses: Vec = (0..10) + .map(|round| web_search_function_call_response_with_id(&format!("call_r{round}"))) + .collect(); + llm_responses.push(support::text_response("continued after incomplete")); + let llm = support::MockServer::start_deque(llm_responses).await; + let exec_ctx = build_exec_ctx(llm.url(), you_url).await; + let web_search: ResponsesTool = serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).unwrap(); + let payload = RequestPayload { + model: "test-model".to_owned(), + input: ResponsesInput::Text("look up rust async".to_owned()), + instructions: None, + previous_response_id: None, + conversation_id: None, + tools: Some(vec![web_search]), + tool_choice: None, + stream: false, + store: true, + include: None, + temperature: None, + top_p: None, + max_output_tokens: Some(1024), + truncation: None, + metadata: None, + }; + + let result = ExecuteRequest::new(payload, Arc::clone(&exec_ctx)).run().await.unwrap(); + let Either::Left(response) = result else { + panic!("non-streaming request should return a payload"); + }; + assert_eq!(response.status, "incomplete"); + + // Continue from the incomplete response — rehydrates the persisted turn. + let continuation_payload = RequestPayload { + model: "test-model".to_owned(), + input: ResponsesInput::Text("continue".to_owned()), + instructions: None, + previous_response_id: Some(response.id), + conversation_id: None, + tools: None, + tool_choice: None, + stream: false, + store: true, + include: None, + temperature: None, + top_p: None, + max_output_tokens: Some(1024), + truncation: None, + metadata: None, + }; + let _ = ExecuteRequest::new(continuation_payload, exec_ctx).run().await.unwrap(); + + // The continuation's upstream request carries the rehydrated conversation. + // Every function_call must pair with a function_call_output — no dangling call. + let request_bodies = llm.request_bodies().await; + let continuation_input = request_bodies + .last() + .expect("continuation request") + .get("input") + .and_then(|input| input.as_array()) + .expect("continuation input is an array"); + let call_ids: std::collections::HashSet<&str> = continuation_input + .iter() + .filter(|item| item["type"] == "function_call") + .filter_map(|item| item["call_id"].as_str()) + .collect(); + let output_call_ids: std::collections::HashSet<&str> = continuation_input + .iter() + .filter(|item| item["type"] == "function_call_output") + .filter_map(|item| item["call_id"].as_str()) + .collect(); + + // The FINAL round (call_r9) executed but the loop stopped on the budget. Its + // call and output must both be persisted — guards the Incomplete arm skipping + // the append, which would drop call_r9's output while its call is in output. + assert!( + call_ids.contains("call_r9"), + "final-round gateway call must be persisted, got calls {call_ids:?}" + ); + assert!( + output_call_ids.contains("call_r9"), + "final-round gateway call has no matching output in the persisted conversation" + ); + for cid in &call_ids { + assert!( + output_call_ids.contains(cid), + "gateway function_call {cid} has no matching output in the persisted conversation" + ); + } +} + +#[tokio::test] +async fn stream_returns_incomplete_after_max_gateway_tool_rounds() { + // Streaming counterpart of the blocking cap test: past the round budget over + // an SSE stream, the final streamed payload must carry status "incomplete". + let (you_url, mut captured_you, _you_handle) = spawn_mock_you().await; + let llm_responses = std::iter::repeat_with(web_search_function_call_sse_response) + .take(10) + .collect(); + let llm = support::MockServer::start_deque(llm_responses).await; + let exec_ctx = build_exec_ctx(llm.url(), you_url).await; + let web_search: ResponsesTool = serde_json::from_value(serde_json::json!({"type": "web_search_preview"})).unwrap(); + let payload = RequestPayload { + model: "test-model".to_owned(), + input: ResponsesInput::Text("look up rust async".to_owned()), + instructions: None, + previous_response_id: None, + conversation_id: None, + tools: Some(vec![web_search]), + tool_choice: None, + stream: true, + store: true, + include: None, + temperature: None, + top_p: None, + max_output_tokens: Some(1024), + truncation: None, + metadata: None, + }; + + let result = ExecuteRequest::new(payload, exec_ctx).run().await.unwrap(); + let Either::Right(stream) = result else { + panic!("expected streaming response"); + }; + let chunks: Vec = stream.collect().await; + + let final_event = chunks + .iter() + .filter_map(|chunk| { + let data = chunk.trim_end_matches('\n').strip_prefix("data: ")?; + (data != "[DONE]").then(|| serde_json::from_str::(data).ok())? + }) + .next_back() + .expect("stream should carry a final response payload"); + // The terminal SSE event wraps the payload: {"type":"response.incomplete","response":{...}} + let response = &final_event["response"]; + assert_eq!(response["status"], "incomplete"); + assert_eq!( + response["incomplete_details"]["reason"], + "gateway tool execution exceeded 10 rounds" + ); + for _ in 0..10 { + captured_you.recv().await.expect("mock You.com should receive request"); + } + assert_eq!(llm.request_bodies().await.len(), 10); +}