diff --git a/crates/agentic-server/Cargo.toml b/crates/agentic-server/Cargo.toml index eaa5b832..e6ea255c 100644 --- a/crates/agentic-server/Cargo.toml +++ b/crates/agentic-server/Cargo.toml @@ -30,7 +30,7 @@ url.workspace = true bytes.workspace = true criterion.workspace = true futures.workspace = true -reqwest = { workspace = true, features = ["json"] } +reqwest = { workspace = true, features = ["json", "stream"] } rand = "0.8" rsa = "0.9" serde_json.workspace = true @@ -44,5 +44,10 @@ uuid = { version = "1", features = ["v7"] } name = "benches" harness = false +[[bench]] +name = "response_provider" +path = "benches/client/response_provider/main.rs" +harness = false + [lints] workspace = true diff --git a/crates/agentic-server/benches/client/response_provider/README.md b/crates/agentic-server/benches/client/response_provider/README.md new file mode 100644 index 00000000..eabdcacc --- /dev/null +++ b/crates/agentic-server/benches/client/response_provider/README.md @@ -0,0 +1,302 @@ +# Responses transport and agentic-workflow benchmark + +The benchmark target at `crates/agentic-server/benches/client/response_provider/` runs native Responses API clients against: + +- Agentic API over one persistent WebSocket per session. +- Agentic API over HTTP with server-sent events (SSE). +- Agentic API over non-streaming HTTP/JSON. +- vLLM directly over HTTP/SSE. +- vLLM directly over non-streaming HTTP/JSON. + +It does not launch Codex CLI and does not use MCP. Function definitions from BFCL are sent directly in each Responses +request's `tools` array. Transport benchmark functions are executed locally in-process, so MCP discovery, process +startup, and stdio IPC are not included in latency measurements. + +By default, the runner starts 5 sessions per selected provider at one synchronization barrier. Requests within one +session remain sequential, while all sessions run concurrently. + +## Workloads + +| Workload | What it measures | Default turns/session | +| --- | --- | ---: | +| `transport` | Repeated model/function/model rounds over a reused connection. | 1 | +| `tool-call` | BFCL function selection and argument accuracy. | 1 | +| `history-rehydration` | Accuracy when each turn depends on the preceding stored response. | 10 | + +For Agentic API sessions, continuation requests contain `previous_response_id` and only the new input items. The +gateway rehydrates stored item history. Direct vLLM sessions have no gateway state, so the benchmark replays the full +accumulated Responses item history in each request. This makes wire-size and history-management differences part of +the deployment comparison. + +### Common execution model + +`SESSIONS` controls concurrency and `REQUESTS_PER_SESSION` controls the number of sequential turns in each session. +Every provider/session task connects first and waits at the same barrier, so the configured sessions begin under +approximately the same load. A session then completes turn N before starting turn N+1. For example, 5 sessions with +50 requests per session produces 250 planned turns per provider, with at most 5 turns in flight at once. + +For `history-rehydration`, `DEPTHS` (a comma-separated list, e.g. `5,10,25`) overrides `REQUESTS_PER_SESSION`: each +depth runs as its own independent batch of `SESSIONS` sessions instead of one long run sliced into buckets +afterward, so sample counts stay even across depths. It also produces `depth_summary.md`/`.json` with request bytes, +response bytes, latency, and accuracy grouped by depth. + +The WebSocket provider keeps one connection open for the complete session. HTTP providers reuse an HTTP client, but +each model response remains a separate Responses request. The benchmark generates prompts once and assigns the same +session and turn indices to every selected provider, allowing successful gateway and direct-vLLM turns to be paired. +Use the same `SEED`, BFCL files, and `DATASET_OFFSET` when the two provider arms are run separately. + +There are two kinds of session state: + +- `transport` and `tool-call` reset conversation state after every logical turn. Repeating these workloads measures + connection reuse and sustained concurrent traffic without allowing one case to affect another. +- `history-rehydration` deliberately retains conversation state between turns. A broken response, + timeout, or provider error stops that session because the next turn would no longer have a trustworthy continuation. + +Rates use all planned turns as their denominator. Therefore, turns skipped after a session-ending failure reduce the +success and correctness rates; `attempted_turns` in `run.json` shows how many requests were actually started. Latency +distributions include only successful turns. Always read latency together with success and correctness: a provider +must not look faster merely because its slow or difficult turns failed. + +### Choosing a comparison + +| Question | Workload and providers | Most useful observations | +| --- | --- | --- | +| What does persistent WebSocket mode change? | Any workload; `agentic-api` versus `agentic-api-http`. | First-output latency, end-to-end latency, continuation-round latency, throughput. | +| What is the gateway cost with the same streaming transport? | Any workload; `agentic-api-http` versus `vllm`. | Paired latency deltas, success, input tokens, throughput. | +| What is the streaming benefit over complete JSON responses? | Same endpoint; SSE/WebSocket versus its JSON provider. | First-output latency and TTFT for streaming, end-to-end latency for both. | +| Does the model emit correct function calls? | `tool-call`; the same BFCL category on both providers. | Tool-call accuracy, success, time to first tool call. | +| Does stored-response continuation preserve recent state? | `history-rehydration`. | Task correctness by turn, input-token growth, latency by turn. | +| How does concurrency affect capacity? | Repeat a workload at increasing `SESSIONS`. | Successful turns/s, p95/p99 latency, timeout and failure rates. | + +`agentic-api` versus `vllm` is an end-to-end deployment comparison: it changes WebSocket versus HTTP/SSE and gateway +managed history versus client replay at the same time. Use `agentic-api` versus `agentic-api-http` to focus more +narrowly on transport within the gateway, or `agentic-api-http` versus `vllm` to compare the gateway and direct path +while holding HTTP/SSE constant. + +### Transport workload + +One transport turn is a deterministic sequential tool loop rather than one model request. With +`TRANSPORT_ROUNDS=4`, the sequence is: + +```text +user prompt + -> model function call benchmark_step(step=1) + -> client-executed function call output + -> model function call benchmark_step(step=2) + -> client-executed function call output + -> model function call benchmark_step(step=3) + -> client-executed function call output + -> model function call benchmark_step(step=4) + -> client-executed function call output containing the final marker + -> model repeats that marker as its final answer +``` + +The function runs in the benchmark process and returns a deterministic marker. It performs no network or application +work, so most measured time comes from inference, event delivery, function-call serialization, continuation, and +gateway state handling. The benchmark requires calls to use the expected run ID, step number, and total step count in +order. A turn is task-correct only if all calls match, the final function call output contains the expected marker, +and the model's final answer is exactly that marker. Parallel tool calls are disabled; this workload intentionally +measures sequential continuation. + +This workload is useful for measuring: + +- the steady-state cost of reusing a transport across repeated independent turns; +- the accumulated cost of a multi-round tool loop; +- whether function calls, function call outputs, and call IDs survive continuation correctly; +- whether gateway storage and rehydration work between inference rounds within one turn; +- throughput and tail latency when many tool loops execute concurrently. + +The primary metrics are `end_to_end_latency_ms`, `initial_model_round_ms`, +`continuation_round_latencies_ms`, `time_to_first_tool_call_ms`, task correctness, and successful turns/s. Compare +multiple `TRANSPORT_ROUNDS` values to estimate how latency grows per continuation round. `mean_tool_duration_ms` +measures the time between the first and completed streaming events for a function call; it does not measure a real +external tool's execution time. This synthetic workload therefore does not predict the latency of database, web, or +other production tools. Connection establishment happens before the start barrier and is excluded, so this workload +also does not measure WebSocket or HTTP connection setup time. + +### Tool-call workload + +Each tool-call turn selects one deterministic row from the Berkeley Function-Calling Leaderboard (BFCL) question +file and joins it to the possible-answer file by case ID. The benchmark sends the row's user request and function +schemas as native Responses function tools, then evaluates the response's `function_call` output items. Selection is +the contiguous range starting at `DATASET_OFFSET`; `SEED` does not shuffle BFCL rows. The required dataset size is: + +```text +SESSIONS * REQUESTS_PER_SESSION + DATASET_OFFSET +``` + +Every BFCL case is independent, conversation state is reset after it, and the workload performs one inference round. +It does not execute the selected function and does not submit a function call output. A call is correct only when the +number of calls and function names match, no unexpected argument keys are present, and each argument equals one of +the accepted top-level BFCL values. Call order is ignored. An empty-string possible answer allows an optional argument +to be omitted. + +Use `simple_python` to test argument construction with one available function. Use `multiple` or `live_multiple` to +test selection of one function from several candidates. `multiple` does not mean parallel function calling. The BFCL +workload currently keeps `parallel_tool_calls=false`, so do not use BFCL `parallel`, `parallel_multiple`, +`live_parallel`, or `live_parallel_multiple` categories. + +The primary metric is `tool_call_accuracy`. `success_rate` is stricter because it additionally requires a completed +response with no timeout or provider error. `time_to_first_tool_call_ms` is the best streaming responsiveness metric; +`ttft_ms` may be absent when a model emits a function call without output text. Run each BFCL category separately to +distinguish schema/argument accuracy from function-selection accuracy. + +This is a lightweight BFCL-compatible comparison, not the complete official BFCL evaluator. It performs exact +top-level accepted-value matching, does not recursively interpret every nested BFCL alternative, and does not run +BFCL executable cases. Use it to compare identical gateway and direct-vLLM traffic, but do not report its aggregate +as an official BFCL leaderboard score. + +### History-rehydration workload + +This workload creates a synthetic chain of secrets unique to each session. Turn 0 tells the model to output marker 0 +and privately remember marker 1. Every later turn intentionally omits the previous secret, asks the model to recall +it exactly, and supplies a new secret for the following turn: + +```text +turn 0 input: output M0; remember M1 expected output: M0 +turn 1 input: recall the previous secret; remember M2 + expected output: M1 +turn 2 input: recall the previous secret; remember M3 + expected output: M2 +``` + +The gateway arm sends `store: true`; after the first turn it sends only the new input plus the previous response ID. +The gateway must load the stored item history, preserve its ordering, and provide it to inference. The direct-vLLM +arm sends `store: false` and replays all accumulated input and output items on every request. Both arms ask the same +semantic question, but their request sizes and state-management responsibilities intentionally differ. + +A turn is correct only when the final answer, after trimming surrounding whitespace or backticks, exactly equals the +hidden marker. This catches missing history, an incorrect previous response ID, reordered items, cross-session state +leakage, and state that was stored but not included in later inference. It is a narrow synthetic recall check: it does +not evaluate summarization, real conversational coherence, tools, compaction, or recall from far back in the session. + +Task correctness and success rate are the primary correctness metrics. Examine `turns.csv` by `turn` for +`end_to_end_latency_ms`, first-output latency, TTFT, input tokens, and `request_bytes`/`response_bytes`. Model-visible +input tokens should grow for both arms because the gateway rehydrates the history before inference; only the +client-to-gateway request stays small for the gateway arm. `request_bytes` is the direct measurement of that +difference: flat across turn depth for the gateway arm, growing for direct vLLM since its client must resend the +full accumulated history every turn. Because every turn has one model response, `initial_model_round_ms` is useful +but continuation-round latency is empty. + +## Run + +The optimized wrapper saves the exact command, live events, diagnostics, and reports in a timestamped directory: + +```bash +MODEL="Qwen/Qwen3.5-35B-A3B-FP8" \ +WORKLOAD=transport \ +SESSIONS=10 \ +REQUESTS_PER_SESSION=10 \ +TRANSPORT_ROUNDS=4 \ +PROVIDER=both \ +./scripts/run-response-provider-benchmark.sh +``` + +`PROVIDER=both` means Agentic API WebSocket versus direct vLLM HTTP/SSE. `PROVIDER=all` runs all five transport/provider +combinations. Individual values are: + +```text +agentic-api Agentic API WebSocket streaming +agentic-api-http Agentic API HTTP/SSE streaming +agentic-api-json Agentic API HTTP/JSON non-streaming +vllm direct vLLM HTTP/SSE streaming +vllm-json direct vLLM HTTP/JSON non-streaming +``` + +To run the gateway first, restart vLLM, and then run the direct arm with identical generated prompts: + +```bash +MODEL="Qwen/Qwen3.5-35B-A3B-FP8" WORKLOAD=transport \ +PROVIDER=agentic-api SEED=20260817 ./scripts/run-response-provider-benchmark.sh + +# Restart vLLM here. + +MODEL="Qwen/Qwen3.5-35B-A3B-FP8" WORKLOAD=transport \ +PROVIDER=vllm SEED=20260817 ./scripts/run-response-provider-benchmark.sh +``` + +For the equivalent JSON comparison, use `PROVIDER=agentic-api-json` and `PROVIDER=vllm-json`. + +The direct Cargo command is: + +```bash +cargo bench -p agentic-server --bench response_provider -- \ + --model "Qwen/Qwen3.5-35B-A3B-FP8" \ + --workload transport \ + --provider both \ + --sessions 10 \ + --requests-per-session 10 \ + --transport-rounds 4 \ + --live-jsonl +``` + +No Codex model catalog or `CODEX_HOME` is needed because the benchmark sends Responses requests directly. + +## BFCL tool-call workload + +Use an official BFCL checkout and select a v4 category: + +```bash +MODEL="Qwen/Qwen3.5-35B-A3B-FP8" \ +WORKLOAD=tool-call \ +PROVIDER=both \ +BFCL_ROOT=/path/to/BFCL \ +BFCL_CATEGORY=simple_python \ +SESSIONS=10 \ +REQUESTS_PER_SESSION=10 \ +./scripts/run-response-provider-benchmark.sh +``` + +The wrapper derives these files: + +```text +berkeley-function-call-leaderboard/bfcl_eval/data/BFCL_v4_.json +berkeley-function-call-leaderboard/bfcl_eval/data/possible_answer/BFCL_v4_.json +``` + +You can instead set `DATASET_QUESTIONS` and `DATASET_ANSWERS` explicitly. `DATASET_OFFSET` chooses the first case. +Cases are assigned deterministically and are paired across providers. Set `PRINT_PROMPTS=1` to validate the dataset +and print the generated cases without contacting either provider. + +## Results + +Each run directory contains: + +```text +command.sh +live-events.jsonl +benchmark.log +run.json +turns.csv +summary.md +events//session-NNN/turn-NNN.responses.jsonl +events//session-NNN/turn-NNN.timestamped.jsonl +events//session-NNN/turn-NNN.errors.log +``` + +`success` is strict: the request must reach a completed Responses terminal event without a timeout or provider error, +all locally handled transport tools must succeed, and the workload-specific correctness check must pass. A BFCL turn +is correct only when the function names, number of calls, and accepted arguments match its ground truth. A history turn +is correct only when the final answer is exactly the expected hidden marker. + +Useful metrics are: + +| Metric | Meaning | +| --- | --- | +| `end_to_end_latency_ms` | Request send through the logical turn's final terminal event, including all tool continuations. | +| `initial_model_round_ms` | Latency of the first model response in a logical turn. | +| `continuation_round_latencies_ms` | Latency of each response after a local function result is submitted. | +| `time_to_first_output_event_ms` | Streaming request send to the first output/reasoning/function event. | +| `ttft_ms` | Streaming request send to the first `response.output_text.delta`. It is null for HTTP/JSON. | +| `time_to_first_tool_call_ms` | Streaming request send to the first function-call event. | +| `request_bytes` | Serialized request-body bytes sent to the provider for the whole logical turn. | +| `response_bytes` | Raw response payload bytes (WS text frames / SSE chunks / JSON body) received for the turn. | +| `tool_call_accuracy` | Fraction of planned turns whose function names and arguments match the expected calls. | +| `task_correctness_rate` | Fraction of planned turns passing the workload-specific semantic check. | +| `success_rate` | Fraction passing transport, completion, execution, and semantic checks together. | +| `successful_turns_per_second` | Successful turns divided by the slowest concurrent session wall time. | + +`p50` is the median sample. `p95` is the value at or below which approximately 95% of samples fall; it highlights tail +latency. `p99` targets rarer tail behavior and needs substantially more samples before it is stable. Compare latency +only between successful turns, while also reporting correctness and failure rates so faster failures are never counted +as wins. diff --git a/crates/agentic-server/benches/client/response_provider/cli.rs b/crates/agentic-server/benches/client/response_provider/cli.rs new file mode 100644 index 00000000..b84aed0a --- /dev/null +++ b/crates/agentic-server/benches/client/response_provider/cli.rs @@ -0,0 +1,107 @@ +use std::path::PathBuf; + +use clap::{Parser, ValueEnum}; + +use crate::types::Workload; + +#[derive(Clone, Copy, Debug, Eq, PartialEq, ValueEnum)] +pub enum ProviderSelection { + /// Gateway WebSocket, HTTP/SSE, and JSON plus direct vLLM HTTP/SSE and JSON. + All, + /// Gateway WebSocket and direct vLLM HTTP/SSE (backwards-compatible pair). + Both, + AgenticApi, + AgenticApiHttp, + AgenticApiJson, + Vllm, + VllmJson, +} + +#[derive(Debug, Parser)] +#[command( + name = "agentic-responses-benchmark", + about = "Run concurrent Responses workloads against Agentic API and direct vLLM" +)] +pub struct Cli { + /// Model slug exposed by each selected provider. + #[arg(long, env = "RESPONSE_PROVIDER_BENCH_MODEL")] + pub model: String, + + /// Capability measured by this run. + #[arg(long, value_enum, default_value_t = Workload::HistoryRehydration)] + pub workload: Workload, + + /// Number of concurrent sessions per selected provider. + #[arg(long, default_value_t = 5)] + pub sessions: usize, + + /// Sequential turns in each session. Defaults depend on `--workload`. + #[arg(long, short = 'n')] + pub requests_per_session: Option, + + /// Fixed session depths (history-rehydration only), each run as its own independent batch of + /// `--sessions` sessions instead of bucketing one long run after the fact. Overrides + /// `--requests-per-session` when set. Example: --depths 1,5,10,25,50,100 + #[arg(long, value_delimiter = ',')] + pub depths: Vec, + + /// Responses created inside each transport workload turn. + #[arg(long, default_value_t = 4)] + pub transport_rounds: usize, + + /// BFCL question JSONL for the tool-call workload. + #[arg(long, requires = "dataset_answers")] + pub dataset_questions: Option, + + /// BFCL possible-answer JSONL paired by case ID. + #[arg(long, requires = "dataset_questions")] + pub dataset_answers: Option, + + /// Zero-based BFCL case offset before selecting deterministic cases. + #[arg(long, default_value_t = 0)] + pub dataset_offset: usize, + + /// Providers to benchmark. `all` includes both gateway transports. + #[arg(long, value_enum, default_value_t = ProviderSelection::Both)] + pub provider: ProviderSelection, + + /// Agentic API OpenAI-compatible base URL. + #[arg(long, default_value = "http://localhost:9000/v1")] + pub agentic_url: String, + + /// Direct vLLM OpenAI-compatible base URL. + #[arg(long, default_value = "http://localhost:5050/v1")] + pub vllm_url: String, + + /// Timeout for one logical turn, including every model/tool round. + #[arg(long, default_value_t = 300)] + pub timeout_seconds: u64, + + /// Seed for deterministic generated prompts and continuation secrets. + #[arg(long, default_value_t = 2_026_081_7)] + pub seed: u64, + + /// Result directory. Defaults to target/response-provider-benchmark/. + #[arg(long)] + pub output_dir: Option, + + /// Stream timestamped, provider-tagged Responses events to stdout. + #[arg(long)] + pub live_jsonl: bool, + + /// Validate inputs, print generated prompt specifications as JSONL, and exit. + #[arg(long)] + pub print_prompts: bool, + + /// Automatically appended by `cargo bench` for custom harnesses. + #[arg(long, hide = true)] + pub bench: bool, +} + +impl Cli { + #[must_use] + pub fn requests_per_session(&self) -> usize { + self.requests_per_session + .unwrap_or_else(|| self.workload.default_requests()) + } +} diff --git a/crates/agentic-server/benches/client/response_provider/main.rs b/crates/agentic-server/benches/client/response_provider/main.rs new file mode 100644 index 00000000..025c0025 --- /dev/null +++ b/crates/agentic-server/benches/client/response_provider/main.rs @@ -0,0 +1,403 @@ +mod cli; +mod prompts; +mod report; +mod runner; +mod types; + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use clap::Parser; +use thiserror::Error; +use tokio::sync::{Barrier, Mutex}; +use tokio::task::JoinSet; + +use crate::cli::{Cli, ProviderSelection}; +use crate::runner::RunnerConfig; +use crate::types::{ProviderSpec, RunConfig, RunReport, SessionResult, Transport, Workload}; + +const TTFT_NOTE: &str = "For WebSocket and HTTP/SSE runs, time_to_first_output_event_ms is measured from request send \ +to the first output event, and ttft_ms is measured to the first output_text delta (token-level \ +TTFT). Non-streaming HTTP/JSON has no token-level TTFT, so those fields are null."; + +#[derive(Debug, Error)] +enum Error { + #[error("{0}")] + InvalidArgument(String), + #[error("failed to access {path}: {source}")] + Io { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("benchmark task failed: {0}")] + Join(#[from] tokio::task::JoinError), + #[error(transparent)] + Prompt(#[from] prompts::PromptError), +} + +#[tokio::main] +#[allow(clippy::too_many_lines)] +async fn main() -> Result<(), Error> { + let cli = Cli::parse(); + validate(&cli)?; + let started_at_unix_ms = unix_millis(); + let dataset_questions = match &cli.dataset_questions { + Some(path) => Some(canonicalize(path).await?), + None => None, + }; + let dataset_answers = match &cli.dataset_answers { + Some(path) => Some(canonicalize(path).await?), + None => None, + }; + let base_output_dir = cli + .output_dir + .clone() + .unwrap_or_else(|| PathBuf::from(format!("target/response-provider-benchmark/{started_at_unix_ms}"))); + + if cli.print_prompts { + let turn_counts = if cli.depths.is_empty() { + vec![cli.requests_per_session()] + } else { + cli.depths.clone() + }; + for turns in turn_counts { + let generated = prompts::generate(&prompts::GenerationConfig { + workload: cli.workload, + seed: cli.seed, + sessions: cli.sessions, + turns, + transport_rounds: cli.transport_rounds, + dataset_questions: dataset_questions.clone(), + dataset_answers: dataset_answers.clone(), + dataset_offset: cli.dataset_offset, + }) + .await?; + for prompt in generated.iter().flat_map(|session| &session.prompts) { + println!( + "{}", + serde_json::to_string(prompt).map_err(|source| Error::Io { + path: PathBuf::from("generated prompt JSON"), + source: std::io::Error::other(source), + })? + ); + } + } + return Ok(()); + } + + if cli.depths.is_empty() { + let run_report = run_pipeline( + &cli, + cli.requests_per_session(), + base_output_dir, + dataset_questions, + dataset_answers, + started_at_unix_ms, + "run", + ) + .await?; + if run_report.summaries.iter().any(|summary| summary.successful_turns == 0) { + return Err(Error::InvalidArgument( + "one or more providers completed zero successful turns; inspect run.json and per-turn error logs" + .to_owned(), + )); + } + return Ok(()); + } + + create_dir_all(&base_output_dir).await?; + let mut depths = cli.depths.clone(); + depths.sort_unstable(); + depths.dedup(); + let mut depth_reports = Vec::with_capacity(depths.len()); + let mut any_zero_success = false; + for depth in depths { + let depth_dir = base_output_dir.join(format!("depth-{depth:05}")); + let label = format!("depth={depth}"); + match run_pipeline( + &cli, + depth, + depth_dir, + dataset_questions.clone(), + dataset_answers.clone(), + started_at_unix_ms, + &label, + ) + .await + { + Ok(run_report) => { + if run_report.summaries.iter().any(|summary| summary.successful_turns == 0) { + any_zero_success = true; + eprintln!("warning: [{label}] one or more providers completed zero successful turns"); + } + depth_reports.push((depth, run_report)); + } + Err(error) => { + any_zero_success = true; + eprintln!("warning: [{label}] failed: {error}"); + } + } + } + + let rows = report::depth_rollup(&depth_reports); + let rollup_markdown = report::depth_rollup_markdown(&rows); + let rollup_json = serde_json::to_vec_pretty(&rows).map_err(|source| Error::Io { + path: base_output_dir.join("depth_summary.json"), + source: std::io::Error::other(source), + })?; + tokio::fs::write(base_output_dir.join("depth_summary.md"), rollup_markdown) + .await + .map_err(|source| Error::Io { + path: base_output_dir.clone(), + source, + })?; + tokio::fs::write(base_output_dir.join("depth_summary.json"), rollup_json) + .await + .map_err(|source| Error::Io { + path: base_output_dir.clone(), + source, + })?; + eprintln!( + "depth-scaling summary: {}", + base_output_dir.join("depth_summary.md").display() + ); + + if any_zero_success { + return Err(Error::InvalidArgument( + "one or more depth runs completed zero successful turns for some provider; inspect each depth-*/run.json" + .to_owned(), + )); + } + Ok(()) +} + +/// Depth-batch mode in `main` calls this once per fixed depth so each depth is an independent, +/// fully-sampled run instead of a bucket sliced out of one long run. +#[allow(clippy::too_many_arguments)] +async fn run_pipeline( + cli: &Cli, + requests_per_session: usize, + output_dir: PathBuf, + dataset_questions: Option, + dataset_answers: Option, + started_at_unix_ms: u64, + label: &str, +) -> Result { + let run_started = Instant::now(); + create_dir_all(&output_dir).await?; + let output_dir = canonicalize(&output_dir).await?; + let providers = selected_providers(cli); + let generated = prompts::generate(&prompts::GenerationConfig { + workload: cli.workload, + seed: cli.seed, + sessions: cli.sessions, + turns: requests_per_session, + transport_rounds: cli.transport_rounds, + dataset_questions: dataset_questions.clone(), + dataset_answers: dataset_answers.clone(), + dataset_offset: cli.dataset_offset, + }) + .await?; + let flat_prompts = generated + .iter() + .flat_map(|session| session.prompts.iter().cloned()) + .collect::>(); + + let runner_config = Arc::new(RunnerConfig { + model: cli.model.clone(), + output_dir: output_dir.clone(), + timeout: Duration::from_secs(cli.timeout_seconds), + live_jsonl: cli.live_jsonl, + }); + + let task_count = providers.len().saturating_mul(cli.sessions); + let start_barrier = Arc::new(Barrier::new(task_count.saturating_add(1))); + let live_output_lock = Arc::new(Mutex::new(())); + let mut tasks = JoinSet::new(); + for provider in &providers { + for session in &generated { + tasks.spawn(runner::run_session( + Arc::clone(&runner_config), + provider.clone(), + session.session_index, + session.prompts.clone(), + Arc::clone(&start_barrier), + Arc::clone(&live_output_lock), + )); + } + } + + eprintln!( + "[{label}] starting {task_count} concurrent Responses sessions ({} per provider, {} turns per session)", + cli.sessions, requests_per_session + ); + start_barrier.wait().await; + let mut sessions: Vec = Vec::with_capacity(task_count); + while let Some(result) = tasks.join_next().await { + sessions.push(result?); + } + sessions.sort_by(|left, right| { + left.provider + .cmp(&right.provider) + .then(left.session_index.cmp(&right.session_index)) + }); + + let summaries = report::summarize(&providers, &sessions, cli.sessions, requests_per_session); + let comparison = report::compare(&sessions); + let run_report = RunReport { + schema_version: 3, + started_at_unix_ms, + elapsed_ms: millis(run_started.elapsed()), + output_dir: output_dir.clone(), + config: RunConfig { + model: cli.model.clone(), + workload: cli.workload, + dataset_questions, + dataset_answers, + sessions_per_provider: cli.sessions, + requests_per_session, + seed: cli.seed, + timeout_seconds: cli.timeout_seconds, + providers, + }, + prompts: flat_prompts, + sessions, + summaries, + comparison, + ttft_note: TTFT_NOTE.to_owned(), + accuracy_note: report::ACCURACY_NOTE.to_owned(), + }; + report::write_reports(&output_dir, &run_report) + .await + .map_err(|source| Error::Io { + path: output_dir.clone(), + source, + })?; + + let summary = report::console_summary(&run_report.summaries, run_report.comparison.as_ref()); + if cli.live_jsonl { + eprintln!("[{label}]\n{summary}"); + } else { + println!("[{label}]\n{summary}"); + } + eprintln!("[{label}] results: {}", output_dir.display()); + Ok(run_report) +} + +fn validate(cli: &Cli) -> Result<(), Error> { + if cli.model.trim().is_empty() { + return Err(Error::InvalidArgument("--model must not be empty".to_owned())); + } + if cli.sessions == 0 { + return Err(Error::InvalidArgument( + "--sessions must be greater than zero".to_owned(), + )); + } + if cli.requests_per_session() == 0 { + return Err(Error::InvalidArgument( + "--requests-per-session must be greater than zero".to_owned(), + )); + } + if cli.timeout_seconds == 0 { + return Err(Error::InvalidArgument( + "--timeout-seconds must be greater than zero".to_owned(), + )); + } + if cli.transport_rounds == 0 { + return Err(Error::InvalidArgument( + "--transport-rounds must be greater than zero".to_owned(), + )); + } + if cli.workload == Workload::ToolCall && (cli.dataset_questions.is_none() || cli.dataset_answers.is_none()) { + return Err(Error::InvalidArgument( + "--workload tool-call requires --dataset-questions and --dataset-answers".to_owned(), + )); + } + if !cli.depths.is_empty() { + if cli.workload != Workload::HistoryRehydration { + return Err(Error::InvalidArgument( + "--depths is only supported for --workload history-rehydration".to_owned(), + )); + } + if cli.depths.contains(&0) { + return Err(Error::InvalidArgument( + "--depths values must be greater than zero".to_owned(), + )); + } + } + Ok(()) +} + +fn selected_providers(cli: &Cli) -> Vec { + let agentic = ProviderSpec { + id: "agentic-api".to_owned(), + name: "agentic-api".to_owned(), + base_url: cli.agentic_url.clone(), + transport: Transport::Websocket, + supports_websockets: true, + }; + let agentic_http = ProviderSpec { + id: "agentic-api-http".to_owned(), + name: "agentic-api-http".to_owned(), + base_url: cli.agentic_url.clone(), + transport: Transport::HttpSse, + supports_websockets: false, + }; + let agentic_json = ProviderSpec { + id: "agentic-api-json".to_owned(), + name: "agentic-api-json".to_owned(), + base_url: cli.agentic_url.clone(), + transport: Transport::HttpJson, + supports_websockets: false, + }; + let vllm = ProviderSpec { + id: "vllm".to_owned(), + name: "vllm".to_owned(), + base_url: cli.vllm_url.clone(), + transport: Transport::HttpSse, + supports_websockets: false, + }; + let vllm_json = ProviderSpec { + id: "vllm-json".to_owned(), + name: "vllm-json".to_owned(), + base_url: cli.vllm_url.clone(), + transport: Transport::HttpJson, + supports_websockets: false, + }; + match cli.provider { + ProviderSelection::All => vec![agentic, agentic_http, agentic_json, vllm, vllm_json], + ProviderSelection::Both => vec![agentic, vllm], + ProviderSelection::AgenticApi => vec![agentic], + ProviderSelection::AgenticApiHttp => vec![agentic_http], + ProviderSelection::AgenticApiJson => vec![agentic_json], + ProviderSelection::Vllm => vec![vllm], + ProviderSelection::VllmJson => vec![vllm_json], + } +} + +async fn canonicalize(path: &Path) -> Result { + tokio::fs::canonicalize(path).await.map_err(|source| Error::Io { + path: path.to_owned(), + source, + }) +} + +async fn create_dir_all(path: &Path) -> Result<(), Error> { + tokio::fs::create_dir_all(path).await.map_err(|source| Error::Io { + path: path.to_owned(), + source, + }) +} + +fn unix_millis() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(millis) + .unwrap_or_default() +} + +fn millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} diff --git a/crates/agentic-server/benches/client/response_provider/prompts.rs b/crates/agentic-server/benches/client/response_provider/prompts.rs new file mode 100644 index 00000000..8c78cb91 --- /dev/null +++ b/crates/agentic-server/benches/client/response_provider/prompts.rs @@ -0,0 +1,359 @@ +use std::collections::{BTreeMap, HashMap}; +use std::path::{Path, PathBuf}; + +use serde::Deserialize; +use serde_json::{Map, Value, json}; +use thiserror::Error; + +use crate::types::{ExpectedToolCall, PromptSpec, SessionSpec, ToolDefinition, TurnExpectation, Workload}; + +const TRANSPORT_TOOL_NAME: &str = "benchmark_step"; + +#[derive(Clone, Debug)] +pub struct GenerationConfig { + pub workload: Workload, + pub seed: u64, + pub sessions: usize, + pub turns: usize, + pub transport_rounds: usize, + pub dataset_questions: Option, + pub dataset_answers: Option, + pub dataset_offset: usize, +} + +#[derive(Debug, Error)] +pub enum PromptError { + #[error("the tool-call workload requires --dataset-questions and --dataset-answers")] + MissingDataset, + #[error("failed to read benchmark dataset {path}: {source}")] + Read { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("invalid JSON on line {line} of {path}: {source}")] + Json { + path: PathBuf, + line: usize, + #[source] + source: serde_json::Error, + }, + #[error("BFCL answer is missing for case {0}")] + MissingAnswer(String), + #[error("BFCL case {0} has no user prompt")] + MissingQuestion(String), + #[error("requested {requested} BFCL cases at offset {offset}, but the dataset contains only {available}")] + DatasetTooSmall { + requested: usize, + offset: usize, + available: usize, + }, +} + +#[derive(Debug, Deserialize)] +struct BfclQuestion { + id: String, + question: Vec>, + #[serde(rename = "function")] + functions: Vec, +} + +#[derive(Debug, Deserialize)] +struct BfclMessage { + role: String, + content: String, +} + +#[derive(Debug, Deserialize)] +struct BfclFunction { + name: String, + #[serde(default)] + description: String, + parameters: Value, +} + +#[derive(Debug, Deserialize)] +struct BfclAnswer { + id: String, + ground_truth: Vec>>>, +} + +pub async fn generate(config: &GenerationConfig) -> Result, PromptError> { + match config.workload { + Workload::Transport => Ok(generate_transport(config)), + Workload::ToolCall => generate_tool_calls(config).await, + Workload::HistoryRehydration => Ok(generate_history(config)), + } +} + +fn generate_transport(config: &GenerationConfig) -> Vec { + let tool = transport_tool(); + (0..config.sessions) + .map(|session_index| { + let prompts = (0..config.turns) + .map(|turn_index| { + let run_id = format!("{:016X}", mixed_marker(config.seed, session_index, turn_index)); + let calls = (1..=config.transport_rounds) + .map(|step| ExpectedToolCall { + name: TRANSPORT_TOOL_NAME.to_owned(), + arguments: BTreeMap::from([ + ("run_id".to_owned(), vec![json!(run_id)]), + ("step".to_owned(), vec![json!(step)]), + ("total_steps".to_owned(), vec![json!(config.transport_rounds)]), + ]), + }) + .collect(); + let final_marker = transport_marker(&run_id, config.transport_rounds, config.transport_rounds); + let prompt = format!( + "This is a transport benchmark. Use only the `{TRANSPORT_TOOL_NAME}` function tool. Call it exactly \ + {rounds} times, one call at a time and in order. For call N, pass run_id `{run_id}`, step N, \ + and total_steps {rounds}. Wait for each tool call output before making the next call. After the \ + last call, reply with exactly the marker returned by that last call and no other text.", + rounds = config.transport_rounds, + ); + PromptSpec { + workload: Workload::Transport, + session_index, + turn_index, + prompt_id: format!("transport-s{session_index:03}-t{turn_index:03}"), + source_id: None, + prompt, + expectation: TurnExpectation::Transport { calls, final_marker }, + tools: vec![tool.clone()], + } + }) + .collect(); + SessionSpec { + session_index, + prompts, + } + }) + .collect() +} + +async fn generate_tool_calls(config: &GenerationConfig) -> Result, PromptError> { + let (Some(question_path), Some(answer_path)) = (&config.dataset_questions, &config.dataset_answers) else { + return Err(PromptError::MissingDataset); + }; + let questions: Vec = read_jsonl(question_path).await?; + let answers: Vec = read_jsonl(answer_path).await?; + let answers: HashMap<_, _> = answers + .into_iter() + .map(|answer| (answer.id, answer.ground_truth)) + .collect(); + let requested = config.sessions.saturating_mul(config.turns); + let end = config.dataset_offset.saturating_add(requested); + if end > questions.len() { + return Err(PromptError::DatasetTooSmall { + requested, + offset: config.dataset_offset, + available: questions.len(), + }); + } + + let selected = &questions[config.dataset_offset..end]; + let mut sessions = Vec::with_capacity(config.sessions); + for session_index in 0..config.sessions { + let mut prompts = Vec::with_capacity(config.turns); + for turn_index in 0..config.turns { + let case = &selected[session_index * config.turns + turn_index]; + let ground_truth = answers + .get(&case.id) + .ok_or_else(|| PromptError::MissingAnswer(case.id.clone()))?; + let prompt = bfcl_prompt(case)?; + let tools = case.functions.iter().map(bfcl_tool).collect::>(); + let calls = ground_truth + .iter() + .flat_map(|call_group| call_group.iter()) + .map(|(name, arguments)| ExpectedToolCall { + name: name.clone(), + arguments: arguments.clone(), + }) + .collect(); + prompts.push(PromptSpec { + workload: Workload::ToolCall, + session_index, + turn_index, + prompt_id: format!("bfcl-{}", case.id), + source_id: Some(case.id.clone()), + prompt, + expectation: TurnExpectation::ToolCalls { calls }, + tools, + }); + } + sessions.push(SessionSpec { session_index, prompts }); + } + Ok(sessions) +} + +fn generate_history(config: &GenerationConfig) -> Vec { + (0..config.sessions) + .map(|session_index| { + let markers: Vec = (0..=config.turns) + .map(|turn_index| marker(config.seed, session_index, turn_index)) + .collect(); + let prompts = (0..config.turns) + .map(|turn_index| { + let expected_marker = markers[turn_index].clone(); + let next_marker = markers[turn_index + 1].clone(); + let prompt = if turn_index == 0 { + initial_history_prompt(&expected_marker, &next_marker) + } else { + continuation_history_prompt(&next_marker) + }; + PromptSpec { + workload: Workload::HistoryRehydration, + session_index, + turn_index, + prompt_id: format!("history-s{session_index:03}-t{turn_index:03}"), + source_id: None, + prompt, + expectation: TurnExpectation::Marker { + marker: expected_marker, + }, + tools: Vec::new(), + } + }) + .collect(); + SessionSpec { session_index, prompts } + }) + .collect() +} + +fn initial_history_prompt(expected_marker: &str, next_marker: &str) -> String { + format!( + "This is a state-continuation benchmark. Reply with exactly `{expected_marker}` and no other text. Privately \ + remember this next-turn secret: `{next_marker}`; do not print it yet." + ) +} + +fn continuation_history_prompt(next_marker: &str) -> String { + format!( + "Recall the next-turn secret from my immediately preceding request; I am intentionally not restating its \ + value. Reply with exactly the recalled value and no other text. \ + Privately remember this new next-turn secret: `{next_marker}`; do not print it yet." + ) +} + +fn bfcl_prompt(case: &BfclQuestion) -> Result { + let user_text = case + .question + .iter() + .flatten() + .filter(|message| message.role == "user") + .map(|message| message.content.as_str()) + .collect::>() + .join("\n"); + if user_text.is_empty() { + return Err(PromptError::MissingQuestion(case.id.clone())); + } + Ok(format!( + "This is a BFCL tool-calling evaluation. Call only the provided function tool or tools needed to satisfy the \ + request. Do not invent arguments. Return the required function call or calls.\n\nUser request: {user_text}" + )) +} + +fn bfcl_tool(function: &BfclFunction) -> ToolDefinition { + let mut parameters = function.parameters.clone(); + normalize_schema(&mut parameters); + let parameters = parameters.as_object().cloned().unwrap_or_else(|| { + Map::from_iter([ + ("type".to_owned(), Value::String("object".to_owned())), + ("properties".to_owned(), Value::Object(Map::new())), + ]) + }); + ToolDefinition { + name: function.name.clone(), + description: function.description.clone(), + parameters, + } +} + +fn normalize_schema(value: &mut Value) { + match value { + Value::Object(object) => { + if let Some(Value::String(kind)) = object.get_mut("type") { + *kind = match kind.as_str() { + "dict" => "object", + "float" => "number", + "list" | "tuple" => "array", + other => other, + } + .to_owned(); + } + for child in object.values_mut() { + normalize_schema(child); + } + } + Value::Array(values) => { + for child in values { + normalize_schema(child); + } + } + _ => {} + } +} + +fn transport_tool() -> ToolDefinition { + ToolDefinition { + name: TRANSPORT_TOOL_NAME.to_owned(), + description: "Advance exactly one round of a deterministic transport benchmark.".to_owned(), + parameters: Map::from_iter([ + ("type".to_owned(), json!("object")), + ( + "properties".to_owned(), + json!({ + "run_id": {"type": "string"}, + "step": {"type": "integer", "minimum": 1}, + "total_steps": {"type": "integer", "minimum": 1} + }), + ), + ("required".to_owned(), json!(["run_id", "step", "total_steps"])), + ("additionalProperties".to_owned(), json!(false)), + ]), + } +} + +pub fn transport_marker(run_id: &str, step: usize, total_steps: usize) -> String { + format!("TRANSPORT_MARKER_{run_id}_{step}_OF_{total_steps}") +} + +async fn read_jsonl(path: &Path) -> Result, PromptError> +where + T: for<'de> Deserialize<'de>, +{ + let text = tokio::fs::read_to_string(path) + .await + .map_err(|source| PromptError::Read { + path: path.to_owned(), + source, + })?; + text.lines() + .enumerate() + .filter(|(_, line)| !line.trim().is_empty()) + .map(|(index, line)| { + serde_json::from_str(line).map_err(|source| PromptError::Json { + path: path.to_owned(), + line: index + 1, + source, + }) + }) + .collect() +} + +fn marker(seed: u64, session_index: usize, turn_index: usize) -> String { + format!("HISTORY_MARKER_{:016X}", mixed_marker(seed, session_index, turn_index)) +} + +fn mixed_marker(seed: u64, session_index: usize, turn_index: usize) -> u64 { + let session = u64::try_from(session_index).unwrap_or(u64::MAX); + let turn = u64::try_from(turn_index).unwrap_or(u64::MAX); + splitmix64(seed ^ session.rotate_left(17) ^ turn.rotate_left(39)) +} + +fn splitmix64(mut value: u64) -> u64 { + value = value.wrapping_add(0x9E37_79B9_7F4A_7C15); + value = (value ^ (value >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + value = (value ^ (value >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + value ^ (value >> 31) +} diff --git a/crates/agentic-server/benches/client/response_provider/report.rs b/crates/agentic-server/benches/client/response_provider/report.rs new file mode 100644 index 00000000..ac068b88 --- /dev/null +++ b/crates/agentic-server/benches/client/response_provider/report.rs @@ -0,0 +1,493 @@ +use std::collections::HashMap; +use std::fmt::Write as _; +use std::path::Path; + +use crate::types::{ + Comparison, DepthSummaryRow, Distribution, ProviderSpec, ProviderSummary, RunReport, SessionResult, Transport, + TurnResult, Workload, +}; + +pub const ACCURACY_NOTE: &str = "Task correctness and tool-call accuracy are correctness/regression checks against \ +the same underlying model, not a performance comparison: a gateway that owns conversation state is expected to \ +present the model with equivalent context, so its accuracy should match direct vLLM. A gap that widens with turn \ +depth indicates a rehydration bug, not a capability advantage."; + +pub fn summarize( + providers: &[ProviderSpec], + sessions: &[SessionResult], + sessions_per_provider: usize, + requests_per_session: usize, +) -> Vec { + providers + .iter() + .map(|provider| summarize_provider(provider, sessions, sessions_per_provider, requests_per_session)) + .collect() +} + +fn summarize_provider( + provider: &ProviderSpec, + sessions: &[SessionResult], + sessions_per_provider: usize, + requests_per_session: usize, +) -> ProviderSummary { + let provider_sessions: Vec<&SessionResult> = sessions + .iter() + .filter(|session| session.provider == provider.id) + .collect(); + let turns: Vec<&TurnResult> = provider_sessions + .iter() + .flat_map(|session| session.turns.iter()) + .collect(); + let successful: Vec<&TurnResult> = turns.iter().copied().filter(|turn| turn.success).collect(); + let attempted_turns = turns.iter().filter(|turn| turn.attempted).count(); + let successful_turns = successful.len(); + let timed_out_turns = turns.iter().filter(|turn| turn.timed_out).count(); + let transport_fallback_turns = turns.iter().filter(|turn| turn.transport_fallback).count(); + let tool_compliant_turns = turns + .iter() + .filter(|turn| turn.tool_calls_completed > 0 && turn.tool_calls_failed == 0) + .count(); + let task_correct_turns = turns.iter().filter(|turn| turn.task_correct).count(); + let tool_call_correct_turns = turns.iter().filter(|turn| turn.tool_call_correct).count(); + let planned_turns = sessions_per_provider.saturating_mul(requests_per_session); + let provider_wall_clock_ms = provider_sessions + .iter() + .map(|session| session.elapsed_ms) + .max() + .unwrap_or_default(); + let denominator = usize_as_f64(planned_turns.max(1)); + + let total_turn_input_tokens = successful + .iter() + .filter_map(|turn| turn.turn_usage.as_ref()) + .map(|usage| usage.input_tokens) + .sum(); + let total_turn_cached_input_tokens = successful + .iter() + .filter_map(|turn| turn.turn_usage.as_ref()) + .map(|usage| usage.cached_input_tokens) + .sum(); + let total_turn_output_tokens: i64 = successful + .iter() + .filter_map(|turn| turn.turn_usage.as_ref()) + .map(|usage| usage.output_tokens) + .sum(); + let total_turn_reasoning_output_tokens = successful + .iter() + .filter_map(|turn| turn.turn_usage.as_ref()) + .map(|usage| usage.reasoning_output_tokens) + .sum(); + let total_latency_ms: u64 = successful.iter().map(|turn| turn.end_to_end_latency_ms).sum(); + let aggregate_effective_output_tokens_per_second = + (total_latency_ms > 0).then_some(i64_as_f64(total_turn_output_tokens) * 1_000.0 / u64_as_f64(total_latency_ms)); + + ProviderSummary { + provider: provider.id.clone(), + transport: provider.transport, + planned_turns, + attempted_turns, + successful_turns, + timed_out_turns, + transport_fallback_turns, + tool_compliant_turns, + task_correct_turns, + tool_call_correct_turns, + success_rate: usize_as_f64(successful_turns) / denominator, + tool_compliance_rate: usize_as_f64(tool_compliant_turns) / denominator, + task_correctness_rate: usize_as_f64(task_correct_turns) / denominator, + tool_call_accuracy: usize_as_f64(tool_call_correct_turns) / denominator, + provider_wall_clock_ms, + successful_turns_per_second: if provider_wall_clock_ms == 0 { + 0.0 + } else { + usize_as_f64(successful_turns) * 1_000.0 / u64_as_f64(provider_wall_clock_ms) + }, + end_to_end_latency_ms: distribution(successful.iter().map(|turn| u64_as_f64(turn.end_to_end_latency_ms))), + time_to_first_output_event_ms: distribution( + successful + .iter() + .filter_map(|turn| turn.time_to_first_output_event_ms.map(u64_as_f64)), + ), + ttft_ms: distribution(successful.iter().filter_map(|turn| turn.ttft_ms.map(u64_as_f64))), + time_to_first_tool_call_ms: distribution( + successful + .iter() + .filter_map(|turn| turn.time_to_first_tool_call_ms.map(u64_as_f64)), + ), + mean_tool_duration_ms: distribution(successful.iter().filter_map(|turn| turn.mean_tool_duration_ms)), + continuation_round_latency_ms: distribution( + successful + .iter() + .flat_map(|turn| turn.continuation_round_latencies_ms.iter().copied().map(u64_as_f64)), + ), + request_bytes: distribution(successful.iter().map(|turn| u64_as_f64(turn.request_bytes))), + response_bytes: distribution(successful.iter().map(|turn| u64_as_f64(turn.response_bytes))), + total_turn_input_tokens, + total_turn_cached_input_tokens, + total_turn_output_tokens, + total_turn_reasoning_output_tokens, + aggregate_effective_output_tokens_per_second, + } +} + +pub fn compare(sessions: &[SessionResult]) -> Option { + let turns: Vec<&TurnResult> = sessions.iter().flat_map(|session| session.turns.iter()).collect(); + let mut agentic = HashMap::new(); + let mut vllm = HashMap::new(); + for turn in turns.into_iter().filter(|turn| turn.success) { + let key = (turn.session_index, turn.turn_index); + match turn.provider.as_str() { + "agentic-api" => { + agentic.insert(key, turn); + } + "vllm" => { + vllm.insert(key, turn); + } + _ => {} + } + } + + let mut latency_deltas = Vec::new(); + let mut latency_ratios = Vec::new(); + let mut first_output_deltas = Vec::new(); + for (key, agentic_turn) in agentic { + let Some(vllm_turn) = vllm.get(&key) else { + continue; + }; + latency_deltas + .push(u64_as_f64(agentic_turn.end_to_end_latency_ms) - u64_as_f64(vllm_turn.end_to_end_latency_ms)); + if vllm_turn.end_to_end_latency_ms > 0 { + latency_ratios + .push(u64_as_f64(agentic_turn.end_to_end_latency_ms) / u64_as_f64(vllm_turn.end_to_end_latency_ms)); + } + if let (Some(agentic_first), Some(vllm_first)) = ( + agentic_turn.time_to_first_output_event_ms, + vllm_turn.time_to_first_output_event_ms, + ) { + first_output_deltas.push(u64_as_f64(agentic_first) - u64_as_f64(vllm_first)); + } + } + if latency_deltas.is_empty() { + return None; + } + + Some(Comparison { + paired_successful_turns: latency_deltas.len(), + median_agentic_minus_vllm_latency_ms: median(latency_deltas), + median_agentic_over_vllm_latency_ratio: median(latency_ratios), + median_agentic_minus_vllm_first_output_ms: median(first_output_deltas), + }) +} + +pub async fn write_reports(output_dir: &Path, report: &RunReport) -> Result<(), std::io::Error> { + let json = serde_json::to_vec_pretty(report).map_err(std::io::Error::other)?; + tokio::fs::write(output_dir.join("run.json"), json).await?; + tokio::fs::write(output_dir.join("turns.csv"), turns_csv(&report.sessions)).await?; + tokio::fs::write(output_dir.join("summary.md"), markdown_summary(report)).await +} + +pub fn console_summary(summaries: &[ProviderSummary], comparison: Option<&Comparison>) -> String { + let mut output = String::new(); + output.push_str("provider success task correct tool accuracy p50 latency p50 continuation turns/s\n"); + for summary in summaries { + let _ = writeln!( + output, + "{:<17} {:>6.1}% {:>10.1}% {:>10.1}% {:>9} ms {:>16} ms {:>7.2}", + summary.provider, + summary.success_rate * 100.0, + summary.task_correctness_rate * 100.0, + summary.tool_call_accuracy * 100.0, + optional_number(summary.end_to_end_latency_ms.p50), + optional_number(summary.continuation_round_latency_ms.p50), + summary.successful_turns_per_second, + ); + } + if let Some(comparison) = comparison { + let _ = writeln!( + output, + "paired turns: {}; median Agentic-vLLM latency: {} ms; median ratio: {}", + comparison.paired_successful_turns, + optional_number(comparison.median_agentic_minus_vllm_latency_ms), + optional_decimal(comparison.median_agentic_over_vllm_latency_ratio), + ); + } + output +} + +fn markdown_summary(report: &RunReport) -> String { + let mut output = format!( + "# Responses provider benchmark: {}\n\n\ + Streaming TTFT is measured at the first `response.output_text.delta`; JSON runs report it as `n/a`. \ + Request/response bytes are the client-visible wire payload per turn: flat across turn depth for a provider \ + that rehydrates history server-side, growing for a provider the client must resend full history to.\n\n\ + | Provider | Transport | Success | Task correctness | Tool-call accuracy | p50 latency (ms) | p95 latency (ms) | p50 continuation round (ms) | p50 first output (ms) | p50 TTFT (ms) | Turns/s | p50 request bytes | p50 response bytes |\n\ + | --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n", + workload_name(report.config.workload), + ); + for summary in &report.summaries { + let _ = writeln!( + output, + "| {} | {} | {:.1}% | {:.1}% | {:.1}% | {} | {} | {} | {} | {} | {:.2} | {} | {} |", + summary.provider, + transport_name(summary.transport), + summary.success_rate * 100.0, + summary.task_correctness_rate * 100.0, + summary.tool_call_accuracy * 100.0, + optional_number(summary.end_to_end_latency_ms.p50), + optional_number(summary.end_to_end_latency_ms.p95), + optional_number(summary.continuation_round_latency_ms.p50), + optional_number(summary.time_to_first_output_event_ms.p50), + optional_number(summary.ttft_ms.p50), + summary.successful_turns_per_second, + optional_number(summary.request_bytes.p50), + optional_number(summary.response_bytes.p50), + ); + } + if let Some(comparison) = &report.comparison { + let _ = writeln!( + output, + "\nPaired successful turns: {}. Median Agentic API minus vLLM latency: {} ms. Median Agentic API / vLLM latency ratio: {}.", + comparison.paired_successful_turns, + optional_number(comparison.median_agentic_minus_vllm_latency_ms), + optional_decimal(comparison.median_agentic_over_vllm_latency_ratio), + ); + } + let _ = writeln!(output, "\n> **Correctness, not competition.** {}", report.accuracy_note); + if report.config.workload == Workload::ToolCall { + output.push_str("\n## Tool-call compatibility (pass/fail by case)\n\n"); + output.push_str(&tool_call_matrix(&report.summaries, &report.sessions)); + } + output +} + +/// Per-BFCL-case pass/fail across providers, in place of a single aggregate accuracy percentage. +/// This is the tool-shape compatibility view: whether the gateway preserved each individual +/// tool-call case through translation, not how often it happens to be right on average. +fn tool_call_matrix(summaries: &[ProviderSummary], sessions: &[SessionResult]) -> String { + let providers: Vec<&str> = summaries.iter().map(|summary| summary.provider.as_str()).collect(); + let mut cases: Vec<(String, HashMap<&str, bool>)> = Vec::new(); + let mut index_by_case: HashMap = HashMap::new(); + for turn in sessions.iter().flat_map(|session| &session.turns) { + let case_id = turn.source_id.clone().unwrap_or_else(|| turn.prompt_id.clone()); + let index = *index_by_case.entry(case_id.clone()).or_insert_with(|| { + cases.push((case_id, HashMap::new())); + cases.len() - 1 + }); + cases[index].1.insert(turn.provider.as_str(), turn.tool_call_correct); + } + cases.sort_by(|left, right| left.0.cmp(&right.0)); + + let mut output = format!("| Case | {} |\n", providers.join(" | ")); + let _ = writeln!(output, "| --- |{}", " ---: |".repeat(providers.len())); + for (case_id, results) in &cases { + let cells = providers + .iter() + .map(|provider| match results.get(provider) { + Some(true) => "✅", + Some(false) => "❌", + None => "n/a", + }) + .collect::>() + .join(" | "); + let _ = writeln!(output, "| {case_id} | {cells} |"); + } + output +} + +/// Roll up independently-run fixed-depth session batches into one table per provider, keyed by +/// turn depth. Each depth is its own complete batch of sessions rather than a bucket sliced out of +/// one long run, so sample counts stay even across depths instead of thinning out at the tail. +#[must_use] +pub fn depth_rollup(depth_reports: &[(usize, RunReport)]) -> Vec { + let mut rows = Vec::new(); + for (depth, report) in depth_reports { + for summary in &report.summaries { + rows.push(DepthSummaryRow { + depth: *depth, + provider: summary.provider.clone(), + transport: summary.transport, + sessions: report.config.sessions_per_provider, + success_rate: summary.success_rate, + task_correctness_rate: summary.task_correctness_rate, + p50_latency_ms: summary.end_to_end_latency_ms.p50, + p50_request_bytes: summary.request_bytes.p50, + p50_response_bytes: summary.response_bytes.p50, + total_request_bytes: sum_u64(&summary.provider, &report.sessions, |turn| turn.request_bytes), + total_response_bytes: sum_u64(&summary.provider, &report.sessions, |turn| turn.response_bytes), + }); + } + } + rows.sort_by(|left, right| left.provider.cmp(&right.provider).then(left.depth.cmp(&right.depth))); + rows +} + +fn sum_u64(provider: &str, sessions: &[SessionResult], field: impl Fn(&TurnResult) -> u64) -> u64 { + sessions + .iter() + .filter(|session| session.provider == provider) + .flat_map(|session| session.turns.iter()) + .filter(|turn| turn.success) + .map(field) + .sum() +} + +#[must_use] +pub fn depth_rollup_markdown(rows: &[DepthSummaryRow]) -> String { + let mut output = String::from( + "# State-scaling: request bytes and accuracy versus turn depth\n\n\ + Each depth below ran as its own independent batch of sessions (not a bucket sliced out of one long run), \ + so sample counts are even across depths instead of thinning out at the tail. Accuracy is a \ + correctness/regression check, not a competition: it should not diverge between providers, and a widening \ + gap at deeper turns points to a rehydration bug.\n\n\ + | Provider | Depth | Sessions | Success | Task correctness | p50 latency (ms) | p50 request bytes | p50 response bytes | total request bytes | total response bytes |\n\ + | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |\n", + ); + for row in rows { + let _ = writeln!( + output, + "| {} | {} | {} | {:.1}% | {:.1}% | {} | {} | {} | {} | {} |", + row.provider, + row.depth, + row.sessions, + row.success_rate * 100.0, + row.task_correctness_rate * 100.0, + optional_number(row.p50_latency_ms), + optional_number(row.p50_request_bytes), + optional_number(row.p50_response_bytes), + row.total_request_bytes, + row.total_response_bytes, + ); + } + output +} + +fn turns_csv(sessions: &[SessionResult]) -> String { + let mut output = String::from( + "provider,transport,workload,session,turn,prompt_id,source_id,success,task_correct,tool_call_correct,timed_out,transport_fallback,latency_ms,first_output_ms,ttft_ms,first_tool_ms,initial_model_round_ms,continuation_round_latencies_ms,tool_duration_ms,tool_calls_completed,tool_calls_failed,observed_tool_calls,expected_tool_calls,request_bytes,response_bytes,input_tokens,cached_input_tokens,output_tokens,reasoning_output_tokens,output_tokens_per_second,response_id,raw_jsonl_path,error_log_path,errors\n", + ); + for turn in sessions.iter().flat_map(|session| &session.turns) { + let usage = turn.turn_usage.as_ref(); + let fields = [ + turn.provider.clone(), + transport_name(turn.transport).to_owned(), + workload_name(turn.workload).to_owned(), + turn.session_index.to_string(), + turn.turn_index.to_string(), + turn.prompt_id.clone(), + turn.source_id.clone().unwrap_or_default(), + turn.success.to_string(), + turn.task_correct.to_string(), + turn.tool_call_correct.to_string(), + turn.timed_out.to_string(), + turn.transport_fallback.to_string(), + turn.end_to_end_latency_ms.to_string(), + optional_u64(turn.time_to_first_output_event_ms), + optional_u64(turn.ttft_ms), + optional_u64(turn.time_to_first_tool_call_ms), + optional_u64(turn.initial_model_round_ms), + serde_json::to_string(&turn.continuation_round_latencies_ms).unwrap_or_default(), + optional_decimal(turn.mean_tool_duration_ms), + turn.tool_calls_completed.to_string(), + turn.tool_calls_failed.to_string(), + serde_json::to_string(&turn.observed_tool_calls).unwrap_or_default(), + serde_json::to_string(&turn.expected_tool_calls).unwrap_or_default(), + turn.request_bytes.to_string(), + turn.response_bytes.to_string(), + usage.map_or_else(String::new, |value| value.input_tokens.to_string()), + usage.map_or_else(String::new, |value| value.cached_input_tokens.to_string()), + usage.map_or_else(String::new, |value| value.output_tokens.to_string()), + usage.map_or_else(String::new, |value| value.reasoning_output_tokens.to_string()), + optional_decimal(turn.effective_output_tokens_per_second), + turn.response_id.clone().unwrap_or_default(), + turn.raw_jsonl_path.clone(), + turn.error_log_path.clone(), + turn.errors.join(" | "), + ]; + output.push_str(&fields.map(|field| csv_escape(&field)).join(",")); + output.push('\n'); + } + output +} + +fn distribution(values: impl Iterator) -> Distribution { + let mut values: Vec = values.filter(|value| value.is_finite()).collect(); + values.sort_by(f64::total_cmp); + Distribution { + count: values.len(), + mean: (!values.is_empty()).then(|| values.iter().sum::() / usize_as_f64(values.len())), + p50: percentile(&values, 0.50), + p95: percentile(&values, 0.95), + p99: percentile(&values, 0.99), + min: values.first().copied(), + max: values.last().copied(), + } +} + +#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] +fn percentile(sorted: &[f64], quantile: f64) -> Option { + if sorted.is_empty() { + return None; + } + let rank = quantile * usize_as_f64(sorted.len().saturating_sub(1)); + let lower = rank.floor() as usize; + let upper = rank.ceil() as usize; + let fraction = rank - usize_as_f64(lower); + Some(sorted[lower] + (sorted[upper] - sorted[lower]) * fraction) +} + +fn median(values: Vec) -> Option { + let mut values: Vec = values.into_iter().filter(|value| value.is_finite()).collect(); + values.sort_by(f64::total_cmp); + percentile(&values, 0.5) +} + +fn csv_escape(value: &str) -> String { + if value.contains([',', '"', '\n', '\r']) { + format!("\"{}\"", value.replace('"', "\"\"")) + } else { + value.to_owned() + } +} + +fn transport_name(transport: Transport) -> &'static str { + match transport { + Transport::Websocket => "responses_websocket", + Transport::HttpSse => "responses_http_sse", + Transport::HttpJson => "responses_http_json", + } +} + +fn workload_name(workload: crate::types::Workload) -> &'static str { + match workload { + crate::types::Workload::Transport => "transport", + crate::types::Workload::ToolCall => "tool_call", + crate::types::Workload::HistoryRehydration => "history_rehydration", + } +} + +fn optional_number(value: Option) -> String { + value.map_or_else(|| "n/a".to_owned(), |number| format!("{number:.0}")) +} + +fn optional_decimal(value: Option) -> String { + value.map_or_else(String::new, |number| format!("{number:.3}")) +} + +fn optional_u64(value: Option) -> String { + value.map_or_else(String::new, |number| number.to_string()) +} + +#[allow(clippy::cast_precision_loss)] +fn i64_as_f64(value: i64) -> f64 { + value as f64 +} + +#[allow(clippy::cast_precision_loss)] +fn u64_as_f64(value: u64) -> f64 { + value as f64 +} + +#[allow(clippy::cast_precision_loss)] +fn usize_as_f64(value: usize) -> f64 { + value as f64 +} diff --git a/crates/agentic-server/benches/client/response_provider/runner.rs b/crates/agentic-server/benches/client/response_provider/runner.rs new file mode 100644 index 00000000..e15ca867 --- /dev/null +++ b/crates/agentic-server/benches/client/response_provider/runner.rs @@ -0,0 +1,1134 @@ +use std::collections::BTreeMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use futures::{SinkExt, StreamExt}; +use reqwest::Client; +use serde_json::{Map, Value, json}; +use thiserror::Error; +use tokio::net::TcpStream; +use tokio::sync::{Barrier, Mutex}; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::{MaybeTlsStream, WebSocketStream, connect_async}; + +use crate::prompts::transport_marker; +use crate::types::{ + ExpectedToolCall, ObservedToolCall, PromptSpec, ProviderSpec, SessionResult, ToolDefinition, Transport, + TurnExpectation, TurnResult, Usage, +}; + +type WsStream = WebSocketStream>; + +#[derive(Clone, Debug)] +pub struct RunnerConfig { + pub model: String, + pub output_dir: PathBuf, + pub timeout: Duration, + pub live_jsonl: bool, +} + +#[derive(Debug, Error)] +enum RunnerError { + #[error("failed to access {path}: {source}")] + Io { + path: PathBuf, + #[source] + source: std::io::Error, + }, + #[error("HTTP request failed: {0}")] + Http(#[from] reqwest::Error), + #[error("WebSocket request failed: {0}")] + WebSocket(#[from] tokio_tungstenite::tungstenite::Error), + #[error("invalid Responses endpoint {0}")] + InvalidEndpoint(String), + #[error("provider returned HTTP {status}: {body}")] + HttpStatus { status: reqwest::StatusCode, body: String }, + #[error("invalid provider event JSON: {0}")] + Json(#[from] serde_json::Error), + #[error("Responses protocol error: {0}")] + Protocol(String), +} + +#[derive(Clone, Debug)] +struct TimedEvent { + elapsed_ms: u64, + value: Value, +} + +#[derive(Debug)] +struct ModelResponse { + events: Vec, + response: Value, + round_latency_ms: u64, + first_output_ms: Option, + first_output_event_type: Option, + first_tool_call_ms: Option, + first_text_delta_ms: Option, + request_bytes: u64, + response_bytes: u64, + errors: Vec, +} + +struct LiveEventContext<'a> { + enabled: bool, + provider: &'a ProviderSpec, + session_index: usize, + turn_index: usize, + output_lock: &'a Mutex<()>, +} + +enum TransportClient { + WebSocket(Box), + HttpSse { client: Client, url: String }, + HttpJson { client: Client, url: String }, +} + +#[derive(Default)] +struct ConversationState { + previous_response_id: Option, + replay_items: Vec, +} + +struct TurnExecution { + events: Vec, + response_id: Option, + saw_completed: bool, + observed_tool_calls: Vec, + tool_calls_failed: usize, + tool_output_marker_found: bool, + final_answer: Option, + first_output_ms: Option, + first_output_event_type: Option, + first_tool_call_ms: Option, + first_text_delta_ms: Option, + round_latencies_ms: Vec, + request_bytes: u64, + response_bytes: u64, + usage: Usage, + errors: Vec, +} + +pub async fn run_session( + config: Arc, + provider: ProviderSpec, + session_index: usize, + prompts: Vec, + start_barrier: Arc, + live_output_lock: Arc>, +) -> SessionResult { + let session_started = Instant::now(); + let mut result = SessionResult { + provider: provider.id.clone(), + session_index, + elapsed_ms: 0, + fatal_error: None, + turns: Vec::with_capacity(prompts.len()), + }; + + let mut client = match TransportClient::connect(&provider).await { + Ok(client) => client, + Err(error) => { + result.fatal_error = Some(error.to_string()); + result.elapsed_ms = millis(session_started.elapsed()); + start_barrier.wait().await; + return result; + } + }; + let mut state = ConversationState::default(); + start_barrier.wait().await; + let workload_started = Instant::now(); + + for prompt in prompts { + let turn = run_turn( + &config, + &provider, + session_index, + &prompt, + &mut client, + &mut state, + &live_output_lock, + ) + .await; + let can_continue = turn.saw_turn_completed && !turn.timed_out && turn.error_events == 0; + if matches!( + prompt.workload, + crate::types::Workload::Transport | crate::types::Workload::ToolCall + ) { + state = ConversationState::default(); + } + eprintln!( + "[{provider_id} s{session_index:03} t{turn_index:03}] success={success} latency={latency}ms tools={tools}", + provider_id = provider.id, + turn_index = prompt.turn_index, + success = turn.success, + latency = turn.end_to_end_latency_ms, + tools = turn.tool_calls_completed, + ); + result.turns.push(turn); + if !can_continue { + result.fatal_error = Some(format!( + "turn {} did not leave a resumable Responses session", + prompt.turn_index + )); + break; + } + } + + result.elapsed_ms = millis(workload_started.elapsed()); + result +} + +#[allow(clippy::too_many_arguments, clippy::too_many_lines)] +async fn run_turn( + config: &RunnerConfig, + provider: &ProviderSpec, + session_index: usize, + prompt: &PromptSpec, + client: &mut TransportClient, + state: &mut ConversationState, + live_output_lock: &Mutex<()>, +) -> TurnResult { + let started = Instant::now(); + let context = LiveEventContext { + enabled: config.live_jsonl, + provider, + session_index, + turn_index: prompt.turn_index, + output_lock: live_output_lock, + }; + let execution = tokio::time::timeout( + config.timeout, + execute_turn(config, provider, prompt, client, state, started, &context), + ) + .await; + + let (execution, timed_out) = match execution { + Ok(Ok(execution)) => (execution, false), + Ok(Err(error)) => (failed_execution(error.to_string()), false), + Err(_) => ( + failed_execution(format!("turn timed out after {} ms", millis(config.timeout))), + true, + ), + }; + let elapsed_ms = millis(started.elapsed()); + let event_dir = config + .output_dir + .join("events") + .join(&provider.id) + .join(format!("session-{session_index:03}")); + let raw_path = event_dir.join(format!("turn-{:03}.responses.jsonl", prompt.turn_index)); + let timestamped_path = event_dir.join(format!("turn-{:03}.timestamped.jsonl", prompt.turn_index)); + let error_path = event_dir.join(format!("turn-{:03}.errors.log", prompt.turn_index)); + if let Err(error) = write_event_files( + &raw_path, + ×tamped_path, + &error_path, + &execution.events, + &execution.errors, + ) + .await + { + eprintln!("warning: failed to write turn event files: {error}"); + } + + let expected_marker = prompt.expectation.expected_marker(); + let final_answer_marker_found = expected_marker.is_some_and(|marker| { + execution + .final_answer + .as_deref() + .is_some_and(|answer| answer.contains(marker)) + }); + let exact_final_answer = expected_marker.is_some_and(|marker| { + execution + .final_answer + .as_deref() + .is_some_and(|answer| answer.trim().trim_matches('`') == marker) + }); + let tool_call_correct = match &prompt.expectation { + TurnExpectation::Marker { .. } => false, + TurnExpectation::ToolCalls { .. } => { + tool_calls_match(prompt.expectation.expected_tool_calls(), &execution.observed_tool_calls) + } + TurnExpectation::Transport { .. } => { + tool_calls_match_ordered(prompt.expectation.expected_tool_calls(), &execution.observed_tool_calls) + } + }; + let task_correct = match &prompt.expectation { + TurnExpectation::Marker { .. } => exact_final_answer, + TurnExpectation::ToolCalls { .. } => tool_call_correct, + TurnExpectation::Transport { .. } => { + tool_call_correct && execution.tool_output_marker_found && exact_final_answer + } + }; + let success = !timed_out + && execution.saw_completed + && execution.errors.is_empty() + && execution.tool_calls_failed == 0 + && task_correct; + let tool_durations = execution.observed_tool_calls.iter().filter_map(|call| { + call.started_at_ms + .map(|started_at| call.completed_at_ms.saturating_sub(started_at)) + }); + let tool_durations = tool_durations.collect::>(); + let mean_tool_duration_ms = (!tool_durations.is_empty()).then(|| { + tool_durations.iter().map(|value| u64_as_f64(*value)).sum::() / usize_as_f64(tool_durations.len()) + }); + let output_tokens = execution.usage.output_tokens; + let effective_output_tokens_per_second = + (elapsed_ms > 0).then_some(i64_as_f64(output_tokens) * 1_000.0 / u64_as_f64(elapsed_ms)); + let initial_model_round_ms = execution.round_latencies_ms.first().copied(); + let continuation_round_latencies_ms = execution.round_latencies_ms.iter().skip(1).copied().collect(); + + TurnResult { + provider: provider.id.clone(), + transport: provider.transport, + workload: prompt.workload, + session_index, + turn_index: prompt.turn_index, + prompt_id: prompt.prompt_id.clone(), + source_id: prompt.source_id.clone(), + expected_marker: expected_marker.map(str::to_owned), + response_id: execution.response_id, + attempted: true, + success, + task_correct, + tool_call_correct, + timed_out, + transport_fallback: false, + saw_turn_completed: execution.saw_completed, + invalid_json_lines: 0, + error_events: execution.errors.len(), + tool_calls_started: execution.observed_tool_calls.len(), + tool_calls_completed: execution.observed_tool_calls.len(), + tool_calls_failed: execution.tool_calls_failed, + observed_tool_calls: execution.observed_tool_calls, + expected_tool_calls: prompt.expectation.expected_tool_calls().to_vec(), + tool_output_marker_found: execution.tool_output_marker_found, + final_answer_marker_found, + exact_final_answer, + time_to_first_output_event_ms: execution.first_output_ms, + first_output_event_type: execution.first_output_event_type, + time_to_first_tool_call_ms: execution.first_tool_call_ms, + ttft_ms: execution.first_text_delta_ms, + mean_tool_duration_ms, + initial_model_round_ms, + continuation_round_latencies_ms, + end_to_end_latency_ms: elapsed_ms, + request_bytes: execution.request_bytes, + response_bytes: execution.response_bytes, + turn_usage: Some(execution.usage), + effective_output_tokens_per_second, + raw_jsonl_path: relative_path(&config.output_dir, &raw_path), + timestamped_jsonl_path: relative_path(&config.output_dir, ×tamped_path), + error_log_path: relative_path(&config.output_dir, &error_path), + errors: execution.errors, + } +} + +fn failed_execution(message: String) -> TurnExecution { + TurnExecution { + events: Vec::new(), + response_id: None, + saw_completed: false, + observed_tool_calls: Vec::new(), + tool_calls_failed: 0, + tool_output_marker_found: false, + final_answer: None, + first_output_ms: None, + first_output_event_type: None, + first_tool_call_ms: None, + first_text_delta_ms: None, + round_latencies_ms: Vec::new(), + request_bytes: 0, + response_bytes: 0, + usage: Usage::default(), + errors: vec![message], + } +} + +#[allow(clippy::too_many_arguments, clippy::too_many_lines)] +async fn execute_turn( + config: &RunnerConfig, + provider: &ProviderSpec, + prompt: &PromptSpec, + client: &mut TransportClient, + state: &mut ConversationState, + turn_started: Instant, + live_context: &LiveEventContext<'_>, +) -> Result { + let gateway_managed_history = provider.id.starts_with("agentic-api"); + let mut new_items = vec![json!({ + "type": "message", + "role": "user", + "content": prompt.prompt, + })]; + let max_rounds = match &prompt.expectation { + TurnExpectation::Transport { calls, .. } => calls.len().saturating_add(1), + TurnExpectation::Marker { .. } | TurnExpectation::ToolCalls { .. } => 1, + }; + let mut events = Vec::new(); + let mut response_id = None; + let mut saw_completed = true; + let mut observed_tool_calls = Vec::new(); + let mut tool_calls_failed = 0; + let mut tool_output_marker_found = false; + let mut final_answer = None; + let mut first_output_ms = None; + let mut first_output_event_type = None; + let mut first_tool_call_ms = None; + let mut first_text_delta_ms = None; + let mut round_latencies_ms = Vec::new(); + let mut request_bytes = 0u64; + let mut response_bytes = 0u64; + let mut usage = Usage::default(); + let mut errors = Vec::new(); + + for _ in 0..max_rounds { + let input = if gateway_managed_history { + new_items.clone() + } else { + state + .replay_items + .iter() + .cloned() + .chain(new_items.iter().cloned()) + .collect() + }; + let body = response_request( + &config.model, + input, + &prompt.tools, + client.is_streaming(), + gateway_managed_history, + gateway_managed_history + .then(|| state.previous_response_id.clone()) + .flatten(), + ); + let model_response = client.request(&body, turn_started, live_context).await?; + first_output_ms = first_output_ms.or(model_response.first_output_ms); + if first_output_event_type.is_none() { + first_output_event_type.clone_from(&model_response.first_output_event_type); + } + first_tool_call_ms = first_tool_call_ms.or(model_response.first_tool_call_ms); + first_text_delta_ms = first_text_delta_ms.or(model_response.first_text_delta_ms); + round_latencies_ms.push(model_response.round_latency_ms); + request_bytes = request_bytes.saturating_add(model_response.request_bytes); + response_bytes = response_bytes.saturating_add(model_response.response_bytes); + errors.extend(model_response.errors.clone()); + events.extend(model_response.events.clone()); + + let current_response_id = model_response + .response + .get("id") + .and_then(Value::as_str) + .map(str::to_owned); + let status = model_response + .response + .get("status") + .and_then(Value::as_str) + .unwrap_or_default(); + saw_completed &= status == "completed"; + let output = model_response + .response + .get("output") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + add_usage(&mut usage, &response_usage(&model_response.response)); + final_answer = response_text(&model_response.response).or(final_answer); + let response_completed_ms = model_response + .events + .last() + .map_or(model_response.round_latency_ms, |event| event.elapsed_ms); + let calls = response_tool_calls(&output, &model_response.events, response_completed_ms); + + if gateway_managed_history { + state.previous_response_id.clone_from(¤t_response_id); + } else { + state.replay_items.extend(new_items); + state.replay_items.extend(output.clone()); + } + response_id = current_response_id; + observed_tool_calls.extend(calls.iter().map(|call| call.observed.clone())); + + if !matches!(prompt.expectation, TurnExpectation::Transport { .. }) || calls.is_empty() { + break; + } + + new_items = calls + .into_iter() + .map(|call| { + let output = match execute_transport_call(&call.observed) { + Ok(output) => { + if prompt + .expectation + .expected_marker() + .is_some_and(|marker| output.contains(marker)) + { + tool_output_marker_found = true; + } + output + } + Err(error) => { + tool_calls_failed += 1; + errors.push(error.clone()); + format!("BENCHMARK_TOOL_ERROR: {error}") + } + }; + json!({ + "type": "function_call_output", + "call_id": call.call_id, + "output": output, + }) + }) + .collect(); + } + + Ok(TurnExecution { + events, + response_id, + saw_completed, + observed_tool_calls, + tool_calls_failed, + tool_output_marker_found, + final_answer, + first_output_ms, + first_output_event_type, + first_tool_call_ms, + first_text_delta_ms, + round_latencies_ms, + request_bytes, + response_bytes, + usage, + errors, + }) +} + +fn response_request( + model: &str, + input: Vec, + tools: &[ToolDefinition], + stream: bool, + store: bool, + previous_response_id: Option, +) -> Value { + let mut body = Map::from_iter([ + ("model".to_owned(), json!(model)), + ("input".to_owned(), Value::Array(input)), + ("stream".to_owned(), json!(stream)), + ("store".to_owned(), json!(store)), + ]); + if let Some(previous_response_id) = previous_response_id { + body.insert("previous_response_id".to_owned(), json!(previous_response_id)); + } + if !tools.is_empty() { + body.insert( + "tools".to_owned(), + Value::Array( + tools + .iter() + .map(|tool| { + json!({ + "type": "function", + "name": tool.name, + "description": tool.description, + "parameters": tool.parameters, + }) + }) + .collect(), + ), + ); + body.insert("tool_choice".to_owned(), json!("auto")); + body.insert("parallel_tool_calls".to_owned(), json!(false)); + } + Value::Object(body) +} + +impl TransportClient { + async fn connect(provider: &ProviderSpec) -> Result { + let endpoint = responses_endpoint(&provider.base_url); + match provider.transport { + Transport::Websocket => { + let websocket_url = websocket_url(&endpoint)?; + let (socket, _) = connect_async(websocket_url.as_str()).await?; + Ok(Self::WebSocket(Box::new(socket))) + } + Transport::HttpSse => Ok(Self::HttpSse { + client: Client::new(), + url: endpoint, + }), + Transport::HttpJson => Ok(Self::HttpJson { + client: Client::new(), + url: endpoint, + }), + } + } + + const fn is_streaming(&self) -> bool { + matches!(self, Self::WebSocket(_) | Self::HttpSse { .. }) + } + + async fn request( + &mut self, + body: &Value, + turn_started: Instant, + live_context: &LiveEventContext<'_>, + ) -> Result { + match self { + Self::WebSocket(socket) => websocket_request(socket, body, turn_started, live_context).await, + Self::HttpSse { client, url } => http_sse_request(client, url, body, turn_started, live_context).await, + Self::HttpJson { client, url } => http_json_request(client, url, body, turn_started, live_context).await, + } + } +} + +async fn websocket_request( + socket: &mut WsStream, + body: &Value, + turn_started: Instant, + live_context: &LiveEventContext<'_>, +) -> Result { + let round_started = Instant::now(); + let mut request = body.clone(); + request + .as_object_mut() + .ok_or_else(|| RunnerError::Protocol("request body must be an object".to_owned()))? + .insert("type".to_owned(), json!("response.create")); + let request_text = request.to_string(); + let request_bytes = byte_len(request_text.len()); + socket.send(Message::Text(request_text.into())).await?; + let mut events = Vec::new(); + let mut response_bytes = 0u64; + loop { + let message = socket + .next() + .await + .ok_or_else(|| RunnerError::Protocol("WebSocket closed before a terminal event".to_owned()))??; + match message { + Message::Text(text) => { + response_bytes = response_bytes.saturating_add(byte_len(text.len())); + let value = serde_json::from_str::(&text)?; + let event = TimedEvent { + elapsed_ms: millis(turn_started.elapsed()), + value, + }; + emit_live_event(live_context, &event).await; + let terminal = is_terminal_event(&event.value); + events.push(event); + if terminal { + break; + } + } + Message::Ping(payload) => socket.send(Message::Pong(payload)).await?, + Message::Pong(_) | Message::Frame(_) => {} + Message::Close(frame) => { + return Err(RunnerError::Protocol(format!( + "WebSocket closed before a terminal event: {frame:?}" + ))); + } + Message::Binary(_) => { + return Err(RunnerError::Protocol( + "WebSocket returned a binary frame instead of JSON text".to_owned(), + )); + } + } + } + model_response_from_stream(events, millis(round_started.elapsed()), request_bytes, response_bytes) +} + +async fn http_sse_request( + client: &Client, + url: &str, + body: &Value, + turn_started: Instant, + live_context: &LiveEventContext<'_>, +) -> Result { + let round_started = Instant::now(); + let request_body = serde_json::to_vec(body)?; + let request_bytes = byte_len(request_body.len()); + let response = client + .post(url) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body(request_body) + .send() + .await?; + let status = response.status(); + if !status.is_success() { + return Err(RunnerError::HttpStatus { + status, + body: response.text().await.unwrap_or_default(), + }); + } + let mut stream = response.bytes_stream(); + let mut pending = Vec::new(); + let mut events = Vec::new(); + let mut terminal = false; + let mut response_bytes = 0u64; + while let Some(chunk) = stream.next().await { + let chunk = chunk?; + response_bytes = response_bytes.saturating_add(byte_len(chunk.len())); + pending.extend_from_slice(&chunk); + while let Some(frame) = take_sse_frame(&mut pending) { + let Some(value) = parse_sse_frame(&frame)? else { + continue; + }; + let event = TimedEvent { + elapsed_ms: millis(turn_started.elapsed()), + value, + }; + emit_live_event(live_context, &event).await; + terminal |= is_terminal_event(&event.value); + events.push(event); + } + if terminal { + break; + } + } + if !terminal { + return Err(RunnerError::Protocol( + "HTTP/SSE stream ended before a terminal event".to_owned(), + )); + } + model_response_from_stream(events, millis(round_started.elapsed()), request_bytes, response_bytes) +} + +async fn http_json_request( + client: &Client, + url: &str, + body: &Value, + turn_started: Instant, + live_context: &LiveEventContext<'_>, +) -> Result { + let round_started = Instant::now(); + let request_body = serde_json::to_vec(body)?; + let request_bytes = byte_len(request_body.len()); + let response = client + .post(url) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .body(request_body) + .send() + .await?; + let status = response.status(); + if !status.is_success() { + return Err(RunnerError::HttpStatus { + status, + body: response.text().await.unwrap_or_default(), + }); + } + let response_body = response.bytes().await?; + let response_bytes = byte_len(response_body.len()); + let response = serde_json::from_slice::(&response_body)?; + let event = TimedEvent { + elapsed_ms: millis(turn_started.elapsed()), + value: response.clone(), + }; + emit_live_event(live_context, &event).await; + let status = response.get("status").and_then(Value::as_str).unwrap_or_default(); + let errors = if status == "completed" { + Vec::new() + } else { + vec![response_error(&response)] + }; + Ok(ModelResponse { + events: vec![event], + response, + round_latency_ms: millis(round_started.elapsed()), + first_output_ms: None, + first_output_event_type: None, + first_tool_call_ms: None, + first_text_delta_ms: None, + request_bytes, + response_bytes, + errors, + }) +} + +fn model_response_from_stream( + events: Vec, + round_latency_ms: u64, + request_bytes: u64, + response_bytes: u64, +) -> Result { + let terminal = events + .last() + .ok_or_else(|| RunnerError::Protocol("provider returned no Responses events".to_owned()))?; + let event_type = terminal.value.get("type").and_then(Value::as_str).unwrap_or_default(); + let response = terminal + .value + .get("response") + .cloned() + .ok_or_else(|| RunnerError::Protocol(format!("terminal event {event_type:?} has no response object")))?; + let first_output_ms = events + .iter() + .find(|event| is_output_event(&event.value)) + .map(|event| event.elapsed_ms); + let first_output_event_type = events + .iter() + .find(|event| is_output_event(&event.value)) + .and_then(|event| event.value.get("type")) + .and_then(Value::as_str) + .map(str::to_owned); + let first_tool_call_ms = events + .iter() + .find(|event| is_tool_event(&event.value)) + .map(|event| event.elapsed_ms); + let first_text_delta_ms = events + .iter() + .find(|event| event.value.get("type").and_then(Value::as_str) == Some("response.output_text.delta")) + .map(|event| event.elapsed_ms); + let errors = if event_type == "response.completed" && response["status"] == "completed" { + Vec::new() + } else { + vec![response_error(&terminal.value)] + }; + Ok(ModelResponse { + events, + response, + round_latency_ms, + first_output_ms, + first_output_event_type, + first_tool_call_ms, + first_text_delta_ms, + request_bytes, + response_bytes, + errors, + }) +} + +fn responses_endpoint(base_url: &str) -> String { + format!("{}/responses", base_url.trim_end_matches('/')) +} + +fn websocket_url(endpoint: &str) -> Result { + let mut url = url::Url::parse(endpoint).map_err(|_| RunnerError::InvalidEndpoint(endpoint.to_owned()))?; + let scheme = match url.scheme() { + "http" => "ws", + "https" => "wss", + "ws" | "wss" => return Ok(url), + _ => return Err(RunnerError::InvalidEndpoint(endpoint.to_owned())), + }; + url.set_scheme(scheme) + .map_err(|()| RunnerError::InvalidEndpoint(endpoint.to_owned()))?; + Ok(url) +} + +fn take_sse_frame(pending: &mut Vec) -> Option> { + let position = pending + .windows(2) + .position(|window| window == b"\n\n") + .or_else(|| pending.windows(4).position(|window| window == b"\r\n\r\n"))?; + let delimiter_len = if pending.get(position..position + 4) == Some(b"\r\n\r\n") { + 4 + } else { + 2 + }; + let frame = pending.drain(..position).collect(); + pending.drain(..delimiter_len); + Some(frame) +} + +fn parse_sse_frame(frame: &[u8]) -> Result, RunnerError> { + let text = std::str::from_utf8(frame) + .map_err(|error| RunnerError::Protocol(format!("SSE frame is not UTF-8: {error}")))?; + let data = text + .lines() + .filter_map(|line| line.strip_prefix("data:")) + .map(str::trim_start) + .collect::>() + .join("\n"); + if data.is_empty() || data == "[DONE]" { + return Ok(None); + } + Ok(Some(serde_json::from_str(&data)?)) +} + +fn is_terminal_event(value: &Value) -> bool { + matches!( + value.get("type").and_then(Value::as_str), + Some("response.completed" | "response.failed" | "response.incomplete" | "error") + ) +} + +fn is_output_event(value: &Value) -> bool { + value.get("type").and_then(Value::as_str).is_some_and(|kind| { + kind.starts_with("response.output_") + || kind.starts_with("response.function_call_arguments.") + || kind.starts_with("response.reasoning_") + }) +} + +fn is_tool_event(value: &Value) -> bool { + let kind = value.get("type").and_then(Value::as_str).unwrap_or_default(); + kind.starts_with("response.function_call_arguments.") + || (kind == "response.output_item.added" + && value.pointer("/item/type").and_then(Value::as_str) == Some("function_call")) +} + +#[derive(Clone)] +struct CompletedToolCall { + call_id: String, + observed: ObservedToolCall, +} + +fn response_tool_calls(output: &[Value], events: &[TimedEvent], fallback_completed_ms: u64) -> Vec { + output + .iter() + .filter(|item| item.get("type").and_then(Value::as_str) == Some("function_call")) + .map(|item| { + let call_id = item + .get("call_id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned(); + let item_id = item.get("id").and_then(Value::as_str); + let started_at_ms = events.iter().find_map(|event| { + let event_item = event.value.get("item")?; + let matches_id = item_id.is_some_and(|id| event_item.get("id").and_then(Value::as_str) == Some(id)); + let matches_call = + !call_id.is_empty() && event_item.get("call_id").and_then(Value::as_str) == Some(call_id.as_str()); + (event.value.get("type").and_then(Value::as_str) == Some("response.output_item.added") + && (matches_id || matches_call)) + .then_some(event.elapsed_ms) + }); + let completed_at_ms = events + .iter() + .find_map(|event| { + let event_item = event.value.get("item")?; + let matches_id = item_id.is_some_and(|id| event_item.get("id").and_then(Value::as_str) == Some(id)); + let matches_call = !call_id.is_empty() + && event_item.get("call_id").and_then(Value::as_str) == Some(call_id.as_str()); + (event.value.get("type").and_then(Value::as_str) == Some("response.output_item.done") + && (matches_id || matches_call)) + .then_some(event.elapsed_ms) + }) + .unwrap_or(fallback_completed_ms); + let arguments = item + .get("arguments") + .and_then(Value::as_str) + .and_then(|text| serde_json::from_str(text).ok()) + .unwrap_or_else(|| item.get("arguments").cloned().unwrap_or(Value::Null)); + CompletedToolCall { + call_id, + observed: ObservedToolCall { + name: item.get("name").and_then(Value::as_str).unwrap_or_default().to_owned(), + arguments, + started_at_ms, + completed_at_ms, + }, + } + }) + .collect() +} + +fn execute_transport_call(call: &ObservedToolCall) -> Result { + if call.name != "benchmark_step" { + return Err(format!("unexpected transport tool {:?}", call.name)); + } + let arguments = call + .arguments + .as_object() + .ok_or_else(|| "benchmark_step arguments must be a JSON object".to_owned())?; + let run_id = arguments + .get("run_id") + .and_then(Value::as_str) + .ok_or_else(|| "benchmark_step run_id must be a string".to_owned())?; + let step = positive_usize(arguments, "step")?; + let total_steps = positive_usize(arguments, "total_steps")?; + Ok(transport_marker(run_id, step, total_steps)) +} + +fn positive_usize(arguments: &Map, name: &str) -> Result { + arguments + .get(name) + .and_then(Value::as_u64) + .and_then(|value| usize::try_from(value).ok()) + .filter(|value| *value > 0) + .ok_or_else(|| format!("benchmark_step {name} must be a positive integer")) +} + +fn response_text(response: &Value) -> Option { + let texts = response + .get("output")? + .as_array()? + .iter() + .filter(|item| item.get("type").and_then(Value::as_str) == Some("message")) + .filter_map(|item| item.get("content").and_then(Value::as_array)) + .flatten() + .filter_map(|content| content.get("text").and_then(Value::as_str)) + .collect::>(); + (!texts.is_empty()).then(|| texts.join("")) +} + +fn response_usage(response: &Value) -> Usage { + let usage = response.get("usage").unwrap_or(&Value::Null); + Usage { + input_tokens: usage.get("input_tokens").and_then(Value::as_i64).unwrap_or_default(), + cached_input_tokens: usage + .pointer("/input_tokens_details/cached_tokens") + .or_else(|| usage.get("cached_input_tokens")) + .and_then(Value::as_i64) + .unwrap_or_default(), + output_tokens: usage.get("output_tokens").and_then(Value::as_i64).unwrap_or_default(), + reasoning_output_tokens: usage + .pointer("/output_tokens_details/reasoning_tokens") + .or_else(|| usage.get("reasoning_output_tokens")) + .and_then(Value::as_i64) + .unwrap_or_default(), + } +} + +fn add_usage(total: &mut Usage, value: &Usage) { + total.input_tokens = total.input_tokens.saturating_add(value.input_tokens); + total.cached_input_tokens = total.cached_input_tokens.saturating_add(value.cached_input_tokens); + total.output_tokens = total.output_tokens.saturating_add(value.output_tokens); + total.reasoning_output_tokens = total + .reasoning_output_tokens + .saturating_add(value.reasoning_output_tokens); +} + +fn response_error(value: &Value) -> String { + value + .pointer("/error/message") + .or_else(|| value.pointer("/response/error/message")) + .or_else(|| value.get("message")) + .and_then(Value::as_str) + .unwrap_or("provider returned a non-completed response") + .to_owned() +} + +async fn emit_live_event(context: &LiveEventContext<'_>, event: &TimedEvent) { + if !context.enabled { + return; + } + let wrapper = json!({ + "provider": context.provider.id, + "transport": context.provider.transport, + "session_index": context.session_index, + "turn_index": context.turn_index, + "elapsed_ms": event.elapsed_ms, + "event": event.value, + }); + let _guard = context.output_lock.lock().await; + println!("{wrapper}"); +} + +async fn write_event_files( + raw_path: &Path, + timestamped_path: &Path, + error_path: &Path, + events: &[TimedEvent], + errors: &[String], +) -> Result<(), RunnerError> { + let parent = raw_path + .parent() + .ok_or_else(|| RunnerError::Protocol("event path has no parent".to_owned()))?; + create_dir_all(parent).await?; + let mut raw = String::new(); + let mut timestamped = String::new(); + for event in events { + raw.push_str(&event.value.to_string()); + raw.push('\n'); + timestamped.push_str(&json!({"elapsed_ms": event.elapsed_ms, "event": event.value}).to_string()); + timestamped.push('\n'); + } + write_file(raw_path, raw.as_bytes()).await?; + write_file(timestamped_path, timestamped.as_bytes()).await?; + write_file(error_path, errors.join("\n").as_bytes()).await +} + +fn tool_calls_match(expected: &[ExpectedToolCall], observed: &[ObservedToolCall]) -> bool { + if expected.len() != observed.len() { + return false; + } + let mut matched = vec![false; observed.len()]; + for expected_call in expected { + let Some((index, _)) = observed.iter().enumerate().find(|(index, observed_call)| { + !matched[*index] + && observed_call.name == expected_call.name + && arguments_match(&expected_call.arguments, &observed_call.arguments) + }) else { + return false; + }; + matched[index] = true; + } + true +} + +fn tool_calls_match_ordered(expected: &[ExpectedToolCall], observed: &[ObservedToolCall]) -> bool { + expected.len() == observed.len() + && expected.iter().zip(observed).all(|(expected_call, observed_call)| { + observed_call.name == expected_call.name + && arguments_match(&expected_call.arguments, &observed_call.arguments) + }) +} + +fn arguments_match(expected: &BTreeMap>, observed: &Value) -> bool { + let parsed; + let observed = if let Some(object) = observed.as_object() { + object + } else if let Some(text) = observed.as_str() { + parsed = serde_json::from_str::(text).ok(); + let Some(object) = parsed.as_ref().and_then(Value::as_object) else { + return expected.is_empty(); + }; + object + } else { + return expected.is_empty(); + }; + + if observed.keys().any(|name| !expected.contains_key(name)) { + return false; + } + expected.iter().all(|(name, accepted)| match observed.get(name) { + Some(value) => accepted.iter().any(|candidate| candidate == value), + None => accepted.iter().any(is_omission_sentinel), + }) +} + +fn is_omission_sentinel(value: &Value) -> bool { + value.as_str() == Some("") +} + +async fn create_dir_all(path: &Path) -> Result<(), RunnerError> { + tokio::fs::create_dir_all(path).await.map_err(|source| RunnerError::Io { + path: path.to_owned(), + source, + }) +} + +async fn write_file(path: &Path, contents: &[u8]) -> Result<(), RunnerError> { + tokio::fs::write(path, contents) + .await + .map_err(|source| RunnerError::Io { + path: path.to_owned(), + source, + }) +} + +fn relative_path(root: &Path, path: &Path) -> String { + path.strip_prefix(root).unwrap_or(path).to_string_lossy().into_owned() +} + +fn millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + +fn byte_len(len: usize) -> u64 { + u64::try_from(len).unwrap_or(u64::MAX) +} + +#[allow(clippy::cast_precision_loss)] +fn i64_as_f64(value: i64) -> f64 { + value as f64 +} + +#[allow(clippy::cast_precision_loss)] +fn u64_as_f64(value: u64) -> f64 { + value as f64 +} + +#[allow(clippy::cast_precision_loss)] +fn usize_as_f64(value: usize) -> f64 { + value as f64 +} diff --git a/crates/agentic-server/benches/client/response_provider/types.rs b/crates/agentic-server/benches/client/response_provider/types.rs new file mode 100644 index 00000000..80e8f78e --- /dev/null +++ b/crates/agentic-server/benches/client/response_provider/types.rs @@ -0,0 +1,282 @@ +use std::collections::BTreeMap; +use std::path::PathBuf; + +use clap::ValueEnum; +use serde::{Deserialize, Serialize}; +use serde_json::{Map, Value}; + +#[derive(Clone, Debug, Serialize)] +pub struct ProviderSpec { + pub id: String, + pub name: String, + pub base_url: String, + pub transport: Transport, + pub supports_websockets: bool, +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum Transport { + Websocket, + HttpSse, + HttpJson, +} + +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize, ValueEnum)] +#[serde(rename_all = "snake_case")] +pub enum Workload { + /// Multi-round turns that exercise connection reuse and streaming transport. + Transport, + /// BFCL function-selection and argument-accuracy cases sent as Responses tools. + ToolCall, + /// Short continuations that require state from the immediately preceding turn. + HistoryRehydration, +} + +impl Workload { + #[must_use] + pub const fn default_requests(self) -> usize { + match self { + Self::Transport | Self::ToolCall => 1, + Self::HistoryRehydration => 10, + } + } +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct ToolDefinition { + pub name: String, + pub description: String, + pub parameters: Map, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct ExpectedToolCall { + pub name: String, + /// BFCL represents each accepted argument value as a list of alternatives. + pub arguments: BTreeMap>, +} + +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum TurnExpectation { + Marker { + marker: String, + }, + ToolCalls { + calls: Vec, + }, + Transport { + calls: Vec, + final_marker: String, + }, +} + +impl TurnExpectation { + #[must_use] + pub fn expected_marker(&self) -> Option<&str> { + match self { + Self::Marker { marker, .. } => Some(marker), + Self::ToolCalls { .. } => None, + Self::Transport { final_marker, .. } => Some(final_marker), + } + } + + #[must_use] + pub fn expected_tool_calls(&self) -> &[ExpectedToolCall] { + match self { + Self::Marker { .. } => &[], + Self::ToolCalls { calls } | Self::Transport { calls, .. } => calls, + } + } +} + +#[derive(Clone, Debug, Serialize)] +pub struct PromptSpec { + pub workload: Workload, + pub session_index: usize, + pub turn_index: usize, + pub prompt_id: String, + pub source_id: Option, + pub prompt: String, + pub expectation: TurnExpectation, + pub tools: Vec, +} + +#[derive(Clone, Debug)] +pub struct SessionSpec { + pub session_index: usize, + pub prompts: Vec, +} + +#[derive(Clone, Debug, Default, Deserialize, Serialize)] +#[allow(clippy::struct_field_names)] +pub struct Usage { + pub input_tokens: i64, + pub cached_input_tokens: i64, + pub output_tokens: i64, + pub reasoning_output_tokens: i64, +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct ObservedToolCall { + pub name: String, + pub arguments: Value, + pub started_at_ms: Option, + pub completed_at_ms: u64, +} + +#[derive(Clone, Debug, Serialize)] +#[allow(clippy::struct_excessive_bools)] +pub struct TurnResult { + pub provider: String, + pub transport: Transport, + pub workload: Workload, + pub session_index: usize, + pub turn_index: usize, + pub prompt_id: String, + pub source_id: Option, + pub expected_marker: Option, + pub response_id: Option, + pub attempted: bool, + pub success: bool, + pub task_correct: bool, + pub tool_call_correct: bool, + pub timed_out: bool, + pub transport_fallback: bool, + pub saw_turn_completed: bool, + pub invalid_json_lines: usize, + pub error_events: usize, + pub tool_calls_started: usize, + pub tool_calls_completed: usize, + pub tool_calls_failed: usize, + pub observed_tool_calls: Vec, + pub expected_tool_calls: Vec, + pub tool_output_marker_found: bool, + pub final_answer_marker_found: bool, + pub exact_final_answer: bool, + pub time_to_first_output_event_ms: Option, + pub first_output_event_type: Option, + pub time_to_first_tool_call_ms: Option, + pub ttft_ms: Option, + pub mean_tool_duration_ms: Option, + pub initial_model_round_ms: Option, + pub continuation_round_latencies_ms: Vec, + pub end_to_end_latency_ms: u64, + /// Bytes of serialized request body sent to the provider, summed across every model round in this turn. + pub request_bytes: u64, + /// Bytes of raw response payload (WS text frames / SSE chunks / JSON body) received from the provider. + pub response_bytes: u64, + pub turn_usage: Option, + pub effective_output_tokens_per_second: Option, + pub raw_jsonl_path: String, + pub timestamped_jsonl_path: String, + pub error_log_path: String, + pub errors: Vec, +} + +#[derive(Clone, Debug, Serialize)] +pub struct SessionResult { + pub provider: String, + pub session_index: usize, + pub elapsed_ms: u64, + pub fatal_error: Option, + pub turns: Vec, +} + +#[derive(Clone, Debug, Serialize)] +pub struct Distribution { + pub count: usize, + pub mean: Option, + pub p50: Option, + pub p95: Option, + pub p99: Option, + pub min: Option, + pub max: Option, +} + +#[derive(Clone, Debug, Serialize)] +pub struct ProviderSummary { + pub provider: String, + pub transport: Transport, + pub planned_turns: usize, + pub attempted_turns: usize, + pub successful_turns: usize, + pub timed_out_turns: usize, + pub transport_fallback_turns: usize, + pub tool_compliant_turns: usize, + pub task_correct_turns: usize, + pub tool_call_correct_turns: usize, + pub success_rate: f64, + pub tool_compliance_rate: f64, + pub task_correctness_rate: f64, + pub tool_call_accuracy: f64, + pub provider_wall_clock_ms: u64, + pub successful_turns_per_second: f64, + pub end_to_end_latency_ms: Distribution, + pub time_to_first_output_event_ms: Distribution, + pub ttft_ms: Distribution, + pub time_to_first_tool_call_ms: Distribution, + pub mean_tool_duration_ms: Distribution, + pub continuation_round_latency_ms: Distribution, + pub request_bytes: Distribution, + pub response_bytes: Distribution, + pub total_turn_input_tokens: i64, + pub total_turn_cached_input_tokens: i64, + pub total_turn_output_tokens: i64, + pub total_turn_reasoning_output_tokens: i64, + pub aggregate_effective_output_tokens_per_second: Option, +} + +#[derive(Clone, Debug, Serialize)] +pub struct Comparison { + pub paired_successful_turns: usize, + pub median_agentic_minus_vllm_latency_ms: Option, + pub median_agentic_over_vllm_latency_ratio: Option, + pub median_agentic_minus_vllm_first_output_ms: Option, +} + +#[derive(Clone, Debug, Serialize)] +pub struct RunConfig { + pub model: String, + pub workload: Workload, + pub dataset_questions: Option, + pub dataset_answers: Option, + pub sessions_per_provider: usize, + pub requests_per_session: usize, + pub seed: u64, + pub timeout_seconds: u64, + pub providers: Vec, +} + +#[derive(Clone, Debug, Serialize)] +pub struct RunReport { + pub schema_version: u32, + pub started_at_unix_ms: u64, + pub elapsed_ms: u64, + pub output_dir: PathBuf, + pub config: RunConfig, + pub prompts: Vec, + pub sessions: Vec, + pub summaries: Vec, + pub comparison: Option, + pub ttft_note: String, + pub accuracy_note: String, +} + +/// One provider's results at one fixed session depth, produced by running that depth as its own +/// independent batch of sessions rather than bucketing a single long-lived run after the fact. +#[derive(Clone, Debug, Serialize)] +pub struct DepthSummaryRow { + pub depth: usize, + pub provider: String, + pub transport: Transport, + pub sessions: usize, + pub success_rate: f64, + pub task_correctness_rate: f64, + pub p50_latency_ms: Option, + pub p50_request_bytes: Option, + pub p50_response_bytes: Option, + pub total_request_bytes: u64, + pub total_response_bytes: u64, +} diff --git a/scripts/run-response-provider-benchmark.sh b/scripts/run-response-provider-benchmark.sh new file mode 100755 index 00000000..3b7ac936 --- /dev/null +++ b/scripts/run-response-provider-benchmark.sh @@ -0,0 +1,197 @@ +#!/usr/bin/env bash +set -Eeuo pipefail + +script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +repository_root="$(cd -- "${script_dir}/.." && pwd)" + +MODEL="${MODEL:-Qwen/Qwen3.5-35B-A3B-FP8}" +WORKLOAD="${WORKLOAD:-history-rehydration}" +REQUESTS_PER_SESSION="${REQUESTS_PER_SESSION:-}" +DEPTHS="${DEPTHS:-}" +SESSIONS="${SESSIONS:-5}" +TRANSPORT_ROUNDS="${TRANSPORT_ROUNDS:-4}" +PROVIDER="${PROVIDER:-both}" +AGENTIC_URL="${AGENTIC_URL:-http://localhost:9000/v1}" +VLLM_URL="${VLLM_URL:-http://localhost:5050/v1}" +TIMEOUT_SECONDS="${TIMEOUT_SECONDS:-300}" +SEED="${SEED:-20260817}" +RESULTS_ROOT="${RESULTS_ROOT:-${repository_root}/target/response-provider-benchmark}" +DATASET_QUESTIONS="${DATASET_QUESTIONS:-}" +DATASET_ANSWERS="${DATASET_ANSWERS:-}" +DATASET_OFFSET="${DATASET_OFFSET:-0}" +BFCL_ROOT="${BFCL_ROOT:-}" +BFCL_CATEGORY="${BFCL_CATEGORY:-simple_python}" +PRINT_PROMPTS="${PRINT_PROMPTS:-0}" + +usage() { + cat <<'EOF' +Run optimized Responses capability benchmarks from the agentic-server bench target. + +Usage: + ./scripts/run-response-provider-benchmark.sh + +Core environment variables: + MODEL Model served by all selected endpoints. + WORKLOAD transport, tool-call, or history-rehydration. + SESSIONS Parallel sessions per provider (default: 5). + REQUESTS_PER_SESSION Sequential turns per session; unset uses the workload default. + DEPTHS Comma-separated fixed turn depths (history-rehydration only), e.g. + 1,5,10,25,50,100. Each depth runs as its own independent batch of SESSIONS + sessions instead of bucketing one long run after the fact, so sample counts + stay even across depths. Overrides REQUESTS_PER_SESSION when set. Produces + depth_summary.md/.json under RESULTS_ROOT with request/response bytes and + accuracy versus turn depth. + PROVIDER all, both, agentic-api, agentic-api-http, agentic-api-json, + vllm, or vllm-json. + AGENTIC_URL Agentic API base URL (default: http://localhost:9000/v1). + VLLM_URL Direct vLLM base URL (default: http://localhost:5050/v1). + TRANSPORT_ROUNDS Sequential model/tool rounds in each transport turn (default: 4). + SEED Deterministic prompt/case seed shared across separate runs. + PRINT_PROMPTS Set to 1 to validate and print generated prompt JSONL without running. + +BFCL tool-call workload: + BFCL_ROOT Checkout of https://github.com/EnlightenedAI/BFCL. + BFCL_CATEGORY v4 filename suffix, such as simple_python or multiple. + DATASET_QUESTIONS Explicit BFCL question JSONL; overrides BFCL_ROOT derivation. + DATASET_ANSWERS Explicit BFCL possible-answer JSONL. + DATASET_OFFSET First deterministic case index (default: 0). + +Other controls: + TIMEOUT_SECONDS RESULTS_ROOT + +Examples: + WORKLOAD=transport PROVIDER=all TRANSPORT_ROUNDS=8 ./scripts/run-response-provider-benchmark.sh + + WORKLOAD=transport PROVIDER=agentic-api-json ./scripts/run-response-provider-benchmark.sh + + WORKLOAD=tool-call PROVIDER=agentic-api BFCL_ROOT=/path/to/BFCL \ + REQUESTS_PER_SESSION=1 ./scripts/run-response-provider-benchmark.sh + +EOF +} + +if [[ "${1:-}" == "-h" || "${1:-}" == "--help" ]]; then + usage + exit 0 +fi +if (( $# != 0 )); then + echo "error: unexpected argument: $1" >&2 + usage >&2 + exit 2 +fi + +for command_name in cargo tee; do + if ! command -v "$command_name" >/dev/null 2>&1; then + echo "error: ${command_name} is required" >&2 + exit 2 + fi +done + +for numeric_value in SESSIONS TRANSPORT_ROUNDS TIMEOUT_SECONDS; do + if [[ ! "${!numeric_value}" =~ ^[1-9][0-9]*$ ]]; then + echo "error: ${numeric_value} must be a positive integer" >&2 + exit 2 + fi +done +if [[ -n "$REQUESTS_PER_SESSION" && ! "$REQUESTS_PER_SESSION" =~ ^[1-9][0-9]*$ ]]; then + echo "error: REQUESTS_PER_SESSION must be a positive integer when set" >&2 + exit 2 +fi +if [[ -n "$DEPTHS" ]]; then + if [[ ! "$DEPTHS" =~ ^[1-9][0-9]*(,[1-9][0-9]*)*$ ]]; then + echo "error: DEPTHS must be a comma-separated list of positive integers" >&2 + exit 2 + fi + if [[ "$WORKLOAD" != "history-rehydration" ]]; then + echo "error: DEPTHS is only supported for WORKLOAD=history-rehydration" >&2 + exit 2 + fi +fi +if [[ ! "$DATASET_OFFSET" =~ ^[0-9]+$ || ! "$SEED" =~ ^[0-9]+$ ]]; then + echo "error: DATASET_OFFSET and SEED must be non-negative integers" >&2 + exit 2 +fi +case "$WORKLOAD" in + transport | tool-call | history-rehydration) ;; + *) + echo "error: unsupported WORKLOAD: ${WORKLOAD}" >&2 + exit 2 + ;; +esac +case "$PROVIDER" in + all | both | agentic-api | agentic-api-http | agentic-api-json | vllm | vllm-json) ;; + *) + echo "error: unsupported PROVIDER: ${PROVIDER}" >&2 + exit 2 + ;; +esac + +if [[ "$WORKLOAD" == "tool-call" && -z "$DATASET_QUESTIONS" && -n "$BFCL_ROOT" ]]; then + bfcl_data_dir="${BFCL_ROOT%/}/berkeley-function-call-leaderboard/bfcl_eval/data" + DATASET_QUESTIONS="${bfcl_data_dir}/BFCL_v4_${BFCL_CATEGORY}.json" + DATASET_ANSWERS="${bfcl_data_dir}/possible_answer/BFCL_v4_${BFCL_CATEGORY}.json" +fi +if [[ "$WORKLOAD" == "tool-call" && ( -z "$DATASET_QUESTIONS" || -z "$DATASET_ANSWERS" ) ]]; then + echo "error: tool-call requires BFCL_ROOT or both DATASET_QUESTIONS and DATASET_ANSWERS" >&2 + exit 2 +fi +for dataset_path in "$DATASET_QUESTIONS" "$DATASET_ANSWERS"; do + if [[ -n "$dataset_path" && ! -f "$dataset_path" ]]; then + echo "error: dataset file not found: ${dataset_path}" >&2 + exit 2 + fi +done +run_stamp="$(date -u +%Y%m%dT%H%M%SZ)-$$" +run_dir="${RESULTS_ROOT%/}/${run_stamp}" +mkdir -p "$run_dir" + +benchmark_command=( + cargo bench --package agentic-server --bench response_provider -- + --model "$MODEL" + --workload "$WORKLOAD" + --sessions "$SESSIONS" + --transport-rounds "$TRANSPORT_ROUNDS" + --provider "$PROVIDER" + --agentic-url "$AGENTIC_URL" + --vllm-url "$VLLM_URL" + --timeout-seconds "$TIMEOUT_SECONDS" + --seed "$SEED" + --dataset-offset "$DATASET_OFFSET" + --output-dir "$run_dir" + --live-jsonl +) +if [[ -n "$DEPTHS" ]]; then + benchmark_command+=(--depths "$DEPTHS") +elif [[ -n "$REQUESTS_PER_SESSION" ]]; then + benchmark_command+=(--requests-per-session "$REQUESTS_PER_SESSION") +fi +if [[ -n "$DATASET_QUESTIONS" ]]; then + benchmark_command+=(--dataset-questions "$DATASET_QUESTIONS" --dataset-answers "$DATASET_ANSWERS") +fi +if [[ "$PRINT_PROMPTS" == "1" ]]; then + benchmark_command+=(--print-prompts) +fi + +{ + printf 'cd %q\n' "$repository_root" + printf '%q ' "${benchmark_command[@]}" + printf '\n' +} >"${run_dir}/command.sh" + +echo "Benchmark workload: ${WORKLOAD}" >&2 +echo "Benchmark results: ${run_dir}" >&2 +echo "Live JSONL: ${run_dir}/live-events.jsonl" >&2 +echo "Diagnostic log: ${run_dir}/benchmark.log" >&2 + +cd "$repository_root" +if "${benchmark_command[@]}" \ + > >(tee "${run_dir}/live-events.jsonl") \ + 2> >(tee "${run_dir}/benchmark.log" >&2); then + benchmark_status=0 +else + benchmark_status=$? +fi + +echo "Benchmark exit status: ${benchmark_status}" >&2 +echo "Summary: ${run_dir}/summary.md" >&2 +exit "$benchmark_status"