fix(moa): account for aborted workers in worker_summaries

PR #566 review feedback (Apr 2026):

> Worker accounting was inconsistent:
> - Similar requests reported different x-moa-workers values.
> - Similar requests reported different x-moa-workers-ok values.
> - Some successful responses used fewer workers than expected.
> - Some churn responses still appeared to report stale worker counts.

`worker_summaries.len()` is what the `x-moa-workers` header reports.
When the arbiter early-exits on consensus, the gateway called
`JoinSet::abort_all` and then drained `join_next()` with
`if let Ok(...)`. `JoinSet::abort_all` causes aborted tasks to
return `Err(JoinError::cancelled)`, with no `(model, role)`
payload \u2014 those tasks were silently dropped from `summaries`. A
4-worker fan-out that early-exited from 2 fast workers reported
`x-moa-workers: 2`, hiding the fact that 2 workers were cancelled
mid-flight. Panicked tasks had the same problem.

## Test (added first, observed failing)

`crates/mesh-mixture-of-agents/tests/sim_worker_accounting.rs` sets
up 4 mock backends, 2 fast/agreeing and 2 slow, and asserts that
`worker_summaries.len() == 4` after early-exit \u2014 i.e. that the
header faithfully reflects the dispatched count.

The test fails against the pre-fix gateway with output

> Got 2 summaries: ["fast-a-3b", "fast-b-3b"]; expected 4.

## Fix

* `fanout.rs` \u2014 `gather_workers_incremental` now takes the
  dispatched-worker list (`&[DispatchedWorker]`) instead of just a
  count. After fan-out finishes (whether via normal completion or
  early-exit drain), `reconcile_dispatched` walks the dispatched
  list and synthesizes a `succeeded: false` summary for any worker
  whose name does not appear in `summaries`. Aborted tasks and
  panicked tasks are now both attributed.
* `lib.rs` \u2014 builds a `Vec<DispatchedWorker>` alongside the
  `JoinSet` and threads it through to `gather_workers_incremental`.

The header `x-moa-workers` now always equals the worker count we
actually dispatched. `x-moa-workers-ok` continues to reflect
genuinely-succeeded workers only.

## Validation

`cargo test -p mesh-mixture-of-agents` \u2014 70 unit + 2 integration
tests pass (new test plus the existing all-workers-fail one).

`cargo test -p mesh-llm-host-runtime --lib` \u2014 1434/1434 pass.

`cargo clippy -p mesh-mixture-of-agents --all-targets -- -D warnings`
\u2014 clean.

`cargo fmt --all -- --check` \u2014 clean.

crates/mesh-mixture-of-agents/src/fanout.rs

@@ -12,9 +12,18 @@ use crate::worker::WorkerRole;
 use crate::{arbiter, normalize, WorkerSummary};
 use normalize::WorkerOutput;
 
