From 3eef9ee92802ee14dda440848fd030cd61042427 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp <19+brynary@users.noreply.github.com> Date: Mon, 4 May 2026 10:46:44 -0700 Subject: [PATCH] Model Test Bounded Concurrency Implementation Plan (#204) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This change makes bulk `fabro model test` run configured model checks concurrently instead of serially. A new `--jobs/-j` flag (defaulting to 4, minimum 1) controls the concurrency bound; the single-model path (`--model `) is unaffected. Under the hood, the serial `for` loop over configured models is replaced with a `futures::stream::buffer_unordered(jobs)` pipeline that clones the shared-state `Client` per request. Completed results carry their original list index and are sorted before rendering, so final stdout table rows and JSON output remain in listing order regardless of which requests finish first. Three new integration tests verify the concurrency behavior using an inline Axum harness with a `ConcurrencyGate` barrier. The gate holds all in-flight requests until the expected number arrive simultaneously, then releases them, letting tests assert `max_in_flight` exactly rather than relying on timing. The ordering test goes further by assigning reverse-listing response delays so the last-listed model always finishes first; if the index sort were dropped, the JSON result order would invert and the assertion would fail. A 15-second gate timeout ensures a regression to serial execution surfaces as a clear `max_in_flight == 1` failure rather than a hung test. Existing behavior is fully preserved: unconfigured models are still skipped without a POST, a configured model returning `skip` after listing is still a failure, `--deep` uses the same `--jobs` value, and `--jobs 1` reproduces the previous serial behavior for users hitting provider rate limits. ### Fabro Details
Ran 9 stages in 30m 52s for $19.61 | Stage | Duration | Cost | Retries | |---|---|---|---| | start | 0s | – | 0 | | toolchain | 1s | – | 0 | | preflight_compile | 2m 8s | – | 0 | | preflight_lint | 2m 21s | – | 0 | | implement | 8m 56s | $3.87 | 0 | | simplify_opus | 7m 55s | $1.43 | 0 | | simplify_gpt | 6m 58s | $14.32 | 0 | | verify | 1m 49s | – | 0 | | fmt | 2s | – | 0 | | **Total** | **30m 52s** | **$19.61** | **0** |
Ran ImplementPlan.fabro (12 nodes and 15 edges) ```dot digraph ImplementPlan { graph [ goal="Implement and simplify", model_stylesheet=" * { model: claude-opus-4-7; } " ] rankdir=LR start [shape=Mdiamond, label="Start"] exit [shape=Msquare, label="Exit"] toolchain [label="Toolchain", shape=parallelogram, script="command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1", max_retries=0] preflight_compile [label="Preflight Compile", shape=parallelogram, script="cargo check -q --workspace 2>&1", max_retries=0] preflight_lint [label="Preflight Lint", shape=parallelogram, script="cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1", max_retries=0] fix_lints [label="Fix Lints", prompt="The preflight lint step failed. Read the build output from context and fix all clippy lint warnings.", max_visits=3] implement [label="Implement", prompt="Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD."] simplify_opus [label="Simplify (Opus)", prompt="@prompts/simplify.md"] simplify_gpt [label="Simplify (GPT-55)", prompt="@prompts/simplify.md", model="gpt-55"] verify [label="Verify", shape=parallelogram, script="cargo +nightly-2026-04-14 clippy -q --workspace --all-targets -- -D warnings 2>&1 && cargo nextest run --cargo-quiet --workspace --status-level fail 2>&1 && cargo dev docs refresh 2>&1 && cargo dev docs check 2>&1", goal_gate=true, retry_target="fixup"] fixup [label="Fixup", prompt="The verify step failed. Read the build output from context and fix all clippy lint warnings, test failures, and generated docs errors.", max_visits=3] fmt [label="Format", shape=parallelogram, script="cargo +nightly-2026-04-14 fmt --all 2>&1", max_retries=0] start -> toolchain toolchain -> preflight_compile [condition="outcome=succeeded"] toolchain -> exit preflight_compile -> preflight_lint [condition="outcome=succeeded"] preflight_compile -> exit preflight_lint -> implement [condition="outcome=succeeded"] preflight_lint -> fix_lints fix_lints -> preflight_lint implement -> simplify_opus -> simplify_gpt -> verify verify -> fmt [condition="outcome=succeeded"] verify -> fixup fixup -> verify fmt -> exit } ```
⚒️ Generated with [Fabro](https://fabro.sh) --------- Co-authored-by: Fabro --- docs/public/reference/cli.mdx | 1 + lib/crates/fabro-cli/src/args.rs | 9 + lib/crates/fabro-cli/src/commands/model.rs | 103 ++++-- .../fabro-cli/tests/it/cmd/model_test.rs | 313 +++++++++++++++++- 4 files changed, 391 insertions(+), 35 deletions(-) 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()); +}