perf(inference): run non-streaming through the batch engine

Non-streaming chat completion was a second, independent inference path
that took the per-model `inference_lock` for the whole request. That
made it invisible to the batch engine (#98): concurrent non-streaming
callers serialized completely.

Measured on beast (TP=2, q6k, `ae994e3`, idle box, identical prompt and
400-token output at every level):

  concurrency   non-streaming        streaming
  1             32.1 tok/s  1.00x    32.0 tok/s  1.00x
  2             32.2 tok/s  1.00x    57.1 tok/s  1.79x
  4             32.1 tok/s  1.00x   105.9 tok/s  3.31x
  8             32.0 tok/s  1.00x   127.1 tok/s  3.98x

Wall time for the non-streaming arm scaled 24.9 -> 49.7 -> 99.7 ->
199.9 s: exactly linear, no coalescing whatever. `max_in_flight = 8`
bought those requests admission and nothing else. Worse, it was
invisible — the neuron reports `in_flight: 8`, admission accepts,
`/health` looks healthy, and throughput is single-stream. Nothing
distinguished it from a legitimately busy node.

The engine's doc-comment said vision and non-streaming "still serialize
safely against the batch", which reads as a chosen invariant. It was
worth re-examining rather than inheriting: nothing about buffering a
result requires bypassing the producer that makes it.

So `chat_completion` now runs `inference_stream` — the same producer the
streaming path uses — and buffers the events into a `ChatCompletionResponse`
via `wire::openai_chat::collect_chat_completion`. Both shapes now share
one producer, so batching, prefix-cache reuse, tool-call extraction, the
reasoning governor and server-measured timing apply identically.

Three things fall out for free:

  - `usage.prompt_tokens_details.cached_tokens` and `usage.helexa_timing`
    now populate on non-streaming. The old path hardcoded both to `None`
    with a comment that timing was "only surfaced on the streaming
    Finish event today (#85)"; the events carried them all along.
  - Non-streaming tool calls now go through the same incremental
    extractor as streaming, closing the known-unhandled non-streaming
    tool-call gap on the Qwen3.6/3.8 XML form.
  - The duplicate path is gone: -1,272 lines against +165. Deleted
    `chat_completion_tp`, `chat_completion_tp_inner`, `run_inference`,
    `split_off_reasoning`, `extract_tool_calls_from_text` and
    `debug_poison_armed`, all reachable only from the path this
    replaces. `parse_tool_call_body` — the actual parser, shared with
    the five streaming call sites — keeps its coverage; only the batch
    framing wrappers and their tests go.

Behaviour deliberately preserved: reasoning text is still dropped from
`content` (counted in `completion_tokens_details.reasoning_tokens`,
never rendered), and `finish_reason` still promotes to `"tool_calls"`
whenever a call was parsed. Both pinned by new tests, along with the
cached-tokens/timing surfacing and the case that matters most — a
stream ending without a `Finish` is an error, not an empty 200, because
that is how a dropped engine slot presents.

`use tracing::Instrument` is now `#[cfg(feature = "cuda")]`: both
remaining `.instrument()` call sites are inside cuda-gated blocks, so an
ungated import warns on the CPU build while an absent one breaks the
CUDA build that only branch CI compiles.

Not yet validated live — the sweep above must be re-run against this
build to confirm the non-streaming column now tracks the streaming one.

Refs #285, #98, #85, #269

.gitea/workflows/build-prerelease.yml

@@ -262,7 +262,7 @@ jobs:
       # fast instead of triggering a slower uncached rebuild; only a real
       # sccache fault drops the cache. See script/ci-cargo-escalate.sh.
       - name: Clippy (sccache escalation)
-        run: script/ci-cargo-escalate.sh cargo clippy --workspace -- -D warnings
+        run: script/ci-cargo-escalate.sh cargo clippy --workspace --all-targets -- -D warnings
 
   test:
     name: Test

.gitea/workflows/ci.yml

@@ -137,6 +137,27 @@ jobs:
       - uses: actions/checkout@v4
       - run: cargo fmt --check --all
 
+  # models.toml and asset/neuron/<host>.toml both describe how a model is
+  # served, and until #283 nothing forced them to agree. #252 is the bill
+  # for that: models.toml omitted `quant`, so a cortex cold-load served
+  # bf16 and could not prefill, while the same model was fine when the
+  # host loaded it from its own config. The sampling override (#283) is a
+  # second field with the same hazard, so the gate lands with it.
+  #
+  # Deliberately ungated by the `changes` classifier: it is a
+  # sub-second Python run with no build, and the whole point is that it
+  # cannot be skipped on the push that introduces the drift. A config
+  # change is neither `rust` nor `web`, so gating it on either would let
+  # a pure-TOML push through unchecked — precisely the push that breaks
+  # this.
+  config:
+    name: Config consistency
+    timeout-minutes: 5
+    runs-on: fedora-43
+    steps:
+      - uses: actions/checkout@v4
+      - run: python3 script/check-config-consistency.py
+
   web:
     name: Web (lint + typecheck + i18n + build)
     timeout-minutes: 20
@@ -179,8 +200,13 @@ jobs:
       # in sync with build-prerelease.yml): a signal death (rustc SIGSEGV
       # / OOM-kill) keeps the cache and fails fast instead of an uncached
       # rebuild; only a real sccache fault drops the cache.
+      # --all-targets so `#[cfg(test)]` modules, tests/, benches/ and
+      # examples/ are linted too (#286). Without it clippy sees only lib
+      # and bin targets, so `-D warnings` never gated a single line of
+      # test code — which is how two tests lost their `#[test]` attribute
+      # and stopped running unnoticed.
       - name: Clippy (sccache escalation)
-        run: script/ci-cargo-escalate.sh cargo clippy --workspace -- -D warnings
+        run: script/ci-cargo-escalate.sh cargo clippy --workspace --all-targets -- -D warnings
 
   test:
     name: Test

.gitea/workflows/deploy.yml

@@ -37,7 +37,31 @@ env:
     ${{ secrets.RSYNC_SSH_KEY }}
 
 jobs:
+  # Gate the whole rollout on the two config files agreeing about every
+  # model they both describe. models.toml (cortex, cold loads) and
+  # asset/neuron/<host>.toml (resident models) each say how a model is
+  # served — quant, and since #283 the sampling override. When they
+  # disagree the model is served one way cold and another way resident,
+  # which is #252.
+  #
+  # Deliberately a *gate*, not a step inside deploy-cortex: an
+  # inconsistent pair must stop the neuron rollout too, and stopping it
+  # before anything restarts is much cheaper than after (a beast restart
+  # costs ~5-6 min of 27B cold-load).
+  config-check:
+    name: Config consistency
+    runs-on: fedora-43
+    if: >-
+      ${{
+        github.event_name == 'workflow_dispatch'
+        || github.event.workflow_run.conclusion == 'success'
+      }}
+    steps:
+      - uses: actions/checkout@v4
+      - run: python3 script/check-config-consistency.py
+
   deploy-cortex:
+    needs: [config-check]
     runs-on: fedora-43
     # Two trigger paths: manual dispatch always runs; workflow_run
     # only runs if the upstream `build-prerelease` actually succeeded.
@@ -47,6 +71,7 @@ jobs:
         || github.event.workflow_run.conclusion == 'success'
       }}
     steps:
+      - uses: actions/checkout@v4
       - name: SSH init
         run: |
           mkdir -p ~/.ssh
@@ -65,6 +90,33 @@ jobs:
           echo "inv=$(ssh gitea_ci@hanzalova.internal \
             'systemctl show -P InvocationID cortex.service' || true)" >> "$GITHUB_OUTPUT"
 
+      # The catalogue is tracked (#283) precisely so it can be rolled out
+      # here rather than hand-copied. Sync it *before* the package step,
+      # so the restart that step performs is the one that picks it up —
+      # and report whether it changed, because a catalogue edit with no
+      # new RPM must still restart cortex or the change never takes
+      # effect.
+      #
+      # Compared by digest rather than rsync's own change detection so the
+      # answer is available to the shell that decides on the restart.
+      - name: Sync gateway catalogue (models.toml)
+        id: catalogue
+        run: |
+          local_sum=$(sha256sum models.toml | cut -d' ' -f1)
+          remote_sum=$(ssh gitea_ci@hanzalova.internal \
+            'sha256sum /etc/cortex/models.toml 2>/dev/null | cut -d" " -f1' || true)
+          if [ "${local_sum}" = "${remote_sum}" ]; then
+            echo "changed=0" >> "$GITHUB_OUTPUT"
+            echo "models.toml unchanged (${local_sum})"
+          else
+            rsync --archive --compress \
+              --chown root:root --chmod 0644 \
+              --rsync-path 'sudo rsync' \
+              models.toml gitea_ci@hanzalova.internal:/etc/cortex/models.toml
+            echo "changed=1" >> "$GITHUB_OUTPUT"
+            echo "models.toml updated: ${remote_sum:-<absent>} -> ${local_sum}"
+          fi
+
       # Gating compares `rpm -q` against the packages.json manifest the
       # publish job maintains — NOT unprivileged `dnf check-update`,
       # which proved unreliable as the gitea_ci user (hung on metadata
@@ -75,6 +127,11 @@ jobs:
           ssh gitea_ci@hanzalova.internal 'bash -s' <<'DEPLOY'
           set -eu
           pkg=cortex
+          # Default to 0 if the sync step somehow produced no output:
+          # `set -u` plus `[ "" -eq 0 ]` would abort the deploy on an
+          # arithmetic error rather than on anything meaningful.
+          config_changed="${{ steps.catalogue.outputs.changed }}"
+          config_changed="${config_changed:-0}"
           # `rpm -q` prints "package X is not installed" on STDOUT and exits
           # 1, so `|| echo not-installed` never fires — it captures the
           # message text instead. Test the exit code.
@@ -90,18 +147,31 @@ jobs:
               p = max(cands, key=lambda p: p.get("buildTime", 0))
               print(p["version"] + "-" + p["release"])
           ' "${pkg}" 2>/dev/null || true)
-          if [ -n "${latest}" ] && [ "${latest}" = "${installed}" ]; then
-            echo "${pkg}-${installed} already current — leaving service untouched"
+          # Skip only when BOTH the package and the catalogue are
+          # unchanged — a models.toml edit (a new sampling override, a
+          # corrected quant) must restart cortex even with no new RPM,
+          # exactly as the neuron drop-in does below.
+          if [ -n "${latest}" ] && [ "${latest}" = "${installed}" ] \
+             && [ "${config_changed}" -eq 0 ]; then
+            echo "${pkg}-${installed} already current; models.toml unchanged — leaving service untouched"
             exit 0
           fi
-          echo "installed=${installed} published=${latest:-unknown} — deploying"
+          pkg_changed=1
+          if [ -n "${latest}" ] && [ "${latest}" = "${installed}" ]; then
+            pkg_changed=0
+          fi
+          echo "installed=${installed} published=${latest:-unknown} pkg_changed=${pkg_changed} config_changed=${config_changed} — deploying"
           if systemctl is-active --quiet cortex.service; then
             sudo /usr/bin/systemctl stop cortex.service
           fi
-          if rpm -q "${pkg}" >/dev/null 2>&1; then
-            sudo /usr/bin/dnf upgrade --refresh --allowerasing -y cortex
-          else
-            sudo /usr/bin/dnf install --refresh --allowerasing -y cortex
+          # A catalogue-only change still restarts, but must not spend a
+          # dnf metadata refresh proving the package is already current.
+          if [ "${pkg_changed}" -eq 1 ]; then
+            if rpm -q "${pkg}" >/dev/null 2>&1; then
+              sudo /usr/bin/dnf upgrade --refresh --allowerasing -y cortex
+            else
+              sudo /usr/bin/dnf install --refresh --allowerasing -y cortex
+            fi
           fi
           sudo /usr/bin/systemctl daemon-reload
           # enable --now: start the service AND enable it for boot so the
@@ -142,7 +212,7 @@ jobs:
           CAPTURE
 
   deploy-neurons:
-    needs: [deploy-cortex]
+    needs: [config-check, deploy-cortex]
     runs-on: fedora-43
     strategy:
       # One neuron failing must not cancel the others. Cortex is up
@@ -177,6 +247,7 @@ jobs:
             load_timeout: 300
             max_prompt_tokens: 16384
     steps:
+      - uses: actions/checkout@v4
       - name: SSH init
         run: |
           mkdir -p ~/.ssh
@@ -192,6 +263,42 @@ jobs:
           echo "inv=$(ssh gitea_ci@${{ matrix.host }} \
             'systemctl show -P InvocationID neuron.service' || true)" >> "$GITHUB_OUTPUT"
 
+      # asset/neuron/<short>.toml is this host's neuron.toml. It carries
+      # the resident-model specs — including the #283 sampling override,
+      # which is inert until the daemon reloads the model, so a change
+      # here has to restart the neuron the same way the context-cap
+      # drop-in does.
+      #
+      # Synced before the package step so the restart that step performs
+      # is the one that picks it up. Digest-compared rather than trusting
+      # rsync's own change detection, because the restart decision needs
+      # the answer and a needless neuron restart costs ~5-6 min of 27B
+      # cold-load (#1).
+      - name: Sync neuron config (${{ matrix.host }})
+        id: neuronconf
+        run: |
+          short="${{ matrix.host }}"; short="${short%%.*}"
+          src="asset/neuron/${short}.toml"
+          if [ ! -f "${src}" ]; then
+            echo "changed=0" >> "$GITHUB_OUTPUT"
+            echo "no ${src} in the repo — leaving this host's config alone"
+            exit 0
+          fi
+          local_sum=$(sha256sum "${src}" | cut -d' ' -f1)
+          remote_sum=$(ssh gitea_ci@${{ matrix.host }} \
+            'sha256sum /etc/neuron/neuron.toml 2>/dev/null | cut -d" " -f1' || true)
+          if [ "${local_sum}" = "${remote_sum}" ]; then
+            echo "changed=0" >> "$GITHUB_OUTPUT"
+            echo "${src} unchanged (${local_sum})"
+          else
+            rsync --archive --compress \
+              --chown root:root --chmod 0644 \
+              --rsync-path 'sudo rsync' \
+              "${src}" gitea_ci@${{ matrix.host }}:/etc/neuron/neuron.toml
+            echo "changed=1" >> "$GITHUB_OUTPUT"
+            echo "${src} updated: ${remote_sum:-<absent>} -> ${local_sum}"
+          fi
+
       # See deploy-cortex for why gating uses the publish manifest and
       # not unprivileged `dnf check-update`.
       - name: Deploy helexa-neuron-${{ matrix.flavour }} (skips when already current)
@@ -200,6 +307,9 @@ jobs:
           set -eu
           pkg=helexa-neuron-${{ matrix.flavour }}
           max_prompt_tokens="${{ matrix.max_prompt_tokens }}"
+          # A neuron.toml change counts as a config change for the
+          # restart decision below, exactly like the drop-in.
+          neuron_toml_changed="${{ steps.neuronconf.outputs.changed }}"
 
           # ── Desired per-model systemd drop-in ─────────────────────────
           # model.conf carries NEURON_MAX_PROMPT_TOKENS so the context cap
@@ -208,7 +318,7 @@ jobs:
           # deploy-managed value wins over any manual local override of the
           # same variable. See doc/context-limits.md.
           conf=/etc/systemd/system/neuron.service.d/model.conf
-          config_changed=0
+          config_changed=${neuron_toml_changed:-0}
           if [ -n "${max_prompt_tokens}" ]; then
             desired=$(printf '%s\n%s\n%s\n%s' \
               "# Managed by .gitea/workflows/deploy.yml - do not edit by hand." \
@@ -242,13 +352,17 @@ jobs:
           # Skip only when BOTH the package and the drop-in are unchanged —
           # a context-cap change must restart the neuron even with no new RPM.
           if [ "${pkg_changed}" -eq 0 ] && [ "${config_changed}" -eq 0 ]; then
-            echo "${pkg}-${installed} current; NEURON_MAX_PROMPT_TOKENS=${max_prompt_tokens:-<unset>} unchanged — leaving service untouched"
+            echo "${pkg}-${installed} current; NEURON_MAX_PROMPT_TOKENS=${max_prompt_tokens:-<unset>} and neuron.toml unchanged — leaving service untouched"
             exit 0
           fi
           echo "installed=${installed} published=${latest:-unknown} pkg_changed=${pkg_changed} config_changed=${config_changed} — deploying"
 
           # Write the drop-in (staged in gitea_ci's dir, installed root-owned).
-          if [ "${config_changed}" -eq 1 ]; then
+          # Guarded on max_prompt_tokens as well as config_changed: since
+          # a neuron.toml change also sets config_changed, an entry with
+          # no max_prompt_tokens would otherwise reach this with `desired`
+          # unset.
+          if [ "${config_changed}" -eq 1 ] && [ -n "${max_prompt_tokens}" ]; then
             printf '%s\n' "${desired}" > /var/lib/gitea_ci/model.conf
             sudo /usr/bin/install -o root -g root -m 0644 -D /var/lib/gitea_ci/model.conf "${conf}"
             rm -f /var/lib/gitea_ci/model.conf
@@ -534,6 +648,7 @@ jobs:
           CAPTURE
 
   deploy-bench:
+    needs: [config-check]
     runs-on: fedora-43
     if: >-
       ${{
@@ -541,6 +656,7 @@ jobs:
         || github.event.workflow_run.conclusion == 'success'
       }}
     steps:
+      - uses: actions/checkout@v4
       - name: SSH init
         run: |
           mkdir -p ~/.ssh
@@ -556,6 +672,28 @@ jobs:
           echo "inv=$(ssh gitea_ci@bob.hanzalova.internal \
             'systemctl show -P InvocationID helexa-bench.service' || true)" >> "$GITHUB_OUTPUT"
 
+      # Same contract as the gateway and neuron configs: the repo is the
+      # only writer, so a config edit must reach the host and restart the
+      # service even when no new RPM landed.
+      - name: Sync bench config
+        id: benchconf
+        run: |
+          src="asset/helexa-bench/bob.toml"
+          local_sum=$(sha256sum "${src}" | cut -d' ' -f1)
+          remote_sum=$(ssh gitea_ci@bob.hanzalova.internal \
+            'sha256sum /etc/helexa-bench/helexa-bench.toml 2>/dev/null | cut -d" " -f1' || true)
+          if [ "${local_sum}" = "${remote_sum}" ]; then
+            echo "changed=0" >> "$GITHUB_OUTPUT"
+            echo "${src} unchanged (${local_sum})"
+          else
+            rsync --archive --compress \
+              --chown root:root --chmod 0644 \
+              --rsync-path 'sudo rsync' \
+              "${src}" gitea_ci@bob.hanzalova.internal:/etc/helexa-bench/helexa-bench.toml
+            echo "changed=1" >> "$GITHUB_OUTPUT"
+            echo "${src} updated: ${remote_sum:-<absent>} -> ${local_sum}"
+          fi
+
       # See deploy-cortex for why gating uses the publish manifest and
       # not unprivileged `dnf check-update`.
       - name: Deploy helexa-bench (skips when already current)
@@ -563,6 +701,8 @@ jobs:
           ssh gitea_ci@bob.hanzalova.internal 'bash -s' <<'DEPLOY'
           set -eu
           pkg=helexa-bench
+          config_changed="${{ steps.benchconf.outputs.changed }}"
+          config_changed="${config_changed:-0}"
           # `rpm -q` prints "package X is not installed" on STDOUT and exits
           # 1, so `|| echo not-installed` never fires — it captures the
           # message text instead. Test the exit code.
@@ -578,18 +718,25 @@ jobs:
               p = max(cands, key=lambda p: p.get("buildTime", 0))
               print(p["version"] + "-" + p["release"])
           ' "${pkg}" 2>/dev/null || true)
-          if [ -n "${latest}" ] && [ "${latest}" = "${installed}" ]; then
-            echo "${pkg}-${installed} already current — leaving service untouched"
+          if [ -n "${latest}" ] && [ "${latest}" = "${installed}" ] \
+             && [ "${config_changed}" -eq 0 ]; then
+            echo "${pkg}-${installed} already current; config unchanged — leaving service untouched"
             exit 0
           fi
-          echo "installed=${installed} published=${latest:-unknown} — deploying"
+          pkg_changed=1
+          if [ -n "${latest}" ] && [ "${latest}" = "${installed}" ]; then
+            pkg_changed=0
+          fi
+          echo "installed=${installed} published=${latest:-unknown} pkg_changed=${pkg_changed} config_changed=${config_changed} — deploying"
           if systemctl is-active --quiet helexa-bench.service; then
             sudo /usr/bin/systemctl stop helexa-bench.service
           fi
-          if rpm -q "${pkg}" >/dev/null 2>&1; then
-            sudo /usr/bin/dnf upgrade --refresh --allowerasing -y helexa-bench
-          else
-            sudo /usr/bin/dnf install --refresh --allowerasing -y helexa-bench
+          if [ "${pkg_changed}" -eq 1 ]; then
+            if rpm -q "${pkg}" >/dev/null 2>&1; then
+              sudo /usr/bin/dnf upgrade --refresh --allowerasing -y helexa-bench
+            else
+              sudo /usr/bin/dnf install --refresh --allowerasing -y helexa-bench
+            fi
           fi
           sudo /usr/bin/systemctl daemon-reload
           # enable --now: start the service AND enable it for boot so the

.gitignore

@@ -9,7 +9,6 @@ helexa.ai/.env.local
 .idea/
 .vscode/
 cortex.toml
-models.toml
 doc/plan/*
 /target-cuda/
 .claude/

CLAUDE.md

@@ -167,7 +167,7 @@ doc-comment; touch points (the `Job` enum, the dispatch handlers, the
 cargo build --release           # build all crates
 cargo run -p cortex-cli -- serve    # run the gateway
 cargo test                      # run all tests
-cargo clippy --workspace        # lint
+cargo clippy --workspace --all-targets   # lint (test code included)
 ```
 
 ## CI
@@ -177,7 +177,7 @@ pass before merging:
 
 ```sh
 cargo fmt --check --all                    # formatting
-cargo clippy --workspace -- -D warnings   # lint (warnings are errors)
+cargo clippy --workspace --all-targets -- -D warnings   # lint (warnings are errors)
 cargo test --workspace                     # tests
 ```
 
@@ -191,7 +191,8 @@ Work each change on its own branch; `main` stays releasable.
 
 1. Implement on a feature branch (`fix/<issue>-…`, `feat/<issue>-…`).
 2. Run the CI triad locally (`cargo fmt --check --all`,
-   `cargo clippy --workspace -- -D warnings`, `cargo test --workspace`).
+   `cargo clippy --workspace --all-targets -- -D warnings`,
+   `cargo test --workspace`).
    Local builds are **CPU-only** — the `#[cfg(feature = "cuda")]` neuron/TP
    paths do NOT compile locally. The branch CI's **CUDA type-check** job is
    the only thing that validates them, so for any neuron change the push to

asset/helexa-bench/bob.toml

@@ -1,3 +1,10 @@
+# ─────────────────────────────────────────────────────────────────────
+# MANAGED BY CI — .gitea/workflows/deploy.yml is the only writer.
+#
+# Edits made directly to the deployed copy on the host are reverted by
+# the next deploy, and until then the fleet runs something no commit
+# describes. Change it here, commit, let the pipeline ship it.
+# ─────────────────────────────────────────────────────────────────────
 # helexa-bench config for bob.hanzalova.internal.
 #
 # Synced to /etc/helexa-bench/helexa-bench.toml by script/infra-setup.sh
@@ -15,6 +22,20 @@ iteration_pause_secs = 2
 request_timeout_secs = 600
 db_path = "/var/lib/helexa-bench/bench.sqlite"
 
+# Identity bench presents to neurons (#288). Without it bench is
+# anonymous, and since #262 anonymous callers are served only from
+# leftover capacity — capped below max_in_flight and parked at the class
+# gate. That moved beast's concurrency:8 from 168.9 tok/s / 0.53s
+# ttft_p95 to 104.6 / 11.77s the day #262 landed and held there, so every
+# number since measured the yield policy, not serving capacity.
+#
+# A dedicated pair, not a borrowed one: bench then holds its own
+# fair-share allocation (#54) and cannot starve interactive traffic —
+# the failure #262 existed to fix.
+[bench.principal]
+account_id = "helexa-bench"
+key_id = "fleet-benchmark"
+
 [scenarios]
 prompt_sizes = [128, 4096]
 max_tokens = 256
@@ -22,6 +43,10 @@ max_tokens = 256
 # carries p95-under-concurrency data before the F3 A/B gate (#94) needs
 # it for comparison. Levels mirror the real a0/hermes/opencode fan-out.
 concurrency_levels = [2, 4, 8]
+# Non-streaming bursts (#285/#288). Kept to the top level: it is where
+# serialization shows most starkly and each level costs a full burst per
+# sample per build.
+concurrency_nonstreaming_levels = [8]
 concurrency_prompt_tokens = 512
 
 # Capability probes (#91) — the reasoning/planning axis the speed

asset/neuron/beast.toml

@@ -1,3 +1,10 @@
+# ─────────────────────────────────────────────────────────────────────
+# MANAGED BY CI — .gitea/workflows/deploy.yml is the only writer.
+#
+# Edits made directly to the deployed copy on the host are reverted by
+# the next deploy, and until then the fleet runs something no commit
+# describes. Change it here, commit, let the pipeline ship it.
+# ─────────────────────────────────────────────────────────────────────
 # neuron.toml for beast.hanzalova.internal
 #
 # 2x RTX 5090 (32 GB each) — TP-2 capable. Pre-warms Qwen3.6-27B with
@@ -35,3 +42,16 @@ harness = "candle"
 quant = "q6k"
 tensor_parallel = 2
 devices = [0, 1]
+
+# Sampling override (#283) — MUST match the `[models.sampling]` block on
+# this model in models.toml. Two files describe one model; only
+# script/check-config-consistency.py forces them to agree, and #252 is
+# what happens when nothing does.
+#
+# The model's own generation_config.json says temperature = 1.0, which
+# measured a 20% structural-defect rate on ~2k-token code generation
+# against 0/60 at <= 0.6 (Fisher exact one-sided p = 0.0031). Precedence
+# is request > operator > model, so a caller naming a temperature still
+# gets what it asked for.
+[default_models.sampling]
+temperature = 0.6

asset/neuron/benjy.toml

@@ -1,3 +1,10 @@
+# ─────────────────────────────────────────────────────────────────────
+# MANAGED BY CI — .gitea/workflows/deploy.yml is the only writer.
+#
+# Edits made directly to the deployed copy on the host are reverted by
+# the next deploy, and until then the fleet runs something no commit
+# describes. Change it here, commit, let the pipeline ship it.
+# ─────────────────────────────────────────────────────────────────────
 # neuron.toml for benjy.hanzalova.internal
 #
 # 1x RTX 4090 (24 GB) — largest single-GPU host on the fleet. Pre-warms

asset/neuron/quadbrat.toml

@@ -1,3 +1,10 @@
+# ─────────────────────────────────────────────────────────────────────
+# MANAGED BY CI — .gitea/workflows/deploy.yml is the only writer.
+#
+# Edits made directly to the deployed copy on the host are reverted by
+# the next deploy, and until then the fleet runs something no commit
+# describes. Change it here, commit, let the pipeline ship it.
+# ─────────────────────────────────────────────────────────────────────
 # neuron.toml for quadbrat.hanzalova.internal
 #
 # 1x RTX 3060 (12 GB) — small / quantised tier. Pre-warms Qwen3-1.7B

bench/src/pages/Trends.tsx

@@ -40,6 +40,80 @@ function Picker({
   );
 }
 
+type SeriesDef = {
+  key: string;
+  name: string;
+  stroke: string;
+  dashed?: boolean;
+};
+
+/** One titled chart over the shared build timeline.
+ *
+ * Every panel draws the same x-axis and the same regime divider, so a
+ * reader can line a change up across metrics — which is the whole point
+ * of having more than two of them. */
+function MetricChart({
+  title,
+  hint,
+  data,
+  lines,
+  divider,
+  unit,
+}: {
+  title: string;
+  hint?: string;
+  data: Record<string, unknown>[];
+  lines: SeriesDef[];
+  divider?: string;
+  unit?: string;
+}) {
+  const hasAny = data.some((d) => lines.some((l) => d[l.key] != null));
+  if (!hasAny) return null;
+  return (
+    <>
+      <h5 className="mt-4">
+        {title}
+        {unit ? <span className="text-muted fw-normal"> ({unit})</span> : null}
+      </h5>
+      {hint && <p className="text-muted small mb-2">{hint}</p>}
+      <ResponsiveContainer width="100%" height={260}>
+        <LineChart data={data} margin={{ top: 8, right: 24, bottom: 8, left: 0 }}>
+          <CartesianGrid strokeDasharray="3 3" />
+          <XAxis dataKey="label" />
+          <YAxis />
+          <Tooltip />
+          <Legend />
+          {divider && (
+            <ReferenceLine
+              x={divider}
+              stroke="#bbb"
+              strokeDasharray="3 3"
+              label={{
+                value: "bench.py → helexa-bench",
+                position: "top",
+                fill: "#999",
+                fontSize: 11,
+              }}
+            />
+          )}
+          {lines.map((l) => (
+            <Line
+              key={l.key}
+              type="monotone"
+              dataKey={l.key}
+              name={l.name}
+              stroke={l.stroke}
+              strokeDasharray={l.dashed ? "5 5" : undefined}
+              connectNulls
+              dot={false}
+            />
+          ))}
+        </LineChart>
+      </ResponsiveContainer>
+    </>
+  );
+}
+
 export default function Trends() {
   const [dims, setDims] = useState<Dimensions | null>(null);
   const [model, setModel] = useState("");
@@ -68,10 +142,7 @@ export default function Trends() {
   // Prepend the pre-helexa-bench baseline (dashed, separate keys) so it
   // anchors the timeline without being merged into the live line. Different
   // measurement regime — see baseline.ts / doc/benchmarks.md.
-  const base = useMemo(
-    () => baselineFor(model, scenario),
-    [model, scenario],
-  );
+  const base = useMemo(() => baselineFor(model, scenario), [model, scenario]);
   const data = useMemo(
     () => [
       ...base.map((p) => ({
@@ -85,6 +156,14 @@ export default function Trends() {
         ttft: p.ttft_s_median,
         decode: p.decode_tps_median,
         total: p.total_s_median,
+        ttftP95: p.ttft_p95_s_median,
+        queueWait: p.queue_wait_ms_median,
+        rejected: p.rejected_median,
+        prefillTps: p.prefill_tps_median,
+        reasoning: p.reasoning_tokens_median,
+        cached: p.cached_tokens_median,
+        completion: p.completion_tokens_median,
+        tpot: p.tpot_p95_ms_median,
       })),
     ],
     [series, base],
@@ -94,6 +173,19 @@ export default function Trends() {
   // first live build, with baseline points to its left).
   const firstLive = series[0]?.git_sha;
   const showDivider = base.length > 0 && series.length > 0;
+  const divider = showDivider ? firstLive : undefined;
+
+  // Anonymous and identified samples are not comparable once #262 is in
+  // the build: an anonymous caller is capped below max_in_flight and
+  // parked at the class gate, so it characterises the yield policy
+  // rather than serving capacity. Say so rather than letting someone
+  // read a step change as an engine regression (#288).
+  const identities = useMemo(
+    () => new Set(series.map((p) => p.principal ?? "anonymous")),
+    [series],
+  );
+  const mixedIdentity = identities.size > 1;
+  const anyAnonymous = identities.has("anonymous");
 
   if (err) return <Alert variant="danger">{err}</Alert>;
   if (!dims) return <Spinner animation="border" />;
@@ -122,6 +214,25 @@ export default function Trends() {
         </p>
       )}
 
+      {mixedIdentity && (
+        <Alert variant="warning" className="py-2">
+          <strong>Mixed measurement identity.</strong> Some builds were
+          sampled anonymously and some under a principal. Since{" "}
+          <code>#262</code> an anonymous caller is capped below{" "}
+          <code>max_in_flight</code> and yields to identified traffic, so
+          those points measure the admission policy rather than serving
+          capacity. A step change across that boundary is the instrument,
+          not the engine — see <code>#288</code>.
+        </Alert>
+      )}
+      {!mixedIdentity && anyAnonymous && (
+        <Alert variant="secondary" className="py-2 small">
+          Sampled anonymously. Since <code>#262</code> anonymous callers are
+          served from leftover capacity, so these numbers understate what an
+          authenticated caller gets (<code>#288</code>).
+        </Alert>
+      )}
+
       {data.length === 0 ? (
         <Alert variant="info">No data for this selection yet.</Alert>
       ) : (
@@ -133,87 +244,95 @@ export default function Trends() {
               see <code>doc/benchmarks.md</code>.
             </p>
           )}
-          <h5 className="mt-3">decode tok/s (higher is better)</h5>
-          <ResponsiveContainer width="100%" height={280}>
-            <LineChart data={data} margin={{ top: 8, right: 24, bottom: 8, left: 0 }}>
-              <CartesianGrid strokeDasharray="3 3" />
-              <XAxis dataKey="label" />
-              <YAxis />
-              <Tooltip />
-              <Legend />
-              {showDivider && firstLive && (
-                <ReferenceLine
-                  x={firstLive}
-                  stroke="#bbb"
-                  strokeDasharray="3 3"
-                  label={{
-                    value: "bench.py → helexa-bench",
-                    position: "top",
-                    fill: "#999",
-                    fontSize: 11,
-                  }}
-                />
-              )}
-              <Line
-                type="monotone"
-                dataKey="decode"
-                name="decode tok/s"
-                stroke="#0d6efd"
-                connectNulls
-              />
-              {base.length > 0 && (
-                <Line
-                  type="monotone"
-                  dataKey="baseDecode"
-                  name="baseline (bench.py · gateway)"
-                  stroke="#888"
-                  strokeDasharray="5 5"
-                  connectNulls
-                />
-              )}
-            </LineChart>
-          </ResponsiveContainer>
 
-          <h5 className="mt-4">TTFT seconds (lower is better)</h5>
-          <ResponsiveContainer width="100%" height={280}>
-            <LineChart data={data} margin={{ top: 8, right: 24, bottom: 8, left: 0 }}>
-              <CartesianGrid strokeDasharray="3 3" />
-              <XAxis dataKey="label" />
-              <YAxis />
-              <Tooltip />
-              <Legend />
-              {showDivider && firstLive && (
-                <ReferenceLine
-                  x={firstLive}
-                  stroke="#bbb"
-                  strokeDasharray="3 3"
-                  label={{
-                    value: "bench.py → helexa-bench",
-                    position: "top",
-                    fill: "#999",
-                    fontSize: 11,
-                  }}
-                />
-              )}
-              <Line
-                type="monotone"
-                dataKey="ttft"
-                name="TTFT (s)"
-                stroke="#dc3545"
-                connectNulls
-              />
-              {base.length > 0 && (
-                <Line
-                  type="monotone"
-                  dataKey="baseTtft"
-                  name="baseline (bench.py · gateway)"
-                  stroke="#888"
-                  strokeDasharray="5 5"
-                  connectNulls
-                />
-              )}
-            </LineChart>
-          </ResponsiveContainer>
+          <MetricChart
+            title="decode tok/s"
+            unit="higher is better"
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "decode", name: "decode tok/s", stroke: "#0d6efd" },
+              ...(base.length > 0
+                ? [
+                    {
+                      key: "baseDecode",
+                      name: "baseline (bench.py · gateway)",
+                      stroke: "#888",
+                      dashed: true,
+                    },
+                  ]
+                : []),
+            ]}
+          />
+
+          <MetricChart
+            title="prefill tok/s"
+            unit="higher is better"
+            hint="The other half of serving speed, derived from prefill_tokens / prefill_ms. A prefix-cache hit shortens prefill_ms while the token count stays whole, so a high rate here is itself the cache-hit signal."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "prefillTps", name: "prefill tok/s", stroke: "#20c997" },
+            ]}
+          />
+
+          <MetricChart
+            title="TTFT"
+            unit="seconds, lower is better"
+            hint="Median and p95 together on purpose. Under concurrency the median is dominated by whichever streams were admitted immediately; the p95 is the one that moves when a caller is made to wait. Charting only the median is how a 0.53 s → 11.77 s tail went unnoticed for a week."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "ttft", name: "TTFT median (s)", stroke: "#dc3545" },
+              { key: "ttftP95", name: "TTFT p95 (s)", stroke: "#fd7e14" },
+              ...(base.length > 0
+                ? [
+                    {
+                      key: "baseTtft",
+                      name: "baseline (bench.py · gateway)",
+                      stroke: "#888",
+                      dashed: true,
+                    },
+                  ]
+                : []),
+            ]}
+          />
+
+          <MetricChart
+            title="inter-token gap p95"
+            unit="ms, lower is better"
+            hint="Stream smoothness. decode tok/s is a mean over the whole window, so a stream that stalls and then catches up is indistinguishable from one that never stalled — this is the number a user feels."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "tpot", name: "inter-token p95 (ms)", stroke: "#6f42c1" },
+            ]}
+          />
+
+          <MetricChart
+            title="admission"
+            unit="queue wait ms · requests shed"
+            hint="Separates “the server is slow” from “you were queued behind someone”. Queue wait is TTFT minus server-measured prefill; rejected counts honest backpressure rather than silent failures."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "queueWait", name: "queue wait (ms)", stroke: "#d63384" },
+              { key: "rejected", name: "rejected (count)", stroke: "#adb5bd" },
+            ]}
+          />
+
+          <MetricChart
+            title="tokens per sample"
+            unit="counts"
+            hint="Cost, not speed. Reasoning tokens are the dominant driver on a reasoning model and move independently of every rate above — a template or sampling change can double what the model thinks before answering while the speed charts stay flat. Cached tokens are why prefill timing varies between otherwise identical samples."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "completion", name: "completion tokens", stroke: "#0dcaf0" },
+              { key: "reasoning", name: "reasoning tokens", stroke: "#ffc107" },
+              { key: "cached", name: "cached prompt tokens", stroke: "#198754" },
+            ]}
+          />
         </>
       )}
     </>

