gateway: close mutation-test gaps; deep mutants in 32 two-stage shards

.cargo/mutants.toml

@@ -21,4 +21,75 @@ exclude_re = [
   # The same guard's operator mutation (it is the only `>=` in `logging` today; re-check this
   # exclusion if another one is added there).
   'replace >= with < in <impl ProxyHttp for AiProxy>::logging',
+  # `pack` ORs fields masked into disjoint bit ranges, where `|` and `^` are the same operation.
+  'replace \| with \^ in CircuitBreaker::pack',
+  # `k < l`: `k` is a dropped item and `l` a kept one, so `k == l` never happens.
+  'replace < with <= in remove_items',
+  # `state << 3 | class`: the shifted state's low three bits are zero and a class is below 8, so
+  # `|` and `^` agree (the TALLY index in `run` and `step`).
+  'replace \| with \^ in InputTally::(run|step)',
+  # `close` clears COLON and ESCAPE (distinct bits, so `^` is `|`). ESCAPE is never set when it
+  # runs, and in valid JSON a `,` `:` `]` or `}` follows every closing quote, so `Structure`
+  # re-sets COLON before reading it: clearing neither (`&`) changes no count on a valid body.
+  'replace \| with (&|\^) in InputTally::close',
+  # The drained `max_tokens` is the last member only when `max_completion_tokens` precedes it
+  # (both are present on this path), so the byte before it is always the separating comma.
+  '''replace match guard body\[comma\] == b',' with true in rename_max_tokens''',
+  # `model_span` and the `stream_options` value are distinct root values: `s == at` never holds.
+  'replace > with >= in force_include_usage',
+  # A one-line delegation to `summary_line` (tested); pingora calls it only on its own error lines.
+  'replace <impl ProxyHttp for AiProxy>::request_summary -> String',
+  # `ArgsEnd::whole`'s first line is a memo: `invalid` (a closed value that failed to parse, which
+  # no later byte can fix) and `failed_at` (this exact length already failed) each skip a re-parse
+  # whose answer is the same `false`. Either operator gives the same result, only slower. (It is
+  # the only `||` in `whole`; re-check this exclusion if another one is added there.)
+  'replace \|\| with && in ArgsEnd::whole',
+  # `KeepInputs::expecting` only words a parse error's message, which `parse_body` discards (it
+  # keeps `None`); `visit_unit`'s `Value::Null` is `Value::default()`.
+  "replace <impl serde::de::Visitor<'de> for KeepInputs>::(expecting|visit_unit)",
+  # `ChatIdentity::key` packs three masked fields into disjoint bit ranges: `|` and `^` agree. The
+  # key is only compared for equality, and XOR with a field's mask is a bijection on the values in
+  # range, so `&` -> `^` keeps every key distinct too.
+  'replace (\||&) with \^ in ChatIdentity::key',
+  # `tally_eager` only picks where the input estimate is made: as the body streams, or in `logging`
+  # from the copy that is still held (pingora's retry buffer or the `FullBody` parent). Both tally
+  # the same request text; widening it (`||`, or `true`) only tallies eagerly where the held copy
+  # would have done.
+  'replace (&& with \|\| in tally_eager|tally_eager -> bool with true)',
+  # `map_response` clones a same-wire body as is: the guard only skips that clone.
+  'replace match guard client == Endpoint::Responses with false in SseBridge::assembled',
+  # `str_is`'s raw compare is a fast path: `escaped_key_is` decodes a string with no escape to
+  # itself, so skipping the raw compare (`false`) gives the same answer, only slower.
+  '''replace match guard !inner\.contains\(&b'\\\\'\) with false in str_is''',
+  # `unserved` ORs `1 << i` for distinct indices `i`, where `|` and `^` are the same operation.
+  'replace \| with \^ in unserved',
+  # The const `TALLY` / `TALLY4` tables pack `state << 3 | class` and `next | add << 4` from
+  # disjoint bit ranges, where `|` and `^` agree. A const initializer's mutant names no function,
+  # only the file and position.
+  'usage\.rs:\d+:\d+: replace \| with \^$',
+  # `de_service_tier`'s visitors for a non-string tier answer `Ok(None)`, which is `Option`'s
+  # default.
+  'replace de_service_tier::<impl serde::de::Visitor<.de> for V>::visit_(bool|i64|u64|f64|none|unit) -> Result<Self::Value, E> with Ok\(Default::default\(\)\)',
+  # A push that leaves the buffer at exactly the cap: `>=` "compacts" it with a self-copy and a
+  # no-op truncate, and `ring()` reads it as a ring either way.
+  'replace > with >= in UsageTail::push',
+  # `release_body` with nothing held: `BodyBudget::release(0)` changes nothing; the guard only skips
+  # an atomic.
+  'replace > with >= in Held::release_body',
+  # `settle_health`'s undecided arm stops the work at the cap or the body's end. Without it the
+  # verdict stays pending, and a pending verdict only re-reads the same capped prefix (it never
+  # counts a failure from it), so nothing observable changes. (The guard's `||` is the only one in
+  # `settle_health`; re-check this exclusion if another one is added there.)
+  'replace match guard end_of_stream \|\| pending\.prefix\.len\(\) >= HEALTH_PREFIX_CAP with false in AiProxy::settle_health',
+  'replace \|\| with && in AiProxy::settle_health',
+  # No `anthropic-beta` header: the rebuild finds no token to keep and removes a header that is not
+  # there. The same headers, one lookup more.
+  'delete match arm \(None, _\) in retain_managed_client_headers',
+  # `mask_all` with `at *= …`: `at` stays 0, so each search restarts at the front, where every hit
+  # so far now reads `[redacted]` (or its `*` padding) and no longer matches: the same hits, found
+  # in quadratic time. Only a pool key spelled like the marker itself could tell them apart.
+  'replace \+= with \*= in mask_all',
+  # `Ctx` implements `Deref` as `DerefMut`'s supertrait: every hook holds the context mutably, so
+  # nothing calls `deref` itself (a panicking body passes the integration suites).
+  'replace <impl std::ops::Deref for Ctx>::deref -> &Self::Target',
 ]

