fix(examples): forward full ensemble requests and cap concurrent explorations

[elyasmnvidian]

let request = LlmRequest {
    model: Some("auto".to_string()),
    messages: vec![
        Message::text(Role::System, "be terse"),
        Message::text(Role::User, "add 2 and 2"),
    ],
    tools: vec![ToolDefinition {
        name: "calc".to_string(),
        description: Some("do math".to_string()),
        parameters: serde_json::json!({"type": "object"}),
        strict: Some(true),
    }],
    tool_choice: Some(ToolChoice::Required),
    sampling: SamplingParams {
        temperature: Some(0.3),
        ..Default::default()
    },
    stream: true,
    ..Default::default()
};

What happens

The ensemble example rebuilds each candidate request from the last user message. The candidates and the model chosen after exploration receive this smaller request:

LlmRequest {
    model: Some("auto".to_string()),
    messages: vec![Message::text(Role::User, "add 2 and 2")],
    tools: vec![],
    tool_choice: None,
    sampling: SamplingParams::default(),
    stream: false,
    ..Default::default()
}

The system message, tools, tool choice, sampling settings, and stream flag are missing, so the candidates answer a different request.

The example also counts an exploration only after the candidate calls finish. With exploration_turns = 1, two requests that arrive together can both explore:

a/model:       2 candidate calls
b/model:       2 candidate calls
judge/haiku:   2 calls

Fix

The example now sends the caller's complete request to every candidate and to the chosen model. It reads candidate streams concurrently before asking the judge to choose, so a slow stream does not delay reading the others.

A request claims an exploration slot before its first model call. Requests wait when all slots are claimed, and a dropped or failed turn returns its slot before a winner is recorded. The completed win counts determine when the example can choose one model.

The example still buffers candidate streams because it must read every candidate response before it can choose a winner.

Evidence

With exploration_turns = 1, two simultaneous requests now make these calls when b/model wins the first request:

a/model:       1 candidate call
b/model:       1 candidate call + 1 chosen-model call
judge/haiku:   1 call

The regression tests also show that:

• candidate and chosen-model calls receive the original messages, tools, tool choice, sampling settings, and stream flag;
• the judge reads the text from streamed candidate responses;
• a failed candidate stream removes only that candidate;
• a failed exploration returns its slot, so the next request does not hang.

This is a correctness fix in an example, not a performance change, so there is no benchmark.

Testing

cargo test -p switchyard-libsy --example ensemble

test result: ok. 14 passed; 0 failed

[coderabbitai]

Walkthrough

The ensemble router now uses reservation-based exploration state with asynchronous completion notifications. It preserves structured requests through candidate and committed paths, aggregates streamed candidate responses for judging, and adds concurrency and failure-handling tests.

Changes

Ensemble router

Layer / File(s)	Summary
Reservation-based routing crates/libsy/examples/ensemble.rs	Per-session state tracks reserved slots, completed explorations, and the committed model. plan_route() and ReservationGuard coordinate capacity, completion, and failure cleanup.
Structured candidate and judge flow crates/libsy/examples/ensemble.rs	Candidate and committed calls preserve the full structured request. Successful streamed responses are buffered before judging.
Routing and streaming validation crates/libsy/examples/ensemble.rs	Tests cover request preservation, streamed outputs, concurrent budget enforcement, reservation release, failed streams, and committed routing.

Estimated code review effort: 4 (Complex) | ~45 minutes

Poem

A rabbit guards each reserved slot,
Structured requests change not a lot.
Streams collect for judges true,
Notify wakes the waiting queue.
Failed turns release their place—
The ensemble hops through every case!

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly summarizes the main changes: forwarding complete ensemble requests and limiting concurrent explorations.

---

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

🧹 Nitpick comments (1)

crates/libsy/examples/ensemble.rs (1)

313-335: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win

Candidate streams are drained serially — fold aggregation into the fan-out futures.

join_all only awaits the call handshake; with stream = true now forwarded to candidates, the actual token streams are consumed one after another here, so candidate n's stream isn't polled until n-1 is fully drained. Aggregating inside each candidate future keeps the whole fan-out concurrent.

♻️ Aggregate each candidate inside its own future

             calls.push(async move {
-                (
-                    model,
-                    driver
-                        .call_llm_target(ctx, &target, call_request, decision)
-                        .await,
-                )
+                let aggregated = match driver
+                    .call_llm_target(ctx, &target, call_request, decision)
+                    .await
+                {
+                    Ok(Response {
+                        llm_response,
+                        metadata,
+                    }) => llm_response.into_agg().await.map(|agg| Response {
+                        llm_response: LlmResponse::Agg(agg),
+                        metadata,
+                    }),
+                    Err(error) => Err(error.into()),
+                };
+                (model, aggregated)
             });
         }
         let results = futures::future::join_all(calls).await;
 
-        // Aggregate each successful candidate before judging: a streamed candidate
-        // must be read to completion so the judge can compare its text, and the
-        // winner is returned as this buffered aggregate. A candidate whose call or
-        // stream fails is excluded rather than failing the turn.
+        // A candidate whose call or stream failed is excluded rather than failing
+        // the turn; the winner is returned as its buffered aggregate.
         let mut survivors: Vec<(String, Response)> = Vec::new();
         for (model, result) in results {
-            let Ok(response) = result else { continue };
-            let Response {
-                llm_response,
-                metadata,
-            } = response;
-            if let Ok(agg) = llm_response.into_agg().await {
-                survivors.push((
-                    model,
-                    Response {
-                        llm_response: LlmResponse::Agg(agg),
-                        metadata,
-                    },
-                ));
-            }
+            let Ok(response) = result else { continue };
+            survivors.push((model, response));
         }

Note the error types on both arms need to line up (Result/LibsyError) — adjust the conversion to whatever call_llm_target and into_agg return.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@crates/libsy/examples/ensemble.rs` around lines 313 - 335, Move
LlmResponse::into_agg aggregation into each candidate future before join_all in
the ensemble fan-out, so stream consumption remains concurrent. Update the
call_llm_target future’s success and failure branches to return a consistent
Result/LibsyError type, then collect only successful aggregated Responses into
survivors while preserving model and metadata.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@crates/libsy/examples/ensemble.rs`:
- Around line 313-335: Move LlmResponse::into_agg aggregation into each
candidate future before join_all in the ensemble fan-out, so stream consumption
remains concurrent. Update the call_llm_target future’s success and failure
branches to return a consistent Result/LibsyError type, then collect only
successful aggregated Responses into survivors while preserving model and
metadata.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Enterprise

Run ID: 64595890-9090-4864-b996-8a6b28b27915

📥 Commits

Reviewing files that changed from the base of the PR and between 6b9a168 and b545e05.

📒 Files selected for processing (1)

• crates/libsy/examples/ensemble.rs

[elyasmnvidian]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.

[github-actions]

PR Preview Action v1.8.1
🚀 View preview at https://NVIDIA-NeMo.github.io/Switchyard/pr-preview/pr-202/
Built to branch gh-pages at 2026-08-04 04:48 UTC. Preview will be ready when the GitHub Pages deployment is complete.