bench/src/types.ts

@@ -41,6 +41,17 @@ export interface SeriesPoint {
   ttft_s_median: number | null;
   decode_tps_median: number | null;
   total_s_median: number | null;
+  ttft_p95_s_median: number | null;
+  queue_wait_ms_median: number | null;
+  rejected_median: number | null;
+  prefill_tps_median: number | null;
+  reasoning_tokens_median: number | null;
+  cached_tokens_median: number | null;
+  completion_tokens_median: number | null;
+  tpot_p95_ms_median: number | null;
+  /** Identity the samples were taken under, or null if anonymous.
+   *  Anonymous and identified points are not comparable (#288). */
+  principal: string | null;
   samples: number;
 }
 

crates/cortex-core/src/anthropic.rs

@@ -20,6 +20,11 @@ pub struct MessagesRequest {
     pub temperature: Option<f64>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub top_p: Option<f64>,
+    /// Anthropic's Messages API carries `top_k` natively, so it maps
+    /// straight onto the OpenAI-side field rather than being dropped in
+    /// translation (#272).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub top_k: Option<usize>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub stream: Option<bool>,
     #[serde(flatten)]

crates/cortex-core/src/catalogue.rs

@@ -1,7 +1,7 @@
 //! Model catalogue — profiles describing how to serve each model.
 
 use crate::discovery::DeviceInfo;
-use crate::harness::{ModelCost, ModelLimit};
+use crate::harness::{ModelCost, ModelLimit, SamplingOverride};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::path::Path;
@@ -57,6 +57,16 @@ pub struct ModelProfile {
     /// on this being explicit per model rather than implicit.
     #[serde(default)]
     pub source: Option<String>,
+    /// Operator sampling override forwarded to neuron on cold load
+    /// (#283). The counterpart to the same field in a host's
+    /// `[[default_models]]` — a model that can be both cold-loaded by
+    /// cortex and resident on a host must carry the same value in both,
+    /// which `script/check-config-consistency.py` enforces in CI.
+    ///
+    /// This is the #252 shape of hazard: two config files describing one
+    /// model, with nothing forcing them to agree.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sampling: Option<SamplingOverride>,
 
     // ── Enrichment (issue #62) ────────────────────────────────
     /// Per-model token budget. When present, advertised in `/v1/models`
@@ -240,6 +250,7 @@ mod tests {
             pinned_on: vec![],
             residency_priority: None,
             source: None,
+            sampling: None,
             limit: None,
             cost: None,
             capabilities: vec![],

crates/cortex-core/src/harness.rs

@@ -26,7 +26,50 @@ pub struct HarnessHealth {
     pub uptime_secs: Option<u64>,
 }
 
+/// Operator-set sampling defaults for one model (#283).
+///
+/// A model's `generation_config.json` is the model authors' statement of
+/// intent, and honouring it (#272) is right by default. It is not always
+/// right for the workload the operator serves: `Qwen/Qwen3.8-27B`
+/// publishes `temperature = 1.0`, which measured a 20% structural-defect
+/// rate on ~2k-token code generation against 0/60 at `<= 0.6`
+/// (#283, p = 0.0031).
+///
+/// This is the operator's explicit, visible correction to that default —
+/// deliberately *not* a heuristic that inspects the model and guesses.
+/// A temperature guesser is a footgun; a number in a config file the
+/// operator wrote is not.
+///
+/// Precedence is **request > operator > model > built-in fallback**: the
+/// override replaces what the model published, but a caller that names a
+/// value still wins, because an explicit API parameter that is silently
+/// ignored is a contract break.
+///
+/// Every field is optional and overlays independently — setting
+/// `temperature` alone leaves the model's `top_p`/`top_k` in force.
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct SamplingOverride {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub temperature: Option<f64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub top_p: Option<f64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub top_k: Option<usize>,
+}
+
+impl SamplingOverride {
+    /// True when the operator set nothing — an empty `[sampling]` table
+    /// must behave exactly as no table at all.
+    pub fn is_empty(&self) -> bool {
+        self.temperature.is_none() && self.top_p.is_none() && self.top_k.is_none()
+    }
+}
+
 /// Specification for loading a model through a harness.
+///
+/// Doubles as the `[[default_models]]` entry shape in `neuron.toml`, so
+/// a field added here is available to both the load API and the
+/// operator's per-host config without a second type to keep in step.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ModelSpec {
     pub model_id: String,
@@ -34,6 +77,15 @@ pub struct ModelSpec {
     pub quant: Option<String>,
     pub tensor_parallel: Option<u32>,
     pub devices: Option<Vec<u32>>,
+    /// Operator sampling override (#283). Travels with the *model*, not
+    /// the request: neuron is the only place sampling is resolved, so
+    /// the override has to reach it whether the load came from cortex's
+    /// catalogue or from this host's own `[[default_models]]`.
+    ///
+    /// `#[serde(default)]` so a cortex or neuron predating this field
+    /// still round-trips the spec.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sampling: Option<SamplingOverride>,
 }
 
 /// Per-model token budget advertised by the catalogue or neuron.
@@ -50,8 +102,48 @@ pub struct ModelLimit {
     /// back to `context − output`.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub input: Option<usize>,
-    /// Maximum number of generation tokens.
+    /// The generation budget a request gets when it names none, and the
+    /// figure `input = context − output` is derived from.
+    ///
+    /// A *reserve*, not a ceiling — a request may ask for more and be
+    /// served it. Named `output` for the shape's history; the number a
+    /// client should plan against is [`ModelLimit::output_ceiling`].
     pub output: usize,
+    /// The largest output a request may name and be served (#278).
+    ///
+    /// Distinct from `output`, which is the default and the KV-planning
+    /// reserve. Publishing the reserve as the ceiling meant a client
+    /// that trusted the advertisement asked for a fraction of what the
+    /// model would happily generate — on a reasoning model, often less
+    /// than its own think block costs.
+    ///
+    /// `#[serde(default)]` so a neuron predating this field still
+    /// deserializes; zero means "not advertised", and consumers fall
+    /// back to `output`.
+    #[serde(default, skip_serializing_if = "is_zero")]
+    pub output_ceiling: usize,
+}
+
+fn is_zero(v: &usize) -> bool {
+    *v == 0
+}
+
+/// What each reasoning effort level costs, in reasoning tokens (#223).
+///
+/// OpenAI-shaped clients can only send `minimal|low|medium|high` — the
+/// ladder is their entire vocabulary, and they cannot express a token
+/// count. So the server picks the numbers, and has to say what they are:
+/// a client choosing `low` with no idea whether that buys 2k tokens or
+/// 20k is guessing, which is the failure #274 exists to end.
+///
+/// Ordered rungs rather than a map, so the ladder reads in the order a
+/// client would climb it.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct ReasoningBudgetRung {
+    /// The effort level as the caller spells it on the wire.
+    pub effort: String,
+    /// Reasoning tokens that level is allowed to spend.
+    pub tokens: usize,
 }
 
 /// Operator-set pricing, **USD per 1,000,000 tokens, as JSON numbers**
@@ -144,6 +236,11 @@ pub struct ModelInfo {
     /// from its own routing table.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub servable: Option<ModelServability>,
+    /// The reasoning-effort ladder this deployment honours (#223).
+    /// Absent when the model does not reason, or the harness has no
+    /// budget to offer.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub reasoning_budget: Vec<ReasoningBudgetRung>,
 }
 
 /// Why a loaded model can or cannot be served right now (#245).

crates/cortex-core/src/node.rs

@@ -44,6 +44,12 @@ pub struct NodeState {
     /// just one whose total topology could. Empty until the first
     /// /health poll.
     pub device_health: Vec<crate::discovery::DeviceHealth>,
+    /// The reasoning-effort ladder this neuron honours (#223), copied
+    /// from its `/models` reply at poll time. Per-node rather than
+    /// per-model because it is host configuration — every model on a
+    /// neuron reports the same rungs — and duplicating it onto each
+    /// model entry would invite the two to disagree.
+    pub reasoning_budget: Vec<crate::harness::ReasoningBudgetRung>,
 }
 
 /// A model registered on a node, with its runtime status.
@@ -187,12 +193,26 @@ pub struct CortexModelEntry {
     /// Served max-seq-len in tokens — mirrors `limit.context`.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub max_model_len: Option<usize>,
+    /// The same window under the two names generic OpenAI clients look
+    /// for (#278). pi-ai's provider discovery, for one, reads
+    /// `context_window` then `context_length` and nothing else — so a
+    /// client probing cortex found no window at all and fell back to a
+    /// number its operator had to invent by hand.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub context_window: Option<usize>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub context_length: Option<usize>,
     /// Usable input budget — mirrors `limit.input` when present.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub max_input_tokens: Option<usize>,
-    /// Maximum generation tokens — mirrors `limit.output`.
+    /// The largest output a request may name — mirrors
+    /// `limit.output_ceiling`, falling back to `limit.output` (#278).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub max_output_tokens: Option<usize>,
+    /// What each effort level buys, in reasoning tokens (#223) — so a
+    /// client picking `low` knows what it asked for.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub reasoning_budget: Vec<crate::harness::ReasoningBudgetRung>,
 }
 
 impl CortexModelEntry {
@@ -204,8 +224,22 @@ impl CortexModelEntry {
     /// dropped the limit.
     pub fn sync_flat_limit(&mut self) {
         self.max_model_len = self.limit.as_ref().map(|l| l.context);
+        self.context_window = self.limit.as_ref().map(|l| l.context);
+        self.context_length = self.limit.as_ref().map(|l| l.context);
         self.max_input_tokens = self.limit.as_ref().and_then(|l| l.input);
-        self.max_output_tokens = self.limit.as_ref().map(|l| l.output);
+        // The ceiling, not the reserve (#278): this field is what
+        // OpenAI-ecosystem clients read to decide what to ask for, and
+        // pi-ai/DSH turns it straight into a per-request cap. Advertising
+        // the reserve here handed a reasoning model less budget than its
+        // own think block needs. Falls back to the reserve for a neuron
+        // too old to publish a ceiling.
+        self.max_output_tokens = self.limit.as_ref().map(|l| {
+            if l.output_ceiling > 0 {
+                l.output_ceiling
+            } else {
+                l.output
+            }
+        });
     }
 }
 

crates/cortex-core/src/openai.rs

@@ -18,6 +18,31 @@ pub struct ChatCompletionRequest {
     pub temperature: Option<f64>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub top_p: Option<f64>,
+    /// Truncate sampling to the `k` most likely tokens (#272).
+    ///
+    /// Not in the OpenAI schema but universal among serving stacks
+    /// (vLLM, SGLang, llama.cpp, Ollama), and required by models that
+    /// publish it — Qwen3 specifies `top_k = 20`. A named field rather
+    /// than `extra`, because a sampling parameter that lands in `extra`
+    /// is accepted and silently discarded.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub top_k: Option<usize>,
+    /// Pin the sampler's RNG for reproducible output (OpenAI core).
+    ///
+    /// Previously ignored on every text path, which called
+    /// `unix_subsec_nanos()` regardless — a caller asking for
+    /// determinism got fresh randomness and a `200`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub seed: Option<u64>,
+    /// Penalty applied to recently-generated tokens; `1.0` disables.
+    /// Defaults to 1.1 when absent (#272).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub repetition_penalty: Option<f32>,
+    /// How many recent tokens `repetition_penalty` considers. Defaults
+    /// to 64 — fine for chat, short for a long reasoning block, which is
+    /// why it is a knob rather than a constant (#272).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub repeat_last_n: Option<usize>,
     /// The deprecated spelling of the output cap. Still what most
     /// clients send; see [`ChatCompletionRequest::effective_max_tokens`].
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -149,8 +174,9 @@ pub struct Usage {
     /// `completion_tokens`, never added into `total_tokens`).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub completion_tokens_details: Option<CompletionTokensDetails>,
-    /// OpenAI-standard breakdown of `prompt_tokens`. Populated once
-    /// prompt caching lands (#11); `None` until then.
+    /// OpenAI-standard breakdown of `prompt_tokens`, carrying
+    /// `cached_tokens` (#269). Omitted when nothing was reused, so
+    /// "absent" stays distinguishable from "measured as zero".
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub prompt_tokens_details: Option<PromptTokensDetails>,
     /// helexa extension (non-OpenAI): server-measured prefill/decode
@@ -182,8 +208,9 @@ pub struct CompletionTokensDetails {
 /// Sub-counts of `Usage::prompt_tokens`.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct PromptTokensDetails {
-    /// Prompt tokens served from cache (cache-read rate). Populated
-    /// once prompt caching lands (#11).
+    /// Prompt tokens served from neuron's prefix KV cache (#11), so a
+    /// caller can see the saving rather than assume none (#269). A
+    /// sub-count of `prompt_tokens`, never added into `total_tokens`.
     pub cached_tokens: u64,
 }
 

crates/cortex-core/src/responses.rs

@@ -55,6 +55,21 @@ pub struct ResponsesRequest {
     pub temperature: Option<f64>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub top_p: Option<f64>,
+    /// Truncate sampling to the `k` most likely tokens (#272). Not in
+    /// the Responses schema, but a sampling parameter that lands in
+    /// `extra` is accepted and silently discarded, which is the defect
+    /// this closes.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub top_k: Option<usize>,
+    /// Pin the sampler's RNG for reproducible output (#272).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub seed: Option<u64>,
+    /// Penalty on recently-generated tokens; `1.0` disables (#272).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub repetition_penalty: Option<f32>,
+    /// Window `repetition_penalty` considers (#272).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub repeat_last_n: Option<usize>,
     /// Chained-conversation identifier. We don't store responses
     /// server-side yet; if this is `Some`, the handler returns 400.
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -130,12 +145,22 @@ pub enum ResponsesInputItem {
     /// string or an array of content parts; the translator renders
     /// either form to text rather than losing the tool result.
     FunctionCallOutput { call_id: String, output: Value },
-    /// Reasoning items emitted by o-series models. Accepted but
-    /// not forwarded to the model — neuron's candle path doesn't
-    /// surface reasoning separately yet.
+    /// A reasoning item the model emitted on an earlier turn, replayed
+    /// by the client so the model can continue its own train of thought.
+    ///
+    /// Both spellings are modelled because both are in the wild and one
+    /// of them is ours: OpenAI's o-series carries the text in `content`
+    /// as `reasoning_text` parts, while neuron emits `summary` with a
+    /// `summary_text` part. Clients round-trip the completed item
+    /// verbatim — pi-ai stores our `response.output_item.done` payload
+    /// and replays it unchanged — so whichever field we emit is the
+    /// field we get back, and reading only one of them silently loses
+    /// the turn's reasoning.
     Reasoning {
         #[serde(default)]
         content: Vec<Value>,
+        #[serde(default)]
+        summary: Vec<Value>,
     },
 }
 
@@ -249,8 +274,8 @@ pub struct ResponsesUsage {
     /// sub-count of `output_tokens`, never added into `total_tokens`).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub output_tokens_details: Option<OutputTokensDetails>,
-    /// OpenAI-standard breakdown of `input_tokens`. Populated once
-    /// prompt caching lands (#11); `None` until then.
+    /// OpenAI-standard breakdown of `input_tokens`, carrying
+    /// `cached_tokens` (#269). Omitted when nothing was reused.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub input_tokens_details: Option<InputTokensDetails>,
 }
@@ -265,8 +290,8 @@ pub struct OutputTokensDetails {
 /// Sub-counts of `ResponsesUsage::input_tokens`.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct InputTokensDetails {
-    /// Input tokens served from cache (cache-read rate). Populated
-    /// once prompt caching lands (#11).
+    /// Input tokens served from neuron's prefix KV cache (#11), so a
+    /// caller can see the saving rather than assume none (#269).
     pub cached_tokens: u64,
 }
 
@@ -289,6 +314,16 @@ pub mod events {
     pub const OUTPUT_TEXT_DONE: &str = "response.output_text.done";
     pub const CONTENT_PART_DONE: &str = "response.content_part.done";
     pub const OUTPUT_ITEM_DONE: &str = "response.output_item.done";
+    /// Reasoning-item event family. A reasoning model's think block is
+    /// its own output item, emitted ahead of the message item, with a
+    /// summary part carrying the text. Streaming these is what keeps the
+    /// connection observably alive while a model thinks — clients time
+    /// out on silence, and SSE comment keep-alives don't count because
+    /// idle timers reset on parsed events, not comments.
+    pub const REASONING_SUMMARY_PART_ADDED: &str = "response.reasoning_summary_part.added";
+    pub const REASONING_SUMMARY_TEXT_DELTA: &str = "response.reasoning_summary_text.delta";
+    pub const REASONING_SUMMARY_TEXT_DONE: &str = "response.reasoning_summary_text.done";
+    pub const REASONING_SUMMARY_PART_DONE: &str = "response.reasoning_summary_part.done";
     pub const FUNCTION_CALL_ARGUMENTS_DELTA: &str = "response.function_call_arguments.delta";
     pub const FUNCTION_CALL_ARGUMENTS_DONE: &str = "response.function_call_arguments.done";
     pub const COMPLETED: &str = "response.completed";

crates/cortex-core/src/translate.rs

@@ -94,6 +94,28 @@ pub fn anthropic_to_openai(req: MessagesRequest) -> ChatCompletionRequest {
         if let Some(tc) = tool_choice {
             obj.insert("tool_choice".into(), tc);
         }
+        // Anthropic's extended-thinking budget is the one quantitative
+        // reasoning control any of our wire formats offers (#223).
+        // Normalised onto `reasoning.max_tokens`, the shape the harness
+        // resolves, so all three inlets converge on one field instead of
+        // each generation loop learning three spellings.
+        let budget = obj
+            .get("thinking")
+            .filter(|t| t.get("type").and_then(Value::as_str) != Some("disabled"))
+            .and_then(|t| t.get("budget_tokens"))
+            .and_then(Value::as_u64)
+            .filter(|n| *n > 0);
+        if let Some(budget) = budget {
+            let reasoning = obj
+                .entry("reasoning")
+                .or_insert_with(|| Value::Object(Default::default()));
+            if !reasoning.is_object() {
+                *reasoning = Value::Object(Default::default());
+            }
+            if let Some(r) = reasoning.as_object_mut() {
+                r.entry("max_tokens").or_insert_with(|| Value::from(budget));
+            }
+        }
     }
 
     ChatCompletionRequest {
@@ -101,6 +123,13 @@ pub fn anthropic_to_openai(req: MessagesRequest) -> ChatCompletionRequest {
         messages,
         temperature: req.temperature,
         top_p: req.top_p,
+        // Anthropic models top_k too, so it survives the hop (#272).
+        top_k: req.top_k,
+        // Anthropic's Messages API has no seed or penalty knobs; absent
+        // means the model's own generation_config decides.
+        seed: None,
+        repetition_penalty: None,
+        repeat_last_n: None,
         max_tokens: Some(req.max_tokens),
         // Anthropic has one spelling; it maps onto the legacy field.
         max_completion_tokens: None,

crates/cortex-gateway/src/handlers.rs

@@ -798,6 +798,10 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
                 max_model_len: None,
                 max_input_tokens: None,
                 max_output_tokens: None,
+                context_window: None,
+                context_length: None,
+                // Catalogue-seeded: no neuron has reported rungs yet.
+                reasoning_budget: Vec::new(),
             },
         );
     }
@@ -839,6 +843,13 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
                     // so a client never overflows the most-constrained
                     // deployment that might serve it.
                     e.limit = tightest_limit(e.limit.take(), entry.limit.clone());
+                    // The reasoning ladder this neuron honours (#223).
+                    // First non-empty wins: a catalogue-seeded entry
+                    // starts with none, and neurons serving the same
+                    // model are configured alike.
+                    if e.reasoning_budget.is_empty() {
+                        e.reasoning_budget = node.reasoning_budget.clone();
+                    }
                 })
                 .or_insert_with(|| CortexModelEntry {
                     id: model_id.clone(),
@@ -858,6 +869,10 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
                     max_model_len: None,
                     max_input_tokens: None,
                     max_output_tokens: None,
+                    context_window: None,
+                    context_length: None,
+                    // What this neuron says each effort level buys (#223).
+                    reasoning_budget: node.reasoning_budget.clone(),
                 });
         }
     }
@@ -919,6 +934,10 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
                     max_model_len: None,
                     max_input_tokens: None,
                     max_output_tokens: None,
+                    context_window: None,
+                    context_length: None,
+                    // What this neuron says each effort level buys (#223).
+                    reasoning_budget: node.reasoning_budget.clone(),
                 });
         }
     }
@@ -956,6 +975,10 @@ async fn list_models(State(fleet): State<Arc<CortexState>>) -> Json<Value> {
                 max_model_len: None,
                 max_input_tokens: None,
                 max_output_tokens: None,
+                context_window: None,
+                context_length: None,
+                // Catalogue-seeded: no neuron has reported rungs yet.
+                reasoning_budget: Vec::new(),
             },
         );
     }

crates/cortex-gateway/src/metering.rs

