diff --git a/docs/public/reference/cli.mdx b/docs/public/reference/cli.mdx index ace79f859..7fd748ec0 100644 --- a/docs/public/reference/cli.mdx +++ b/docs/public/reference/cli.mdx @@ -532,6 +532,7 @@ fabro model test [OPTIONS] | Option | Description | | --- | --- | | `--deep` | Run a multi-turn tool-use test (catches reasoning round-trip bugs) | +| `-j, --jobs ` | Number of model tests to run concurrently in bulk mode
Default: `4` | | `-m, --model ` | Test a specific model | | `-p, --provider ` | Filter by provider | | `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | diff --git a/lib/crates/fabro-cli/src/args.rs b/lib/crates/fabro-cli/src/args.rs index 52ea3a87a..331b0eb34 100644 --- a/lib/crates/fabro-cli/src/args.rs +++ b/lib/crates/fabro-cli/src/args.rs @@ -876,6 +876,15 @@ pub(crate) struct ModelTestArgs { #[arg(short, long)] pub(crate) model: Option, + /// Number of model tests to run concurrently in bulk mode + #[arg( + short = 'j', + long, + default_value_t = 4, + value_parser = clap::builder::RangedU64ValueParser::::new().range(1..) + )] + pub(crate) jobs: usize, + /// Run a multi-turn tool-use test (catches reasoning round-trip bugs) #[arg(long)] pub(crate) deep: bool, diff --git a/lib/crates/fabro-cli/src/commands/model.rs b/lib/crates/fabro-cli/src/commands/model.rs index e66fbdbfd..9ccfc20a4 100644 --- a/lib/crates/fabro-cli/src/commands/model.rs +++ b/lib/crates/fabro-cli/src/commands/model.rs @@ -4,6 +4,7 @@ use cli_table::{Cell, CellStruct, Color, Style, Table}; use fabro_api::types as api_types; use fabro_model::{Catalog, Model, ModelTestMode, Provider}; use fabro_util::terminal::Styles; +use futures::{StreamExt, stream}; use serde::Serialize; use crate::args::{ModelListArgs, ModelTestArgs, ModelsCommand}; @@ -37,6 +38,13 @@ struct ModelTestOutput { skipped: u32, } +struct CompletedModelTest { + index: usize, + model: Model, + result_color: Color, + status: String, +} + pub(crate) async fn execute( command: Option, base_ctx: &CommandContext, @@ -155,6 +163,28 @@ fn print_models_table(models: &[Model], styles: &Styles) { ); } +fn configured_model_test_status( + result: Result, +) -> (Color, String, bool) { + match result { + Ok(resp) if resp.status == api_types::ModelTestResultStatus::Ok => { + (Color::Green, "ok".to_string(), false) + } + Ok(resp) if resp.status == api_types::ModelTestResultStatus::Skip => ( + Color::Red, + "error: provider became unconfigured after listing".to_string(), + true, + ), + Ok(resp) => { + let message = resp + .error_message + .unwrap_or_else(|| "unknown error".to_string()); + (Color::Red, format!("error: {message}"), true) + } + Err(err) => (Color::Red, format!("error: {err}"), true), + } +} + fn model_test_row_from_status(model: &Model, status: &str, result_color: Color) -> ModelTestRow { let trimmed = status.trim(); match result_color { @@ -197,6 +227,7 @@ async fn test_models_via_server( provider: Option<&str>, model: Option<&str>, deep: bool, + jobs: usize, styles: &Styles, json_output: bool, ) -> Result<()> { @@ -289,48 +320,50 @@ async fn test_models_via_server( } } - for info in &configured { - if !json_output { - eprint!("Testing {}...", info.id); - } - let result = client.test_model(&info.id, request_mode).await; - if !json_output { - eprintln!(" done"); - } - - let (result_color, status) = match result { - Ok(resp) if resp.status == api_types::ModelTestResultStatus::Ok => { - (Color::Green, "ok".to_string()) - } - Ok(resp) if resp.status == api_types::ModelTestResultStatus::Skip => { - failures += 1; + let mut completed = stream::iter(configured.into_iter().enumerate()) + .map(|(index, info)| { + let client = client.clone(); + async move { + let result = client.test_model(&info.id, request_mode).await; + if !json_output { + eprintln!("Testing {}... done", info.id); + } + let (result_color, status, failed) = configured_model_test_status(result); ( - Color::Red, - "error: provider became unconfigured after listing".to_string(), + CompletedModelTest { + index, + model: info, + result_color, + status, + }, + failed, ) } - Ok(resp) => { - failures += 1; - let message = resp - .error_message - .unwrap_or_else(|| "unknown error".to_string()); - (Color::Red, format!("error: {message}")) - } - Err(err) => { - failures += 1; - (Color::Red, format!("error: {err}")) - } - }; + }) + .buffer_unordered(jobs) + .collect::>() + .await; - let mut row = model_row(info, use_color); + completed.sort_by_key(|(completed, _)| completed.index); + + for (completed, failed) in completed { + if failed { + failures += 1; + } + + let mut row = model_row(&completed.model, use_color); + json_rows.push(model_test_row_from_status( + &completed.model, + &completed.status, + completed.result_color, + )); row.push( - status - .clone() + completed + .status .cell() - .foreground_color(color_if(use_color, result_color)), + .foreground_color(color_if(use_color, completed.result_color)), ); rows.push(row); - json_rows.push(model_test_row_from_status(info, &status, result_color)); } } @@ -404,6 +437,7 @@ async fn run_models( provider, model, deep, + jobs, .. }) => { test_models_via_server( @@ -411,6 +445,7 @@ async fn run_models( provider.as_deref(), model.as_deref(), deep, + jobs, &styles, json_output, ) diff --git a/lib/crates/fabro-cli/tests/it/cmd/model_test.rs b/lib/crates/fabro-cli/tests/it/cmd/model_test.rs index 5707c21f1..6020dac3d 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/model_test.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/model_test.rs @@ -1,6 +1,17 @@ +use std::collections::HashMap; +use std::net::SocketAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::time::Duration; + use assert_cmd::Command; +use axum::extract::{Path, State}; +use axum::routing::{get, post}; +use axum::{Json, Router}; use fabro_test::{fabro_snapshot, test_context}; use httpmock::MockServer; +use tokio::net::TcpListener; +use tokio::sync::{Semaphore, oneshot}; fn remove_provider_env(cmd: &mut Command) -> &mut Command { cmd.env_remove("ANTHROPIC_API_KEY") @@ -78,8 +89,9 @@ fn help() { -p, --provider Filter by provider -m, --model Test a specific model --no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true] - --deep Run a multi-turn tool-use test (catches reasoning round-trip bugs) + -j, --jobs Number of model tests to run concurrently in bulk mode [default: 4] --quiet Suppress non-essential output [env: FABRO_QUIET=] + --deep Run a multi-turn tool-use test (catches reasoning round-trip bugs) --verbose Enable verbose output [env: FABRO_VERBOSE=] -h, --help Print help ----- stderr ----- @@ -337,3 +349,302 @@ fn model_test_json_partitions_skip_and_fail() { "provider became unconfigured after listing" ); } + +#[derive(Clone)] +struct ConcurrentModelServerState { + models: Vec, + gate: Arc, + response_delays: Arc>, +} + +struct ConcurrentModelServer { + base_url: String, + gate: Arc, + shutdown_tx: Option>, + join_handle: Option>, +} + +impl Drop for ConcurrentModelServer { + fn drop(&mut self) { + if let Some(shutdown_tx) = self.shutdown_tx.take() { + let _ = shutdown_tx.send(()); + } + if let Some(join_handle) = self.join_handle.take() { + join_handle + .join() + .expect("concurrent model test server thread should not panic"); + } + } +} + +struct ConcurrencyGate { + expected: usize, + arrived: AtomicUsize, + in_flight: AtomicUsize, + max_in_flight: AtomicUsize, + released: AtomicBool, + timed_out: AtomicBool, + release: Semaphore, +} + +impl ConcurrencyGate { + fn new(expected: usize) -> Self { + assert!(expected > 0, "ConcurrencyGate requires expected > 0"); + Self { + expected, + arrived: AtomicUsize::new(0), + in_flight: AtomicUsize::new(0), + max_in_flight: AtomicUsize::new(0), + released: AtomicBool::new(false), + timed_out: AtomicBool::new(false), + release: Semaphore::new(0), + } + } + + async fn enter(&self) { + let in_flight = self.in_flight.fetch_add(1, Ordering::SeqCst) + 1; + self.max_in_flight.fetch_max(in_flight, Ordering::SeqCst); + + if self.released.load(Ordering::SeqCst) { + return; + } + + let arrived = self.arrived.fetch_add(1, Ordering::SeqCst) + 1; + if arrived >= self.expected { + // The expected-th arrival releases the (expected - 1) tasks already + // blocked on `release.acquire()`. Late arrivals short-circuit on the + // `released` check above and never touch the semaphore. + if !self.released.swap(true, Ordering::SeqCst) { + self.release.add_permits(self.expected - 1); + } + return; + } + + let permit = self.release.acquire(); + if self.released.load(Ordering::SeqCst) { + return; + } + + if tokio::time::timeout(Duration::from_secs(15), permit) + .await + .is_err() + { + self.timed_out.store(true, Ordering::SeqCst); + if !self.released.swap(true, Ordering::SeqCst) { + self.release.add_permits(self.expected - 1); + } + } + } + + fn exit(&self) { + self.in_flight.fetch_sub(1, Ordering::SeqCst); + } + + fn max_in_flight(&self) -> usize { + self.max_in_flight.load(Ordering::SeqCst) + } + + fn timed_out(&self) -> bool { + self.timed_out.load(Ordering::SeqCst) + } +} + +fn start_concurrent_model_server( + models: Vec, + gate_expected: usize, + response_delays: HashMap, +) -> ConcurrentModelServer { + #[expect( + clippy::disallowed_types, + reason = "Bind synchronously so we can read the listening port before spawning the \ + runtime thread; converted to tokio::net::TcpListener inside the runtime." + )] + let std_listener = std::net::TcpListener::bind("127.0.0.1:0").expect("test server should bind"); + std_listener + .set_nonblocking(true) + .expect("test server listener should be nonblocking"); + let addr: SocketAddr = std_listener + .local_addr() + .expect("test server should have addr"); + let gate = Arc::new(ConcurrencyGate::new(gate_expected)); + let state = ConcurrentModelServerState { + models, + gate: Arc::clone(&gate), + response_delays: Arc::new(response_delays), + }; + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + + #[expect( + clippy::disallowed_methods, + reason = "Owns a dedicated OS thread that hosts a fresh Tokio runtime so the test server \ + is independent of any caller runtime and joinable via Drop." + )] + let join_handle = std::thread::spawn(move || { + let runtime = tokio::runtime::Runtime::new().expect("test runtime should start"); + runtime.block_on(async move { + let listener = + TcpListener::from_std(std_listener).expect("test listener should convert"); + let app = Router::new() + .route("/api/v1/models", get(concurrent_list_models)) + .route("/api/v1/models/{id}/test", post(concurrent_test_model)) + .with_state(state); + let _ = axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = shutdown_rx.await; + }) + .await; + }); + }); + + ConcurrentModelServer { + base_url: format!("http://{addr}"), + gate, + shutdown_tx: Some(shutdown_tx), + join_handle: Some(join_handle), + } +} + +async fn concurrent_list_models( + State(state): State, +) -> Json { + Json(serde_json::json!({ + "data": state.models, + "meta": { "has_more": false } + })) +} + +async fn concurrent_test_model( + State(state): State, + Path(id): Path, +) -> Json { + state.gate.enter().await; + if let Some(delay) = state.response_delays.get(&id) { + tokio::time::sleep(*delay).await; + } + state.gate.exit(); + + Json(serde_json::json!({ + "model_id": id, + "status": "ok" + })) +} + +const FIVE_ANTHROPIC_MODEL_IDS: [&str; 5] = [ + "claude-opus-4-7", + "claude-opus-4-6", + "claude-sonnet-4-5", + "claude-sonnet-4-6", + "claude-haiku-4-5", +]; + +fn five_anthropic_models() -> Vec { + FIVE_ANTHROPIC_MODEL_IDS + .iter() + .map(|id| model_json(id, "anthropic", true)) + .collect() +} + +#[test] +fn model_test_default_jobs_runs_four_concurrently() { + let context = test_context!(); + let server = start_concurrent_model_server(five_anthropic_models(), 4, HashMap::new()); + context.set_http_target(&server.base_url); + + let mut cmd = context.command(); + remove_provider_env(&mut cmd); + cmd.args(["model", "test"]); + let output = cmd.output().expect("command should execute"); + + assert!( + output.status.success(), + "model test should succeed:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + !server.gate.timed_out(), + "concurrency gate timed out before four requests arrived" + ); + assert_eq!( + server.gate.max_in_flight(), + 4, + "default jobs should run four model tests concurrently before the gate releases" + ); +} + +#[test] +fn model_test_explicit_jobs_two_runs_two_concurrently() { + let context = test_context!(); + let server = start_concurrent_model_server(five_anthropic_models(), 2, HashMap::new()); + context.set_http_target(&server.base_url); + + let mut cmd = context.command(); + remove_provider_env(&mut cmd); + cmd.args(["model", "test", "--jobs", "2"]); + let output = cmd.output().expect("command should execute"); + + assert!( + output.status.success(), + "model test should succeed:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + !server.gate.timed_out(), + "concurrency gate timed out before two requests arrived" + ); + assert_eq!( + server.gate.max_in_flight(), + 2, + "--jobs 2 should run two model tests concurrently before the gate releases" + ); +} + +#[test] +fn model_test_json_preserves_listing_order_under_concurrency() { + let context = test_context!(); + // Reverse-order delays force completion order opposite to listing order so + // the test fails if the configured-list `index` sort is dropped. + let response_delays = FIVE_ANTHROPIC_MODEL_IDS + .iter() + .enumerate() + .map(|(i, id)| { + ( + (*id).to_string(), + Duration::from_millis(50 * (FIVE_ANTHROPIC_MODEL_IDS.len() - i) as u64), + ) + }) + .collect(); + let server = start_concurrent_model_server(five_anthropic_models(), 5, response_delays); + context.set_http_target(&server.base_url); + + let mut cmd = context.command(); + remove_provider_env(&mut cmd); + cmd.args(["model", "test", "--jobs", "5", "--json"]); + let output = cmd.output().expect("command should execute"); + + assert!( + output.status.success(), + "model test should succeed:\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!( + !server.gate.timed_out(), + "concurrency gate timed out before five requests arrived" + ); + assert_eq!( + server.gate.max_in_flight(), + 5, + "ordering test should have all five model requests in flight" + ); + let json: serde_json::Value = + serde_json::from_slice(&output.stdout).expect("invalid JSON output"); + let models = json["results"] + .as_array() + .expect("results should be an array") + .iter() + .map(|row| row["model"].as_str().expect("model should be a string")) + .collect::>(); + assert_eq!(models, FIVE_ANTHROPIC_MODEL_IDS.to_vec()); +}