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

[elyasmnvidian]

What the ensemble example does

crates/libsy/examples/ensemble.rs is a runnable demo of a custom router built on libsy's Algorithm trait — the interface a router implements, where run handles one request (cargo run --example ensemble). For the first few requests it sends each request to several models at once, asks a separate model to pick the best answer, and then commits to whichever model won most often — every later request goes straight to that one model.

Terms used below:

• ensemble request (turn): one request the router sends to every candidate at once and then has judged.
• candidate: one of the models the router is trying out (this demo has three). A candidate call is the single call to one candidate during a turn.
• judge: a separate model that reads the candidates' answers and picks the best one.
• committed route: once the exploration turns are done, the router picks the model that won most often and sends every later request straight to it — no candidates, no judge.

Problem

The candidate models never saw the caller's real request. Each candidate call kept only the last user message as plain text; the caller's system message, tools, tool choice, sampling settings, and stream flag were dropped before the call. A caller who sent this:

let request = LlmRequest {
    messages: vec![
        Message::text(Role::System, "be terse"),
        Message::text(Role::User, "add 2 and 2"),
    ],
    tools: vec![calc_tool()],
    tool_choice: Some(ToolChoice::Required),
    sampling: SamplingParams { temperature: Some(0.3), ..Default::default() },
    stream: true,
    ..Default::default()
};

What happens

Every candidate call — and the committed route — was rebuilt as text_request(model, prompt_text(request)), which keeps only the last user message. So no matter what the caller sent, each candidate received just this:

LlmRequest {
    messages: vec![Message::text(Role::User, "add 2 and 2")],
    // system message, tools, tool_choice gone;
    // sampling back to default; stream = false
    ..Default::default()
}

Every candidate answered a smaller, different question than the caller asked, and the winning model used the same stripped request. Separately, the exploration counter advanced only after the concurrent candidate calls returned, so two requests that arrived together both explored past exploration_turns; and a turn that errored never returned its slot, so the next request waited forever for the exploration budget to free up.

The fix

Two changes, both in the example:

1. Forward the caller's request unchanged. Candidates and the committed route now send the caller's request as-is, so every model answers the real question — same messages, tools, tool choice, sampling, and stream. Because stream now reaches candidates, each candidate's answer is read to the end inside its own future before judging. That does two things: the judge compares the real text instead of empty text, and the candidate streams are read at the same time instead of one after another.
2. Cap concurrent explorations and free a slot on failure. The router counts how many explorations are in flight behind a mutex. A request now reserves a slot in that count before any async work, and a drop-guard returns the slot if the turn errors or its future is dropped (a client disconnect). Concurrent requests now run exactly exploration_turns explorations, and a failed turn can no longer hang the next request.

Evidence

The judge scores the candidates' real streamed answers instead of empty text:

Response 1: answer from a/model
Response 2: answer from b/model

• Two requests arriving together run exactly one exploration when exploration_turns = 1: the judge is consulted once, and the second request commits from the finished tally instead of starting a second exploration.
• A turn whose candidates all fail returns its slot, so the next request still explores instead of hanging.

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

How tested

crates/libsy/examples/ensemble.rs, each test driving the real Algorithm::run:

• candidate_calls_preserve_the_full_structured_request / committed_route_preserves_the_full_structured_request — candidates and the committed route receive the caller's LlmRequest unchanged (messages, tools, tool_choice, sampling, stream).
• streamed_candidates_are_judged_on_their_real_text — the judge sees both streamed answers; the winner returns as buffered text.
• a_candidate_whose_stream_fails_is_excluded_not_fatal — a stream that breaks partway excludes that candidate instead of failing the turn.
• concurrent_requests_do_not_exceed_the_exploration_budget — two concurrent requests run one exploration.
• a_failed_exploration_frees_the_slot_for_the_next_request — a failed turn returns its slot; the next request finishes within a timeout instead of deadlocking.

[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 directly summarizes the two main changes: forwarding complete ensemble requests to candidates and implementing concurrent exploration budget limits.

---

_{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.