@@ -141,6 +141,15 @@ pub fn usage_sink(
 /// returns a guard the caller settles with actual usage; on refusal returns
 /// the #63 envelope (`rate_limit_exceeded` + `Retry-After` for a resetting
 /// window, `insufficient_quota` for a hard balance — never `402`).
+// The `Err` variant is the #63 envelope itself, which is a data-carrying
+// struct (status, type, code, message, param, retry hint, diagnostics) and
+// is passed by value everywhere it is built, matched and rendered. Boxing it
+// at this one boundary would push `Box` conversions through three handlers
+// and both HTTP adapters to save an allocation on a path taken once per
+// rejected request. clippy 1.98 lowered what it considers large enough to
+// flag; the shape is deliberate. Same call as the existing allows on the
+// figment-returning config loaders.
+#[allow(clippy::result_large_err)]
 pub async fn reserve_or_reject(
     provider: Arc<dyn EntitlementProvider>,
     principal: &Principal,
@@ -279,7 +288,7 @@ mod tests {
         // Requested output dominates; prompt estimate is small for a tiny body.
         let body = br#"{"model":"m","max_tokens":1000}"#;
         let est = reservation_estimate(body, Some(8192));
-        assert!(est >= 1000 && est < 1100, "est was {est}");
+        assert!((1000..1100).contains(&est), "est was {est}");
     }
 
     #[test]

crates/cortex-gateway/src/metrics.rs

@@ -4,7 +4,7 @@
 //! in Prometheus text format.
 
 use anyhow::Result;
-use metrics_exporter_prometheus::PrometheusBuilder;
+use metrics_exporter_prometheus::{Matcher, PrometheusBuilder};
 use std::net::SocketAddr;
 
 /// Install the Prometheus metrics recorder and return a handle.
@@ -12,8 +12,7 @@ use std::net::SocketAddr;
 pub fn install(listen: &str) -> Result<()> {
     let addr: SocketAddr = listen.parse()?;
 
-    PrometheusBuilder::new()
-        .with_http_listener(addr)
+    with_buckets(PrometheusBuilder::new().with_http_listener(addr))?
         .install()
         .map_err(|e| anyhow::anyhow!("failed to install Prometheus exporter: {e}"))?;
 
@@ -25,13 +24,77 @@ pub fn install(listen: &str) -> Result<()> {
 /// Install a recorder for testing (no HTTP listener). Returns a handle
 /// that can render the current metrics as Prometheus text.
 pub fn install_test_recorder() -> Result<metrics_exporter_prometheus::PrometheusHandle> {
-    let handle = PrometheusBuilder::new()
+    let handle = with_buckets(PrometheusBuilder::new())?
         .install_recorder()
         .map_err(|e| anyhow::anyhow!("failed to install test recorder: {e}"))?;
     describe_metrics();
     Ok(handle)
 }
 
+/// Give every histogram explicit buckets, so it is exported as a
+/// Prometheus **histogram** rather than a summary.
+///
+/// Without this, `metrics-exporter-prometheus` renders every
+/// `histogram!` as a summary: `{quantile="0.95"}` series and no
+/// `_bucket` series at all. Three things break as a result, and the
+/// third is what made it visible.
+///
+/// 1. `histogram_quantile()` has nothing to read, so any dashboard
+///    panel written the idiomatic way returns *No data* forever. The
+///    fleet dashboard's TTFT panel did exactly that while cortex was
+///    serving thousands of requests.
+/// 2. Summary quantiles are computed per process over a rolling
+///    window, so they **cannot be aggregated**. Averaging p95 across
+///    two gateways is not p95 of anything.
+/// 3. They decay: with no samples in the window every quantile reads
+///    `0`, which is indistinguishable from "genuinely instant" on a
+///    graph.
+///
+/// Buckets are per-metric because the quantities differ by orders of
+/// magnitude — a request lasting minutes and a TTFT of milliseconds
+/// have no useful shared scale. Ranges are chosen from observed fleet
+/// behaviour rather than round numbers: decode runs for minutes on a
+/// long turn, prefill is seconds on a long prompt, and decode
+/// throughput sits in the tens of tokens/sec.
+fn with_buckets(builder: PrometheusBuilder) -> Result<PrometheusBuilder> {
+    let seconds_short = &[
+        0.05, 0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0, 20.0, 30.0, 60.0, 120.0,
+    ];
+    let seconds_long = &[
+        0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0, 600.0, 1200.0,
+    ];
+    let tokens_per_second = &[
+        1.0, 2.5, 5.0, 10.0, 15.0, 20.0, 30.0, 40.0, 60.0, 80.0, 120.0, 200.0,
+    ];
+    builder
+        // Whole-request latency: a long agentic turn legitimately runs
+        // for many minutes, so the tail has to reach there.
+        .set_buckets_for_metric(
+            Matcher::Full("cortex_request_duration_seconds".into()),
+            seconds_long,
+        )?
+        // Prefill: sub-second for a short prompt, seconds for a long
+        // one. Anything past a minute is pathological and belongs in
+        // the overflow bucket.
+        .set_buckets_for_metric(
+            Matcher::Full("cortex_time_to_first_token_seconds".into()),
+            seconds_short,
+        )?
+        // Image generation is inherently slower than a text turn and
+        // scales with resolution and step count.
+        .set_buckets_for_metric(
+            Matcher::Full("cortex_images_generation_seconds".into()),
+            seconds_long,
+        )?
+        // Not a duration — decode throughput, tens of tokens/sec on
+        // this fleet.
+        .set_buckets_for_metric(
+            Matcher::Full("cortex_tokens_per_second".into()),
+            tokens_per_second,
+        )
+        .map_err(|e| anyhow::anyhow!("failed to configure histogram buckets: {e}"))
+}
+
 fn describe_metrics() {
     metrics::describe_histogram!(
         "cortex_request_duration_seconds",
@@ -58,6 +121,10 @@ fn describe_metrics() {
         "cortex_request_errors_total",
         "Total number of failed proxy requests"
     );
+    metrics::describe_counter!(
+        "cortex_cached_prompt_tokens_total",
+        "Prompt tokens served from neuron's prefix KV cache; over cortex_prompt_tokens_total this is the cache-hit rate (#269)"
+    );
     metrics::describe_counter!("cortex_evictions_total", "Total number of model evictions");
     metrics::describe_counter!(
         "cortex_cold_starts_total",
@@ -130,3 +197,43 @@ fn describe_metrics() {
         "Live prefill throughput per neuron:model, tokens/sec EMA (#137)"
     );
 }
+
+#[cfg(test)]
+mod bucket_tests {
+    /// Histograms must export as Prometheus **histograms**, with
+    /// `_bucket` series — not as summaries.
+    ///
+    /// This asserts the exported *shape*, not the recording call,
+    /// because the recording was never the problem: cortex measured
+    /// TTFT correctly for months while the fleet dashboard's panel sat
+    /// on "No data", since `histogram_quantile()` reads `_bucket`
+    /// series and a summary has none. A metric that is collected but
+    /// cannot be queried is indistinguishable from one that was never
+    /// collected — from the graph, and from the operator's chair.
+    #[test]
+    fn histograms_export_buckets_not_summary_quantiles() {
+        let handle = match super::install_test_recorder() {
+            Ok(h) => h,
+            // Another test in this binary owns the global recorder;
+            // it installs the same buckets, so skipping is honest.
+            Err(_) => return,
+        };
+        metrics::histogram!("cortex_time_to_first_token_seconds", "node" => "n", "model" => "m")
+            .record(0.42);
+        let rendered = handle.render();
+
+        assert!(
+            rendered.contains("cortex_time_to_first_token_seconds_bucket"),
+            "no _bucket series — histogram_quantile() cannot read this:\n{rendered}"
+        );
+        assert!(
+            rendered.contains("# TYPE cortex_time_to_first_token_seconds histogram"),
+            "exported as the wrong metric type:\n{rendered}"
+        );
+        assert!(
+            !rendered
+                .contains(r#"cortex_time_to_first_token_seconds{node="n",model="m",quantile="#),
+            "still exporting summary quantiles, which cannot be aggregated across gateways"
+        );
+    }
+}

crates/cortex-gateway/src/poller.rs

@@ -130,6 +130,17 @@ async fn poll_neuron(fleet: &CortexState, name: &str, endpoint: &str) {
         Ok(resp) if resp.status().is_success() => {
             match resp.json::<Vec<ModelInfo>>().await {
                 Ok(models) => {
+                    // The reasoning-effort ladder is host configuration:
+                    // every model on a neuron reports the same rungs, so
+                    // it is held once per node rather than copied onto
+                    // each model entry where the two could drift (#223).
+                    // Taken from the first model that offers one — a
+                    // non-reasoning model advertises none.
+                    node.reasoning_budget = models
+                        .iter()
+                        .map(|m| m.reasoning_budget.clone())
+                        .find(|rungs| !rungs.is_empty())
+                        .unwrap_or_default();
                     let mut seen = std::collections::HashSet::new();
                     for upstream in &models {
                         seen.insert(upstream.id.clone());

crates/cortex-gateway/src/proxy.rs

@@ -171,6 +171,9 @@ impl IntoResponse for ProxyError {
 //       over the decode window (first→last chunk); falls back to the
 //       full request duration for single-chunk (non-streaming) bodies
 //   cortex_prompt_tokens_total / cortex_completion_tokens_total (counters)
+//   cortex_cached_prompt_tokens_total   (counter) — prompt tokens served
+//       from neuron's prefix cache; over prompt_tokens_total this is the
+//       fleet's cache-hit rate (#269)
 
 /// Cap on the retained body tail. The usage object rides on the final
 /// chunk, so a generous tail is plenty; the cap bounds memory on huge
@@ -243,6 +246,15 @@ impl ChunkObserver for CortexMetrics {
             if let Some(prompt) = prompt {
                 metrics::counter!("cortex_prompt_tokens_total", &self.labels).increment(prompt);
             }
+            // Prefix-cache reuse (#269), read from the same usage object.
+            // Against `cortex_prompt_tokens_total` this gives fleet-wide
+            // cache-hit rate as a ratio, which was previously only
+            // obtainable by grepping `reused=` out of a neuron journal —
+            // per-host, unaggregated, and gone on log rotation.
+            if let Some(cached) = last_count_for(self.tail.as_str(), "cached_tokens") {
+                metrics::counter!("cortex_cached_prompt_tokens_total", &self.labels)
+                    .increment(cached);
+            }
             if let Some(completion) = completion.filter(|c| *c > 0) {
                 metrics::counter!("cortex_completion_tokens_total", &self.labels)
                     .increment(completion);

crates/cortex-gateway/src/router.rs

@@ -535,6 +535,10 @@ async fn profile_to_spec(
         quant: profile.quant.clone(),
         tensor_parallel,
         devices: Some(devices),
+        // The catalogue's operator override rides along with the load
+        // (#283), so a cold-loaded model samples the same way as one
+        // this host was configured to hold resident.
+        sampling: profile.sampling.clone(),
     }
 }
 
@@ -642,6 +646,7 @@ mod tests {
             pinned_on: vec![],
             residency_priority: None,
             source: source.map(String::from),
+            sampling: None,
             limit: None,
             cost: None,
             capabilities: vec![],

crates/cortex-gateway/src/state.rs

@@ -66,6 +66,9 @@ impl CortexState {
                     endpoint: nc.endpoint.clone(),
                     healthy: false,
                     models: HashMap::new(),
+                    // Filled from the neuron's /models reply at the
+                    // first poll (#223).
+                    reasoning_budget: Vec::new(),
                     lifecycle_cycles: 0,
                     last_poll: None,
                     discovery: None,
@@ -98,8 +101,29 @@ impl CortexState {
             neuron_configs: config.neurons.clone(),
             eviction: config.eviction.clone(),
             catalogue,
+            // READ timeout, not a total one. reqwest's `.timeout()` is a
+            // total deadline — "from when the request starts connecting
+            // until the response body has finished" — which for a streamed
+            // completion is a hard cap on how long a model may generate,
+            // not a liveness check. At 300s and ~13 tok/s that capped every
+            // stream at roughly 3.9k tokens: cortex severed the connection
+            // mid-answer, the client reported the stream as terminated, and
+            // neuron carried on generating into a socket nobody was reading.
+            // A long agentic turn could therefore never complete through the
+            // gateway however healthy the fleet was, and the failure looked
+            // like a model fault because the model was blameless.
+            //
+            // `.read_timeout()` resets on every successful read, so it times
+            // out a stalled upstream while letting a producing one run as
+            // long as it needs. The same 300s is kept deliberately: it is
+            // the headroom a cold model load needs before its first byte,
+            // which is what the original total timeout was reaching for.
+            //
+            // This pairs with neuron's 1s SSE keep-alive: a healthy stream
+            // puts bytes on the wire every second, so idle here means idle,
+            // and the deadline finally measures what it was meant to.
             http_client: reqwest::Client::builder()
-                .timeout(std::time::Duration::from_secs(300))
+                .read_timeout(std::time::Duration::from_secs(300))
                 .build()
                 .expect("failed to build HTTP client"),
             entitlements,

crates/cortex-gateway/tests/budget_enforcement.rs

@@ -1,9 +1,9 @@
 //! Integration tests for budget enforcement (#52) — the A0 seatbelt.
 //!
 //! A reservation over the key's hard cap is refused *before* neuron is hit,
-//! with the #63 code matching the cap-window semantics (rate_limit_exceeded
-//! + Retry-After for a resetting window, insufficient_quota for a hard
-//! balance). Spend never exceeds the cap. No 402, ever.
+//! with the #63 code matching the cap-window semantics: `rate_limit_exceeded`
+//! plus `Retry-After` for a resetting window, `insufficient_quota` for a hard
+//! balance. Spend never exceeds the cap. No 402, ever.
 
 use axum::Json;
 use axum::extract::Path;

crates/cortex-gateway/tests/model_limits.rs

@@ -69,6 +69,18 @@ capabilities = ["text"]
         let mut nodes = fleet.nodes.write().await;
         let node = nodes.get_mut("mock-node").expect("node exists");
         node.healthy = true;
+        // The ladder the neuron reported at poll time (#223) — host
+        // configuration, held once per node.
+        node.reasoning_budget = ["minimal", "low", "medium", "high"]
+            .into_iter()
+            .zip([1024usize, 4096, 12288, 32768])
+            .map(
+                |(effort, tokens)| cortex_core::harness::ReasoningBudgetRung {
+                    effort: effort.into(),
+                    tokens,
+                },
+            )
+            .collect();
         node.models.insert(
             "test-model".into(),
             ModelEntry {
@@ -85,6 +97,9 @@ capabilities = ["text"]
                     context: 49152,
                     input: Some(40960),
                     output: 8192,
+                    // The ceiling a request may name, distinct from the
+                    // 8192-token reserve above (#278).
+                    output_ceiling: 32768,
                 }),
                 servable: None,
             },
@@ -147,7 +162,34 @@ capabilities = ["text"]
     // vLLM-convention probes (Hermes Agent) auto-detect the window.
     assert_eq!(entry["max_model_len"], 49152);
     assert_eq!(entry["max_input_tokens"], 40960);
-    assert_eq!(entry["max_output_tokens"], 8192);
+    // The ceiling, not the reserve (#278). pi-ai's discovery — and every
+    // client like it — turns this field into the cap it sends on each
+    // request; advertising the 8192 reserve told a reasoning model's
+    // harness to stop below the cost of a single think block.
+    assert_eq!(entry["max_output_tokens"], 32768);
+    // What each effort level buys (#223). A client can only send the
+    // rung names — it has no way to express a token count — so a ladder
+    // it cannot see is a ladder it has to guess at.
+    let rungs = entry["reasoning_budget"]
+        .as_array()
+        .expect("reasoning_budget advertised for a reasoning model");
+    let names: Vec<&str> = rungs
+        .iter()
+        .map(|r| r["effort"].as_str().expect("effort"))
+        .collect();
+    assert_eq!(names, ["minimal", "low", "medium", "high"]);
+    assert!(
+        rungs
+            .iter()
+            .all(|r| r["tokens"].as_u64().is_some_and(|t| t > 0)),
+        "every advertised rung must name a token count"
+    );
+
+    // The window under the two names generic clients actually read.
+    // pi-ai looks for `context_window` then `context_length` and gives
+    // up if neither is present — `max_model_len` is invisible to it.
+    assert_eq!(entry["context_window"], 49152);
+    assert_eq!(entry["context_length"], 49152);
 
     // No limit → flat fields omitted entirely, never 0 or a guess.
     let unknown = body["data"]
@@ -157,7 +199,13 @@ capabilities = ["text"]
         .find(|m| m["id"] == "no-limit-model")
         .expect("no-limit-model present in /v1/models");
     assert!(unknown.get("limit").is_none());
-    for key in ["max_model_len", "max_input_tokens", "max_output_tokens"] {
+    for key in [
+        "max_model_len",
+        "max_input_tokens",
+        "max_output_tokens",
+        "context_window",
+        "context_length",
+    ] {
         assert!(
             unknown.get(key).is_none(),
             "{key} must be omitted when the window is unknown"

crates/cortex-gateway/tests/prompt_prevalidation.rs

@@ -96,6 +96,7 @@ async fn spawn_gateway(neuron: &str, context: usize) -> String {
                     context,
                     input: None,
                     output: 16,
+                    output_ceiling: 0,
                 }),
                 servable: None,
             },

crates/helexa-bench/src/client.rs

@@ -6,6 +6,7 @@ use crate::config::{TargetConfig, TargetKind};
 use anyhow::{Context, Result, anyhow};
 use cortex_core::build_info::BuildInfo;
 use cortex_core::discovery::{DiscoveryResponse, HealthResponse};
+use cortex_core::entitlements::{HEADER_ACCOUNT_ID, HEADER_KEY_ID};
 use cortex_core::harness::{ModelInfo, ModelSpec};
 use cortex_core::openai::ModelsResponse;
 use std::time::Duration;
@@ -19,10 +20,49 @@ pub struct TargetClient {
 
 impl TargetClient {
     pub fn new(request_timeout: Duration) -> Result<Self> {
-        let http = reqwest::Client::builder()
-            .timeout(request_timeout)
-            .build()
-            .context("building HTTP client")?;
+        Self::with_principal(request_timeout, None)
+    }
+
+    /// Build a client that stamps `principal`'s identity headers on every
+    /// request (#288).
+    ///
+    /// Set on the client rather than at each call site so no scenario can
+    /// be added later that silently measures as anonymous — which is the
+    /// failure this fixes, and it was invisible for a week.
+    ///
+    /// `None` keeps the historical anonymous behaviour, which is correct
+    /// for `openai` targets (the headers mean nothing to a foreign
+    /// engine) and for a fresh install with no principal configured.
+    pub fn with_principal(
+        request_timeout: Duration,
+        principal: Option<&crate::config::PrincipalSettings>,
+    ) -> Result<Self> {
+        let mut builder = reqwest::Client::builder().timeout(request_timeout);
+        if let Some(p) = principal {
+            let mut headers = reqwest::header::HeaderMap::new();
+            headers.insert(
+                HEADER_ACCOUNT_ID,
+                reqwest::header::HeaderValue::from_str(&p.account_id)
+                    .context("principal.account_id is not a valid header value")?,
+            );
+            headers.insert(
+                HEADER_KEY_ID,
+                reqwest::header::HeaderValue::from_str(&p.key_id)
+                    .context("principal.key_id is not a valid header value")?,
+            );
+            tracing::info!(
+                account_id = %p.account_id,
+                key_id = %p.key_id,
+                "bench authenticates: measurements reflect identified-caller capacity (#288)"
+            );
+            builder = builder.default_headers(headers);
+        } else {
+            tracing::warn!(
+                "bench has no [bench.principal]: requests are anonymous, so since #262 \
+                 they measure the anonymous-yield policy rather than serving capacity (#288)"
+            );
+        }
+        let http = builder.build().context("building HTTP client")?;
         Ok(TargetClient { http })
     }
 
@@ -169,6 +209,14 @@ impl TargetClient {
             quant: None,
             tensor_parallel: (info.devices.len() > 1).then_some(info.devices.len() as u32),
             devices: Some(info.devices.clone()),
+            // `GET /models` does not report the operator sampling
+            // override (#283), so a spec reconstructed here cannot carry
+            // it — same limitation as `quant` directly above. Benign
+            // while this is only used to re-load a model the host
+            // already has a `[[default_models]]` entry for, since that
+            // entry supplies the override; it would matter if this were
+            // ever used to load a model the host has no config for.
+            sampling: None,
         })
     }
 
@@ -236,6 +284,7 @@ impl TargetClient {
                         // A generic OpenAI endpoint says nothing about
                         // serviceability, and absent means "no opinion".
                         servable: None,
+                        reasoning_budget: Vec::new(),
                     })
                     .collect())
             }

crates/helexa-bench/src/config.rs

@@ -67,6 +67,44 @@ pub struct BenchSettings {
     /// SQLite system-of-record path.
     #[serde(default = "default_db_path")]
     pub db_path: String,
+    /// Identity bench presents to a neuron (#288).
+    ///
+    /// Without this bench is *anonymous*, and since #262 anonymous
+    /// callers are served only from capacity left over once identified
+    /// traffic is satisfied — they are capped below `max_in_flight` and
+    /// park at the class gate ahead of every other gate. Measured on
+    /// beast, that moved `concurrency:8` from 168.9 tok/s / 0.53 s
+    /// ttft_p95 to 104.6 / 11.77 s the day #262 landed, and held there.
+    ///
+    /// So an unauthenticated bench does not measure serving capacity; it
+    /// measures the yield policy. Real traffic arrives through cortex
+    /// with a principal and is subject to none of it.
+    ///
+    /// Deliberately not an admission *exemption*: bench should be
+    /// subject to the same rules as any caller, or it measures something
+    /// no user experiences — the same class of error in the other
+    /// direction. Give it a principal and let it queue like everyone
+    /// else.
+    #[serde(default)]
+    pub principal: Option<PrincipalSettings>,
+}
+
+/// The account/key pair bench stamps on inference requests (#288).
+///
+/// These are the headers cortex asserts after a bearer resolves (#49).
+/// Bench talks to neurons **directly** over the WireGuard mesh, so it
+/// stamps them itself — the same trust model the link already relies on,
+/// since a neuron accepts them from cortex on exactly that basis.
+///
+/// Give bench its *own* pair rather than borrowing a real caller's: it
+/// then holds its own fair-share allocation (#54) and cannot starve
+/// interactive traffic, which is the failure #262 was fixing.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PrincipalSettings {
+    /// Stamped as `x-helexa-account-id`.
+    pub account_id: String,
+    /// Stamped as `x-helexa-key-id`.
+    pub key_id: String,
 }
 
 impl Default for BenchSettings {
@@ -77,6 +115,7 @@ impl Default for BenchSettings {
             iteration_pause_secs: default_iter_pause(),
             request_timeout_secs: default_timeout(),
             db_path: default_db_path(),
+            principal: None,
         }
     }
 }
@@ -110,6 +149,18 @@ pub struct ScenarioConfig {
     /// deliberately, e.g. `concurrency_levels = [2, 4, 8]`.
     #[serde(default)]
     pub concurrency_levels: Vec<u32>,
+    /// Concurrency levels to run **non-streaming** (#285). Separate from
+    /// `concurrency_levels` because the two shapes are not
+    /// interchangeable: streaming multiplexes through the batch engine,
+    /// non-streaming did not until #285 and serialized completely — a
+    /// measured 1.00x aggregate throughput at every level against 3.98x
+    /// streamed.
+    ///
+    /// Defaults to the top level only. The signal is starkest there and
+    /// each extra level costs a full burst per sample per build; an
+    /// operator who wants the whole curve can list it.
+    #[serde(default = "default_concurrency_nonstreaming_levels")]
+    pub concurrency_nonstreaming_levels: Vec<u32>,
     /// Square image sizes (px per side) — one `image:<px>` scenario per
     /// entry (#203), run only against models advertising the `image`
     /// capability. Defaults to `[1024]`.
@@ -148,6 +199,7 @@ impl Default for ScenarioConfig {
             prompt_sizes: default_prompt_sizes(),
             max_tokens: default_max_tokens(),
             concurrency_levels: Vec::new(),
+            concurrency_nonstreaming_levels: Vec::new(),
             concurrency_prompt_tokens: default_concurrency_prompt_tokens(),
             image_sizes: default_image_sizes(),
             capability_probes: Vec::new(),
@@ -215,6 +267,10 @@ fn default_timeout() -> u64 {
 fn default_db_path() -> String {
     "/var/lib/helexa-bench/bench.sqlite".to_string()
 }
+fn default_concurrency_nonstreaming_levels() -> Vec<u32> {
+    vec![8]
+}
+
 fn default_api_enabled() -> bool {
     true
 }
@@ -243,6 +299,44 @@ fn default_capability_max_tokens() -> u64 {
 // figment's, not ours, so suppress the lint here.
 #[allow(clippy::result_large_err)]
 mod tests {
+
+    /// #288: a config with no `[bench.principal]` must still parse — the
+    /// anonymous behaviour is the historical one and correct for
+    /// `openai` targets — but a configured one must round-trip, because
+    /// a silently-dropped principal reverts the measurement to the
+    /// anonymous-yield path without changing a visible number.
+    #[test]
+    fn principal_is_optional_and_round_trips() {
+        let parse = |t: &str| -> BenchConfig {
+            figment::Figment::new()
+                .merge(figment::providers::Toml::string(t))
+                .extract()
+                .expect("parse")
+        };
+        let none = parse("[bench]\nsamples_per_version = 3\n");
+        assert!(none.bench.principal.is_none());
+
+        let some = parse(
+            "[bench]\nsamples_per_version = 3\n\n\
+             [bench.principal]\naccount_id = \"acct\"\nkey_id = \"k1\"\n",
+        );
+        let p = some.bench.principal.expect("principal");
+        assert_eq!((p.account_id.as_str(), p.key_id.as_str()), ("acct", "k1"));
+    }
+
+    /// The non-streaming burst defaults on (#285). If it defaulted off,
+    /// the path that serialized completely would stay unmeasured on every
+    /// fleet that had not opted in — which is how it went unnoticed.
+    #[test]
+    fn nonstreaming_concurrency_defaults_to_the_top_level() {
+        let cfg: BenchConfig = figment::Figment::new()
+            .merge(figment::providers::Toml::string(
+                "[scenarios]\nprompt_sizes = [128]\n",
+            ))
+            .extract()
+            .expect("parse");
+        assert_eq!(cfg.scenarios.concurrency_nonstreaming_levels, vec![8]);
+    }
     use super::*;
     use figment::Jail;
 

crates/helexa-bench/src/scenario.rs

@@ -9,6 +9,7 @@
 use crate::config::ScenarioConfig;
 use anyhow::{Context, Result, anyhow};
 use async_trait::async_trait;
+use cortex_core::entitlements::HEADER_KEY_ID;
 use cortex_core::harness::ModelInfo;
 use cortex_core::openai::ChatCompletionChunk;
 use eventsource_stream::Eventsource;
@@ -46,6 +47,17 @@ pub struct RunCtx<'a> {
     pub model_id: String,
     pub max_tokens: u64,
     pub timeout: Duration,
+    /// Base `x-helexa-key-id` bench presents (#288), when configured.
+    ///
+    /// The client already sends this as a default header, so every
+    /// scenario is identified without having to remember. A concurrency
+    /// burst overrides it per stream — see
+    /// [`ConcurrencyScenario`] — because fair-share (#54) bounds one
+    /// principal to `max_per_principal` in-flight-or-queued requests,
+    /// which defaults to 2. Eight streams under one identity would be
+    /// capped at two and the rest rejected, measuring the fair-share cap
+    /// instead of the server.
+    pub principal_key_id: Option<String>,
 }
 
 /// Operator-felt metrics for a single measured request.
@@ -72,6 +84,34 @@ pub struct ScenarioMetrics {
     pub decode_ms: Option<u64>,
     /// Tokens submitted to prefill — the denominator for prefill tok/s.
     pub prefill_tokens: Option<u64>,
+    /// Tokens spent inside the reasoning span, from
+    /// `usage.completion_tokens_details.reasoning_tokens` (#223). A
+    /// sub-count of `completion_tokens`.
+    ///
+    /// Worth a trend of its own: on a reasoning model this is the
+    /// dominant cost driver and it moves independently of speed. A
+    /// template change, a sampling change (#283) or a reasoning-budget
+    /// change can double what the model thinks before answering while
+    /// every tok/s number stays flat — the bill doubles and no chart
+    /// moves.
+    pub reasoning_tokens: Option<u64>,
+    /// p95 inter-token arrival gap in milliseconds — the tail of the
+    /// stream's smoothness, client-observed.
+    ///
+    /// `decode_tps` is a mean over the whole decode window, so a stream
+    /// that stalls for a second and then catches up is indistinguishable
+    /// from one that never stalled. This is the number a user actually
+    /// feels, and the one a batching stall or a mid-stream rebatch shows
+    /// up in. `None` for non-streaming, which has no inter-token gaps by
+    /// construction, and for streams too short to have a tail.
+    pub tpot_p95_ms: Option<f64>,
+    /// Prompt tokens served from neuron's prefix KV cache (#269), from
+    /// `usage.prompt_tokens_details.cached_tokens`.
+    ///
+    /// The reason prefill timing varies between otherwise identical
+    /// samples. #269 exists because the saving was invisible and every
+    /// client reported a 0% hit rate; bench then threw the number away.
+    pub cached_tokens: Option<u64>,
     // ── Concurrency / agentic-load fields (#89) ──────────────────────────
     // Set only by the concurrency scenario, which fans out N simultaneous
     // streams to characterize the real a0/hermes/opencode workload that
@@ -203,6 +243,9 @@ impl Scenario for ImageLatencyScenario {
                 .as_u64()
                 .map(|d| d + timing["decode_ms"].as_u64().unwrap_or(0)),
             prefill_tokens: None,
+            reasoning_tokens: None,
+            cached_tokens: None,
+            tpot_p95_ms: None,
             concurrency: None,
             ttft_p95_s: None,
             queue_wait_ms_median: None,
@@ -233,6 +276,18 @@ pub fn build_scenarios(cfg: &ScenarioConfig) -> Vec<Box<dyn Scenario>> {
             id: format!("concurrency:{n}"),
             concurrency: n,
             approx_prompt_tokens: cfg.concurrency_prompt_tokens,
+            streaming: true,
+        }) as Box<dyn Scenario>);
+    }
+    // Non-streaming bursts (#285). A separate cell id so the two shapes
+    // never average together — they measured 1.00x against 3.98x before
+    // #285, and a combined number would have hidden both.
+    for &n in &cfg.concurrency_nonstreaming_levels {
+        scenarios.push(Box::new(ConcurrencyScenario {
+            id: format!("concurrency:{n}:nostream"),
+            concurrency: n,
+            approx_prompt_tokens: cfg.concurrency_prompt_tokens,
+            streaming: false,
         }) as Box<dyn Scenario>);
     }
     for &side in &cfg.image_sizes {
@@ -257,7 +312,7 @@ pub fn build_scenarios(cfg: &ScenarioConfig) -> Vec<Box<dyn Scenario>> {
 pub async fn cold_probe(ctx: &RunCtx<'_>) -> Result<ScenarioMetrics> {
     let prompt = build_prompt(128);
     let payload = chat_payload(ctx, &prompt);
-    tokio::time::timeout(ctx.timeout, stream_and_measure(ctx, &payload))
+    tokio::time::timeout(ctx.timeout, stream_and_measure(ctx, &payload, None))
         .await
         .map_err(|_| anyhow!("cold probe timed out after {:?}", ctx.timeout))?
 }
@@ -295,7 +350,7 @@ impl Scenario for ChatLatencyScenario {
     async fn run(&self, ctx: &RunCtx) -> Result<ScenarioMetrics> {
         let prompt = build_prompt(self.approx_prompt_tokens);
         let payload = chat_payload(ctx, &prompt);
-        let fut = stream_and_measure(ctx, &payload);
+        let fut = stream_and_measure(ctx, &payload, None);
         tokio::time::timeout(ctx.timeout, fut)
             .await
             .map_err(|_| anyhow!("request timed out after {:?}", ctx.timeout))?
@@ -313,6 +368,16 @@ pub struct ConcurrencyScenario {
     id: String,
     concurrency: u32,
     approx_prompt_tokens: u32,
+    /// Whether the burst streams (#285).
+    ///
+    /// Both shapes matter and they are not interchangeable. Streaming
+    /// multiplexes through the batch engine; non-streaming did not until
+    /// #285, and serialized completely — 1.00x aggregate throughput at
+    /// every concurrency level against 3.98x streamed. A bench that only
+    /// streams cannot see that, which is why it went unmeasured: the
+    /// neuron reports `in_flight: 8`, admission accepts, `/health` looks
+    /// healthy, and throughput is single-stream.
+    streaming: bool,
 }
 
 #[async_trait]
@@ -327,13 +392,26 @@ impl Scenario for ConcurrencyScenario {
 
     async fn run(&self, ctx: &RunCtx) -> Result<ScenarioMetrics> {
         let prompt = build_prompt(self.approx_prompt_tokens);
-        let payload = chat_payload(ctx, &prompt);
+        let payload = if self.streaming {
+            chat_payload(ctx, &prompt)
+        } else {
+            nonstreaming_payload(ctx, &prompt)
+        };
 
-        // Fire all streams at once; each is independently timed and capped by
-        // the per-request timeout so one hung stream can't stall the burst.
+        // Fire all requests at once; each is independently timed and capped
+        // by the per-request timeout so one hung request can't stall the
+        // burst.
         let burst_start = Instant::now();
-        let futs = (0..self.concurrency).map(|_| async {
-            tokio::time::timeout(ctx.timeout, stream_and_measure(ctx, &payload)).await
+        let streaming = self.streaming;
+        let payload = &payload;
+        let futs = (0..self.concurrency).map(move |i| async move {
+            // Each stream presents its own identity — see
+            // `with_stream_identity`.
+            if streaming {
+                tokio::time::timeout(ctx.timeout, stream_and_measure(ctx, payload, Some(i))).await
+            } else {
+                tokio::time::timeout(ctx.timeout, request_and_measure(ctx, payload, Some(i))).await
+            }
         });
         let results = futures::future::join_all(futs).await;
         let burst_window = burst_start.elapsed().as_secs_f64();
@@ -351,7 +429,7 @@ impl Scenario for ConcurrencyScenario {
         }
         if streams.is_empty() {
             return Err(anyhow!(
-                "all {} concurrent streams failed ({rejected} shed by admission)",
+                "all {} concurrent requests failed ({rejected} shed by admission)",
                 self.concurrency
             ));
         }
@@ -383,6 +461,18 @@ impl Scenario for ConcurrencyScenario {
             prefill_ms: None,
             decode_ms: None,
             prefill_tokens: None,
+            // Summed across the burst, like `completion_tokens`: the cell
+            // describes the whole burst, so a per-stream figure here
+            // would not compose with it.
+            reasoning_tokens: sum_opt(&streams, |m| m.reasoning_tokens),
+            cached_tokens: sum_opt(&streams, |m| m.cached_tokens),
+            // The worst stream in the burst, not a median of medians:
+            // under load the question is how bad it got for somebody,
+            // and averaging tails is how a stall stays invisible.
+            tpot_p95_ms: streams
+                .iter()
+                .filter_map(|m| m.tpot_p95_ms)
+                .max_by(|a, b| a.total_cmp(b)),
             concurrency: Some(self.concurrency),
             ttft_p95_s: percentile(&ttfts, 95.0),
             queue_wait_ms_median: median(&queue_waits),
@@ -393,6 +483,146 @@ impl Scenario for ConcurrencyScenario {
     }
 }
 
+/// Sum an optional per-stream count across a burst, `None` when no
+/// stream reported it — so "nobody measured it" stays distinguishable
+/// from "measured as zero", the same distinction #269 exists to
+/// preserve.
+fn sum_opt(
+    streams: &[ScenarioMetrics],
+    f: impl Fn(&ScenarioMetrics) -> Option<u64>,
+) -> Option<u64> {
+    let vals: Vec<u64> = streams.iter().filter_map(f).collect();
+    (!vals.is_empty()).then(|| vals.iter().sum())
+}
+
+/// Per-stream identity for a concurrency burst (#288/#54).
+///
+/// Fair-share bounds *one principal* to `max_per_principal`
+/// in-flight-or-queued requests, defaulting to 2. A burst of 8 under a
+/// single identity would therefore have 6 rejected with `PrincipalCap`,
+/// and the scenario would measure the fair-share cap rather than the
+/// server.
+///
+/// Giving each stream its own key is also the more faithful model: a
+/// concurrency-8 burst is meant to stand in for eight simultaneous
+/// *callers*, not one caller issuing eight requests. Fair-share is still
+/// exercised — each identity is subject to it — just not self-inflicted.
+fn with_stream_identity(
+    rb: reqwest::RequestBuilder,
+    ctx: &RunCtx<'_>,
+    stream: Option<u32>,
+) -> reqwest::RequestBuilder {
+    match (ctx.principal_key_id.as_deref(), stream) {
+        (Some(base), Some(i)) => rb.header(HEADER_KEY_ID, format!("{base}-{i}")),
+        // No principal configured, or a single-request scenario: the
+        // client's default headers already carry the base identity.
+        _ => rb,
+    }
+}
+
+/// The non-streaming counterpart to [`chat_payload`] (#285).
+///
+/// `stream_options` is deliberately absent: usage is part of the body on
+/// this shape, and sending the streaming-only option to a strict server
+/// is a needless compatibility risk.
+fn nonstreaming_payload(ctx: &RunCtx, prompt: &str) -> serde_json::Value {
+    json!({
+        "model": ctx.model_id,
+        "messages": [{"role": "user", "content": prompt}],
+        "max_tokens": ctx.max_tokens,
+        "temperature": 0,
+        "stream": false,
+    })
+}
+
+/// Time one non-streaming chat completion (#285).
+///
+/// There is no first-chunk moment on this shape — the whole body arrives
+/// at once — so `ttft_s` is the full request wall-clock rather than a
+/// separate measurement. That is the honest reading: for a non-streaming
+/// caller, time-to-first-token *is* time-to-everything, and it is exactly
+/// what makes the serialization this scenario exists to detect so
+/// expensive.
+///
+/// `decode_tps` is per-request; the burst aggregate is computed by the
+/// caller over the whole window, the same way the streaming path does it.
+async fn request_and_measure(
+    ctx: &RunCtx<'_>,
+    payload: &serde_json::Value,
+    stream: Option<u32>,
+) -> Result<ScenarioMetrics> {
+    let start = Instant::now();
+    let resp = with_stream_identity(ctx.client.post(&ctx.chat_url), ctx, stream)
+        .json(payload)
+        .send()
+        .await
+        .context("sending non-streaming chat request")?;
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let body = resp.text().await.unwrap_or_default();
+        return Err(anyhow!("upstream returned {status}: {}", body.trim()));
+    }
+    let v: serde_json::Value = resp
+        .json()
+        .await
+        .context("non-streaming chat response was not JSON")?;
+    let total_s = start.elapsed().as_secs_f64();
+
+    let usage = v.get("usage");
+    let completion_tokens = usage
+        .and_then(|u| u.get("completion_tokens"))
+        .and_then(|t| t.as_u64())
+        .unwrap_or(0);
+    let prompt_tokens = usage
+        .and_then(|u| u.get("prompt_tokens"))
+        .and_then(|t| t.as_u64());
+    let reasoning_tokens = usage
+        .and_then(|u| u.get("completion_tokens_details"))
+        .and_then(|d| d.get("reasoning_tokens"))
+        .and_then(|t| t.as_u64());
+    let cached_tokens = usage
+        .and_then(|u| u.get("prompt_tokens_details"))
+        .and_then(|d| d.get("cached_tokens"))
+        .and_then(|t| t.as_u64());
+    let timing = usage.and_then(|u| u.get("helexa_timing"));
+    let field = |name: &str| timing.and_then(|t| t.get(name)).and_then(|x| x.as_u64());
+    let (prefill_ms, decode_ms, prefill_tokens) = (
+        field("prefill_ms"),
+        field("decode_ms"),
+        field("prefill_tokens"),
+    );
+
+    // Prefer the server's own decode window when it reports one (#85);
+    // it excludes prefill, so it is comparable with the streaming path's
+    // decode-window rate rather than being diluted by it.
+    let decode_tps = match decode_ms {
+        Some(ms) if ms > 200 => Some(completion_tokens as f64 / (ms as f64 / 1000.0)),
+        _ if total_s > 0.2 => Some(completion_tokens as f64 / total_s),
+        _ => None,
+    };
+
+    Ok(ScenarioMetrics {
+        ttft_s: total_s,
+        decode_tps,
+        total_s,
+        prompt_tokens,
+        completion_tokens,
+        prefill_ms,
+        decode_ms,
+        prefill_tokens,
+        reasoning_tokens,
+        cached_tokens,
+        // Non-streaming has no inter-token gaps: the body arrives whole.
+        tpot_p95_ms: None,
+        concurrency: None,
+        ttft_p95_s: None,
+        queue_wait_ms_median: None,
+        rejected: None,
+        artifact: None,
+        image_units: None,
+    })
+}
+
 /// Quality probe (#91): runs a fixed prompt and stores the full generated
 /// text as an artifact for later scoring (manual now, LLM-judge later). The
 /// point is to compare reasoning/planning quality across models — the axis
@@ -425,7 +655,7 @@ impl Scenario for CapabilityScenario {
             "stream": true,
             "stream_options": {"include_usage": true},
         });
-        let fut = stream_and_measure_inner(ctx, &payload, true);
+        let fut = stream_and_measure_inner(ctx, &payload, true, None);
         tokio::time::timeout(ctx.timeout, fut)
             .await
             .map_err(|_| anyhow!("capability probe timed out after {:?}", ctx.timeout))?
@@ -468,8 +698,9 @@ fn percentile(values: &[f64], p: f64) -> Option<f64> {
 async fn stream_and_measure(
     ctx: &RunCtx<'_>,
     payload: &serde_json::Value,
+    stream: Option<u32>,
 ) -> Result<ScenarioMetrics> {
-    stream_and_measure_inner(ctx, payload, false).await
+    stream_and_measure_inner(ctx, payload, false, stream).await
 }
 
 /// As [`stream_and_measure`] but accumulates the full visible text when
@@ -479,11 +710,10 @@ async fn stream_and_measure_inner(
     ctx: &RunCtx<'_>,
     payload: &serde_json::Value,
     capture_text: bool,
+    stream_ix: Option<u32>,
 ) -> Result<ScenarioMetrics> {
     let start = Instant::now();
-    let resp = ctx
-        .client
-        .post(&ctx.chat_url)
+    let resp = with_stream_identity(ctx.client.post(&ctx.chat_url), ctx, stream_ix)
         .json(payload)
         .send()
         .await
@@ -503,6 +733,10 @@ async fn stream_and_measure_inner(
     let mut prefill_ms: Option<u64> = None;
     let mut decode_ms: Option<u64> = None;
     let mut prefill_tokens: Option<u64> = None;
+    let mut reasoning_tokens: Option<u64> = None;
+    let mut cached_tokens: Option<u64> = None;
+    // Inter-token arrival gaps, for the p95 tail.
+    let mut gaps_ms: Vec<f64> = Vec::new();
     let mut captured = String::new();
 
     while let Some(event) = stream.next().await {
@@ -540,6 +774,11 @@ async fn stream_and_measure_inner(
             if content.is_some() || reasoning.is_some() {
                 if first.is_none() {
                     first = Some(now);
+                } else if let Some(prev) = last {
+                    // Gaps only *between* generated deltas: the wait for
+                    // the first one is TTFT and belongs to prefill, not
+                    // to stream smoothness.
+                    gaps_ms.push(now.duration_since(prev).as_secs_f64() * 1000.0);
                 }
                 last = Some(now);
                 chunk_count += 1;
@@ -551,6 +790,14 @@ async fn stream_and_measure_inner(
         if let Some(usage) = chunk.usage {
             prompt_tokens = Some(usage.prompt_tokens);
             completion_tokens = Some(usage.completion_tokens);
+            reasoning_tokens = usage
+                .completion_tokens_details
+                .as_ref()
+                .map(|d| d.reasoning_tokens);
+            cached_tokens = usage
+                .prompt_tokens_details
+                .as_ref()
+                .map(|d| d.cached_tokens);
             if let Some(t) = usage.helexa_timing {
                 prefill_ms = Some(t.prefill_ms);
                 decode_ms = Some(t.decode_ms);
@@ -589,6 +836,9 @@ async fn stream_and_measure_inner(
         prefill_ms,
         decode_ms,
         prefill_tokens,
+        reasoning_tokens,
+        cached_tokens,
+        tpot_p95_ms: percentile(&gaps_ms, 95.0),
         // Concurrency fields unset on the single-request path; the
         // concurrency scenario builds its own aggregate (#89).
         concurrency: None,
@@ -604,6 +854,67 @@ async fn stream_and_measure_inner(
 mod tests {
     use super::*;
 
+    /// #288/#54: a burst must present one identity *per stream*.
+    ///
+    /// Fair-share bounds a single principal to `max_per_principal`
+    /// in-flight-or-queued requests, which defaults to 2. Eight streams
+    /// under one key would have six rejected with `PrincipalCap`, and
+    /// `concurrency:8` would silently measure the fair-share cap instead
+    /// of the server — a worse failure than the anonymity it replaced,
+    /// because it looks like backpressure.
+    #[test]
+    fn a_burst_presents_one_identity_per_stream() {
+        let client = reqwest::Client::new();
+        let ctx = RunCtx {
+            client: &client,
+            chat_url: "http://x/v1/chat/completions".into(),
+            model_id: "m".into(),
+            max_tokens: 16,
+            timeout: Duration::from_secs(1),
+            principal_key_id: Some("fleet-benchmark".into()),
+        };
+        let key_of = |stream: Option<u32>| -> Option<String> {
+            with_stream_identity(client.post(&ctx.chat_url), &ctx, stream)
+                .build()
+                .expect("build")
+                .headers()
+                .get(HEADER_KEY_ID)
+                .map(|v| v.to_str().expect("utf8").to_string())
+        };
+        let keys: Vec<Option<String>> = (0..3).map(|i| key_of(Some(i))).collect();
+        assert_eq!(
+            keys,
+            vec![
+                Some("fleet-benchmark-0".to_string()),
+                Some("fleet-benchmark-1".to_string()),
+                Some("fleet-benchmark-2".to_string()),
+            ]
+        );
+        // Distinct, or the cap bites anyway.
+        let uniq: std::collections::HashSet<_> = keys.iter().collect();
+        assert_eq!(uniq.len(), 3);
+    }
+
+    /// A single-request scenario adds no override: the client's default
+    /// headers already carry the base identity, and a per-request copy
+    /// would be a second place for it to drift.
+    #[test]
+    fn a_single_request_scenario_keeps_the_client_identity() {
+        let client = reqwest::Client::new();
+        let ctx = RunCtx {
+            client: &client,
+            chat_url: "http://x/v1/chat/completions".into(),
+            model_id: "m".into(),
+            max_tokens: 16,
+            timeout: Duration::from_secs(1),
+            principal_key_id: Some("fleet-benchmark".into()),
+        };
+        let req = with_stream_identity(client.post(&ctx.chat_url), &ctx, None)
+            .build()
+            .expect("build");
+        assert!(req.headers().get(HEADER_KEY_ID).is_none());
+    }
+
     #[test]
     fn prompt_grows_with_token_target() {
         let small = build_prompt(128);
@@ -646,6 +957,7 @@ mod tests {
             prompt_sizes: vec![128],
             max_tokens: 64,
             concurrency_levels: vec![2, 8],
+            concurrency_nonstreaming_levels: vec![8],
             concurrency_prompt_tokens: 512,
             capability_probes: vec![CapabilityProbe {
                 name: "plan".into(),

crates/helexa-bench/src/store.rs

@@ -14,6 +14,22 @@ use std::path::Path;
 #[derive(Debug, Clone)]
 pub struct RunRecord {
     pub ts: String, // RFC3339
+    /// Identity this sample was taken under (#288), `account/key`, or
+    /// `None` for an anonymous run. Anonymous samples measure the
+    /// admission yield policy rather than serving capacity once #262 is
+    /// in the build, so this is the column that says whether two rows
+    /// are comparable at all.
+    pub principal: Option<String>,
+    /// Tokens spent reasoning (#223) — the dominant cost driver on a
+    /// reasoning model, and one that moves independently of every tok/s
+    /// number on the dashboard.
+    pub reasoning_tokens: Option<u64>,
+    /// Prompt tokens served from the prefix cache (#269) — the reason
+    /// prefill timing varies between otherwise identical samples.
+    pub cached_tokens: Option<u64>,
+    /// p95 inter-token arrival gap (ms) — stream smoothness, which a
+    /// mean tok/s cannot express.
+    pub tpot_p95_ms: Option<f64>,
     // target
     pub target_name: String,
     pub target_kind: String,
@@ -204,6 +220,18 @@ impl Store {
                 ("artifact", "TEXT"),
                 ("quality_score", "REAL"),
                 ("scorer", "TEXT"),
+                // #288: which identity the sample was taken under. Rows
+                // predating this are NULL, which is honest — they were
+                // anonymous, and since #262 that measures a different
+                // thing. Without the column the 2026-08-18 discontinuity
+                // is folklore rather than data.
+                ("principal", "TEXT"),
+                // #223 / #269: both are on the wire and were being discarded.
+                ("reasoning_tokens", "INTEGER"),
+                ("cached_tokens", "INTEGER"),
+                // #285: inter-token tail. `decode_tps` is a mean and hides
+                // a stall that later catches up.
+                ("tpot_p95_ms", "REAL"),
             ],
         )?;
         Ok(())
@@ -264,6 +292,7 @@ impl Store {
                 swap_unload_ms, swap_load_ms,
                 artifact, quality_score, scorer,
                 image_units,
+                principal, reasoning_tokens, cached_tokens, tpot_p95_ms,
                 ok, error
             ) VALUES (
                 ?1, ?2, ?3, ?4,
@@ -280,7 +309,8 @@ impl Store {
                 ?42, ?43,
                 ?44, ?45, ?46,
                 ?47,
-                ?48, ?49
+                ?48, ?49, ?50, ?51,
+                ?52, ?53
             )",
             params![
                 r.ts,
@@ -330,6 +360,10 @@ impl Store {
                 r.quality_score,
                 r.scorer,
                 r.image_units,
+                r.principal,
+                r.reasoning_tokens,
+                r.cached_tokens,
+                r.tpot_p95_ms,
                 r.ok as i64,
                 r.error,
             ],
@@ -703,7 +737,15 @@ impl Store {
             }
         };
         let mut stmt = self.conn.prepare(
-            "SELECT git_sha, build_timestamp, package_version, ttft_s, decode_tps, total_s, ts
+            "SELECT git_sha, build_timestamp, package_version, ttft_s, decode_tps, total_s,
+                    ttft_p95_s, queue_wait_ms, rejected,
+                    -- Prefill rate derived here rather than stored: the
+                    -- inputs are already recorded and a stored rate
+                    -- would be a third thing to keep consistent.
+                    CASE WHEN prefill_ms > 0 AND prefill_tokens IS NOT NULL
+                         THEN (prefill_tokens * 1000.0) / prefill_ms END AS prefill_tps,
+                    reasoning_tokens, cached_tokens, completion_tokens,
+                    tpot_p95_ms, principal, ts
              FROM runs
              WHERE ok=1 AND target_name=?1 AND model_id=?2 AND scenario_id=?3
              ORDER BY id",
@@ -717,7 +759,16 @@ impl Store {
                     ttft_s: r.get(3)?,
                     decode_tps: r.get(4)?,
                     total_s: r.get(5)?,
-                    ts: r.get(6)?,
+                    ttft_p95_s: r.get(6)?,
+                    queue_wait_ms: r.get(7)?,
+                    rejected: r.get(8)?,
+                    prefill_tps: r.get(9)?,
+                    reasoning_tokens: r.get(10)?,
+                    cached_tokens: r.get(11)?,
+                    completion_tokens: r.get(12)?,
+                    tpot_p95_ms: r.get(13)?,
+                    principal: r.get(14)?,
+                    ts: r.get(15)?,
                 })
             })?
             .collect::<rusqlite::Result<_>>()?;
@@ -838,6 +889,33 @@ pub struct SeriesPoint {
     pub ttft_s_median: Option<f64>,
     pub decode_tps_median: Option<f64>,
     pub total_s_median: Option<f64>,
+    // ── Added because the two above cannot show what changed ──────
+    /// Tail TTFT. The median above stayed flat at ~0.3 s across the
+    /// build where p95 went 0.53 s → 11.77 s (#288); charting only the
+    /// median is why that went unnoticed for a week.
+    pub ttft_p95_s_median: Option<f64>,
+    /// Admission queue wait — separates "the server is slow" from
+    /// "you were queued behind someone".
+    pub queue_wait_ms_median: Option<f64>,
+    /// Requests shed by admission during the burst.
+    pub rejected_median: Option<f64>,
+    /// Prefill throughput, derived from `prefill_tokens / prefill_ms`.
+    /// The other half of serving speed; only decode was ever charted.
+    pub prefill_tps_median: Option<f64>,
+    /// Tokens spent reasoning (#223) — the dominant cost driver on a
+    /// reasoning model, and independent of every speed metric here.
+    pub reasoning_tokens_median: Option<f64>,
+    /// Prompt tokens served from the prefix cache (#269).
+    pub cached_tokens_median: Option<f64>,
+    /// Total generated tokens — catches a model that starts truncating
+    /// or rambling, which no rate metric shows.
+    pub completion_tokens_median: Option<f64>,
+    /// p95 inter-token gap (ms) — stream smoothness.
+    pub tpot_p95_ms_median: Option<f64>,
+    /// Identity the samples were taken under (#288), or `None` if
+    /// anonymous. Anonymous and identified points are **not
+    /// comparable** once #262 is in the build.
+    pub principal: Option<String>,
     pub samples: usize,
 }
 
@@ -848,6 +926,15 @@ struct SeriesRaw {
     ttft_s: Option<f64>,
     decode_tps: Option<f64>,
     total_s: Option<f64>,
+    ttft_p95_s: Option<f64>,
+    queue_wait_ms: Option<f64>,
+    rejected: Option<f64>,
+    prefill_tps: Option<f64>,
+    reasoning_tokens: Option<f64>,
+    cached_tokens: Option<f64>,
+    completion_tokens: Option<f64>,
+    tpot_p95_ms: Option<f64>,
+    principal: Option<String>,
     ts: String,
 }
 
@@ -880,6 +967,19 @@ fn aggregate_series(raws: Vec<SeriesRaw>) -> Vec<SeriesPoint> {
                 ttft_s_median: median(rows.iter().filter_map(|r| r.ttft_s)),
                 decode_tps_median: median(rows.iter().filter_map(|r| r.decode_tps)),
                 total_s_median: median(rows.iter().filter_map(|r| r.total_s)),
+                ttft_p95_s_median: median(rows.iter().filter_map(|r| r.ttft_p95_s)),
+                queue_wait_ms_median: median(rows.iter().filter_map(|r| r.queue_wait_ms)),
+                rejected_median: median(rows.iter().filter_map(|r| r.rejected)),
+                prefill_tps_median: median(rows.iter().filter_map(|r| r.prefill_tps)),
+                reasoning_tokens_median: median(rows.iter().filter_map(|r| r.reasoning_tokens)),
+                cached_tokens_median: median(rows.iter().filter_map(|r| r.cached_tokens)),
+                completion_tokens_median: median(rows.iter().filter_map(|r| r.completion_tokens)),
+                tpot_p95_ms_median: median(rows.iter().filter_map(|r| r.tpot_p95_ms)),
+                // Any row's principal: a build's samples are all taken
+                // under one identity, and a mixture would mean the
+                // config changed mid-build, which the chart should show
+                // as a discontinuity rather than average away.
+                principal: rows.iter().find_map(|r| r.principal.clone()),
                 samples: rows.len(),
             };
             (sort_key, point)
@@ -1281,6 +1381,10 @@ mod tests {
     fn rec(target: &str, sha: &str, model: &str, scenario: &str, ok: bool) -> RunRecord {
         RunRecord {
             ts: "2026-06-13T00:00:00Z".into(),
+            principal: None,
+            reasoning_tokens: None,
+            cached_tokens: None,
+            tpot_p95_ms: None,
             target_name: target.into(),
             target_kind: "neuron".into(),
             endpoint: "http://x:13131".into(),

crates/helexa-bench/src/sweep.rs

@@ -78,7 +78,10 @@ pub struct Sweeper {
 
 impl Sweeper {
     pub fn new(cfg: BenchConfig) -> Result<Self> {
-        let client = TargetClient::new(cfg.bench.request_timeout())?;
+        let client = TargetClient::with_principal(
+            cfg.bench.request_timeout(),
+            cfg.bench.principal.as_ref(),
+        )?;
         let store = Store::open(&cfg.bench.db_path)?;
         Ok(Sweeper { cfg, client, store })
     }
@@ -169,6 +172,7 @@ impl Sweeper {
             model_id: model.id.clone(),
             max_tokens: self.cfg.scenarios.max_tokens,
             timeout: self.cfg.bench.request_timeout(),
+            principal_key_id: self.cfg.bench.principal.as_ref().map(|p| p.key_id.clone()),
         };
         let cold = crate::scenario::cold_probe(&ctx).await;
         let swap = SwapTiming { unload_ms, load_ms };
@@ -252,6 +256,7 @@ impl Sweeper {
                     model_id: model.id.clone(),
                     max_tokens: self.cfg.scenarios.max_tokens,
                     timeout: self.cfg.bench.request_timeout(),
+                    principal_key_id: self.cfg.bench.principal.as_ref().map(|p| p.key_id.clone()),
                 };
 
                 // One unmeasured warmup when the cell is empty (matches
@@ -346,6 +351,15 @@ impl Sweeper {
 
         RunRecord {
             ts: chrono::Utc::now().to_rfc3339(),
+            // #288: stamp the identity this sample was taken under, so a
+            // later reader can tell an anonymous row from an identified
+            // one instead of inferring it from the date.
+            principal: self
+                .cfg
+                .bench
+                .principal
+                .as_ref()
+                .map(|p| format!("{}/{}", p.account_id, p.key_id)),
             target_name: target.name.clone(),
             target_kind: kind_str(target.kind).to_string(),
             endpoint: target.endpoint.clone(),
@@ -383,6 +397,9 @@ impl Sweeper {
             prefill_ms: m.and_then(|m| m.prefill_ms),
             decode_ms: m.and_then(|m| m.decode_ms),
             prefill_tokens: m.and_then(|m| m.prefill_tokens),
+            reasoning_tokens: m.and_then(|x| x.reasoning_tokens),
+            cached_tokens: m.and_then(|x| x.cached_tokens),
+            tpot_p95_ms: m.and_then(|x| x.tpot_p95_ms),
             vram_used_mb: health.map(|h| h.vram_used_mb),
             gpu_util_pct: health.map(|h| h.gpu_util_pct),
             gpu_temp_c: health.map(|h| h.gpu_temp_c),

crates/helexa-bench/tests/api.rs

@@ -15,6 +15,10 @@ fn rec(
     ok: bool,
 ) -> RunRecord {
     RunRecord {
+        principal: None,
+        reasoning_tokens: None,
+        cached_tokens: None,
+        tpot_p95_ms: None,
         ts: "2026-06-13T00:00:00Z".into(),
         target_name: host.into(),
         target_kind: "neuron".into(),

crates/helexa-bench/tests/sweep_integration.rs

@@ -80,6 +80,7 @@ async fn spawn_mock(sha: &str) -> (String, Arc<Mutex<String>>) {
 fn config_for(endpoint: String, db_path: String) -> BenchConfig {
     BenchConfig {
         bench: BenchSettings {
+            principal: None,
             sweep_interval_secs: 1,
             samples_per_version: 2,
             iteration_pause_secs: 0,
@@ -91,6 +92,7 @@ fn config_for(endpoint: String, db_path: String) -> BenchConfig {
             prompt_sizes: vec![128], // single scenario keeps assertions simple
             max_tokens: 16,
             concurrency_levels: Vec::new(),
+            concurrency_nonstreaming_levels: vec![],
             concurrency_prompt_tokens: 512,
             capability_probes: Vec::new(),
         },

crates/helexa-router/src/catalogue.rs

@@ -80,10 +80,19 @@ fn router_entry(cortex: &str, e: &CortexModelEntry) -> CortexModelEntry {
         cost: e.cost.clone(),
         tool_call: e.tool_call,
         reasoning: e.reasoning,
-        // Derived from `limit` by the final sync pass in aggregate_models.
-        max_model_len: None,
-        max_input_tokens: None,
-        max_output_tokens: None,
+        // Every flat ecosystem field is derived from `limit` by the
+        // final sync pass in aggregate_models, so they are cleared here
+        // rather than copied. Spread from `e` keeps a field added to the
+        // entry later from being silently dropped at this hop.
+        ..CortexModelEntry {
+            max_model_len: None,
+            max_input_tokens: None,
+            max_output_tokens: None,
+            context_window: None,
+            context_length: None,
+            reasoning_budget: Vec::new(),
+            ..e.clone()
+        }
     }
 }
 
@@ -164,6 +173,9 @@ mod tests {
             max_model_len: None,
             max_input_tokens: None,
             max_output_tokens: None,
+            context_window: None,
+            context_length: None,
+            reasoning_budget: Vec::new(),
         }
     }
 
@@ -224,6 +236,7 @@ mod tests {
             context: 32_768,
             input: None,
             output: 4096,
+            output_ceiling: 0,
         });
         a.cost = Some(ModelCost {
             input: 0.50,
@@ -236,6 +249,7 @@ mod tests {
             context: 16_384, // tighter
             input: None,
             output: 4096,
+            output_ceiling: 0,
         });
         b.cost = Some(ModelCost {
             input: 0.20, // cheaper

crates/helexa-router/tests/dispatch.rs

@@ -42,6 +42,9 @@ fn model_entry(loaded: bool, feasible: bool) -> CortexModelEntry {
         max_model_len: None,
         max_input_tokens: None,
         max_output_tokens: None,
+        context_window: None,
+        context_length: None,
+        reasoning_budget: Vec::new(),
     }
 }
 

crates/helexa-tools/src/config.rs

@@ -58,6 +58,11 @@ mod tests {
         assert!(cfg.max_redirects > 0);
     }
 
+    // The closure's error type is `figment::Error`, which is 208 bytes and
+    // not ours to shrink; `Jail::expect_with` dictates the signature. The
+    // usual remedy (box the Err) is not available through a third-party
+    // callback contract, and this is a test closure run once.
+    #[allow(clippy::result_large_err)]
     #[test]
     fn loads_toml_with_env_override() {
         figment::Jail::expect_with(|jail| {

crates/neuron/src/api.rs

@@ -21,7 +21,7 @@ use futures::stream::{self, StreamExt};
 use serde_json::{Value, json};
 use std::convert::Infallible;
 use std::sync::Arc;
-use std::time::{SystemTime, UNIX_EPOCH};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
 use tokio::sync::RwLock;
 use tokio_stream::wrappers::ReceiverStream;
 
@@ -272,6 +272,35 @@ fn default_enable_thinking(req: &mut ChatCompletionRequest, include_thinking: bo
 /// client-supplied copy and asserts the authoritative value, so over the
 /// trusted WireGuard link these are safe to key fair-share on. `None` for an
 /// unauthenticated/direct request — exempt from the per-principal cap.
+/// SSE keep-alive for the inference streams.
+///
+/// A streamed completion emits its `role` delta immediately and then says
+/// **nothing at all until prefill finishes** — there is no token to send
+/// yet. On a long prompt that silence is seconds long: a 7.4k-token prompt
+/// on a TP-2 27B prefills in about six.
+///
+/// `KeepAlive::default()` is axum's 15s, which is longer than almost every
+/// prefill we serve, so in practice it never fired and the gap went out on
+/// the wire as dead air. Clients read dead air as a dropped stream: the
+/// DeepSeek Harness gives up after roughly a second and reports
+/// `terminated`, then retries — and because the abandoned request keeps its
+/// admission permit and the TP pool lock until it finishes, each retry
+/// queues behind the last one and waits *longer*, so the session never
+/// recovers on its own. Every attempt failed while the model was healthy
+/// and, by then, three sequences deep in work nobody was listening to.
+///
+/// One second is chosen to sit under the shortest client idle timeout we
+/// have met rather than to be tuned to any particular one. The cost is a
+/// 3-byte comment frame per idle second, which SSE parsers discard by
+/// spec; the benefit is that a prefill of any length looks alive.
+///
+/// This does not make a slow prefill fast — see the `chunked prefill`
+/// path for that — it stops a slow prefill being indistinguishable from a
+/// dead connection.
+fn prefill_keep_alive() -> KeepAlive {
+    KeepAlive::new().interval(Duration::from_secs(1))
+}
+
 fn principal_key(headers: &axum::http::HeaderMap) -> Option<String> {
     let account = headers.get(HEADER_ACCOUNT_ID)?.to_str().ok()?;
     let key = headers.get(HEADER_KEY_ID)?.to_str().ok()?;
@@ -346,7 +375,7 @@ async fn chat_completions(
                 let done_stream =
                     stream::once(async { Ok::<_, Infallible>(Event::default().data("[DONE]")) });
                 Sse::new(body_stream.chain(done_stream))
-                    .keep_alive(KeepAlive::default())
+                    .keep_alive(prefill_keep_alive())
                     .into_response()
             }
             Err(e) => inference_error_response(e),
@@ -419,7 +448,7 @@ async fn responses(
                     Ok::<_, Infallible>(Event::default().event(frame.event_name).data(body))
                 });
                 Sse::new(body_stream)
-                    .keep_alive(KeepAlive::default())
+                    .keep_alive(prefill_keep_alive())
                     .into_response()
             }
             Err(e) => inference_error_response(e),
@@ -1057,3 +1086,46 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod keep_alive_tests {
+    use super::*;
+    use futures::StreamExt;
+
+    /// A stream that emits its opening frame and then goes quiet — the
+    /// shape of every streamed completion while it prefills — must keep
+    /// putting bytes on the wire.
+    ///
+    /// The deadline here is the point: with axum's 15s default this test
+    /// times out, which is exactly the production failure. A 7.4k-token
+    /// prompt prefills in ~6s, so the default never fired once, the
+    /// connection sat silent for the whole prefill, and clients dropped it
+    /// as dead — the DeepSeek Harness after ~1s, reporting `terminated`.
+    ///
+    /// Asserted against observable output rather than the configured
+    /// interval because `KeepAlive` exposes no getter, and because what
+    /// broke was the behaviour, not the number.
+    #[tokio::test]
+    async fn a_silent_stream_still_sends_bytes_while_prefilling() {
+        let silent = stream::pending::<Result<Event, Infallible>>();
+        let response = Sse::new(silent)
+            .keep_alive(prefill_keep_alive())
+            .into_response();
+
+        let mut body = response.into_body().into_data_stream();
+        let frame = tokio::time::timeout(Duration::from_secs(3), body.next())
+            .await
+            .expect("a silent stream must emit a keep-alive well inside any client idle timeout")
+            .expect("stream ended instead of keeping alive")
+            .expect("keep-alive frame");
+
+        // SSE comment frames start with ':' and are discarded by parsers
+        // per spec, so this costs the client nothing but proves liveness.
+        assert_eq!(
+            frame.first(),
+            Some(&b':'),
+            "expected an SSE comment frame, got {:?}",
+            String::from_utf8_lossy(&frame)
+        );
+    }
+}

crates/neuron/src/config.rs

@@ -86,6 +86,16 @@ pub struct CandleHarnessConfig {
     #[serde(default)]
     pub context_limit: ContextLimitConfig,
 
+    /// What each OpenAI effort level costs, in reasoning tokens (#223).
+    ///
+    /// OpenAI-shaped clients cannot send a token count — their whole
+    /// vocabulary is the `minimal|low|medium|high` ladder — so the
+    /// numbers live here and are advertised on `/models`. Operator
+    /// configurable because the right values depend on the model and the
+    /// hardware.
+    #[serde(default)]
+    pub reasoning_budget: crate::harness::reasoning_budget::ReasoningBudgetLadder,
+
     /// Admission control (#53): bounds the per-model wait queue so a busy
     /// model returns a fast, retryable `429`/`503` instead of stalling new
     /// requests until their client times out.
@@ -333,8 +343,24 @@ pub struct ContextLimitConfig {
     /// Generation reserve (tokens) left below the context wall:
     /// `input = context − output_reserve_tokens`. Defaults to neuron's
     /// default `max_tokens`.
+    ///
+    /// This is the budget a request gets when it names none, and the
+    /// figure the input/output split is derived from. It is **not** the
+    /// most a request may ask for — see `max_output_tokens` (#278).
     #[serde(default = "default_output_reserve_tokens")]
     pub output_reserve_tokens: usize,
+
+    /// The largest output a request may name, advertised as
+    /// `max_output_tokens` on `/v1/models` (#278).
+    ///
+    /// `None` derives it as `min(max_position_embeddings,
+    /// DEFAULT_OUTPUT_CEILING)`. Separate from the reserve because the
+    /// two answer different questions and were previously the same
+    /// number: publishing the reserve told every client that read the
+    /// advertisement to cap itself at the default, which on a reasoning
+    /// model can be less than one think block.
+    #[serde(default)]
+    pub max_output_tokens: Option<usize>,
 }
 
 impl Default for ContextLimitConfig {
@@ -346,6 +372,7 @@ impl Default for ContextLimitConfig {
             activation_headroom_mb: default_activation_headroom_mb(),
             min_free_floor_mb: default_context_min_free_floor_mb(),
             output_reserve_tokens: default_output_reserve_tokens(),
+            max_output_tokens: None,
         }
     }
 }

crates/neuron/src/harness/admission.rs

@@ -1051,7 +1051,7 @@ mod tests {
 
         // Fill the single queue slot with a waiter that parks on the semaphore.
         let ctrl2 = Arc::clone(&ctrl);
-        let waiter = tokio::spawn(async move { ctrl2.enter(None).await.map(|p| drop(p)) });
+        let waiter = tokio::spawn(async move { ctrl2.enter(None).await.map(drop) });
         // Give the waiter a moment to occupy the queue slot.
         tokio::time::sleep(Duration::from_millis(50)).await;
         assert_eq!(ctrl.queue_depth(), 1);

crates/neuron/src/harness/arch/qwen3_5/snapshot.rs

@@ -562,9 +562,9 @@ mod tests {
                 .forward_batch_decode(&input, &positions, mask.as_ref())
                 .expect("batched step");
             assert_eq!(h.dims()[0], 3);
-            for row in 0..3 {
+            for (row, exp) in expected.iter().enumerate() {
                 let got: Vec<f32> = h.i((row, 0, ..)).unwrap().to_vec1().unwrap();
-                let diff = max_abs_diff(&expected[row][t], &got);
+                let diff = max_abs_diff(&exp[t], &got);
                 assert!(diff < 1e-4, "row {row} step {t} diverged: {diff}");
             }
         }
@@ -610,9 +610,9 @@ mod tests {
         let h = model
             .forward_batch_decode(&input, &[3, 3], None)
             .expect("step");
-        for row in 0..2 {
+        for (row, exp) in expected.iter().enumerate() {
             let got: Vec<f32> = h.i((row, 0, ..)).unwrap().to_vec1().unwrap();
-            let diff = max_abs_diff(&expected[row], &got);
+            let diff = max_abs_diff(exp, &got);
             assert!(diff < 1e-4, "row {row} diverged: {diff}");
         }
     }
@@ -674,9 +674,9 @@ mod tests {
             let h = model
                 .forward_batch_decode(&input, &positions, mask.as_ref())
                 .expect("batched step");
-            for row in 0..2 {
+            for (row, exp) in expected.iter().enumerate() {
                 let got: Vec<f32> = h.i((row, 0, ..)).unwrap().to_vec1().unwrap();
-                let diff = max_abs_diff(&expected[row][t], &got);
+                let diff = max_abs_diff(&exp[t], &got);
                 assert!(diff < 1e-4, "batched row {row} step {t}: {diff}");
             }
         }

crates/neuron/src/harness/chat_template.rs

@@ -233,7 +233,24 @@ pub fn render_chat_template(
                 MessageContent::Null => Value::Null,
             };
             let mut obj = serde_json::Map::new();
-            obj.insert("role".into(), Value::String(m.role.clone()));
+            // `developer` is OpenAI's successor to `system` and is what
+            // a client sends once a model is declared reasoning-capable
+            // (pi-ai: `model.reasoning && supportsDeveloperRole`). Most
+            // chat templates — every one on this fleet — only know the
+            // older spelling and `raise_exception` on anything else, so
+            // an unmodified client got a 422 for a system prompt it sent
+            // correctly.
+            //
+            // Rewritten only when the template does not handle the role
+            // itself: a template that knows `developer` may treat it as
+            // distinct from `system`, and substituting there would be us
+            // deciding something the model's authors already decided.
+            let role = if m.role == "developer" && !template.contains("developer") {
+                "system"
+            } else {
+                m.role.as_str()
+            };
+            obj.insert("role".into(), Value::String(role.to_string()));
             obj.insert("content".into(), content_value);
             // Forward extras (e.g. tool_calls on assistant turns,
             // tool_call_id on tool result turns). HF templates that
@@ -498,6 +515,78 @@ mod tests {
     /// Minimal Qwen3-style template — enough surface to confirm
     /// our renderer threads role + content correctly without
     /// loading a real model's tokenizer_config.json.
+    /// A reasoning-capable model makes OpenAI clients send the system
+    /// prompt as `developer` — and every chat template on this fleet
+    /// raises on an unknown role, so the caller got a 422 for a prompt
+    /// it had sent correctly. Observed live: pi with `reasoning: true`,
+    /// `template_render_failed: Unexpected message role`.
+    #[test]
+    fn the_developer_role_reaches_a_template_that_only_knows_system() {
+        const SYSTEM_ONLY: &str = "{%- for message in messages -%}\
+{%- if message.role == 'system' -%}SYS:{{ message.content }}\
+{%- elif message.role == 'user' -%}USR:{{ message.content }}\
+{%- else -%}{{ raise_exception('Unexpected message role.') }}\
+{%- endif -%}\
+{%- endfor -%}";
+        let messages = vec![
+            ChatMessage {
+                role: "developer".into(),
+                content: MessageContent::Text("be terse".into()),
+                extra: Value::Object(Default::default()),
+            },
+            user_msg("hi"),
+        ];
+        let out = render_chat_template(SYSTEM_ONLY, &messages, &Value::Null, &Value::Null)
+            .expect("a developer role must not blow up a system-only template");
+        assert!(out.contains("SYS:be terse"), "got {out:?}");
+    }
+
+    /// A template that models `developer` itself keeps it: the role may
+    /// mean something distinct there, and substituting would override a
+    /// decision the model's authors already made.
+    #[test]
+    fn a_template_that_knows_developer_keeps_it() {
+        const KNOWS_DEVELOPER: &str = "{%- for message in messages -%}\
+{%- if message.role == 'developer' -%}DEV:{{ message.content }}\
+{%- else -%}{{ message.role }}:{{ message.content }}\
+{%- endif -%}\
+{%- endfor -%}";
+        let messages = vec![ChatMessage {
+            role: "developer".into(),
+            content: MessageContent::Text("be terse".into()),
+            extra: Value::Object(Default::default()),
+        }];
+        let out = render_chat_template(KNOWS_DEVELOPER, &messages, &Value::Null, &Value::Null)
+            .expect("render");
+        assert!(out.contains("DEV:be terse"), "got {out:?}");
+    }
+
+    /// The fetcher and the reader must agree on the filenames, or the
+    /// reader looks for something nobody downloaded and falls back to a
+    /// built-in prompt format without anything failing (#280 — observed
+    /// live: every model neuron downloaded itself was served with the
+    /// fallback because no sidecar was ever fetched).
+    #[test]
+    fn every_template_source_this_reads_is_one_the_loader_fetches() {
+        for name in crate::harness::candle::CandleHarness::SIDECAR_FILES {
+            assert!(
+                !name.is_empty(),
+                "sidecar names are the contract between fetch and read"
+            );
+        }
+        // The three the reader consults, in its own priority order.
+        for source in [
+            "chat_template.jinja",
+            "chat_template.json",
+            "tokenizer_config.json",
+        ] {
+            assert!(
+                crate::harness::candle::CandleHarness::SIDECAR_FILES.contains(&source),
+                "{source} is read by load_chat_template_alongside but never fetched"
+            );
+        }
+    }
+
     const QWEN3_LIKE: &str = "{%- for message in messages -%}\
 <|im_start|>{{ message.role }}\n{{ message.content }}<|im_end|>\n\
 {%- endfor -%}\

crates/neuron/src/harness/context_limit.rs

@@ -234,10 +234,23 @@ fn round_down(tokens: usize, granularity: usize) -> usize {
 
 const CONTEXT_GRANULARITY: usize = 1024;
 
+/// Default ceiling on what a request may name as its output budget.
+///
+/// Not derived from anything — a deliberate figure, and configurable
+/// per host (`[context_limit] max_output_tokens`). The evidence behind
+/// it: a reasoning model asked to build a small application front-loads
+/// one long think block before it acts, measured at 27,504 tokens for a
+/// frontier model on a task Qwen3.8-27B was cut off attempting at
+/// 24,433 (#271, #278). A ceiling below that turns a capable model into
+/// one that never reaches its first tool call, so the default sits
+/// above the observed cost of that class of turn.
+const DEFAULT_OUTPUT_CEILING: usize = 32_768;
+
 /// Derive `limit{context,input,output}` for a loaded model.
 ///
 /// ```text
-/// output  = output_reserve_tokens
+/// output         = output_reserve_tokens          (default + KV planning)
+/// output_ceiling = min(max_output_tokens, max_position_embeddings)
 /// vram_ceiling       = (free_tightest − activation_headroom − min_free_floor) / kv_bytes_per_token_per_card
 /// throughput_ceiling = target_prefill_latency_secs × prefill_tok_per_sec
 /// context = min(max_position_embeddings, vram_ceiling, throughput_ceiling) [clamped by `hard_ceiling` if set]
@@ -318,10 +331,26 @@ pub fn derive_limit(
     );
 
     let input = context.saturating_sub(output);
+    // The ceiling a request may name, distinct from the reserve above
+    // (#278). Bounded by the model's own position limit rather than by
+    // the live context, because the advertisement is read once and
+    // cached by clients: a ceiling that shrank with fleet VRAM would be
+    // stale in whichever direction hurt — either rejecting what we said
+    // we would serve, or leaving budget unclaimed. A request that will
+    // not actually fit is refused at request time, with the numbers,
+    // instead of being under-promised here.
+    let output_ceiling = round_down(
+        cfg.max_output_tokens
+            .unwrap_or(DEFAULT_OUTPUT_CEILING)
+            .min(profile.max_position_embeddings),
+        CONTEXT_GRANULARITY,
+    )
+    .max(output);
     ModelLimit {
         context,
         input: Some(input),
         output,
+        output_ceiling,
     }
 }
 
@@ -329,6 +358,82 @@ pub fn derive_limit(
 mod tests {
     use super::*;
 
+    /// The ceiling a client is told to plan against must not be the
+    /// reserve. Advertising the reserve is what told a reasoning model's
+    /// harness to cap itself below the cost of one think block (#278).
+    #[test]
+    fn the_advertised_ceiling_is_not_the_reserve() {
+        let cfg = ContextLimitConfig::default();
+        let profile = ContextProfile {
+            max_position_embeddings: 131_072,
+            kv_bytes_per_token_per_card: 65_536,
+            world_size: 2,
+        };
+        let limit = derive_limit(&profile, 0, 0.0, None, &cfg);
+        assert_eq!(limit.output, cfg.output_reserve_tokens, "reserve unchanged");
+        assert_eq!(limit.output_ceiling, DEFAULT_OUTPUT_CEILING);
+        assert!(limit.output_ceiling > limit.output);
+    }
+
+    /// A model whose own position limit is below the default ceiling
+    /// cannot generate past it, so the advertisement must not claim it.
+    #[test]
+    fn the_ceiling_never_exceeds_the_models_own_position_limit() {
+        let cfg = ContextLimitConfig::default();
+        let profile = ContextProfile {
+            max_position_embeddings: 16_384,
+            kv_bytes_per_token_per_card: 65_536,
+            world_size: 1,
+        };
+        let limit = derive_limit(&profile, 0, 0.0, None, &cfg);
+        assert_eq!(limit.output_ceiling, 16_384);
+    }
+
+    /// The operator's number wins, which is what makes the default a
+    /// starting point rather than a guess baked into the binary.
+    #[test]
+    fn a_configured_ceiling_overrides_the_default() {
+        let cfg = ContextLimitConfig {
+            max_output_tokens: Some(65_536),
+            ..ContextLimitConfig::default()
+        };
+        let profile = ContextProfile {
+            max_position_embeddings: 131_072,
+            kv_bytes_per_token_per_card: 65_536,
+            world_size: 2,
+        };
+        assert_eq!(
+            derive_limit(&profile, 0, 0.0, None, &cfg).output_ceiling,
+            65_536
+        );
+    }
+
+    /// The ceiling is deliberately *not* a function of live VRAM: it is
+    /// read once and cached by clients, so it has to mean the same thing
+    /// tomorrow. The live context still moves — that is what a request
+    /// is judged against — but the contract does not.
+    #[test]
+    fn the_ceiling_holds_still_while_the_context_moves() {
+        let cfg = ContextLimitConfig::default();
+        let profile = ContextProfile {
+            max_position_embeddings: 131_072,
+            kv_bytes_per_token_per_card: 65_536,
+            world_size: 2,
+        };
+        let roomy = derive_limit(&profile, 60_000, 5_000.0, None, &cfg);
+        // Tight enough that the VRAM term binds — the same shape as the
+        // fleet advertising 109568 one hour and 131072 the next.
+        let tight = derive_limit(&profile, 6_000, 5_000.0, None, &cfg);
+        assert!(
+            tight.context < roomy.context,
+            "the live window should shrink under VRAM pressure"
+        );
+        assert_eq!(
+            tight.output_ceiling, roomy.output_ceiling,
+            "the advertised ceiling must not move with fleet load"
+        );
+    }
+
     /// beast Qwen3.6-27B: 16 full-attn layers, 4 kv heads, head_dim 256,
     /// f16 (2 B), TP=2 → 64 KiB/token total, 32 KiB/token/card.
     fn beast_profile() -> ContextProfile {

crates/neuron/src/harness/engine.rs

@@ -297,14 +297,23 @@ async fn op_drop_snap(cfg: &EngineConfig, session: &mut ActiveSession, id: u64)
 
 /// Prefill one joining request at B=1 through the backend's existing
 /// chunked-prefill + prefix-cache paths. Returns the prefill logits.
+/// Prefill, returning the logits **and** how many leading tokens the
+/// prefix cache let us skip (#269).
+///
+/// The reuse count is computed here and was previously only logged, so
+/// no caller could report it — every client saw a 0% cache hit rate on
+/// a cache that was working. Returning it is what lets `usage` carry
+/// `cached_tokens`.
 async fn op_prefill(
     cfg: &EngineConfig,
     session: &mut ActiveSession,
     prompt_tokens: &[u32],
-) -> Result<Vec<f32>> {
+) -> Result<(Vec<f32>, usize)> {
     let prompt_len = prompt_tokens.len();
     let prefill_start = std::time::Instant::now();
-    let logits = match (&cfg.backend, session) {
+    // Each arm yields its logits plus how many leading tokens the
+    // prefix cache served, so neither can be forgotten.
+    let (logits, cached_tokens) = match (&cfg.backend, session) {
         (
             BackendConfig::Single {
                 worker,
@@ -340,7 +349,7 @@ async fn op_prefill(
                 None => chunked_prefill_via_worker(worker, *handle, prompt_tokens, reused).await?,
             };
             prefill_rate.record(prompt_len, prefill_start.elapsed());
-            logits
+            (logits, reused)
         }
         #[cfg(feature = "cuda")]
         (BackendConfig::Tp { .. }, ActiveSession::Tp { tp, pool }) => {
@@ -384,12 +393,12 @@ async fn op_prefill(
                 }
             };
             tp.prefill_rate.record(prompt_len, prefill_start.elapsed());
-            logits
+            (logits, reused)
         }
         #[cfg(feature = "cuda")]
         _ => anyhow::bail!("engine backend/session mismatch"),
     };
-    Ok(logits)
+    Ok((logits, cached_tokens))
 }
 
 /// One queued request. Admission has already been passed — the permit
@@ -397,9 +406,14 @@ async fn op_prefill(
 pub struct EngineRequest {
     pub prompt_tokens: Vec<u32>,
     pub max_new: usize,
-    pub temperature: f64,
-    pub top_p: Option<f64>,
-    pub seed: u64,
+    /// Sampling resolved once from request → model `generation_config`
+    /// → fallback (#272), rather than loose scalars each caller could
+    /// default differently.
+    pub sampling: crate::harness::sampling::SamplingParams,
+    /// Reasoning-token budget for this request (#223). `None` is
+    /// unbounded — the default, and the behaviour before it existed.
+    /// Per-request because slots in one batch can differ.
+    pub reasoning_budget: Option<usize>,
     pub eos_id: Option<u32>,
     pub tool_schemas: ToolSchemas,
     pub tx: mpsc::Sender<InferenceEvent>,
@@ -439,6 +453,8 @@ enum RouterMsg {
         reason: FinishReason,
         prompt_tokens: u32,
         completion_tokens: u32,
+        /// Leading prompt tokens the prefix cache let us skip (#269).
+        cached_tokens: u32,
         timing: FinishTiming,
     },
 }
@@ -455,6 +471,12 @@ struct Slot {
     max_new: usize,
     eos_id: Option<u32>,
     lp: LogitsProcessor,
+    /// The slot's resolved sampling (#272). Held per-slot because the
+    /// repetition penalty and its window are per-request knobs, and
+    /// slots in one batch can carry different ones.
+    sampling: crate::harness::sampling::SamplingParams,
+    /// Bounds this slot's think block (#223); `None` = unbounded.
+    governor: Option<crate::harness::reasoning_budget::ReasoningGovernor>,
     router: mpsc::Sender<RouterMsg>,
     /// Set by the router when the consumer hangs up; the engine stops
     /// feeding the slot and compacts it out.
@@ -462,6 +484,9 @@ struct Slot {
     finished: Option<FinishReason>,
     prefill_ms: u32,
     prefill_tokens: u32,
+    /// Prompt tokens served from the prefix cache (#269), reported to
+    /// the caller as `cached_tokens` so a working cache is visible.
+    cached_tokens: u32,
     decode_start: std::time::Instant,
     _admit: AdmissionPermit,
 }
@@ -559,7 +584,13 @@ async fn run_engine(cfg: EngineConfig, mut rx: mpsc::Receiver<EngineRequest>) {
                     break;
                 }
             };
-            let nt = match sample_with_penalty(&logits, &slot.generated, &mut slot.lp) {
+            let nt = match sample_with_penalty(
+                &logits,
+                &slot.generated,
+                &mut slot.lp,
+                &slot.sampling,
+                slot.governor.as_mut(),
+            ) {
                 Ok(t) => t,
                 Err(e) => {
                     let health = logits_health_slice(&logits_vec);
@@ -627,6 +658,7 @@ async fn finish_slot(slot: &mut Slot, reason: FinishReason, rate: &PrefillRateEm
             reason,
             prompt_tokens: slot.prompt_len as u32,
             completion_tokens: slot.generated.len() as u32,
+            cached_tokens: slot.cached_tokens,
             timing: FinishTiming {
                 prefill_ms: slot.prefill_ms,
                 decode_ms: slot.decode_start.elapsed().as_millis() as u32,
@@ -759,14 +791,11 @@ async fn prefill_join(
     session: &mut ActiveSession,
     req: EngineRequest,
 ) -> Result<Option<(Slot, u64)>> {
-    use candle_transformers::generation::Sampling;
-
     let EngineRequest {
         prompt_tokens,
         max_new,
-        temperature,
-        top_p,
-        seed,
+        sampling,
+        reasoning_budget,
         eos_id,
         tool_schemas,
         tx,
@@ -774,37 +803,37 @@ async fn prefill_join(
         span,
     } = req;
 
-    let mut lp = {
-        let sampling = if temperature <= 0.0 {
-            Sampling::ArgMax
-        } else {
-            match top_p {
-                Some(p) => Sampling::TopP { p, temperature },
-                None => Sampling::All { temperature },
-            }
-        };
-        LogitsProcessor::from_sampling(seed, sampling)
-    };
+    let mut lp = LogitsProcessor::from_sampling(sampling.seed, sampling.to_sampling());
+    // Built before prefill's first sample so a prompt that opens the
+    // think block is governed from its very first token (#223).
+    let mut governor = crate::harness::reasoning_budget::ReasoningGovernor::new(
+        reasoning_budget,
+        cfg.reasoning_tokens
+            .as_ref()
+            .map(|p| (p.open_id, p.close_id)),
+        prompt_opens_reasoning(&prompt_tokens, cfg.reasoning_tokens.as_ref()),
+    );
 
     let prompt_len = prompt_tokens.len();
     let prefill_start = std::time::Instant::now();
-    let logits_vec = op_prefill(cfg, session, &prompt_tokens).await?;
+    let (logits_vec, cached_tokens) = op_prefill(cfg, session, &prompt_tokens).await?;
     let prefill_elapsed = prefill_start.elapsed();
 
     // First token from the prefill logits.
     let generated: Vec<u32> = Vec::new();
     let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
-    let first = match sample_with_penalty(&logits, &generated, &mut lp) {
-        Ok(t) => t,
-        Err(e) => {
-            let health = logits_health_slice(&logits_vec);
-            tracing::warn!(
-                ?health,
-                "batch engine: prefill sample failed; logits unhealthy"
-            );
-            return Err(e);
-        }
-    };
+    let first =
+        match sample_with_penalty(&logits, &generated, &mut lp, &sampling, governor.as_mut()) {
+            Ok(t) => t,
+            Err(e) => {
+                let health = logits_health_slice(&logits_vec);
+                tracing::warn!(
+                    ?health,
+                    "batch engine: prefill sample failed; logits unhealthy"
+                );
+                return Err(e);
+            }
+        };
 
     // Router task for this slot.
     let hangup = Arc::new(AtomicBool::new(false));
@@ -827,11 +856,14 @@ async fn prefill_join(
     let mut slot = Slot {
         prefix_len: prompt_len,
         prompt_len,
+        cached_tokens: cached_tokens as u32,
         generated,
         next_token: first,
         max_new,
         eos_id,
         lp,
+        sampling,
+        governor,
         router: router_tx,
         hangup,
         finished: None,
@@ -890,6 +922,9 @@ async fn run_router(
     let mut tool_call_idx: usize = 0;
     let mut emitted_tool_call = false;
     let mut consumer_alive = true;
+    // Whether a Finish was seen, so the fall-through below can tell a
+    // clean end from a dropped slot.
+    let mut finished = false;
 
     while let Some(msg) = rx.recv().await {
         match msg {
@@ -978,17 +1013,43 @@ async fn run_router(
                 mut reason,
                 prompt_tokens,
                 completion_tokens,
+                cached_tokens,
                 timing,
             } => {
                 if emitted_tool_call && reason == FinishReason::Stop {
                     reason = FinishReason::ToolCalls;
                 }
+                // Close the journal trail for this turn (#268). The
+                // non-streaming path has always logged this; streamed
+                // turns logged nothing at all, so whether one finished,
+                // hit its output cap, or was abandoned mid-flight was
+                // unanswerable afterwards — and an absence of lines was
+                // repeatedly misread as an absence of work.
+                //
+                // `run_router` is instrumented with the request's span,
+                // so req_id and model come along without being passed.
+                tracing::info!(
+                    prompt_tokens,
+                    completion_tokens,
+                    reasoning_tokens = reasoning_token_count,
+                    cached_tokens,
+                    finish_reason = ?reason,
+                    prefill_ms = timing.prefill_ms,
+                    decode_ms = timing.decode_ms,
+                    // The turn ran to completion with nobody reading it:
+                    // it held its admission permit and the pool lock the
+                    // whole time. Invisible until now.
+                    client_gone = !consumer_alive,
+                    "chat_completion (stream): done"
+                );
+                finished = true;
                 let _ = tx
                     .send(InferenceEvent::Finish {
                         reason,
                         prompt_tokens,
                         completion_tokens,
                         reasoning_tokens: reasoning_token_count,
+                        cached_tokens,
                         timing: Some(timing),
                     })
                     .await;
@@ -996,6 +1057,17 @@ async fn run_router(
             }
         }
     }
+    // Fell out of the loop without a Finish: the engine dropped this
+    // slot's channel (fatal worker error, shutdown). The client's stream
+    // ends either way, but silently — so say so rather than leaving the
+    // turn's last trace to be a prefill line (#268).
+    if !finished {
+        tracing::warn!(
+            reasoning_tokens = reasoning_token_count,
+            client_gone = !consumer_alive,
+            "chat_completion (stream): ended without a finish event — engine dropped the slot"
+        );
+    }
 }
 
 /// `tracing::Instrument` without importing the trait at every use
@@ -1013,6 +1085,92 @@ mod tests {
     use crate::config::AdmissionConfig;
     use crate::harness::admission::AdmissionController;
 
+    /// A streamed turn must close its own journal trail (#268).
+    ///
+    /// Streaming used to end silently: no token counts, no
+    /// `finish_reason`, no duration. That made a finished turn, a turn
+    /// that hit its output cap, and a turn abandoned by a disconnected
+    /// client indistinguishable after the fact — and during the #267
+    /// investigation an absence of log lines was twice mistaken for an
+    /// absence of work on a daemon that was healthy and decoding.
+    ///
+    /// Asserts the emitted line rather than the struct, because a
+    /// `Finish` event that reaches the client but leaves no server-side
+    /// record is exactly the state this fixes.
+    #[tokio::test]
+    async fn a_streamed_turn_logs_how_it_ended() {
+        use std::sync::{Arc, Mutex};
+        use tracing_subscriber::prelude::*;
+
+        #[derive(Clone, Default)]
+        struct Capture(Arc<Mutex<Vec<u8>>>);
+        impl std::io::Write for Capture {
+            fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+                self.0
+                    .lock()
+                    .expect("capture poisoned")
+                    .extend_from_slice(buf);
+                Ok(buf.len())
+            }
+            fn flush(&mut self) -> std::io::Result<()> {
+                Ok(())
+            }
+        }
+        impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for Capture {
+            type Writer = Self;
+            fn make_writer(&'a self) -> Self::Writer {
+                self.clone()
+            }
+        }
+
+        let capture = Capture::default();
+        let subscriber = tracing_subscriber::registry()
+            .with(tracing_subscriber::fmt::layer().with_writer(capture.clone()));
+        let _guard = tracing::subscriber::set_default(subscriber);
+
+        let (router_tx, router_rx) = mpsc::channel::<RouterMsg>(8);
+        let (tx, mut rx) = mpsc::channel::<InferenceEvent>(8);
+        let handle = tokio::spawn(run_router(
+            tiny_tokenizer(8),
+            None,
+            None,
+            ToolSchemas::default(),
+            false,
+            tx,
+            Arc::new(AtomicBool::new(false)),
+            router_rx,
+        ));
+
+        router_tx
+            .send(RouterMsg::Finish {
+                reason: FinishReason::Length,
+                prompt_tokens: 7413,
+                completion_tokens: 24576,
+                cached_tokens: 2068,
+                timing: FinishTiming {
+                    prefill_ms: 6000,
+                    decode_ms: 900_000,
+                    prefill_tokens: 7413,
+                },
+            })
+            .await
+            .expect("router accepts finish");
+        drop(router_tx);
+        handle.await.expect("router task joins");
+        while rx.recv().await.is_some() {}
+
+        let logged = String::from_utf8(capture.0.lock().expect("capture poisoned").clone())
+            .expect("utf8 log output");
+        assert!(
+            logged.contains("chat_completion (stream): done"),
+            "streamed turn left no completion line:\n{logged}"
+        );
+        // The counts are the point — "it ended" without them cannot
+        // distinguish a complete answer from an exhausted budget.
+        assert!(logged.contains("24576"), "no completion_tokens:\n{logged}");
+        assert!(logged.contains("Length"), "no finish_reason:\n{logged}");
+    }
+
     /// A WordLevel tokenizer whose vocab covers the whole fixture
     /// vocab (`w0`..`w511`), so every decoded token maps to a unique
     /// word and the emitted text uniquely encodes the token sequence.
@@ -1046,9 +1204,20 @@ mod tests {
             .submit(EngineRequest {
                 prompt_tokens: prompt,
                 max_new,
-                temperature: 0.0, // greedy — deterministic
-                top_p: None,
-                seed: 0,
+                // Greedy, deterministic: the fixture asserts exact
+                // token sequences.
+                sampling: crate::harness::sampling::SamplingParams::resolve(
+                    &crate::harness::sampling::RequestedSampling {
+                        temperature: Some(0.0),
+                        seed: Some(0),
+                        ..Default::default()
+                    },
+                    &crate::harness::sampling::ModelGenerationDefaults::default(),
+                    0,
+                ),
+                // Unbounded: this fixture asserts exact token sequences,
+                // so nothing may substitute a close marker (#223).
+                reasoning_budget: None,
                 eos_id: None,
                 tool_schemas: ToolSchemas::new(),
                 tx,

crates/neuron/src/harness/mod.rs

@@ -12,6 +12,8 @@ pub mod image;
 pub mod prefix_cache;
 pub mod preflight;
 pub mod preprocess;
+pub mod reasoning_budget;
+pub mod sampling;
 pub mod speculative;
 pub mod tp;
 

crates/neuron/src/harness/preflight.rs

@@ -401,6 +401,7 @@ mod tests {
             quant: quant.map(String::from),
             tensor_parallel: tp,
             devices: None,
+            sampling: None,
         }
     }
 

crates/neuron/src/harness/reasoning_budget.rs

@@ -0,0 +1,279 @@
+//! Bounding how long a model may think before it must answer (#223).
+//!
+//! A reasoning model with no stopping criterion will fill whatever
+//! output budget it is given. Measured on Qwen3.8-27B: given 16,384
+//! tokens it used 16,384; given 32,768 it used 32,768; both times it was
+//! still opening new topics when the cap bit, and both times the caller
+//! received no answer at all.
+//!
+//! The fix is not to stop the model thinking — the run that did produce
+//! files designed for 27,157 tokens first, and that design was good. It
+//! is to bound the thinking so the rest of the output budget survives
+//! for the answer.
+//!
+//! Two halves live here:
+//!
+//! - [`ReasoningBudget`] — how many reasoning tokens this request may
+//!   spend, resolved from whichever control the caller's wire format
+//!   gave them.
+//! - [`ReasoningGovernor`] — the per-request state that enforces it, by
+//!   substituting the model's own `</think>` token when the budget runs
+//!   out so generation transitions to answering instead of being cut off.
+
+use serde::{Deserialize, Serialize};
+
+/// Per-model mapping from an OpenAI effort level to a token budget.
+///
+/// OpenAI-shaped clients cannot send a token count — pi-ai's vocabulary
+/// is the ladder `off | minimal | low | medium | high | xhigh | max` and
+/// nothing else — so the numbers have to live on the server. Operator
+/// configurable, because the right values depend on the model and the
+/// hardware; a derived default would be a guess baked into the binary.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ReasoningBudgetLadder {
+    pub minimal: usize,
+    pub low: usize,
+    pub medium: usize,
+    pub high: usize,
+    /// `xhigh` and `max` are only offered by a client when the model
+    /// declares them; absent here means "not offered", and a request
+    /// naming one falls back to `high` rather than being rejected.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub xhigh: Option<usize>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub max: Option<usize>,
+}
+
+impl Default for ReasoningBudgetLadder {
+    /// Rungs sized against observed behaviour rather than round numbers:
+    /// a long agentic design turn on this class of model runs to ~27k
+    /// tokens, so `high` sits above that (bounded but rarely binding),
+    /// `medium` at roughly half (enough to plan a whole implementation),
+    /// and `low` at the scale of a focused single-file change.
+    fn default() -> Self {
+        Self {
+            minimal: 1_024,
+            low: 4_096,
+            medium: 12_288,
+            high: 32_768,
+            xhigh: None,
+            max: None,
+        }
+    }
+}
+
+impl ReasoningBudgetLadder {
+    /// Resolve an effort level to a token budget.
+    ///
+    /// `none`/`off` are handled before this — they disable thinking
+    /// outright via the chat template — so they have no budget here.
+    /// An unrecognised level resolves to `None` (unbounded) rather than
+    /// to a guess: silently substituting a budget nobody asked for is
+    /// the failure mode this whole area has been full of.
+    pub fn for_effort(&self, effort: &str) -> Option<usize> {
+        match effort.trim().to_ascii_lowercase().as_str() {
+            "minimal" => Some(self.minimal),
+            "low" => Some(self.low),
+            "medium" => Some(self.medium),
+            "high" => Some(self.high),
+            "xhigh" => Some(self.xhigh.unwrap_or(self.high)),
+            "max" => Some(self.max.or(self.xhigh).unwrap_or(self.high)),
+            _ => None,
+        }
+    }
+
+    /// The mapping as a client should see it on `/v1/models` — the rungs
+    /// this deployment actually offers, so a caller picking `low` knows
+    /// what it bought instead of guessing (#274).
+    pub fn advertised(&self) -> Vec<(String, usize)> {
+        let mut out = vec![
+            ("minimal".to_string(), self.minimal),
+            ("low".to_string(), self.low),
+            ("medium".to_string(), self.medium),
+            ("high".to_string(), self.high),
+        ];
+        if let Some(v) = self.xhigh {
+            out.push(("xhigh".to_string(), v));
+        }
+        if let Some(v) = self.max {
+            out.push(("max".to_string(), v));
+        }
+        out
+    }
+}
+
+/// Enforces a resolved budget over one request's token stream.
+///
+/// Fed every sampled token; returns the token that should actually be
+/// emitted. While the model is inside a think block and over budget, it
+/// returns the close marker instead, exactly once — after which the
+/// model is answering and the governor is inert.
+#[derive(Debug, Clone)]
+pub struct ReasoningGovernor {
+    limit: usize,
+    open_id: u32,
+    close_id: u32,
+    inside: bool,
+    used: usize,
+    forced: bool,
+}
+
+impl ReasoningGovernor {
+    /// `None` when the request named no budget, or the model has no
+    /// reasoning markers to enforce one with — in both cases generation
+    /// proceeds exactly as it did before this existed.
+    pub fn new(
+        limit: Option<usize>,
+        markers: Option<(u32, u32)>,
+        prompt_opened_reasoning: bool,
+    ) -> Option<Self> {
+        let limit = limit?;
+        let (open_id, close_id) = markers?;
+        Some(Self {
+            limit,
+            open_id,
+            close_id,
+            inside: prompt_opened_reasoning,
+            used: 0,
+            forced: false,
+        })
+    }
+
+    /// Whether the forced close has already been spent.
+    pub fn forced(&self) -> bool {
+        self.forced
+    }
+
+    /// Reasoning tokens counted so far.
+    pub fn used(&self) -> usize {
+        self.used
+    }
+
+    /// Govern one sampled token, returning the token to emit.
+    pub fn govern(&mut self, sampled: u32) -> u32 {
+        if sampled == self.open_id {
+            self.inside = true;
+            return sampled;
+        }
+        if sampled == self.close_id {
+            self.inside = false;
+            return sampled;
+        }
+        if !self.inside {
+            return sampled;
+        }
+        if self.used >= self.limit {
+            // Budget spent: hand the model its own close marker so it
+            // transitions to answering. Cutting the stream here instead
+            // would reproduce the very failure this exists to prevent.
+            self.inside = false;
+            self.forced = true;
+            return self.close_id;
+        }
+        self.used += 1;
+        sampled
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const OPEN: u32 = 100;
+    const CLOSE: u32 = 101;
+
+    fn governor(limit: usize, opened: bool) -> ReasoningGovernor {
+        ReasoningGovernor::new(Some(limit), Some((OPEN, CLOSE)), opened).expect("governor")
+    }
+
+    #[test]
+    fn a_request_with_no_budget_is_ungoverned() {
+        assert!(ReasoningGovernor::new(None, Some((OPEN, CLOSE)), true).is_none());
+    }
+
+    /// A model with no detected markers cannot be governed — better
+    /// unbounded than truncated at an arbitrary token.
+    #[test]
+    fn a_model_without_markers_is_ungoverned() {
+        assert!(ReasoningGovernor::new(Some(10), None, true).is_none());
+    }
+
+    /// The case that motivated this: the chat template left the think
+    /// block open, so the model is reasoning from the first token.
+    #[test]
+    fn a_prompt_opened_block_is_bounded_and_force_closed() {
+        let mut g = governor(3, true);
+        assert_eq!(g.govern(7), 7);
+        assert_eq!(g.govern(8), 8);
+        assert_eq!(g.govern(9), 9);
+        assert_eq!(g.govern(10), CLOSE, "the fourth token must close the block");
+        assert!(g.forced());
+        assert_eq!(g.used(), 3);
+        // Answering now — the governor is inert.
+        assert_eq!(g.govern(11), 11);
+        assert_eq!(g.govern(12), 12);
+    }
+
+    /// Content before the block opens is never counted against the
+    /// reasoning budget.
+    #[test]
+    fn only_tokens_inside_the_block_are_counted() {
+        let mut g = governor(2, false);
+        assert_eq!(g.govern(5), 5);
+        assert_eq!(g.govern(6), 6);
+        assert_eq!(g.used(), 0);
+        assert_eq!(g.govern(OPEN), OPEN);
+        assert_eq!(g.govern(7), 7);
+        assert_eq!(g.govern(8), 8);
+        assert_eq!(g.govern(9), CLOSE);
+        assert_eq!(g.used(), 2);
+    }
+
+    /// A model that finishes thinking on its own is untouched — the
+    /// budget is a ceiling, not a target.
+    #[test]
+    fn a_model_that_stops_on_its_own_is_not_forced() {
+        let mut g = governor(100, true);
+        for t in 20..30 {
+            assert_eq!(g.govern(t), t);
+        }
+        assert_eq!(g.govern(CLOSE), CLOSE);
+        assert!(!g.forced());
+        for t in 40..50 {
+            assert_eq!(g.govern(t), t);
+        }
+        assert!(!g.forced(), "content after the block must not be governed");
+    }
+
+    #[test]
+    fn the_ladder_resolves_the_levels_clients_can_send() {
+        let l = ReasoningBudgetLadder::default();
+        assert_eq!(l.for_effort("minimal"), Some(1_024));
+        assert_eq!(l.for_effort("LOW"), Some(4_096));
+        assert_eq!(l.for_effort(" medium "), Some(12_288));
+        assert_eq!(l.for_effort("high"), Some(32_768));
+        // Not offered by default → falls back to high rather than
+        // rejecting a level the client was told it could send.
+        assert_eq!(l.for_effort("xhigh"), Some(32_768));
+        assert_eq!(l.for_effort("max"), Some(32_768));
+        // Unknown → unbounded, never a substituted guess.
+        assert_eq!(l.for_effort("enthusiastic"), None);
+        assert_eq!(l.for_effort("none"), None);
+    }
+
+    #[test]
+    fn the_advertised_ladder_lists_only_what_is_offered() {
+        let l = ReasoningBudgetLadder::default();
+        let names: Vec<_> = l.advertised().into_iter().map(|(k, _)| k).collect();
+        assert_eq!(names, ["minimal", "low", "medium", "high"]);
+        let l2 = ReasoningBudgetLadder {
+            xhigh: Some(49_152),
+            ..ReasoningBudgetLadder::default()
+        };
+        assert!(
+            l2.advertised()
+                .iter()
+                .any(|(k, v)| k == "xhigh" && *v == 49_152)
+        );
+    }
+}

crates/neuron/src/harness/sampling.rs

@@ -0,0 +1,538 @@
+//! Sampling parameter resolution (#272, #283).
+//!
+//! Four sources, in priority order:
+//!
+//! 1. what the **request** asked for,
+//! 2. what the **operator** set for this model (#283),
+//! 3. what the **model** published in `generation_config.json`,
+//! 4. a built-in fallback.
+//!
+//! Tier (2) exists because honouring the model's own file is right by
+//! default and still wrong sometimes: `Qwen/Qwen3.8-27B` publishes
+//! `temperature = 1.0`, which measured a 20% structural-defect rate on
+//! ~2k-token code generation against 0/60 at `<= 0.6` (#283,
+//! p = 0.0031). The operator's correction is explicit and lives in a
+//! config file — never a heuristic that inspects the model and guesses,
+//! which would be a footgun.
+//!
+//! The operator sits *below* the request on purpose. An explicit API
+//! parameter that is silently ignored is a contract break; the observed
+//! failure came from callers that named nothing, so a default-tier
+//! override is sufficient to fix it.
+//!
+//! Mechanically, tier (2) is folded into tier (3) once at load time by
+//! [`ModelGenerationDefaults::with_operator_override`], so the
+//! per-request resolve path stays a two-tier lookup and every call site
+//! gets the override for free — there is no site that can forget it.
+//!
+//! Before this, (2) did not exist and (3) was a hardcoded
+//! `temperature = 0.7` with `top_p = None` — and `None` selects
+//! `Sampling::All`, meaning no truncation at all. So a caller that
+//! omitted sampling fields (the common case: the DeepSeek Harness sends
+//! only `temperature`) got untruncated sampling across the full ~250k
+//! vocabulary on a model whose authors published `top_k = 20,
+//! top_p = 0.95, temperature = 1.0`.
+//!
+//! Picking "no truncation" for an absent field is not a neutral default,
+//! it is the widest possible one. The neutral default is what the model
+//! shipped with.
+//!
+//! Resolved **once** per request and threaded whole. The eight
+//! `LogitsProcessor` construction sites this replaces could each have
+//! disagreed about a default — the same shape of bug as #252, where two
+//! config files described one model and drifted.
+
+use serde::{Deserialize, Serialize};
+use std::path::Path;
+
+/// Fallback temperature when neither the request nor the model says.
+/// Retained from the previous hardcoded value so a model without a
+/// `generation_config.json` behaves exactly as it did before.
+const FALLBACK_TEMPERATURE: f64 = 0.7;
+
+/// Repetition penalty applied to recently-generated tokens before
+/// sampling. 1.0 disables it; >1.0 makes recently-emitted tokens less
+/// likely. mistral.rs and llama.cpp default to 1.1, which is enough to
+/// stop small quantized models from degenerating into "Wait, no, no..."
+/// loops without distorting normal output.
+pub const DEFAULT_REPEAT_PENALTY: f32 = 1.1;
+
+/// Number of recently-generated tokens fed into the repetition penalty.
+/// Matches the candle quantized-qwen3 example default.
+///
+/// Defensible for chat and poorly suited to a 24k-token reasoning block,
+/// where a derivation repeated 500 tokens later is well outside the
+/// window — which is why it is now a per-request knob rather than a
+/// constant.
+pub const DEFAULT_REPEAT_LAST_N: usize = 64;
+
+/// What a model published in its `generation_config.json`.
+///
+/// Every field optional: the file is a HuggingFace convention, not a
+/// schema, and models populate different subsets of it.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct ModelGenerationDefaults {
+    pub temperature: Option<f64>,
+    pub top_p: Option<f64>,
+    pub top_k: Option<usize>,
+}
+
+impl ModelGenerationDefaults {
+    /// Read `generation_config.json` from a model snapshot directory.
+    ///
+    /// Absent or unparseable is not an error — we fall back exactly as
+    /// before — but it is logged, because silently guessing the model's
+    /// sampling is how the untruncated default went unnoticed.
+    pub fn load_from_dir(dir: &Path) -> Self {
+        let path = dir.join("generation_config.json");
+        let text = match std::fs::read_to_string(&path) {
+            Ok(t) => t,
+            Err(e) => {
+                tracing::debug!(
+                    path = %path.display(),
+                    error = %e,
+                    "sampling: no generation_config.json; using built-in fallbacks"
+                );
+                return Self::default();
+            }
+        };
+        match serde_json::from_str::<Self>(&text) {
+            Ok(d) => {
+                tracing::info!(
+                    path = %path.display(),
+                    temperature = ?d.temperature,
+                    top_p = ?d.top_p,
+                    top_k = ?d.top_k,
+                    "sampling: model generation defaults loaded"
+                );
+                d
+            }
+            Err(e) => {
+                tracing::warn!(
+                    path = %path.display(),
+                    error = %e,
+                    "sampling: generation_config.json did not parse; using built-in fallbacks"
+                );
+                Self::default()
+            }
+        }
+    }
+
+    /// Overlay the operator's per-model override (#283) onto what the
+    /// model published, field by field.
+    ///
+    /// Applied **once at load**, not per request, so the override cannot
+    /// be forgotten at one of the resolve sites — the same reasoning
+    /// that made #272 resolve sampling once rather than at eight
+    /// `LogitsProcessor` construction sites.
+    ///
+    /// Each field overlays independently: setting only `temperature`
+    /// leaves the model's `top_p`/`top_k` in force, because an operator
+    /// correcting one number should not silently blank the other two
+    /// back to "no truncation".
+    ///
+    /// Logged at `info` whenever it changes something — an operator
+    /// override that is in force but invisible is how a fleet ends up
+    /// with two configs that disagree and nobody noticing (#252).
+    pub fn with_operator_override(
+        mut self,
+        model_id: &str,
+        operator: Option<&cortex_core::harness::SamplingOverride>,
+    ) -> Self {
+        let Some(op) = operator else {
+            return self;
+        };
+        if op.is_empty() {
+            return self;
+        }
+        let before = self.clone();
+        if let Some(t) = op.temperature {
+            self.temperature = Some(t);
+        }
+        if let Some(p) = op.top_p {
+            self.top_p = Some(p);
+        }
+        if let Some(k) = op.top_k {
+            self.top_k = Some(k);
+        }
+        tracing::info!(
+            model = %model_id,
+            model_temperature = ?before.temperature,
+            model_top_p = ?before.top_p,
+            model_top_k = ?before.top_k,
+            temperature = ?self.temperature,
+            top_p = ?self.top_p,
+            top_k = ?self.top_k,
+            "sampling: operator override applied over the model's published defaults (#283)"
+        );
+        self
+    }
+}
+
+/// What the caller asked for. All optional — absent means "no opinion",
+/// which defers to the model's published value.
+#[derive(Debug, Clone, Default)]
+pub struct RequestedSampling {
+    pub temperature: Option<f64>,
+    pub top_p: Option<f64>,
+    pub top_k: Option<usize>,
+    pub seed: Option<u64>,
+    pub repeat_penalty: Option<f32>,
+    pub repeat_last_n: Option<usize>,
+}
+
+/// Fully-resolved sampling for one request.
+#[derive(Debug, Clone, PartialEq)]
+pub struct SamplingParams {
+    pub temperature: f64,
+    pub top_p: Option<f64>,
+    pub top_k: Option<usize>,
+    pub seed: u64,
+    pub repeat_penalty: f32,
+    pub repeat_last_n: usize,
+}
+
+impl SamplingParams {
+    /// Resolve request over model defaults over fallback.
+    ///
+    /// `random_seed` is the per-request seed used when the caller did
+    /// not pin one; it is passed in rather than generated here so this
+    /// stays a pure function and the "same seed reproduces" property is
+    /// testable without a clock.
+    pub fn resolve(
+        requested: &RequestedSampling,
+        model: &ModelGenerationDefaults,
+        random_seed: u64,
+    ) -> Self {
+        Self {
+            temperature: requested
+                .temperature
+                .or(model.temperature)
+                .unwrap_or(FALLBACK_TEMPERATURE),
+            // No `unwrap_or` here on purpose: `None` still means "do not
+            // truncate", but it is now only reachable when neither the
+            // caller nor the model expressed a preference, rather than
+            // whenever the caller stayed quiet.
+            top_p: requested.top_p.or(model.top_p),
+            top_k: requested.top_k.or(model.top_k),
+            seed: requested.seed.unwrap_or(random_seed),
+            repeat_penalty: requested.repeat_penalty.unwrap_or(DEFAULT_REPEAT_PENALTY),
+            repeat_last_n: requested.repeat_last_n.unwrap_or(DEFAULT_REPEAT_LAST_N),
+        }
+    }
+
+    /// The candle sampling strategy these parameters select.
+    ///
+    /// `temperature <= 0` is greedy regardless of the rest — a caller
+    /// asking for determinism that way should not have it silently
+    /// softened by a `top_k` the model happened to publish.
+    pub fn to_sampling(&self) -> candle_transformers::generation::Sampling {
+        use candle_transformers::generation::Sampling;
+        if self.temperature <= 0.0 {
+            return Sampling::ArgMax;
+        }
+        let temperature = self.temperature;
+        match (self.top_k, self.top_p) {
+            (Some(k), Some(p)) => Sampling::TopKThenTopP { k, p, temperature },
+            (Some(k), None) => Sampling::TopK { k, temperature },
+            (None, Some(p)) => Sampling::TopP { p, temperature },
+            (None, None) => Sampling::All { temperature },
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use candle_transformers::generation::Sampling;
+    use cortex_core::harness::SamplingOverride;
+
+    fn qwen3_defaults() -> ModelGenerationDefaults {
+        ModelGenerationDefaults {
+            temperature: Some(1.0),
+            top_p: Some(0.95),
+            top_k: Some(20),
+        }
+    }
+
+    /// The case that motivated #271: a caller that names no sampling
+    /// parameters must get the model's published ones, not untruncated
+    /// sampling at a hardcoded 0.7.
+    #[test]
+    fn a_silent_request_inherits_the_models_published_sampling() {
+        let p = SamplingParams::resolve(&RequestedSampling::default(), &qwen3_defaults(), 42);
+        assert_eq!(p.temperature, 1.0);
+        assert_eq!(p.top_p, Some(0.95));
+        assert_eq!(p.top_k, Some(20));
+        assert_eq!(
+            p.to_sampling(),
+            Sampling::TopKThenTopP {
+                k: 20,
+                p: 0.95,
+                temperature: 1.0
+            },
+            "the combination Qwen3 asks for must be expressible"
+        );
+    }
+
+    /// The caller still wins. A harness that deliberately sets a low
+    /// temperature must not have it overridden by the model's file.
+    #[test]
+    fn the_request_overrides_the_model() {
+        let requested = RequestedSampling {
+            temperature: Some(0.2),
+            top_k: Some(5),
+            ..Default::default()
+        };
+        let p = SamplingParams::resolve(&requested, &qwen3_defaults(), 42);
+        assert_eq!(p.temperature, 0.2);
+        assert_eq!(p.top_k, Some(5));
+        // Untouched fields still fall through to the model.
+        assert_eq!(p.top_p, Some(0.95));
+    }
+
+    // ── Operator override (#283) ──────────────────────────────
+
+    fn op(temperature: Option<f64>, top_p: Option<f64>, top_k: Option<usize>) -> SamplingOverride {
+        SamplingOverride {
+            temperature,
+            top_p,
+            top_k,
+        }
+    }
+
+    fn with_op(o: SamplingOverride) -> ModelGenerationDefaults {
+        qwen3_defaults().with_operator_override("Qwen/Qwen3.8-27B", Some(&o))
+    }
+
+    /// The #283 case: the model publishes `temperature = 1.0`, which
+    /// measured a 20% structural-defect rate on code generation. The
+    /// operator sets 0.6 and a silent caller must get 0.6.
+    #[test]
+    fn the_operator_override_replaces_what_the_model_published() {
+        let p = SamplingParams::resolve(
+            &RequestedSampling::default(),
+            &with_op(op(Some(0.6), None, None)),
+            42,
+        );
+        assert_eq!(p.temperature, 0.6);
+    }
+
+    /// Precedence is request > operator > model. An explicit API
+    /// parameter that is silently ignored is a contract break, so a
+    /// caller naming a temperature still wins over the operator.
+    #[test]
+    fn an_explicit_request_still_beats_the_operator() {
+        let requested = RequestedSampling {
+            temperature: Some(0.9),
+            ..Default::default()
+        };
+        let p = SamplingParams::resolve(&requested, &with_op(op(Some(0.6), None, None)), 42);
+        assert_eq!(p.temperature, 0.9);
+    }
+
+    /// Overriding one field must not blank the others back to "no
+    /// truncation" — that was the #271 failure mode, reintroduced by a
+    /// wholesale replace instead of a field-by-field overlay.
+    #[test]
+    fn overriding_temperature_alone_leaves_top_p_and_top_k_intact() {
+        let d = with_op(op(Some(0.6), None, None));
+        assert_eq!(d.temperature, Some(0.6));
+        assert_eq!(d.top_p, Some(0.95), "the model's top_p must survive");
+        assert_eq!(d.top_k, Some(20), "the model's top_k must survive");
+    }
+
+    /// Each field overlays independently.
+    #[test]
+    fn every_field_can_be_overridden_independently() {
+        let d = with_op(op(None, Some(0.8), Some(40)));
+        assert_eq!(d.temperature, Some(1.0), "untouched, so the model's");
+        assert_eq!(d.top_p, Some(0.8));
+        assert_eq!(d.top_k, Some(40));
+    }
+
+    /// No override, and an override that sets nothing, must both be
+    /// indistinguishable from the pre-#283 behaviour.
+    #[test]
+    fn absent_or_empty_override_changes_nothing() {
+        let base = qwen3_defaults();
+        let none = qwen3_defaults().with_operator_override("m", None);
+        let empty =
+            qwen3_defaults().with_operator_override("m", Some(&SamplingOverride::default()));
+        for d in [none, empty] {
+            assert_eq!(d.temperature, base.temperature);
+            assert_eq!(d.top_p, base.top_p);
+            assert_eq!(d.top_k, base.top_k);
+        }
+    }
+
+    /// An operator override on a model that ships no
+    /// `generation_config.json` must still take effect, rather than
+    /// falling through to the built-in 0.7.
+    #[test]
+    fn the_override_applies_even_without_a_generation_config() {
+        let d = ModelGenerationDefaults::default()
+            .with_operator_override("m", Some(&op(Some(0.6), None, None)));
+        let p = SamplingParams::resolve(&RequestedSampling::default(), &d, 42);
+        assert_eq!(p.temperature, 0.6);
+        assert_ne!(p.temperature, FALLBACK_TEMPERATURE);
+    }
+
+    /// `temperature = 0` from an operator must select greedy decoding,
+    /// not be treated as "unset" by an `unwrap_or`-shaped bug.
+    #[test]
+    fn an_operator_temperature_of_zero_is_greedy_not_unset() {
+        let p = SamplingParams::resolve(
+            &RequestedSampling::default(),
+            &with_op(op(Some(0.0), None, None)),
+            42,
+        );
+        assert_eq!(p.temperature, 0.0);
+        assert_eq!(p.to_sampling(), Sampling::ArgMax);
+    }
+
+    /// A model with no `generation_config.json` must behave exactly as
+    /// it did before this landed, so nothing regresses on the models
+    /// that do not ship one.
+    #[test]
+    fn without_model_defaults_the_old_behaviour_is_preserved() {
+        let p = SamplingParams::resolve(
+            &RequestedSampling::default(),
+            &ModelGenerationDefaults::default(),
+            42,
+        );
+        assert_eq!(p.temperature, FALLBACK_TEMPERATURE);
+        assert_eq!(p.top_p, None);
+        assert_eq!(p.top_k, None);
+        assert_eq!(p.to_sampling(), Sampling::All { temperature: 0.7 });
+    }
+
+    /// A pinned seed is honoured; an absent one takes the per-request
+    /// random value. Reproducibility is the whole point — the text paths
+    /// previously called `unix_subsec_nanos()` unconditionally, so
+    /// `seed` was accepted and discarded.
+    #[test]
+    fn seed_is_pinned_when_given_and_random_otherwise() {
+        let pinned = SamplingParams::resolve(
+            &RequestedSampling {
+                seed: Some(7),
+                ..Default::default()
+            },
+            &ModelGenerationDefaults::default(),
+            42,
+        );
+        assert_eq!(pinned.seed, 7);
+
+        let unpinned = SamplingParams::resolve(
+            &RequestedSampling::default(),
+            &ModelGenerationDefaults::default(),
+            42,
+        );
+        assert_eq!(unpinned.seed, 42);
+    }
+
+    /// Greedy beats every truncation setting: a caller asking for
+    /// `temperature = 0` wants determinism, not a softened version of it.
+    #[test]
+    fn zero_temperature_is_greedy_even_with_model_top_k() {
+        let requested = RequestedSampling {
+            temperature: Some(0.0),
+            ..Default::default()
+        };
+        let p = SamplingParams::resolve(&requested, &qwen3_defaults(), 42);
+        assert_eq!(p.to_sampling(), Sampling::ArgMax);
+    }
+
+    /// Each truncation knob alone selects the matching strategy.
+    #[test]
+    fn single_knobs_select_their_own_strategy() {
+        let only_k = SamplingParams::resolve(
+            &RequestedSampling {
+                top_k: Some(20),
+                ..Default::default()
+            },
+            &ModelGenerationDefaults::default(),
+            1,
+        );
+        assert_eq!(
+            only_k.to_sampling(),
+            Sampling::TopK {
+                k: 20,
+                temperature: 0.7
+            }
+        );
+
+        let only_p = SamplingParams::resolve(
+            &RequestedSampling {
+                top_p: Some(0.9),
+                ..Default::default()
+            },
+            &ModelGenerationDefaults::default(),
+            1,
+        );
+        assert_eq!(
+            only_p.to_sampling(),
+            Sampling::TopP {
+                p: 0.9,
+                temperature: 0.7
+            }
+        );
+    }
+
+    /// Repetition penalty and its window are now per-request, defaulting
+    /// to the values that were previously constants.
+    #[test]
+    fn repetition_knobs_default_to_the_former_constants() {
+        let d = SamplingParams::resolve(
+            &RequestedSampling::default(),
+            &ModelGenerationDefaults::default(),
+            1,
+        );
+        assert_eq!(d.repeat_penalty, DEFAULT_REPEAT_PENALTY);
+        assert_eq!(d.repeat_last_n, DEFAULT_REPEAT_LAST_N);
+
+        let tuned = SamplingParams::resolve(
+            &RequestedSampling {
+                repeat_penalty: Some(1.05),
+                repeat_last_n: Some(512),
+                ..Default::default()
+            },
+            &ModelGenerationDefaults::default(),
+            1,
+        );
+        assert_eq!(tuned.repeat_penalty, 1.05);
+        assert_eq!(tuned.repeat_last_n, 512);
+    }
+
+    #[test]
+    fn absent_generation_config_is_not_an_error() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let d = ModelGenerationDefaults::load_from_dir(dir.path());
+        assert!(d.temperature.is_none() && d.top_p.is_none() && d.top_k.is_none());
+    }
+
+    #[test]
+    fn a_real_generation_config_parses() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        // Verbatim from Qwen/Qwen3.8-27B, extra fields included — the
+        // reader must tolerate the ones it does not model.
+        std::fs::write(
+            dir.path().join("generation_config.json"),
+            r#"{"bos_token_id":248044,"do_sample":true,"eos_token_id":[248046,248044],
+                "pad_token_id":248044,"temperature":1.0,"top_k":20,"top_p":0.95}"#,
+        )
+        .expect("write");
+        let d = ModelGenerationDefaults::load_from_dir(dir.path());
+        assert_eq!(d.temperature, Some(1.0));
+        assert_eq!(d.top_p, Some(0.95));
+        assert_eq!(d.top_k, Some(20));
+    }
+
+    #[test]
+    fn malformed_generation_config_falls_back_rather_than_failing() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        std::fs::write(dir.path().join("generation_config.json"), "{not json").expect("write");
+        let d = ModelGenerationDefaults::load_from_dir(dir.path());
+        assert!(d.temperature.is_none());
+    }
+}

crates/neuron/src/harness/tp/tp_qwen3_5.rs

@@ -2445,7 +2445,7 @@ mod tests {
     }
 
     #[cfg(not(feature = "cuda"))]
-    fn vb_over(path: &std::path::Path) -> ShardedVarBuilder {
+    fn vb_over(path: &std::path::Path) -> ShardedVarBuilder<'_> {
         // SAFETY: mmap of a file the test just wrote; nothing mutates it.
         unsafe {
             candle_nn::var_builder::ShardedSafeTensors::var_builder(

crates/neuron/src/wire/event.rs

@@ -84,6 +84,13 @@ pub enum InferenceEvent {
         /// `output_tokens_details.reasoning_tokens` (responses).
         /// Zero for non-reasoning models.
         reasoning_tokens: u32,
+        /// Leading prompt tokens served from the prefix cache (#269) — a
+        /// sub-count of `prompt_tokens`, never added to `total_tokens`.
+        /// Surfaced as `prompt_tokens_details.cached_tokens` (chat) /
+        /// `input_tokens_details.cached_tokens` (responses). The cache
+        /// has worked since #11; without this the saving was invisible
+        /// and every client reported a 0% hit rate.
+        cached_tokens: u32,
         /// Server-measured prefill/decode timing for the request, or
         /// `None` on paths that don't measure it (CPU fallback that
         /// doesn't instrument, tests). Streaming projectors surface

crates/neuron/src/wire/openai_chat.rs

@@ -27,7 +27,8 @@
 //! propagate without us writing any logic.
 
 use cortex_core::openai::{
-    ChatCompletionChunk, ChunkChoice, CompletionTokensDetails, HelexaTiming, Usage,
+    ChatCompletionChoice, ChatCompletionChunk, ChatCompletionResponse, ChatMessage, ChunkChoice,
+    CompletionTokensDetails, HelexaTiming, MessageContent, PromptTokensDetails, Usage,
 };
 use serde_json::json;
 use tokio::sync::mpsc;
@@ -139,11 +140,35 @@ pub fn project_chat_stream_with(
                 }
                 InferenceEvent::ReasoningDelta(text) => {
                     if !config.include_thinking {
-                        // Default path — reasoning has no slot in
-                        // chat completions, so it's dropped. Naïve
-                        // clients (Zed commit-message generator,
-                        // any vanilla OpenAI client) get clean
-                        // output.
+                        // Default path. This used to `continue` —
+                        // dropping the token and emitting nothing at
+                        // all — to keep model-internal scratchpad out
+                        // of naïve clients' UIs. That goal was right;
+                        // the method silenced the stream.
+                        //
+                        // A reasoning model can think for minutes on a
+                        // hard prompt, and for that whole span not one
+                        // SSE event went out. Clients time that out: the
+                        // DeepSeek Harness reports `pi-ai stream idle
+                        // timeout after 300000ms` and retries, so a task
+                        // that reasons for over five minutes could never
+                        // complete no matter how healthy the model was.
+                        // SSE comment keep-alives do not help — idle
+                        // timers reset on events, not comments.
+                        //
+                        // Emitting `reasoning_content` instead keeps the
+                        // original guarantee (it is not `content`, so
+                        // unaware clients still show clean output) while
+                        // the stream stays visibly alive, and hands
+                        // reasoning-aware clients the field they are
+                        // already looking for.
+                        if text.is_empty() {
+                            continue;
+                        }
+                        let chunk = reasoning_chunk(&id, created, &model_id, &text);
+                        if tx.send(chunk).await.is_err() {
+                            return;
+                        }
                         continue;
                     }
                     let Some(markers) = config.reasoning_markers.as_ref() else {
@@ -195,6 +220,7 @@ pub fn project_chat_stream_with(
                     prompt_tokens,
                     completion_tokens,
                     reasoning_tokens,
+                    cached_tokens,
                     timing,
                 } => {
                     // The finish_reason chunk, then an OpenAI-style
@@ -209,9 +235,12 @@ pub fn project_chat_stream_with(
                             &id,
                             created,
                             &model_id,
-                            prompt_tokens,
-                            completion_tokens,
-                            reasoning_tokens,
+                            UsageCounts {
+                                prompt_tokens,
+                                completion_tokens,
+                                reasoning_tokens,
+                                cached_tokens,
+                            },
                             timing,
                         ),
                     ]
@@ -246,6 +275,35 @@ fn role_chunk(id: &str, created: u64, model_id: &str) -> ChatCompletionChunk {
     }
 }
 
+/// A reasoning delta, in the field reasoning-aware clients read.
+///
+/// `reasoning_content` is not in the OpenAI spec, but it is the
+/// de-facto slot: DeepSeek's API emits it, and so do vLLM and SGLang
+/// for reasoning models. Our own consumers already expect it —
+/// `helexa-acp`'s OpenAI provider reads `choice.delta.reasoning_content`,
+/// and `helexa-bench` times a stream's first reasoning token from it.
+///
+/// Crucially it is NOT `content`, so a client that has never heard of
+/// reasoning ignores the unknown field and sees clean output — the
+/// property the old drop-it-entirely default was protecting, kept
+/// without the silence that came with it.
+fn reasoning_chunk(id: &str, created: u64, model_id: &str, text: &str) -> ChatCompletionChunk {
+    ChatCompletionChunk {
+        id: id.into(),
+        object: "chat.completion.chunk".into(),
+        created,
+        model: model_id.into(),
+        choices: vec![ChunkChoice {
+            index: 0,
+            delta: json!({ "reasoning_content": text }),
+            finish_reason: None,
+            extra: serde_json::Value::Object(Default::default()),
+        }],
+        usage: None,
+        extra: serde_json::Value::Object(Default::default()),
+    }
+}
+
 fn content_chunk(id: &str, created: u64, model_id: &str, text: &str) -> ChatCompletionChunk {
     ChatCompletionChunk {
         id: id.into(),
@@ -332,15 +390,32 @@ fn final_chunk(
 /// ignore the empty-choices chunk; clients that do (opencode, and
 /// cortex's Anthropic translator) get the token counts they need to
 /// track context.
+/// The token counts a finished turn reports. Grouped rather than passed
+/// loose: adding `cached_tokens` (#269) put the list past the point
+/// where four adjacent bare `u32`s are safe to read at a call site.
+struct UsageCounts {
+    prompt_tokens: u32,
+    completion_tokens: u32,
+    /// Sub-count of `completion_tokens` — tokens spent reasoning.
+    reasoning_tokens: u32,
+    /// Sub-count of `prompt_tokens` — tokens served from the prefix
+    /// cache. Neither sub-count is added into `total_tokens`.
+    cached_tokens: u32,
+}
+
 fn usage_chunk(
     id: &str,
     created: u64,
     model_id: &str,
-    prompt_tokens: u32,
-    completion_tokens: u32,
-    reasoning_tokens: u32,
+    counts: UsageCounts,
     timing: Option<FinishTiming>,
 ) -> ChatCompletionChunk {
+    let UsageCounts {
+        prompt_tokens,
+        completion_tokens,
+        reasoning_tokens,
+        cached_tokens,
+    } = counts;
     ChatCompletionChunk {
         id: id.into(),
         object: "chat.completion.chunk".into(),
@@ -356,7 +431,12 @@ fn usage_chunk(
             completion_tokens_details: (reasoning_tokens > 0).then_some(CompletionTokensDetails {
                 reasoning_tokens: reasoning_tokens as u64,
             }),
-            prompt_tokens_details: None,
+            // Prefix-cache reuse (#269). Omitted at zero so a client
+            // sees unchanged JSON when nothing was cached, and so
+            // "absent" stays distinguishable from "measured as none".
+            prompt_tokens_details: (cached_tokens > 0).then_some(PromptTokensDetails {
+                cached_tokens: cached_tokens as u64,
+            }),
             // helexa extension (#85): server-measured prefill/decode
             // timing for the bench harness. Omitted on paths that don't
             // measure it so standard clients see unchanged JSON.
@@ -370,10 +450,283 @@ fn usage_chunk(
     }
 }
 
+/// Drain an [`InferenceEvent`] stream into a single non-streaming
+/// [`ChatCompletionResponse`] (#285).
+///
+/// Non-streaming used to be a *separate inference path* that took the
+/// per-model `inference_lock` for the whole request. That made it
+/// invisible to the batch engine (#98), so concurrent non-streaming
+/// callers serialized completely — measured on beast at 1.00x aggregate
+/// throughput across 1/2/4/8 concurrency, against 3.98x for the same
+/// work streamed. `max_in_flight` bought those requests admission and
+/// nothing else.
+///
+/// Expressing non-streaming as "run the stream, buffer the result"
+/// collapses the two paths onto one producer, so batching, prefix-cache
+/// reuse, tool-call extraction, the reasoning governor and server-side
+/// timing all apply identically to both. It also removes a second place
+/// where the same wire object could be assembled differently — the
+/// shape of defect #252 and #272 both were.
+///
+/// Semantics deliberately preserved from the path this replaces:
+/// reasoning text is **dropped** from `content` (it is counted in
+/// `usage.completion_tokens_details.reasoning_tokens` but never
+/// rendered), and `finish_reason` becomes `"tool_calls"` whenever any
+/// tool call was parsed.
+///
+/// Two fields that the old path hardcoded to `None` now populate,
+/// because the events carry them and the struct already had the slots:
+/// `prompt_tokens_details.cached_tokens` and `helexa_timing`.
+///
+/// A stream that ends without a `Finish` is an error, not an empty
+/// reply: that is how a dropped engine slot or a fatal worker fault
+/// presents, and returning a 200 with partial text would hide it.
+pub async fn collect_chat_completion(
+    mut rx: mpsc::Receiver<InferenceEvent>,
+    id: String,
+    created: u64,
+    model_id: String,
+) -> Result<ChatCompletionResponse, String> {
+    let mut content = String::new();
+    let mut tool_calls: Vec<serde_json::Value> = Vec::new();
+    let mut finish: Option<(FinishReason, u32, u32, u32, u32, Option<FinishTiming>)> = None;
+
+    while let Some(event) = rx.recv().await {
+        match event {
+            InferenceEvent::Start => {}
+            InferenceEvent::TextDelta(t) => content.push_str(&t),
+            // Counted, never rendered — see the doc comment.
+            InferenceEvent::ReasoningDelta(_) => {}
+            InferenceEvent::ToolCall {
+                index,
+                id: call_id,
+                name,
+                arguments,
+            } => tool_calls.push(json!({
+                "index": index,
+                "id": call_id,
+                "type": "function",
+                "function": { "name": name, "arguments": arguments },
+            })),
+            InferenceEvent::Finish {
+                reason,
+                prompt_tokens,
+                completion_tokens,
+                reasoning_tokens,
+                cached_tokens,
+                timing,
+            } => {
+                finish = Some((
+                    reason,
+                    prompt_tokens,
+                    completion_tokens,
+                    reasoning_tokens,
+                    cached_tokens,
+                    timing,
+                ));
+                break;
+            }
+        }
+    }
+
+    let Some((reason, prompt_tokens, completion_tokens, reasoning_tokens, cached_tokens, timing)) =
+        finish
+    else {
+        return Err(
+            "inference stream ended without a finish event (engine slot dropped?)".to_string(),
+        );
+    };
+
+    // A parsed tool call outranks whatever the sampler stopped on, so a
+    // model that emits a call and then hits EOS still reports
+    // `tool_calls` and the client dispatches instead of rendering.
+    let finish_reason = if tool_calls.is_empty() {
+        reason.as_openai_str().to_string()
+    } else {
+        "tool_calls".to_string()
+    };
+    let message_extra = if tool_calls.is_empty() {
+        serde_json::Value::Object(Default::default())
+    } else {
+        json!({ "tool_calls": tool_calls })
+    };
+
+    Ok(ChatCompletionResponse {
+        id,
+        object: "chat.completion".into(),
+        created,
+        model: model_id,
+        choices: vec![ChatCompletionChoice {
+            index: 0,
+            message: ChatMessage {
+                role: "assistant".into(),
+                content: MessageContent::Text(content),
+                extra: message_extra,
+            },
+            finish_reason: Some(finish_reason),
+            extra: serde_json::Value::Object(Default::default()),
+        }],
+        usage: Some(Usage {
+            prompt_tokens: prompt_tokens as u64,
+            completion_tokens: completion_tokens as u64,
+            total_tokens: (prompt_tokens + completion_tokens) as u64,
+            completion_tokens_details: (reasoning_tokens > 0).then_some(CompletionTokensDetails {
+                reasoning_tokens: reasoning_tokens as u64,
+            }),
+            prompt_tokens_details: (cached_tokens > 0).then_some(PromptTokensDetails {
+                cached_tokens: cached_tokens as u64,
+            }),
+            helexa_timing: timing.map(|t| HelexaTiming {
+                prefill_ms: t.prefill_ms as u64,
+                decode_ms: t.decode_ms as u64,
+                prefill_tokens: t.prefill_tokens as u64,
+            }),
+        }),
+        extra: serde_json::Value::Object(Default::default()),
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    // ── Non-streaming collector (#285) ────────────────────────────
+
+    async fn collect_from(events: Vec<InferenceEvent>) -> Result<ChatCompletionResponse, String> {
+        let (tx, rx) = mpsc::channel(64);
+        for e in events {
+            tx.send(e).await.expect("send");
+        }
+        drop(tx);
+        collect_chat_completion(rx, "chatcmpl-test".into(), 1_700_000_000, "m".into()).await
+    }
+
+    fn finish(reason: FinishReason) -> InferenceEvent {
+        InferenceEvent::Finish {
+            reason,
+            prompt_tokens: 10,
+            completion_tokens: 4,
+            reasoning_tokens: 0,
+            cached_tokens: 0,
+            timing: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn collector_concatenates_text_deltas() {
+        let r = collect_from(vec![
+            InferenceEvent::Start,
+            InferenceEvent::TextDelta("hel".into()),
+            InferenceEvent::TextDelta("lo".into()),
+            finish(FinishReason::Stop),
+        ])
+        .await
+        .expect("collected");
+        let c = &r.choices[0];
+        assert!(matches!(&c.message.content, MessageContent::Text(t) if t == "hello"));
+        assert_eq!(c.finish_reason.as_deref(), Some("stop"));
+        assert_eq!(r.object, "chat.completion");
+    }
+
+    /// Reasoning is counted but never rendered — the behaviour of the
+    /// path this replaced. A regression here would leak scratchpad into
+    /// `content` for every non-streaming caller.
+    #[tokio::test]
+    async fn collector_drops_reasoning_text_but_keeps_the_count() {
+        let (tx, rx) = mpsc::channel(64);
+        tx.send(InferenceEvent::ReasoningDelta("secret".into()))
+            .await
+            .expect("send");
+        tx.send(InferenceEvent::TextDelta("answer".into()))
+            .await
+            .expect("send");
+        tx.send(InferenceEvent::Finish {
+            reason: FinishReason::Stop,
+            prompt_tokens: 10,
+            completion_tokens: 9,
+            reasoning_tokens: 5,
+            cached_tokens: 0,
+            timing: None,
+        })
+        .await
+        .expect("send");
+        drop(tx);
+        let r = collect_chat_completion(rx, "id".into(), 1, "m".into())
+            .await
+            .expect("collected");
+        let c = &r.choices[0];
+        assert!(matches!(&c.message.content, MessageContent::Text(t) if t == "answer"));
+        let u = r.usage.expect("usage");
+        assert_eq!(
+            u.completion_tokens_details
+                .expect("details")
+                .reasoning_tokens,
+            5
+        );
+    }
+
+    /// A parsed tool call outranks the sampler's stop reason, matching
+    /// the streaming projector.
+    #[tokio::test]
+    async fn collector_promotes_finish_reason_to_tool_calls() {
+        let r = collect_from(vec![
+            InferenceEvent::ToolCall {
+                index: 0,
+                id: "call_1".into(),
+                name: "read_file".into(),
+                arguments: r#"{"path":"a.txt"}"#.into(),
+            },
+            finish(FinishReason::Stop),
+        ])
+        .await
+        .expect("collected");
+        let c = &r.choices[0];
+        assert_eq!(c.finish_reason.as_deref(), Some("tool_calls"));
+        let calls = c.message.extra.get("tool_calls").expect("tool_calls");
+        assert_eq!(calls[0]["function"]["name"], "read_file");
+        assert_eq!(calls[0]["type"], "function");
+        assert_eq!(calls[0]["id"], "call_1");
+    }
+
+    /// cached_tokens and helexa_timing were hardcoded to `None` on the
+    /// old non-streaming path; the events carried them all along.
+    #[tokio::test]
+    async fn collector_surfaces_cached_tokens_and_timing() {
+        let (tx, rx) = mpsc::channel(8);
+        tx.send(InferenceEvent::Finish {
+            reason: FinishReason::Stop,
+            prompt_tokens: 100,
+            completion_tokens: 5,
+            reasoning_tokens: 0,
+            cached_tokens: 64,
+            timing: Some(FinishTiming {
+                prefill_ms: 12,
+                decode_ms: 34,
+                prefill_tokens: 100,
+            }),
+        })
+        .await
+        .expect("send");
+        drop(tx);
+        let r = collect_chat_completion(rx, "id".into(), 1, "m".into())
+            .await
+            .expect("collected");
+        let u = r.usage.expect("usage");
+        assert_eq!(u.prompt_tokens_details.expect("details").cached_tokens, 64);
+        let t = u.helexa_timing.expect("timing");
+        assert_eq!((t.prefill_ms, t.decode_ms, t.prefill_tokens), (12, 34, 100));
+    }
+
+    /// A dropped engine slot ends the channel with no Finish. Returning
+    /// a 200 with partial text would hide a fatal worker fault.
+    #[tokio::test]
+    async fn collector_errors_when_the_stream_ends_without_finish() {
+        let err = collect_from(vec![InferenceEvent::TextDelta("partial".into())])
+            .await
+            .expect_err("must not succeed");
+        assert!(err.contains("without a finish"), "unexpected: {err}");
+    }
+
     /// Drain the projection's output into a Vec for assertion.
     async fn collect(mut rx: mpsc::Receiver<ChatCompletionChunk>) -> Vec<ChatCompletionChunk> {
         let mut out = Vec::new();
@@ -405,6 +758,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -443,6 +797,7 @@ mod tests {
             prompt_tokens: 128,
             completion_tokens: 64,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: Some(FinishTiming {
                 prefill_ms: 200,
                 decode_ms: 1500,
@@ -467,6 +822,73 @@ mod tests {
         assert_eq!(timing.prefill_tokens, 128);
     }
 
+    /// A cache hit must reach the caller (#269).
+    ///
+    /// Prefix caching has worked since #11 — neuron logs `reused=2068`
+    /// on prefills — but nothing carried the number onto the wire, so
+    /// every client reported a 0% hit rate on a cache that was saving
+    /// real prefill. dsh showed `Cache hit 0%` against a 42.1K-token
+    /// context, and an agentic harness that believes nothing is cached
+    /// may restructure the very prefix that was being reused.
+    #[tokio::test]
+    async fn cached_tokens_reach_the_usage_chunk() {
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
+        let out_rx = project_chat_stream(rx, "id".into(), 1, "m".into());
+        tx.send(InferenceEvent::Finish {
+            reason: FinishReason::Stop,
+            prompt_tokens: 7413,
+            completion_tokens: 12,
+            reasoning_tokens: 0,
+            cached_tokens: 2068,
+            timing: None,
+        })
+        .await
+        .unwrap();
+        drop(tx);
+
+        let out = collect(out_rx).await;
+        let usage = out
+            .iter()
+            .find_map(|c| c.usage.as_ref())
+            .expect("usage chunk present");
+        let details = usage
+            .prompt_tokens_details
+            .as_ref()
+            .expect("prompt_tokens_details populated when the cache was hit");
+        assert_eq!(details.cached_tokens, 2068);
+        // A sub-count of prompt_tokens, never added into the total —
+        // otherwise a cache hit would inflate what a caller is billed.
+        assert_eq!(usage.prompt_tokens, 7413);
+        assert_eq!(usage.total_tokens, 7425);
+    }
+
+    /// No reuse means no `prompt_tokens_details` at all, so a client
+    /// sees byte-identical JSON to before this landed — and "absent"
+    /// stays distinguishable from "measured as zero".
+    #[tokio::test]
+    async fn no_cache_hit_omits_the_details_object() {
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
+        let out_rx = project_chat_stream(rx, "id".into(), 1, "m".into());
+        tx.send(InferenceEvent::Finish {
+            reason: FinishReason::Stop,
+            prompt_tokens: 10,
+            completion_tokens: 1,
+            reasoning_tokens: 0,
+            cached_tokens: 0,
+            timing: None,
+        })
+        .await
+        .unwrap();
+        drop(tx);
+
+        let usage = collect(out_rx)
+            .await
+            .iter()
+            .find_map(|c| c.usage.clone())
+            .expect("usage chunk present");
+        assert!(usage.prompt_tokens_details.is_none());
+    }
+
     #[tokio::test]
     async fn empty_text_delta_is_dropped() {
         let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
@@ -488,6 +910,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -500,10 +923,24 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn reasoning_delta_is_dropped_in_chat_projection() {
+    /// Reasoning is streamed as `reasoning_content`, never as `content`.
+    ///
+    /// Both halves matter. Keeping it out of `content` is what stops
+    /// scratchpad material appearing in the UI of a client that has
+    /// never heard of reasoning — the guarantee the previous
+    /// drop-everything default provided.
+    ///
+    /// But emitting *something* is what keeps the connection alive. A
+    /// model can think for minutes; while it did, this projection sent
+    /// no events at all, and clients closed the stream as dead. The
+    /// DeepSeek Harness reports `pi-ai stream idle timeout after
+    /// 300000ms`, so any task whose reasoning outlasts the client's
+    /// idle timeout could never complete. SSE comment keep-alives do
+    /// not help, because idle timers reset on events, not comments.
+    async fn reasoning_delta_is_streamed_as_reasoning_content() {
         let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
         let out_rx = project_chat_stream(rx, "id".into(), 1, "m".into());
-        tx.send(InferenceEvent::ReasoningDelta("<think>".into()))
+        tx.send(InferenceEvent::ReasoningDelta("pondering".into()))
             .await
             .unwrap();
         tx.send(InferenceEvent::TextDelta("real".into()))
@@ -511,8 +948,27 @@ mod tests {
             .unwrap();
         drop(tx);
         let out = collect(out_rx).await;
-        assert_eq!(out.len(), 1);
-        assert_eq!(out[0].choices[0].delta["content"], "real");
+        assert_eq!(out.len(), 2, "reasoning must produce a chunk, not silence");
+        assert_eq!(out[0].choices[0].delta["reasoning_content"], "pondering");
+        assert!(
+            out[0].choices[0].delta.get("content").is_none(),
+            "reasoning must never arrive as content"
+        );
+        assert_eq!(out[1].choices[0].delta["content"], "real");
+    }
+
+    /// An empty reasoning delta is still nothing to say — it must not
+    /// manufacture a chunk, matching the text path.
+    #[tokio::test]
+    async fn empty_reasoning_delta_is_dropped() {
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
+        let out_rx = project_chat_stream(rx, "id".into(), 1, "m".into());
+        tx.send(InferenceEvent::ReasoningDelta(String::new()))
+            .await
+            .unwrap();
+        drop(tx);
+        let out = collect(out_rx).await;
+        assert!(out.is_empty(), "empty reasoning deltas produce no chunks");
     }
 
     fn pair() -> ReasoningTokenPair {
@@ -551,6 +1007,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -603,6 +1060,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -649,6 +1107,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -667,10 +1126,12 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn include_thinking_off_drops_reasoning_even_with_markers() {
-        // Default behaviour even when markers happen to be
-        // configured. The flag is the gate, not the marker
-        // presence.
+    /// With the flag off, reasoning stays out of `content` whether or
+    /// not markers are configured — the flag is the gate, not the
+    /// marker presence. It now travels as `reasoning_content` instead
+    /// of being discarded, so this asserts the containment guarantee
+    /// only: nothing model-internal reaches a naive client's `content`.
+    async fn include_thinking_off_keeps_reasoning_out_of_content() {
         let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
         let out_rx = project_chat_stream_with(
             rx,
@@ -693,6 +1154,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -721,6 +1183,7 @@ mod tests {
             prompt_tokens: 42,
             completion_tokens: 5,
             reasoning_tokens: 2,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -755,6 +1218,7 @@ mod tests {
             prompt_tokens: 10,
             completion_tokens: 7,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await

crates/neuron/tests/activation.rs

@@ -29,6 +29,7 @@ async fn test_load_default_models_skips_unknown_harness() {
             quant: None,
             tensor_parallel: None,
             devices: None,
+            sampling: None,
         },
         ModelSpec {
             model_id: "model-b".into(),
@@ -36,6 +37,7 @@ async fn test_load_default_models_skips_unknown_harness() {
             quant: None,
             tensor_parallel: None,
             devices: None,
+            sampling: None,
         },
     ];
 
@@ -95,6 +97,7 @@ async fn test_load_default_models_skipped_on_driver_mismatch() {
         quant: Some("q6k".into()),
         tensor_parallel: Some(2),
         devices: None,
+        sampling: None,
     }];
     let activation = ActivationTracker::new(&specs);
     let reason = "host NVIDIA driver/library mismatch (userspace NVML 580.159 vs loaded \
@@ -196,6 +199,7 @@ fn qwen_spec() -> ModelSpec {
         quant: None,
         tensor_parallel: None,
         devices: None,
+        sampling: None,
     }
 }
 

crates/neuron/tests/candle_lifecycle.rs

@@ -53,6 +53,7 @@ async fn test_candle_qwen3_load_unload_lifecycle() {
         quant: Some(quant),
         tensor_parallel: None,
         devices: Some(vec![0]),
+        sampling: None,
     };
 
     registry

crates/neuron/tests/numerical_reference.rs

@@ -60,7 +60,9 @@ fn read_f32(path: &Path) -> Vec<f32> {
     let bytes = std::fs::read(path).unwrap_or_else(|e| panic!("read {path:?}: {e}"));
     assert!(bytes.len().is_multiple_of(4), "truncated f32 file {path:?}");
     bytes
-        .chunks_exact(4)
+        .as_chunks::<4>()
+        .0
+        .iter()
         .map(|b| f32::from_le_bytes([b[0], b[1], b[2], b[3]]))
         .collect()
 }

crates/neuron/tests/preflight.rs

@@ -106,6 +106,7 @@ fn spec(model_id: &str, tp: Option<u32>, quant: Option<&str>) -> ModelSpec {
         quant: quant.map(String::from),
         tensor_parallel: tp,
         devices: None,
+        sampling: None,
     }
 }
 

crates/neuron/tests/qwen3_next_parity.rs

@@ -37,7 +37,9 @@ fn fixture_dir() -> PathBuf {
 fn read_f32(path: &Path) -> Vec<f32> {
     let bytes = std::fs::read(path).unwrap_or_else(|e| panic!("read {path:?}: {e}"));
     bytes
-        .chunks_exact(4)
+        .as_chunks::<4>()
+        .0
+        .iter()
         .map(|c| f32::from_le_bytes([c[0], c[1], c[2], c[3]]))
         .collect()
 }

models.example.toml

@@ -29,6 +29,27 @@
 #
 #                        Omit only when full precision is genuinely what
 #                        you want served.
+#   sampling           - operator sampling override (#283). Optional
+#                        sub-table `[models.sampling]` with any of
+#                        `temperature`, `top_p`, `top_k`.
+#
+#                        Precedence is request > operator > model's
+#                        generation_config.json > built-in fallback: this
+#                        replaces what the model published, but a caller
+#                        that names a value still wins.
+#
+#                        Use when a model's own published sampling is
+#                        wrong for the workload you serve. Qwen3.8-27B
+#                        publishes temperature = 1.0, which measured a
+#                        20% structural-defect rate on ~2k-token code
+#                        generation against 0/60 at <= 0.6 (Fisher exact
+#                        one-sided p = 0.0031, #283).
+#
+#                        Like `quant`, this must MATCH the same model's
+#                        `[default_models.sampling]` in the host's
+#                        neuron.toml — the same two-paths-to-one-model
+#                        hazard that made #252 possible.
+#                        `script/check-config-consistency.py` gates it.
 #   vram_mb            - rough estimate; advisory only, not enforced.
 #   min_devices        - GPU count this profile needs. TP profiles use
 #                        the same value as the tensor-parallel size.

models.toml

@@ -0,0 +1,219 @@
+# models.toml — helexa fleet catalogue
+#
+# Operator-owned and **tracked in git** (#283). It carries no secrets —
+# cortex.toml holds the API keys — and CI needs to see it: this file and
+# `asset/neuron/<host>.toml` both describe how a model is served, and
+# nothing forced them to agree until `script/check-config-consistency.py`
+# began gating on it. That gap is what #252 was.
+#
+# Deployed to the gateway by `.gitea/workflows/deploy.yml`, which is the
+# ONLY writer (operator decision 2026-08-25). Editing
+# /etc/cortex/models.toml on the host by hand is pointless — the next
+# deploy overwrites it from this file — and dangerous, because until it
+# does, the fleet is running something no commit describes.
+#
+# Fleet shape (sourced from asset/manifest.yml):
+#   beast      = 2x RTX 5090 (32 GB each)  → tp-2 capable
+#   benjy      = 1x RTX 4090 (24 GB)        → large single-GPU
+#   quadbrat   = 1x RTX 3060 (12 GB)        → small / quantised
+
+# Qwen3-Next single-GPU validation target (Stage 8c). Same architecture
+# as Qwen3.6-27B (`model_type = "qwen3_5"`, same layer_types pattern,
+# same Gated DeltaNet hyperparams) but small enough to fit on any
+# fleet GPU in bf16 (~1.6 GB). Stepping stone — if this answers the
+# Tbilisi probe correctly, the qwen3_5 code path is right and TP-2 on
+# Qwen3.6-27B is purely a sharding/collective problem.
+[[models]]
+id = "Qwen/Qwen3.5-0.8B"
+harness = "candle"
+vram_mb = 2000
+min_devices = 1
+min_device_vram_mb = 4000
+# Illustrative pricing (USD per 1M tokens). helexa has no paying customers
+# yet; these differ per model purely to validate spend surfaces in clients.
+cost.input = 0.02
+cost.output = 0.08
+
+# Frontier MoE target (epic #84, F2 #93): 80B total / 3B active,
+# 512 experts / 10 active + shared expert, same Gated DeltaNet hybrid
+# as the 27B. Weights ~45 GB @ Q4_K_M ISQ under TP-2 (~22.5 GB/GPU) —
+# beast only, and NOT co-resident with the 27B (cold-swap per F4e #99).
+# The serving quant belongs HERE as well as on any load request (#252):
+# a cortex cold-load builds its spec from this file, and without it the
+# model loads unquantised — ~149 GB, which cannot fit beast at all.
+# tensor_parallel is derived from min_devices, so only quant is needed.
+[[models]]
+id = "Qwen/Qwen3-Coder-Next"
+harness = "candle"
+quant = "q4k"
+vram_mb = 46000
+min_devices = 2
+min_device_vram_mb = 24000
+# Shares the big-node class with the flagship: either may cold-swap the
+# other off beast (#99), in both directions, so a comparison between
+# them can move traffic each way on demand.
+residency_priority = 300
+# Illustrative pricing (USD per 1M tokens) — frontier tier.
+cost.input = 0.40
+cost.output = 1.60
+
+# Thinking sibling of Coder-Next (same 80B-A3B arch) — on beast for the
+# F3 A/B evaluation (#94); one of the two frontier variants competes for
+# residency, the other stays cold-loadable.
+[[models]]
+id = "Qwen/Qwen3-Next-80B-A3B-Thinking"
+harness = "candle"
+quant = "q4k"
+vram_mb = 46000
+min_devices = 2
+min_device_vram_mb = 24000
+# Shares the big-node class with the flagship: either may cold-swap the
+# other off beast (#99), in both directions, so a comparison between
+# them can move traffic each way on demand.
+residency_priority = 300
+# Illustrative pricing (USD per 1M tokens) — frontier tier.
+cost.input = 0.40
+cost.output = 1.60
+
+# Tensor-parallel target. Beast is the only neuron with two big GPUs.
+[[models]]
+id = "Qwen/Qwen3.6-27B"
+harness = "candle"
+# REQUIRED, not cosmetic (#252). Without it a cortex cold-load builds an
+# unquantised request: the model loads bf16, leaves ~2 GB free per card
+# against a ~3.9 GB prefill requirement, reports `loaded`, and then fails
+# every request. Activation from neuron's default_models has always said
+# q6k; this is the same fact on the path cortex uses.
+quant = "q6k"
+# Measured q6k residency across the pair (17.9 + 22.9 GB), not the 51.7 GB
+# bf16 checkpoint size. The old 54000 described a model we do not serve.
+vram_mb = 42000
+min_devices = 2
+min_device_vram_mb = 24000
+pinned_on = ["beast"]
+# The big-node class. Above the everyday class, so image generation can
+# never take beast however idle the flagship is; equal to the frontier
+# tier, so those two swap freely. `pinned_on` here is affinity only --
+# it no longer implies immunity, so this number is what protects it.
+residency_priority = 300
+# Illustrative pricing (USD per 1M tokens) — top tier, the 27B flagship.
+cost.input = 0.30
+cost.output = 1.20
+# DELIBERATE, TEMPORARY operator override (issue #67): unblocks the opencode
+# helexa-provider implementation by giving /v1/models a limit to consume on
+# beast. Mirrors the enforced NEURON_MAX_PROMPT_TOKENS=131072 there.
+# To be REMOVED once neuron self-derives these — its removal is the
+# acceptance test that the computed values replace this hand-set one.
+#limit.context = 131072
+#limit.input = 122880
+#limit.output = 8192
+
+# Qwen3.8-27B (#251). Architecturally identical to the 3.6-27B — same
+# model_type, same layer counts, same vision tower; the two configs
+# differ only in `transformers_version` — so it serves through the same
+# qwen3_5 TP path with the same KV arithmetic.
+#
+# Same residency class as the 3.6-27B *deliberately*: equal rank is what
+# lets the two displace each other in BOTH directions, so whichever is
+# asked for takes beast and the other can take it back. A higher number
+# here would let this one claim the node permanently and make the
+# comparison between them impossible.
+[[models]]
+id = "Qwen/Qwen3.8-27B"
+harness = "candle"
+quant = "q6k"
+vram_mb = 42000
+min_devices = 2
+min_device_vram_mb = 24000
+pinned_on = ["beast"]
+residency_priority = 300
+cost.input = 0.30
+cost.output = 1.20
+
+# Sampling override (#283). The model publishes `temperature = 1.0` in
+# its generation_config.json, and honouring that (#272) is correct by
+# default — but measured on this stack it produced a 20% structural-
+# defect rate on ~2k-token code generation (4/20 samples failed
+# `node --check` or dropped required exports) against 0/60 at
+# temperature <= 0.6. Fisher exact one-sided p = 0.0031.
+#
+# Extrapolated at constant hazard that is a ~79% chance of at least one
+# defect in a 15k-token file, which is why this model had never once
+# completed the tower-defence task two cloud models did in ten minutes.
+#
+# 0.6 is Qwen's own published *inference* guidance for thinking mode —
+# a number that lives in a model card no machine can read, which is
+# exactly why it has to be set here by hand.
+#
+# Precedence is request > operator > model, so a caller that names a
+# temperature still gets it. MUST match asset/neuron/beast.toml;
+# script/check-config-consistency.py fails CI if it drifts.
+[models.sampling]
+temperature = 0.6
+
+# Mid-size dense bf16 — fits on benjy (24 GB) or beast (32 GB ea.).
+[[models]]
+id = "Qwen/Qwen3-8B"
+harness = "candle"
+vram_mb = 18000
+min_devices = 1
+min_device_vram_mb = 16000
+# Everyday class, shared with Z-Image — they take turns on benjy.
+residency_priority = 200
+# Illustrative pricing (USD per 1M tokens) — mid tier (helexa/balanced).
+cost.input = 0.10
+cost.output = 0.40
+
+# Single-GPU dense bf16 — fits anywhere with ≥4 GB.
+[[models]]
+id = "Qwen/Qwen3-1.7B"
+harness = "candle"
+vram_mb = 4000
+min_devices = 1
+min_device_vram_mb = 4000
+# Illustrative pricing (USD per 1M tokens) — small tier (helexa/small).
+cost.input = 0.04
+cost.output = 0.16
+
+# Small GGUF quantised — runs on quadbrat (12 GB) without breaking a sweat.
+[[models]]
+id = "unsloth/Qwen3-0.6B-GGUF"
+harness = "candle"
+quant = "Q4_K_M"
+vram_mb = 500
+min_devices = 1
+min_device_vram_mb = 4000
+# Illustrative pricing (USD per 1M tokens) — cheapest, the tiny GGUF.
+cost.input = 0.015
+cost.output = 0.06
+
+# -- Tier aliases ------------------------------------------------------------
+# Clients can request inference against an alias and cortex transparently
+# routes to the concrete model id below. Useful so client code can pick a
+# latency/quality tier without hardcoding specific model ids — operators
+# can swap the targets here without touching client config.
+[aliases]
+"helexa/small" = "Qwen/Qwen3-1.7B"
+"helexa/balanced" = "Qwen/Qwen3-8B"
+"helexa/large" = "Qwen/Qwen3.8-27B"
+"helexa/image" = "Tongyi-MAI/Z-Image-Turbo"
+
+# Text-to-image (#197/#203): Z-Image-Turbo, 6B S3-DiT + Qwen3-4B text
+# encoder (runs on host CPU by default — see [harness.candle.image] in
+# neuron.toml) + VAE. BF16 DiT+VAE resident ≈ 14.6 GB; fits benjy's
+# 4090 via cold-swap with Qwen3-8B (operator decision 2026-07-29,
+# #197). Not default-activated anywhere — cortex cold-loads it on
+# first images request and the evictor swaps it back out under text
+# pressure. Metered in megapixel-steps (#202), not tokens; the token
+# cost fields deliberately stay unset.
+[[models]]
+id = "Tongyi-MAI/Z-Image-Turbo"
+harness = "candle"
+vram_mb = 15000
+min_devices = 1
+min_device_vram_mb = 20000
+# Everyday class, shared with Qwen3-8B: an image request cold-swaps the
+# 8B off benjy and the next text request swaps it straight back. Below
+# the big-node class, so it can never take beast -- its device
+# constraints alone would otherwise let it land there.
+residency_priority = 200

neuron.example.toml

@@ -110,6 +110,33 @@ name = "candle"
 # harness = "candle"
 # quant = "Q4_K_M"
 # devices = [0]
+#
+# Operator sampling override (#283) — optional, per model.
+#
+# By default neuron honours what the model published in its
+# `generation_config.json`; that is the model authors' stated intent and
+# is right most of the time. It is not always right for the workload you
+# serve: Qwen3.8-27B publishes `temperature = 1.0`, which measured a 20%
+# structural-defect rate on ~2k-token code generation against 0/60 at
+# <= 0.6 (Fisher exact one-sided p = 0.0031).
+#
+# Precedence is request > operator > model > built-in fallback, so a
+# caller that names a temperature still gets what it asked for; this only
+# changes what a caller who names nothing receives. Each field overlays
+# independently — setting `temperature` alone leaves the model's `top_p`
+# and `top_k` in force.
+#
+# Deliberately explicit. neuron will never inspect a model and guess a
+# "better" temperature: a guesser is a footgun, a number you wrote is not.
+#
+# If cortex can also cold-load this model, the same values must appear in
+# its `models.toml` entry — `script/check-config-consistency.py` gates on
+# that, because two files describing one model is how #252 happened.
+#
+# [default_models.sampling]
+# temperature = 0.6
+# top_p = 0.95
+# top_k = 20
 
 # Text-to-image serving (#198). Applies to diffusers-format image
 # models (Z-Image family). All fields optional.

script/check-config-consistency.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""Fail when the two files describing one model disagree.
+
+`models.toml` (cortex's catalogue, used for cold loads) and
+`asset/neuron/<host>.toml` (`[[default_models]]`, used for models a host
+holds resident) both describe *how a model is served*. Nothing forced
+them to agree, and #252 is what that costs: models.toml omitted `quant`,
+so a cortex cold-load served bf16 and could not prefill, while the same
+model was fine when the host loaded it from its own config.
+
+#283 adds a second such field — the operator sampling override — so this
+check exists before the second instance of the bug rather than after it.
+
+Checked, per model id present in both files:
+
+  * ``quant``      — the #252 field
+  * ``sampling``   — the #283 field, compared field-by-field
+
+A model in only one file is not an error: plenty of catalogue entries are
+never resident anywhere, and a host may hold a model the catalogue does
+not list. Only *disagreement* fails.
+
+Usage::
+
+    check-config-consistency.py                       # repo defaults
+    check-config-consistency.py --models /etc/cortex/models.toml \\
+                               --neuron asset/neuron/beast.toml
+
+Exit status is 0 when consistent, 1 on any disagreement, 2 on bad input.
+"""
+
+from __future__ import annotations
+
+import argparse
+import glob
+import os
+import sys
+
+try:
+    import tomllib
+except ModuleNotFoundError:  # pragma: no cover - Python < 3.11
+    print("error: needs Python 3.11+ for tomllib", file=sys.stderr)
+    raise SystemExit(2)
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+# Fields that must agree wherever both files mention the same model.
+# Extend this list whenever a serving-critical field lands in both.
+SCALAR_FIELDS = ["quant"]
+TABLE_FIELDS = ["sampling"]
+
+
+def load(path: str) -> dict:
+    try:
+        with open(path, "rb") as fh:
+            return tomllib.load(fh)
+    except FileNotFoundError:
+        print(f"error: {path} not found", file=sys.stderr)
+        raise SystemExit(2)
+    except tomllib.TOMLDecodeError as exc:
+        print(f"error: {path} is not valid TOML: {exc}", file=sys.stderr)
+        raise SystemExit(2)
+
+
+def catalogue_entries(path: str) -> dict[str, dict]:
+    """model id -> profile, from models.toml's `[[models]]`."""
+    return {m["id"]: m for m in load(path).get("models", []) if "id" in m}
+
+
+def resident_entries(path: str) -> dict[str, dict]:
+    """model id -> spec, from a neuron config's `[[default_models]]`."""
+    return {
+        m["model_id"]: m
+        for m in load(path).get("default_models", [])
+        if "model_id" in m
+    }
+
+
+def label(path: str) -> str:
+    """Repo-relative when the file is in the repo, basename otherwise.
+
+    Deploy runs pass paths outside the tree (a models.toml fetched off
+    the gateway), and `../../../..`-prefixed relpaths make the diff
+    unreadable in a CI log.
+    """
+    rel = os.path.relpath(path, REPO)
+    return rel if not rel.startswith("..") else os.path.basename(path)
+
+
+def compare(model_id: str, cat: dict, res: dict, catalogue_path: str, neuron_path: str):
+    """Yield one human-readable line per disagreement."""
+    cat_name = label(catalogue_path)
+    neu_name = label(neuron_path)
+
+    for field in SCALAR_FIELDS:
+        a, b = cat.get(field), res.get(field)
+        # Absent on one side is "unspecified", not "different" -- but for
+        # quant specifically that IS the #252 bug, so say so loudly.
+        if a != b:
+            if a is None or b is None:
+                detail = (
+                    "  (one side leaves it unset -- this is exactly #252: "
+                    "an unset quant makes a cold load serve bf16)"
+                    if field == "quant"
+                    else "  (one side leaves it unset)"
+                )
+            else:
+                detail = ""
+            yield (
+                f"{model_id}: {field} differs\n"
+                f"    {cat_name}: {a!r}\n"
+                f"    {neu_name}: {b!r}{detail}"
+            )
+
+    for field in TABLE_FIELDS:
+        a = cat.get(field) or {}
+        b = res.get(field) or {}
+        if a == b:
+            continue
+        keys = sorted(set(a) | set(b))
+        rows = "\n".join(
+            f"      {k}: {cat_name}={a.get(k)!r}  {neu_name}={b.get(k)!r}"
+            for k in keys
+            if a.get(k) != b.get(k)
+        )
+        yield (
+            f"{model_id}: {field} differs\n"
+            f"    a model must sample the same way whether cortex cold-loads it\n"
+            f"    or the host holds it resident:\n{rows}"
+        )
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--models", default=os.path.join(REPO, "models.toml"),
+                    help="path to models.toml (default: repo root)")
+    ap.add_argument("--neuron", nargs="*", default=None,
+                    help="neuron config(s) (default: asset/neuron/*.toml)")
+    args = ap.parse_args()
+
+    neuron_paths = args.neuron
+    if not neuron_paths:
+        neuron_paths = sorted(glob.glob(os.path.join(REPO, "asset", "neuron", "*.toml")))
+    if not neuron_paths:
+        print("error: no neuron configs found", file=sys.stderr)
+        return 2
+
+    catalogue = catalogue_entries(args.models)
+    problems: list[str] = []
+    compared = 0
+
+    for npath in neuron_paths:
+        for model_id, spec in resident_entries(npath).items():
+            profile = catalogue.get(model_id)
+            if profile is None:
+                # A host may hold a model the catalogue does not list.
+                # Not an error, but worth seeing -- cortex cannot route
+                # to what it has no profile for.
+                print(f"note: {model_id} is resident in {label(npath)} "
+                      f"but absent from {label(args.models)}")
+                continue
+            compared += 1
+            problems.extend(compare(model_id, profile, spec, args.models, npath))
+
+    if problems:
+        print(f"\nconfig inconsistency: {len(problems)} disagreement(s)\n",
+              file=sys.stderr)
+        for p in problems:
+            print(f"  {p}\n", file=sys.stderr)
+        print("Two config files describe one model. Make them agree, or the\n"
+              "model is served one way on a cold load and another when\n"
+              "resident -- see #252 and #283.", file=sys.stderr)
+        return 1
+
+    print(f"config consistency OK: {compared} model/host pairing(s) agree "
+          f"on {', '.join(SCALAR_FIELDS + TABLE_FIELDS)}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

script/infra-setup.sh

@@ -138,14 +138,28 @@ if ! ssh "${bench_host}" '
     echo "  failed to prepare ${bench_host} for helexa-bench"
 fi
 
-# Push application config to the fleet. The deploy workflow is
-# scoped to package install + service restart; config changes ride
-# along with this script instead, since:
-#   - cortex.toml and models.toml are gitignored (operator-owned, may
-#     include secrets), so CI never sees them
-#   - asset/neuron/<short>.toml is tracked but iterating locally is
-#     faster than pushing a commit and waiting for build-prerelease
-#     to roll over
+# Push the one config CI cannot own.
+#
+# Operator decision 2026-08-25 (#283): the automated deployment path is
+# the ONLY path. `.gitea/workflows/deploy.yml` now ships models.toml to
+# the gateway, asset/neuron/<short>.toml to each neuron, and
+# asset/helexa-bench/<host>.toml to the bench host — digest-compared,
+# restarting the service when the file changes.
+#
+# Those three syncs USED to live here, and that is exactly how the fleet
+# drifted: benjy and quadbrat were still running pre-#197 configs months
+# after the tracked assets had moved on, because deploying them meant
+# remembering to re-run this script. A hand-sync that exists "for local
+# iteration" is a footgun that silently overwrites, or silently fails to
+# overwrite, whatever CI just placed. Do not add one back.
+#
+# What remains is cortex.toml, which carries the inference API keys
+# ([[entitlements.keys]]) and therefore cannot be tracked in git for CI
+# to read. It is the last hand-synced config on the fleet and the last
+# place this drift can still happen; closing it needs the secrets to
+# come from Gitea secrets at deploy time rather than from a working
+# copy.
+#
 # Missing source files are skipped silently — re-run after editing.
 # Optional 4th/5th args set the destination group and mode. Configs that
 # carry secrets (API keys, JWT signing keys, SMTP/DB passwords) must land
@@ -174,23 +188,15 @@ sync_config() {
     fi
 }
 
-echo "==> ${cortex_host}: syncing gateway configs"
-# cortex.toml holds the inference API keys ([[entitlements.keys]]).
+echo "==> ${cortex_host}: syncing gateway secrets config"
+# cortex.toml holds the inference API keys ([[entitlements.keys]]), so it
+# is gitignored and CI cannot see it. The only config still synced by
+# hand — everything else is deploy.yml's.
 sync_config "${cortex_host}" "${repo_path}/cortex.toml" /etc/cortex/cortex.toml cortex 0640
-sync_config "${cortex_host}" "${repo_path}/models.toml" /etc/cortex/models.toml
 
-for neuron_host in "${neuron_hosts[@]}"; do
-    short="${neuron_host%%.*}"
-    echo "==> ${neuron_host}: syncing per-host neuron config"
-    sync_config "${neuron_host}" \
-        "${repo_path}/asset/neuron/${short}.toml" \
-        /etc/neuron/neuron.toml
-done
-
-echo "==> ${bench_host}: syncing bench config"
-sync_config "${bench_host}" \
-    "${repo_path}/asset/helexa-bench/${bench_host%%.*}.toml" \
-    /etc/helexa-bench/helexa-bench.toml
+# models.toml, asset/neuron/<short>.toml and asset/helexa-bench/<host>.toml
+# are deliberately NOT synced here. deploy.yml owns them; see the note
+# above sync_config().
 
 # ── bench UI: public nginx vhost on the gateway (bench.helexa.ai) ──────
 # The built SPA is rsynced to the webroot by deploy.yml; nginx serves it
@@ -600,6 +606,24 @@ if ! ssh "${svc_host}" '
     echo "  failed to prepare ${svc_host}"
 fi
 
+# ── STILL HAND-SYNCED, and tracked as such ────────────────────────────
+# Everything below this line is a config CI does not yet own, which by
+# the 2026-08-25 decision (#283) makes it a footgun still loaded. Two
+# distinct reasons, needing two distinct fixes:
+#
+#   * helexa-router.toml / helexa-upstream.toml / helexa-angels.toml are
+#     gitignored because they carry JWT secrets, SMTP passwords and
+#     database URLs. Same class as cortex.toml: CI cannot read them until
+#     the secrets come from Gitea secrets and the file is rendered at
+#     deploy time.
+#   * asset/searxng/{settings.yml,searxng.container,helexa-searxng.xml}
+#     ARE tracked and could move to deploy.yml today but for one thing:
+#     asset/sudoers.d/gallumbits-host.conf grants gitea_ci no rsync at
+#     all, so the deploy would fail until the grants land on the host.
+#     Grants first, deploy step second — wiring it the other way round
+#     breaks every deploy in between.
+#
+# See the tracking issue; do not close the gap by adding syncs here.
 echo "==> ${svc_host}: syncing service configs"
 sync_config "${svc_host}" "${repo_path}/helexa-router.toml" /etc/helexa-router/helexa-router.toml
 # helexa-upstream.toml holds the session JWT secret, the SMTP submission