.github/scripts/survivor-diff.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+# Turn a cargo-mutants survivor list (file:line:col: desc) into a unified diff that marks exactly
+# those lines as added, for `cargo mutants --in-diff`. Usage: survivor-diff.py SRC survivors.txt > d.diff
+import sys, collections, difflib
+src, surv = sys.argv[1], sys.argv[2]
+lines = collections.defaultdict(set)
+for l in open(surv):
+    l = l.strip()
+    if not l: continue
+    f, ln = l.split(':')[0], int(l.split(':')[1])
+    lines[f].add(ln)
+for f, lns in sorted(lines.items()):
+    new = open(f"{src}/{f}").read().splitlines(keepends=True)
+    old = [x for i, x in enumerate(new, 1) if i not in lns]
+    sys.stdout.writelines(difflib.unified_diff(old, new, f"a/{f}", f"b/{f}", n=0))

.github/workflows/ci.yml

@@ -318,6 +318,10 @@ jobs:
         with:
           tool: nextest,cargo-mutants
       - name: mutate the lines this PR changed
+        # No debug info: mutants only need pass/fail, and each mutant rebuilds the test binaries.
+        env:
+          CARGO_PROFILE_DEV_DEBUG: 0
+          CARGO_PROFILE_TEST_DEBUG: 0
         run: |
           git diff "origin/${{ github.base_ref }}...HEAD" -- crates/gateway/src crates/providers/src > "$RUNNER_TEMP/pr.diff"
           if [ ! -s "$RUNNER_TEMP/pr.diff" ]; then echo "no gateway or catalog source changes"; exit 0; fi

.github/workflows/deep.yml

@@ -6,6 +6,15 @@ on:
   schedule:
     - cron: "0 7 * * 0"
   workflow_dispatch:
+    inputs:
+      shards:
+        description: "JSON list of mutants shards to run (default: all 32)"
+        required: false
+        default: ""
+      only_mutants:
+        description: "Skip the property and catalog jobs"
+        type: boolean
+        default: false
 env:
   CARGO_TERM_COLOR: always
   CARGO_INCREMENTAL: "0"
@@ -14,36 +23,68 @@ jobs:
   # job stays well under GitHub's 6-hour ceiling. A surviving mutant is code no test checks: the job
   # fails and lists it.
   mutants:
-    name: mutants (shard ${{ matrix.shard }}/8)
+    name: mutants (shard ${{ matrix.shard }}/32)
     runs-on: ubuntu-latest
     timeout-minutes: 300
     strategy:
       fail-fast: false
       matrix:
-        shard: [0, 1, 2, 3, 4, 5, 6, 7]
+        shard: ${{ fromJSON(inputs.shards || '[0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31]') }}
     steps:
       - uses: actions/checkout@v7
       - uses: jdx/mise-action@v4
       - uses: Swatinem/rust-cache@v2
       - uses: taiki-e/install-action@v2
         with:
           tool: nextest,cargo-mutants
