Model Test Bounded Concurrency Implementation Plan (#204)

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 <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

<details>
<summary>Ran 9 stages in 30m 52s for $19.61</summary>

| 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** |

</details>

<details>
<summary>Ran <code>ImplementPlan.fabro</code> (12 nodes and 15
edges)</summary>

```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
}

```

</details>

⚒️ Generated with [Fabro](https://fabro.sh)

---------

Co-authored-by: Fabro <noreply@fabro.sh>

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 <jobs>` | Number of model tests to run concurrently in bulk mode<br />Default: `4` |
 | `-m, --model <model>` | Test a specific model |
 | `-p, --provider <provider>` | Filter by provider |
 | `--server <server>` | Fabro server target: http(s) URL or absolute Unix socket path |

lib/crates/fabro-cli/src/args.rs

@@ -876,6 +876,15 @@ pub(crate) struct ModelTestArgs {
     #[arg(short, long)]
     pub(crate) model: Option<String>,
 
+    /// Number of model tests to run concurrently in bulk mode
+    #[arg(
+        short = 'j',
+        long,
+        default_value_t = 4,
+        value_parser = clap::builder::RangedU64ValueParser::<usize>::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,

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<ModelsCommand>,
     base_ctx: &CommandContext,
@@ -155,6 +163,28 @@ fn print_models_table(models: &[Model], styles: &Styles) {
     );
 }
 
+fn configured_model_test_status(
+    result: Result<api_types::ModelTestResult>,
+) -> (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::<Vec<_>>()
+            .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,
             )

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 <PROVIDER>  Filter by provider
       -m, --model <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 <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<serde_json::Value>,
+    gate:            Arc<ConcurrencyGate>,
+    response_delays: Arc<HashMap<String, Duration>>,
+}
+
+struct ConcurrentModelServer {
+    base_url:    String,
+    gate:        Arc<ConcurrencyGate>,
+    shutdown_tx: Option<oneshot::Sender<()>>,
+    join_handle: Option<std::thread::JoinHandle<()>>,
+}
+
+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<serde_json::Value>,
+    gate_expected: usize,
+    response_delays: HashMap<String, Duration>,
+) -> 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<ConcurrentModelServerState>,
+) -> Json<serde_json::Value> {
+    Json(serde_json::json!({
+        "data": state.models,
+        "meta": { "has_more": false }
+    }))
+}
+
+async fn concurrent_test_model(
+    State(state): State<ConcurrentModelServerState>,
+    Path(id): Path<String>,
+) -> Json<serde_json::Value> {
+    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<serde_json::Value> {
+    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::<Vec<_>>();
+    assert_eq!(models, FIVE_ANTHROPIC_MODEL_IDS.to_vec());
+}