+/// Identifier for a worker we dispatched. Used to reconcile the
+/// per-worker accounting at the end of fan-out so the — possibly
+/// aborted or panicked — task's existence still shows up in
+/// `worker_summaries`.
+pub(crate) struct DispatchedWorker {
+    pub model: String,
+    pub role: WorkerRole,
+}
+
 pub(crate) async fn gather_workers_incremental(
     join_set: &mut tokio::task::JoinSet<(String, WorkerRole, Result<String, String>, u64)>,
-    total_workers: usize,
+    dispatched: &[DispatchedWorker],
     has_tools: bool,
     allowed_tools: &[String],
 ) -> (
@@ -22,6 +31,7 @@ pub(crate) async fn gather_workers_incremental(
     Vec<WorkerSummary>,
     Option<arbiter::Decision>,
 ) {
+    let total_workers = dispatched.len();
     let mut outputs = Vec::new();
     let mut summaries = Vec::new();
     let mut total_finished: usize = 0;
@@ -55,19 +65,8 @@ pub(crate) async fn gather_workers_incremental(
                 if let Some(decision) =
                     arbiter::try_early_decision(&outputs, total_workers, total_finished, has_tools)
                 {
-                    join_set.abort_all();
-                    while let Some(leftover) = join_set.join_next().await {
-                        if let Ok((m, r, result, el)) = leftover {
-                            summaries.push(WorkerSummary {
-                                model: m,
-                                role: r,
-                                succeeded: result.is_ok(),
-                                elapsed_ms: el,
-                                output_kind: None,
-                                confidence: None,
-                            });
-                        }
-                    }
+                    drain_after_early_exit(join_set, &mut summaries).await;
+                    reconcile_dispatched(dispatched, &mut summaries);
                     return (outputs, summaries, Some(decision));
                 }
             }
@@ -92,28 +91,67 @@ pub(crate) async fn gather_workers_incremental(
                 if let Some(decision) =
                     arbiter::try_early_decision(&outputs, total_workers, total_finished, has_tools)
                 {
-                    join_set.abort_all();
-                    while let Some(leftover) = join_set.join_next().await {
-                        if let Ok((m, r, result, el)) = leftover {
-                            summaries.push(WorkerSummary {
-                                model: m,
-                                role: r,
-                                succeeded: result.is_ok(),
-                                elapsed_ms: el,
-                                output_kind: None,
-                                confidence: None,
-                            });
-                        }
-                    }
+                    drain_after_early_exit(join_set, &mut summaries).await;
+                    reconcile_dispatched(dispatched, &mut summaries);
                     return (outputs, summaries, Some(decision));
                 }
             }
             Err(e) => {
                 total_finished += 1;
-                tracing::warn!("moa: worker task panicked: {e}");
+                tracing::warn!("moa: worker task panicked or was cancelled: {e}");
+                // No (model, role) payload available from a JoinError, so
+                // we cannot attribute this slot here. `reconcile_dispatched`
+                // at the end picks up any dispatched worker that has not
+                // produced a summary by name.
             }
         }
     }
 
+    reconcile_dispatched(dispatched, &mut summaries);
     (outputs, summaries, None)
 }
+
+/// After `abort_all`, drain any tasks that did finish before the abort
+/// reached them, recording each as a summary. Aborted tasks produce a
+/// `JoinError::cancelled` which carries no `(model, role)` payload —
+/// those are reconciled by [`reconcile_dispatched`] using the dispatch
+/// list.
+async fn drain_after_early_exit(
+    join_set: &mut tokio::task::JoinSet<(String, WorkerRole, Result<String, String>, u64)>,
+    summaries: &mut Vec<WorkerSummary>,
+) {
+    join_set.abort_all();
+    while let Some(leftover) = join_set.join_next().await {
+        if let Ok((m, r, result, el)) = leftover {
+            summaries.push(WorkerSummary {
+                model: m,
+                role: r,
+                succeeded: result.is_ok(),
+                elapsed_ms: el,
+                output_kind: None,
+                confidence: None,
+            });
+        }
+    }
+}
+
+/// Ensure every dispatched worker appears in `summaries`. Anything we
+/// dispatched that didn't produce a summary by name (aborted by
+/// early-exit, panicked, or otherwise lost) gets a synthesized
+/// `succeeded: false` entry so the `x-moa-workers` header faithfully
+/// reflects the dispatched count.
+fn reconcile_dispatched(dispatched: &[DispatchedWorker], summaries: &mut Vec<WorkerSummary>) {
+    for w in dispatched {
+        if summaries.iter().any(|s| s.model == w.model) {
+            continue;
+        }
+        summaries.push(WorkerSummary {
+            model: w.model.clone(),
+            role: w.role,
+            succeeded: false,
+            elapsed_ms: 0,
+            output_kind: None,
+            confidence: None,
+        });
+    }
+}

crates/mesh-mixture-of-agents/src/lib.rs

@@ -203,6 +203,7 @@ async fn handle_query(
     );
 
     let mut join_set = tokio::task::JoinSet::new();