-      - run: |
-          cargo mutants -p beyond-ai --test-tool nextest --shard ${{ matrix.shard }}/8 \
-            --file crates/gateway/src/usage.rs --file crates/gateway/src/translate.rs \
+      # No debug info: mutants only need pass/fail, and each mutant rebuilds the test binaries, so
+      # full debug info writes several GB per mutant (it saturated the local SSD, and likely filled
+      # runner disks: shards were cancelled mid-run).
+      # Two stages per shard: every mutant against the fast lib unit tests, then only the
+      # survivors against the full suite. Stage 2's missed.txt is the real list.
+      - env:
+          CARGO_PROFILE_DEV_DEBUG: 0
+          CARGO_PROFILE_TEST_DEBUG: 0
+        run: |
+          FILES="--file crates/gateway/src/usage.rs --file crates/gateway/src/translate.rs \
             --file crates/gateway/src/proxy.rs --file crates/gateway/src/peek.rs \
-            --file crates/gateway/src/circuit_breaker.rs --file crates/gateway/src/route.rs \
-            --jobs 2 --timeout-multiplier 2 </dev/null
+            --file crates/gateway/src/circuit_breaker.rs --file crates/gateway/src/route.rs"
+          cargo mutants -p beyond-ai --test-tool nextest --shard ${{ matrix.shard }}/32 $FILES \
+            --jobs 2 --timeout-multiplier 3 --minimum-test-timeout 180 --output s1 -- --lib </dev/null || true
+          cat s1/mutants.out/missed.txt s1/mutants.out/timeout.txt 2>/dev/null | sort -u > survivors.txt || true
+          echo "stage 1 survivors: $(wc -l < survivors.txt)"
+          if [ -s survivors.txt ]; then
+            # Select exactly the survivors: `--in-diff` narrows to their lines, an anchored `--re` to
+            # their full names (a description alone matches the same text elsewhere). Keep $FILES:
+            # without it the toml's examine_globs apply and route.rs is skipped.
+            python3 .github/scripts/survivor-diff.py . survivors.txt > survivors.diff
+            RE=$(python3 -c "import re,sys; print('|'.join('^'+re.escape(l.strip())+'\$' for l in open(sys.argv[1]) if l.strip()))" survivors.txt)
+            cargo mutants -p beyond-ai --test-tool nextest $FILES --in-diff survivors.diff --re "$RE" \
+              --jobs 2 --timeout-multiplier 3 --minimum-test-timeout 600 --output s2 </dev/null || rc=$?
+            mkdir -p mutants.out && cp s2/mutants.out/*.txt mutants.out/ 2>/dev/null || true
+            [ "${rc:-0}" = 3 ] && exit 0
+            exit "${rc:-0}"
+          fi
       - name: missed mutants
         if: failure()
         run: cat mutants.out/missed.txt 2>/dev/null || true
+      - name: upload results
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: mutants-shard-${{ matrix.shard }}
+          path: |
+            s1/mutants.out/*.txt
+            s2/mutants.out/*.txt
+          if-no-files-found: ignore
   # The property tests (translator, usage parsers, body scanners) at 50x the PR case count. Every
   # property runs all 100,000 cases: PROPS_SECS (each property's wall-clock box, 60s by default) is
   # raised past anything a property needs, and the `deep` nextest profile lifts `ci`'s 180s
   # per-test kill and 10-minute run cap to match. The box stays as a backstop below the per-test
   # kill, so a property that somehow runs away still reports how many cases it ran.
   properties:
     name: property tests (deep)
+    if: ${{ !inputs.only_mutants }}
     runs-on: ubuntu-latest
     timeout-minutes: 330
     steps:
@@ -63,6 +104,7 @@ jobs:
   # ahead, so the row is removed before the vendor's shutdown.
   catalog-retirements:
     name: catalog retirements (strict)
+    if: ${{ !inputs.only_mutants }}
     runs-on: ubuntu-latest
     timeout-minutes: 30
     steps:

crates/gateway/ARCHITECTURE.md

@@ -510,9 +510,10 @@ stays a byte relay. `/{provider}/…` never translates.
 **Stream-only candidates.** Together serves Qwen3.6 Plus, Qwen3.7 Plus, Qwen3.7 Max and Qwen3.8
 Flash only as streams: a request without `"stream": true` is a 400 `streaming_required` (D147).
 `providers::catalog::stream_only` names those candidates. When one serves an attempt whose client
-did not ask for a stream, the body goes out with `"stream":true,"stream_options":{"include_usage":true}`
-spliced first and any `stream` / `stream_options` the client sent cut out by span
-(`translate::force_stream`; onto a Messages path only `stream`). Every other byte is unchanged.
+did not ask for a stream, the body goes out with `"stream":true` spliced first and any `stream` the
+client sent cut out by span (`translate::force_stream`). Its usage is asked for by the splice every
+Chat Completions stream gets (an injected `stream_options`, or the client's own made to include
+usage). Every other byte is unchanged.
 The streamed answer is
 assembled into the client's ordinary JSON body: `TranslateState::assemble` gives the attempt a
 `SseBridge::assembling` bridge. That is the bridge a Responses client's stream already uses, which

crates/gateway/src/circuit_breaker.rs

@@ -149,15 +149,6 @@ impl CircuitBreakerConfig {
         self.half_open_permits = permits;
         self
     }
-
-    /// Get the failure threshold from the policy.
-    #[allow(dead_code)]
-    fn threshold(&self) -> u32 {
-        match &self.failure_policy {
-            FailurePolicy::Consecutive { threshold } => *threshold,
-            FailurePolicy::Windowed { threshold, .. } => *threshold,
-        }
-    }
 }
 
 /// Lock-free circuit breaker.
@@ -460,7 +451,7 @@ impl CircuitBreaker {
                 // Consecutive, an expired window, or HALF_OPEN (a probe succeeded → close the
                 // circuit): reset the counts and re-anchor the window.
                 (STATE_CLOSED | STATE_HALF_OPEN, _) => Self::pack(STATE_CLOSED, 0, 0, now),
-                (STATE_OPEN, _) => return, // Shouldn't record success while open
+                // OPEN records no success; any other word is invalid.
                 _ => return,
             };
 
@@ -542,7 +533,7 @@ impl CircuitBreaker {
                 STATE_HALF_OPEN if Self::stale(scope, failures) => return,
                 // Keep the generation; the next half-open bumps it.
                 STATE_HALF_OPEN => Self::pack(STATE_OPEN, failures, 0, now),
-                STATE_OPEN => return,
+                // OPEN is already open; any other word is invalid.
                 _ => return,
             };
 
@@ -606,7 +597,7 @@ impl CircuitBreaker {
                 }
                 STATE_HALF_OPEN if Self::stale(scope, failures) => return,
                 STATE_HALF_OPEN => Self::pack(STATE_OPEN, failures, 0, now),
-                STATE_OPEN => return,
+                // OPEN is already open; any other word is invalid.
                 _ => return,
             };
 
@@ -1713,4 +1704,46 @@ mod mutation_gaps {
         assert_eq!(cb.state(), CircuitState::Closed { failure_count: 0 });
         assert!(cb.allow().is_ok());
     }
+
+    /// A probe generation wraps through 0 (the 14-bit field), and a probe of generation 0 still
+    /// closes the breaker on success: the no-op fast path is for a clean CLOSED word only.
+    /// claim: R6, REL-6
+    #[test]
+    fn a_generation_zero_probe_success_closes_the_breaker() {
+        // Opening at t = 16383 seeds the generation at the field's max, so half-opening wraps it
+        // to 0.
+        fn clock() -> u64 {
+            FAILURE_MASK
+        }
+        let cb = CircuitBreaker::with_clock(
+            CircuitBreakerConfig::consecutive(1)
+                .reset_timeout(Duration::ZERO)
+                .half_open_permits(1),
+            clock,
+        );
+        cb.record_failure();
+        let probe = cb.allow().expect("half-open probe");
+        assert_eq!(probe, Permit::probe(0), "the generation wrapped to 0");
+        cb.record_success_for(probe);
+        assert_eq!(cb.state(), CircuitState::Closed { failure_count: 0 });
+    }
+
+    /// The breaker's `Debug` prints its config, not an empty struct.
+    /// claim: R6
+    #[test]
+    fn debug_names_the_breaker_and_its_config() {
+        let cb = CircuitBreaker::new(CircuitBreakerConfig::consecutive(7));
+        let s = format!("{cb:?}");
+        assert!(
+            s.starts_with("CircuitBreaker {") && s.contains("threshold: 7"),
+            "{s}"
+        );
+    }
+
+    /// A refusal reads as what it is when it reaches a log line or an error chain.
+    /// claim: R6
+    #[test]
+    fn an_open_circuit_says_so() {
+        assert_eq!(CircuitOpen.to_string(), "circuit breaker is open");
+    }
 }

crates/gateway/src/control.rs

@@ -142,6 +142,13 @@ impl Walk {
         (i < self.len).then_some(self.indices[i as usize])
     }
 
+    /// The walked catalog indices as a bitmask (bit `orig` set for `arms[orig]`).
+    pub fn mask(self) -> u8 {
+        (0..self.len)
+            .filter_map(|i| self.catalog_index(i))
+            .fold(0u8, |m, orig| m | (1 << orig))
+    }
+
     /// `ProviderId::index` bytes in walk order — the exact-match cache's per-arm discriminator.
     pub fn provider_ids(self, candidates: &[Candidate]) -> ([u8; MAX_CANDIDATES], u8) {
         let mut ids = [0u8; MAX_CANDIDATES];