+    let mut dispatched: Vec<fanout::DispatchedWorker> = Vec::with_capacity(assignments.len());
 
     for assignment in &assignments {
         let packed = context::pack_for_worker(session, assignment.role, has_tools);
@@ -211,6 +212,11 @@ async fn handle_query(
         let backend = config.backends[assignment.backend_index].clone();
         let timeout = config.worker_timeout;
 
+        dispatched.push(fanout::DispatchedWorker {
+            model: model_name.clone(),
+            role,
+        });
+
         join_set.spawn(async move {
             let t0 = Instant::now();
             let result = call_backend(
@@ -228,9 +234,8 @@ async fn handle_query(
         });
     }
 
-    let total_workers = join_set.len();
     let (outputs, summaries, early_decision) =
-        gather_workers_incremental(&mut join_set, total_workers, has_tools, allowed_tools).await;
+        gather_workers_incremental(&mut join_set, &dispatched, has_tools, allowed_tools).await;
 
     if outputs.is_empty() {
         return TurnResult {

crates/mesh-mixture-of-agents/tests/sim_worker_accounting.rs

@@ -0,0 +1,189 @@
+//! Pin the worker-accounting contract for `TurnResult.worker_summaries`.
+//!
+//! Background — PR #566 review feedback (Apr 2026):
+//!
+//! > Worker accounting was inconsistent:
+//! > - Similar requests reported different `x-moa-workers` values.
+//! > - Similar requests reported different `x-moa-workers-ok` values.
+//! > - Some successful responses used fewer workers than expected.
+//! > - Some churn responses still appeared to report stale worker counts.
+//! > - One concurrency response reported zero successful workers.
+//!
+//! `worker_summaries.len()` is what the `x-moa-workers` header reports.
+//! If the gateway dispatches N workers, the summaries must reflect all
+//! N of them once the turn finishes — including ones that were aborted
+//! by early-exit consensus. Otherwise a client cannot tell whether the
+//! response came from 2 workers because we only dispatched 2 or because
+//! 2 more were cancelled mid-flight.
+//!
+//! Two regressions this file pins:
+//!
+//! 1. **Early-exit aborts must still be accounted for.** When the
+//!    arbiter decides from the first two workers and aborts the rest,
+//!    the cancelled workers' summaries are silently dropped today
+//!    because `gather_workers_incremental` drains `join_next()` with
+//!    `if let Ok(...)` and `JoinSet::abort_all` returns
+//!    `Err(JoinError::cancelled)` for the aborted tasks.
+//!
+//! 2. **All-fail must still attribute every worker.** When every
+//!    backend errors, `worker_summaries.len()` must still equal the
+//!    number of workers dispatched. (This one passes today because the
+//!    `Err` branch in `gather_workers_incremental` pushes a summary
+//!    regardless — `sim_all_workers_fail.rs` already covers it.)
+
+use async_trait::async_trait;
+use mesh_mixture_of_agents as moa;
+use serde_json::{json, Value};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+/// Backend that returns a deterministic answer text after a configurable
+/// delay. Used to set up early-exit consensus.
+struct DelayedAnswerBackend {
+    text: String,
+    delay: Duration,
+    calls: AtomicUsize,
+}
+
+impl DelayedAnswerBackend {
+    fn new(text: impl Into<String>, delay: Duration) -> Arc<Self> {
+        Arc::new(Self {
+            text: text.into(),
+            delay,
+            calls: AtomicUsize::new(0),
+        })
+    }
+}
+
+#[async_trait]
+impl moa::ModelBackend for DelayedAnswerBackend {
+    async fn chat_completion(
+        &self,
+        _model: &str,
+        _messages: &[Value],
+        _tools: Option<&Value>,
+        _max_tokens: u32,
+        _timeout: Duration,
+        _sampling: moa::SamplingParams,
+    ) -> Result<Value, String> {
+        self.calls.fetch_add(1, Ordering::SeqCst);
+        tokio::time::sleep(self.delay).await;
+        Ok(json!({
+            "choices": [{"message": {"content": self.text}}],
+        }))
+    }
+}
+
+fn four_workers_two_fast_consensus() -> moa::GatewayConfig {
+    // Two fast workers that agree on the same answer ("Tokyo") → arbiter
+    // should early-exit on consensus and abort the two slow workers.
+    let fast_a = DelayedAnswerBackend::new("Tokyo", Duration::from_millis(20));
+    let fast_b = DelayedAnswerBackend::new("Tokyo", Duration::from_millis(40));
+    let slow_a = DelayedAnswerBackend::new("Tokyo", Duration::from_secs(5));
+    let slow_b = DelayedAnswerBackend::new("Tokyo", Duration::from_secs(5));
+
+    let backends: Vec<Arc<dyn moa::ModelBackend>> = vec![fast_a, fast_b, slow_a, slow_b];
+    let models = vec![
+        moa::ModelEntry {
+            name: "fast-a-3b".into(),
+            backend_index: 0,
+        },
+        moa::ModelEntry {
+            name: "fast-b-3b".into(),
+            backend_index: 1,
+        },
+        // The two big-tier models — one of them will be picked as
+        // `Strong`. They are deliberately slow so the early-exit path
+        // is hit before they finish.
+        moa::ModelEntry {
+            name: "slow-a-32b".into(),
+            backend_index: 2,
+        },
+        moa::ModelEntry {
+            name: "slow-b-32b".into(),
+            backend_index: 3,
+        },
+    ];
+
+    moa::GatewayConfig {
+        backends,
+        models,
+        worker_timeout: Duration::from_secs(10),
+        hedge_delay: Duration::from_millis(200),
+        reducer_timeout: Duration::from_secs(2),
+    }
+}
+
+fn user_turn(content: &str) -> Value {
+    json!({
+        "model": "mesh",
+        "messages": [{"role": "user", "content": content}],
+        "max_tokens": 32,
+    })
+}
+
+#[tokio::test]
+async fn early_exit_summaries_account_for_aborted_workers() {
+    // Bias the runtime to advance through sleeps as fast as possible —
+    // we just want the order: fast_a finishes, fast_b finishes, consensus
+    // → abort the two slow workers.
+    let config = four_workers_two_fast_consensus();
+    let body = user_turn("What is the capital of Japan? One word only.");
+
+    let result = moa::handle_turn(&config, &body).await;
+
+    // Sanity: this should be the early-exit path.
+    assert_eq!(
+        result.turn_kind,
+        moa::TurnKind::EarlyExit,
+        "two fast agreeing workers should produce TurnKind::EarlyExit; got {:?}",
+        result.turn_kind
+    );
+
+    // The contract: we dispatched 4 workers. Every dispatched worker
+    // must appear in `worker_summaries`, with `succeeded` reflecting
+    // its true fate (succeeded / failed / aborted). The header
+    // `x-moa-workers` is built from `worker_summaries.len()`; if it is
+    // less than the dispatched count, the client cannot tell whether
+    // we ran 2 workers or 4-with-2-aborted.
+    let dispatched = 4;
+    let summary_count = result.worker_summaries.len();
+    let summary_models: Vec<&str> = result
+        .worker_summaries
+        .iter()
+        .map(|w| w.model.as_str())
+        .collect();
+    assert_eq!(
+        summary_count, dispatched,
+        "worker_summaries.len() must equal the dispatched worker count even when \
+         early-exit aborts cancelled the remaining workers. Got {summary_count} \
+         summaries: {summary_models:?}; expected {dispatched}."
+    );
+
+    // Every dispatched model must be named.
+    for name in ["fast-a-3b", "fast-b-3b", "slow-a-32b", "slow-b-32b"] {
+        assert!(
+            summary_models.contains(&name),
+            "worker_summaries is missing model {name:?}; got {summary_models:?}"
+        );
+    }
+
+    // `succeeded` should be true for the workers that produced output
+    // before consensus, and false (or otherwise "not succeeded") for
+    // the ones we cancelled. We do not assert exactly which two — the
+    // contract is just "the count is honest."
+    let succeeded_count = result
+        .worker_summaries
+        .iter()
+        .filter(|w| w.succeeded)
+        .count();
+    assert!(
+        succeeded_count >= 2,
+        "at least the two fast workers must be marked succeeded; got {succeeded_count}"
+    );
+    assert!(
+        succeeded_count <= dispatched,
+        "succeeded_count {succeeded_count} cannot exceed dispatched {dispatched}"
+    );
+}