fix(sampling): apply the repetition knobs, and fetch the config that sets defaults

Two gaps in #272, both the failure shape the epic names: work done
correctly in one place, not honoured at the edge.

**The repetition knobs never reached the sampler.**
`repetition_penalty` and `repeat_last_n` were modelled on the wire,
resolved into `SamplingParams` and threaded through all four generation
loops — and then `sample_with_penalty` read the module constants
`REPEAT_PENALTY` / `REPEAT_LAST_N` instead, so nothing ever read the
resolved values. Accepted, answered 200, discarded. Observed live: the
same seed at penalty 1.0 and at 1.9 over a 512-token window produced
byte-identical output.

`sample_with_penalty` now takes the resolved `SamplingParams`; the
constants are gone, so there is nothing left to drift from. A zero
window is also honoured as "no penalty" rather than slicing the history
backwards.

**The model's published sampling was usually not on disk to read.**
`ModelGenerationDefaults::load_from_dir` looks beside `tokenizer.json`,
but the loader only ever fetched `config.json`, `tokenizer.json` and the
weights — never `generation_config.json`. So the file was present only
where a model had been staged as a full repo clone. Qwen3.8-27B on beast
was staged that way and does log `top_k=20 top_p=0.95 temperature=1.0`;
Qwen3-1.7B, downloaded by neuron itself, logs "no generation_config.json"
and silently falls back to 0.7 with no truncation — the exact default
#272 set out to stop choosing.

Both resolvers now fetch it alongside the tokenizer: from the model repo
for dense loads, and from the base repo for GGUF, which is where the
tokenizer comes from too (a GGUF-only repo ships neither). A miss stays
a debug line, not a load failure.

Three tests pin the penalty path: a caller-set penalty changes the
sample, the window decides what is penalised, and a zero window is a
no-op.

Refs #271, #272.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

.gitea/workflows/build-prerelease.yml

@@ -105,6 +105,11 @@ jobs:
           echo "commit_timestamp=${COMMIT_TIMESTAMP}" >> "$GITHUB_OUTPUT"
 
       - id: changes
+        env:
+          # The push's own range. Empty on workflow_dispatch and all
+          # zeroes for a new branch; `decide` treats both as unusable and
+          # falls back to the full gate.
+          PUSH_BEFORE: ${{ github.event.before }}
         run: |
           set -ux
           # Default: build everything. Detection only ever narrows
@@ -189,9 +194,30 @@ jobs:
               BUILD_TOOLS=$(decide "$TOOLS_BASE" "$TOOLS_RE")
               BUILD_ANGELS=$(decide "$ANGELS_BASE" "$ANGELS_RE")
               if [ "$BUILD_CORTEX" = "true" ] || [ "$BUILD_NEURON" = "true" ] || [ "$BUILD_BENCH" = "true" ] || [ "$BUILD_UPSTREAM" = "true" ] || [ "$BUILD_ROUTER" = "true" ] || [ "$BUILD_TOOLS" = "true" ] || [ "$BUILD_ANGELS" = "true" ]; then
+                # Anything we are about to build gets linted and tested.
+                # This is the invariant that matters: no artefact reaches
+                # the fleet unchecked, however the branch below behaves.
                 CHECK_RUST=true
               else
-                CHECK_RUST=$(decide "$CORTEX_BASE" "$RUST_RE")
+                # Nothing is being built, so nothing will be published.
+                # The only question left is whether THIS PUSH touched
+                # Rust — and the honest base for that is the push's own
+                # range, not a package's publication lag.
+                #
+                # This used to ask "has anything Rust changed since
+                # cortex was last published". Cortex is only rebuilt when
+                # cortex paths change, so after a run of neuron-only work
+                # its base sits days behind and the diff from it sweeps in
+                # every unrelated Rust commit since. A one-character
+                # whitespace fix to a doc then re-ran fmt, clippy and the
+                # full test suite over a delta that had already been
+                # linted on the branch that introduced it (2026-08-18,
+                # base fec5b33).
+                #
+                # `decide` returns true for an unusable base, so a new
+                # branch (all-zero before-sha) or a force-push still gets
+                # the full gate.
+                CHECK_RUST=$(decide "$PUSH_BEFORE" "$RUST_RE")
               fi
             fi
           fi

.gitea/workflows/ci.yml

@@ -39,24 +39,34 @@ env:
   CARGO_TARGET_DIR: target-${{ github.job }}
 
 jobs:
+  # Two independent questions: does this change need the Rust gate, and
+  # does it need the web gate.
+  #
   # The SPA under helexa.ai/ is built and shipped by its own deploy job and
   # is not read by any crate, build script or spec file — so a change
   # confined to it cannot alter a Rust result. Waiting up to 70 minutes for
   # a CUDA type-check to prove that is pure latency, and the wait is what
-  # tempts people to merge front-end work unchecked.
+  # tempts people to merge front-end work unchecked. Documentation is inert
+  # the same way, and more obviously: no crate `include_str!`s a markdown
+  # file, no spec packages `doc/`, no build script reads one.
   #
-  # The classification is deliberately one-directional: it only ever
-  # narrows the gate when *every* changed path is front-end, and every
+  # Conversely a Rust-only change cannot break the SPA, so it need not
+  # spend three minutes on `npm ci` + lint + typecheck + build.
+  #
+  # The classification stays deliberately one-directional: it only ever
+  # narrows the gate when *every* changed path is known-inert, and every
   # other outcome — a tag, an unknown base, an empty diff, a mixed
   # changeset — falls through to the full gate. A new top-level directory
   # is therefore covered on the day it appears, without anyone
-  # remembering to add it here.
+  # remembering to add it here. Adding an inert pattern is the only way to
+  # widen what gets skipped, which is the property worth keeping.
   changes:
     name: Classify changes
     timeout-minutes: 10
     runs-on: fedora-43
     outputs:
       rust: ${{ steps.classify.outputs.rust }}
+      web: ${{ steps.classify.outputs.web }}
     steps:
       - uses: actions/checkout@v4
         with:
@@ -64,7 +74,12 @@ jobs:
       - id: classify
         run: |
           set -u
-          full() { echo "→ full gate: $1"; echo "rust=true" >> "$GITHUB_OUTPUT"; exit 0; }
+          full() {
+            echo "→ full gate: $1"
+            echo "rust=true" >> "$GITHUB_OUTPUT"
+            echo "web=true" >> "$GITHUB_OUTPUT"
+            exit 0
+          }
 
           # Tags build the SRPMs from the whole tree, and the srpm jobs
           # gate on fmt/clippy/test/cuda-check. Never narrow there.
@@ -86,11 +101,31 @@ jobs:
           echo "changed files:"
           printf '%s\n' "${files}" | sed 's/^/  /'
 
-          other="$(printf '%s\n' "${files}" | grep -v '^helexa\.ai/' || true)"
-          [ -z "${other}" ] || full "changes outside helexa.ai/"
+          # Inert for Rust: the SPA, and documentation. Markdown is
+          # matched only at the repo root and under doc/ — a .md inside
+          # crates/ is left to fall through, since `include_str!` could
+          # make one load-bearing without this file knowing.
+          rust_relevant="$(printf '%s\n' "${files}" \
+            | grep -v '^helexa\.ai/' \
+            | grep -v '^doc/' \
+            | grep -vE '^[^/]+\.md$' \
+            || true)"
+          if [ -n "${rust_relevant}" ]; then
+            echo "rust=true" >> "$GITHUB_OUTPUT"
+            echo "→ Rust gate: required"
+          else
+            echo "rust=false" >> "$GITHUB_OUTPUT"
+            echo "→ Rust gate: skipped (front-end and/or docs only)"
+          fi
 
-          echo "→ front-end only: skipping the Rust gate"
-          echo "rust=false" >> "$GITHUB_OUTPUT"
+          # The web gate is needed only when the SPA itself changed.
+          if printf '%s\n' "${files}" | grep -q '^helexa\.ai/'; then
+            echo "web=true" >> "$GITHUB_OUTPUT"
+            echo "→ web gate: required"
+          else
+            echo "web=false" >> "$GITHUB_OUTPUT"
+            echo "→ web gate: skipped (nothing under helexa.ai/)"
+          fi
 
   fmt:
     name: Format
@@ -106,6 +141,8 @@ jobs:
     name: Web (lint + typecheck + i18n + build)
     timeout-minutes: 20
     runs-on: fedora-43
+    needs: changes
+    if: needs.changes.outputs.web == 'true'
     defaults:
       run:
         working-directory: helexa.ai

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/discovery.rs

@@ -109,6 +109,23 @@ pub struct ModelLoad {
     /// (#54/#137). `#[serde(default)]` for back-compat.
     #[serde(default)]
     pub rejected_per_principal: u64,
+    /// Anonymous requests currently holding a seat (#262). Read against
+    /// `in_flight` it says how much of this model's load is unattributable;
+    /// against `anon_max_in_flight`, how close that traffic is to the
+    /// ceiling that keeps it from starving identified callers.
+    #[serde(default)]
+    pub anon_in_flight: usize,
+    /// Seats anonymous traffic may hold at once (#262); `0` = refused
+    /// outright. `#[serde(default)]` (→ 0) for pre-#262 neurons, which
+    /// applied no ceiling at all — cortex skips the gauge rather than
+    /// publish a 0 that reads as the opposite of the truth.
+    #[serde(default)]
+    pub anon_max_in_flight: usize,
+    /// Cumulative anonymous requests refused for want of leftover capacity
+    /// (#262). Rising here with a flat `rejected_timeout` means the
+    /// reservation is working, not that the model is overloaded.
+    #[serde(default)]
+    pub rejected_anon_yield: u64,
     /// Cumulative requests that waited for KV budget and never got it
     /// (#257) — the model is saturated by long-context sequences rather
     /// than by request count. `#[serde(default)]` for back-compat.
@@ -171,6 +188,9 @@ mod health_load_tests {
                 kv_available_mb: 0,
                 rejected_timeout: 0,
                 rejected_per_principal: 0,
+                anon_in_flight: 2,
+                anon_max_in_flight: 7,
+                rejected_anon_yield: 0,
                 tok_s_prefill: 0.0,
                 tok_s_decode: 0.0,
             }],
@@ -182,6 +202,8 @@ mod health_load_tests {
         assert_eq!(back.models[0].queue_depth, 3);
         assert_eq!(back.models[0].max_in_flight, 8);
         assert_eq!(back.models[0].max_queue_depth, 8);
+        assert_eq!(back.models[0].anon_in_flight, 2);
+        assert_eq!(back.models[0].anon_max_in_flight, 7);
     }
 
     #[test]
@@ -193,6 +215,10 @@ mod health_load_tests {
         assert_eq!(m.in_flight, 2);
         assert_eq!(m.max_in_flight, 0);
         assert_eq!(m.max_queue_depth, 0);
+        // Likewise a pre-#262 neuron, which enforced no anonymous ceiling.
+        assert_eq!(m.anon_in_flight, 0);
+        assert_eq!(m.anon_max_in_flight, 0);
+        assert_eq!(m.rejected_anon_yield, 0);
     }
 }
 

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")]
@@ -249,8 +264,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 +280,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 +304,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

@@ -101,6 +101,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/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,

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",
@@ -111,7 +178,15 @@ fn describe_metrics() {
     );
     metrics::describe_counter!(
         "cortex_model_rejections_total",
-        "Admission rejections per neuron:model by reason: queue_full / wait_timeout / per_principal — the load-shedding signal (#137)"
+        "Admission rejections per neuron:model by reason: queue_full / wait_timeout / per_principal / anon_yield — the load-shedding signal (#137, #262)"
+    );
+    metrics::describe_gauge!(
+        "cortex_model_anon_in_flight",
+        "Anonymous (unattributable) requests holding a seat on a neuron:model (#262)"
+    );
+    metrics::describe_gauge!(
+        "cortex_model_anon_max_in_flight",
+        "Seats anonymous traffic may hold at once, so it cannot starve identified callers (#262)"
     );
     metrics::describe_gauge!(
         "cortex_model_tok_s_decode",
@@ -122,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

@@ -380,6 +380,21 @@ fn export_health_metrics(node: &str, h: &HealthResponse) {
         counter!("cortex_model_rejections_total",
             "node" => node.to_string(), "model" => m.id.clone(), "reason" => "per_principal")
         .absolute(m.rejected_per_principal);
+        counter!("cortex_model_rejections_total",
+            "node" => node.to_string(), "model" => m.id.clone(), "reason" => "anon_yield")
+        .absolute(m.rejected_anon_yield);
+        // How much of this model's load is unattributable (#262), and the
+        // ceiling that keeps it from starving identified callers. 0 for a
+        // pre-#262 neuron, which had no ceiling — skip rather than publish
+        // a 0 that reads as "anonymous is refused" when it means the
+        // opposite.
+        gauge!("cortex_model_anon_in_flight", "node" => node.to_string(), "model" => m.id.clone())
+            .set(m.anon_in_flight as f64);
+        if m.anon_max_in_flight > 0 {
+            gauge!("cortex_model_anon_max_in_flight",
+                "node" => node.to_string(), "model" => m.id.clone())
+            .set(m.anon_max_in_flight as f64);
+        }
     }
     for d in &h.devices {
         let device = d.index.to_string();

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/state.rs

@@ -98,8 +98,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/load_routing.rs

@@ -55,6 +55,9 @@ async fn seed_loaded(fleet: &CortexState, node: &str, in_flight: usize, queue_de
             kv_available_mb: 0,
             rejected_timeout: 0,
             rejected_per_principal: 0,
+            anon_in_flight: 0,
+            anon_max_in_flight: 7,
+            rejected_anon_yield: 0,
             tok_s_prefill: 0.0,
             tok_s_decode: 0.0,
         },

crates/helexa-upstream/tests/web_pg.rs

@@ -53,18 +53,6 @@ fn unique_email() -> String {
     format!("u-{}@test.local", uuid::Uuid::new_v4())
 }
 
-/// A fixture value that is unique per run.
-///
-/// Rows keyed on a hash of these (`email_tokens.token_hash`,
-/// `api_keys.key_hash`) are uniquely indexed, so a hard-coded literal
-/// passes against a virgin database and then fails with `23505` on every
-/// run afterwards — the tests only looked deterministic because CI always
-/// gets a fresh one. Anything inserted under a unique constraint gets a
-/// fresh value here.
-fn unique_secret(prefix: &str) -> String {
-    format!("{prefix}-{}", uuid::Uuid::new_v4())
-}
-
 async fn post(url: String, body: Value, bearer: Option<&str>) -> reqwest::Response {
     let c = reqwest::Client::new();
     let mut req = c.post(url).json(&body);
@@ -97,13 +85,13 @@ async fn verify_endpoint_consumes_token_once() {
         .await
         .unwrap()
         .get("id");
-    let raw = unique_secret("verify-raw-token");
+    let raw = "verify-raw-token-xyz";
     pool.execute(
         sqlx::query(
             "INSERT INTO email_tokens (token_hash, user_id, kind, expires_at) \
              VALUES ($1, $2, 'verify', now() + interval '1 hour')",
         )
-        .bind(sha256(&raw))
+        .bind(sha256(raw))
         .bind(user_id),
     )
     .await
@@ -285,13 +273,13 @@ async fn fingerprint_abuse_silently_deactivates_all_no_clue() {
         .await
         .unwrap()
         .get("id");
-    let raw = unique_secret("sk-helexa-deactivated-probe");
+    let raw = "sk-helexa-deactivated-probe";
     pool.execute(
         sqlx::query(
             "INSERT INTO api_keys (account_id, key_hash, key_prefix) VALUES ($1, $2, 'sk-helexa-')",
         )
         .bind(acct)
-        .bind(sha256(&raw)),
+        .bind(sha256(raw)),
     )
     .await
     .unwrap();

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),
@@ -729,6 +758,18 @@ fn inference_error_response(err: InferenceError) -> axum::response::Response {
             "too many concurrent requests for this key; retry shortly",
         )
         .with_retry_after(retry_after_secs),
+        // Draining (#256). 503 + Retry-After, like any transient capacity
+        // condition: the node is coming back, and a well-behaved client
+        // should retry rather than surface a failure. Deliberately not
+        // `rate_limit_exceeded` — nothing about this is the caller's
+        // fault or their rate.
+        InferenceError::ShuttingDown => OpenAiError::new(
+            503,
+            "api_error",
+            "shutting_down",
+            "neuron is shutting down; retry shortly",
+        )
+        .with_retry_after(15),
         InferenceError::Other(e) => OpenAiError::without_code(500, "api_error", format!("{e:#}")),
     };
     // Every failed request says why, once, here (#63). Before this the
@@ -1045,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

@@ -180,9 +180,36 @@ pub struct AdmissionConfig {
     /// for any single principal (resolved from the `x-helexa-*` headers
     /// cortex stamps), so one client can't monopolize the queue while others
     /// wait. Over-cap → `429 rate_limit_exceeded` + `Retry-After`. `0`
-    /// disables the cap; anonymous requests are always exempt.
+    /// disables the cap.
+    ///
+    /// Anonymous requests carry no principal and so cannot be capped by it.
+    /// They are bounded instead by `anon_max_in_flight` / `anon_max_pending`
+    /// below (#262).
     #[serde(default = "default_admission_max_per_principal")]
     pub max_per_principal: usize,
+    /// Ceiling on **anonymous** requests holding an in-flight seat (#262).
+    ///
+    /// Anonymous traffic is served from capacity left over once identified
+    /// traffic is satisfied: it never takes a seat while an identified
+    /// request is waiting, and it may never hold every seat at once — which
+    /// is what this bounds, because a request already running cannot be
+    /// preempted. Reserving one seat is what caps how long a newly arrived
+    /// identified request waits: one request's duration, not all of them.
+    ///
+    /// Unset (the default) derives `max_in_flight - 1`, floored at 1 so a
+    /// single-seat model still serves anonymous callers. `0` refuses
+    /// anonymous traffic outright.
+    #[serde(default)]
+    pub anon_max_in_flight: Option<usize>,
+    /// Ceiling on anonymous requests in flight **plus** queued (#262).
+    ///
+    /// The queue needs the same reservation as the seats: without it an
+    /// anonymous flood fills the queue and identified callers are refused at
+    /// the door, before priority can help them.
+    ///
+    /// Unset derives `max_in_flight + max_queue_depth - 1`, floored at 1.
+    #[serde(default)]
+    pub anon_max_pending: Option<usize>,
     /// Maximum seconds a request waits for enough **KV budget** to free
     /// before it is refused (#257). Deliberately far longer than
     /// `max_wait_secs`: that one waits for a slot to turn over, this one
@@ -200,6 +227,8 @@ impl Default for AdmissionConfig {
             max_queue_depth: default_admission_max_queue_depth(),
             max_wait_secs: default_admission_max_wait_secs(),
             max_per_principal: default_admission_max_per_principal(),
+            anon_max_in_flight: None,
+            anon_max_pending: None,
             kv_max_wait_secs: default_admission_kv_max_wait_secs(),
         }
     }

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,10 @@ 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,
     pub eos_id: Option<u32>,
     pub tool_schemas: ToolSchemas,
     pub tx: mpsc::Sender<InferenceEvent>,
@@ -439,6 +449,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 +467,10 @@ 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,
     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 +478,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,22 +578,23 @@ async fn run_engine(cfg: EngineConfig, mut rx: mpsc::Receiver<EngineRequest>) {
                     break;
                 }
             };
-            let nt = match sample_with_penalty(&logits, &slot.generated, &mut slot.lp) {
-                Ok(t) => t,
-                Err(e) => {
-                    let health = logits_health_slice(&logits_vec);
-                    tracing::warn!(
-                        ?health,
-                        error = %e,
-                        "batch engine: sample failed; logits unhealthy"
-                    );
-                    // Unhealthy logits are a device-level problem —
-                    // fail the whole engine, mirroring the B=1 path's
-                    // poison classification.
-                    fatal = Some(e);
-                    break;
-                }
-            };
+            let nt =
+                match sample_with_penalty(&logits, &slot.generated, &mut slot.lp, &slot.sampling) {
+                    Ok(t) => t,
+                    Err(e) => {
+                        let health = logits_health_slice(&logits_vec);
+                        tracing::warn!(
+                            ?health,
+                            error = %e,
+                            "batch engine: sample failed; logits unhealthy"
+                        );
+                        // Unhealthy logits are a device-level problem —
+                        // fail the whole engine, mirroring the B=1 path's
+                        // poison classification.
+                        fatal = Some(e);
+                        break;
+                    }
+                };
             if Some(nt) == slot.eos_id {
                 finish_slot(slot, FinishReason::Stop, active_rate(&cfg, sess)).await;
                 continue;
@@ -627,6 +647,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 +780,10 @@ 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,
         eos_id,
         tool_schemas,
         tx,
@@ -774,27 +791,17 @@ 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());
 
     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) {
+    let first = match sample_with_penalty(&logits, &generated, &mut lp, &sampling) {
         Ok(t) => t,
         Err(e) => {
             let health = logits_health_slice(&logits_vec);
@@ -827,11 +834,13 @@ 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,
         router: router_tx,
         hangup,
         finished: None,
@@ -890,6 +899,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 +990,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 +1034,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 +1062,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 +1181,17 @@ 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,
+                ),
                 eos_id: None,
                 tool_schemas: ToolSchemas::new(),
                 tx,

crates/neuron/src/harness/mod.rs

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

crates/neuron/src/harness/sampling.rs

@@ -0,0 +1,368 @@
+//! Sampling parameter resolution (#272).
+//!
+//! Three sources, in priority order:
+//!
+//! 1. what the **request** asked for,
+//! 2. what the **model** published in `generation_config.json`,
+//! 3. a built-in fallback.
+//!
+//! 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()
+            }
+        }
+    }
+}
+
+/// 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;
+
+    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));
+    }
+
+    /// 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/mod.rs

@@ -30,6 +30,10 @@ pub mod worker;
 use anyhow::{Context, Result};
 use std::path::{Path, PathBuf};
 use std::process::Stdio;
+#[cfg(any(feature = "cuda", test))]
+use std::sync::Arc;
+#[cfg(any(feature = "cuda", test))]
+use std::sync::atomic::{AtomicBool, Ordering};
 use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, Lines};
 use tokio::process::{Child, ChildStdin, ChildStdout, Command};
 
@@ -319,6 +323,87 @@ pub struct WorkerPool {
     leader_comm: Option<nccl_state::SendComm>,
 }
 
+/// How long the watchdog waits for `ncclCommAbort` to return before
+/// concluding the process is unrecoverable (#265). Generous relative to a
+/// healthy abort (milliseconds) and short relative to a human noticing.
+#[cfg(any(feature = "cuda", test))]
+fn tp_abort_deadline() -> std::time::Duration {
+    let secs = std::env::var("NEURON_TP_ABORT_DEADLINE_S")
+        .ok()
+        .and_then(|v| v.parse::<u64>().ok())
+        .filter(|s| *s > 0)
+        .unwrap_or(60);
+    std::time::Duration::from_secs(secs)
+}
+
+/// Wait for the abort thread to signal completion. `true` if it returned
+/// in time (recovery proceeds normally), `false` if the deadline elapsed
+/// (the process is unrecoverable).
+///
+/// Split out from the deadman thread so the decision is testable without
+/// a GPU: the branch that ends the process is the one most worth
+/// exercising, and the one hardest to reach by accident.
+#[cfg(any(feature = "cuda", test))]
+fn wait_for_abort(done: &AtomicBool, deadline: std::time::Duration) -> bool {
+    let start = std::time::Instant::now();
+    while start.elapsed() < deadline {
+        if done.load(Ordering::Acquire) {
+            return true;
+        }
+        std::thread::sleep(std::time::Duration::from_millis(25));
+    }
+    // One last look: the abort may have landed inside the final sleep.
+    done.load(Ordering::Acquire)
+}
+
+/// Exit non-zero so systemd's `Restart=on-failure` restarts us (#265).
+///
+/// Called only when the process cannot recover on its own: the NCCL
+/// collective is wedged and the abort that would unblock it has itself
+/// hung. The daemon then serves nothing — on beast 2026-08-17 it sat in
+/// that state for 70 minutes with GPUs idle, weights resident, 27
+/// connections queued on the listener, and `systemctl` reporting it
+/// active. A process that knows it is unrecoverable should not wait for a
+/// human.
+///
+/// **Everything diagnostic is logged first.** A restart destroys the
+/// state that explains the hang, so trading a diagnosable outage for an
+/// undiagnosable self-healing one would be a poor bargain.
+///
+/// `process::exit` deliberately skips destructors and the graceful-drain
+/// path. That is normally the wrong instinct, but draining a wedged
+/// process cannot work — and `TimeoutStopSec=120s` plus
+/// `TimeoutStopFailureMode=abort` would add two more minutes of outage on
+/// top of the wedge (#256).
+#[cfg(feature = "cuda")]
+fn exit_unrecoverable(model_id: &str, reason: &str) -> ! {
+    // Read straight from the driver rather than any cached value: the
+    // cache is refreshed by the very machinery that is wedged.
+    let vram = std::process::Command::new("nvidia-smi")
+        .args([
+            "--query-gpu=index,memory.used,memory.free,utilization.gpu",
+            "--format=csv,noheader",
+        ])
+        .output()
+        .ok()
+        .map(|o| String::from_utf8_lossy(&o.stdout).replace('\n', " | "))
+        .unwrap_or_else(|| "unavailable".into());
+    tracing::error!(
+        model = %model_id,
+        reason = %reason,
+        vram = %vram,
+        pid = std::process::id(),
+        "tp watchdog: UNRECOVERABLE — exiting non-zero so systemd restarts this daemon (#265). \
+         The GPU state above is from immediately before the exit; the wedged collective and its \
+         request id are in the preceding watchdog lines."
+    );
+    // Give the logging layer a moment to flush before the process dies —
+    // an exit that eats its own explanation is the failure mode this
+    // whole change exists to avoid.
+    std::thread::sleep(std::time::Duration::from_millis(500));
+    std::process::exit(70) // EX_SOFTWARE
+}
+
 /// Per-step deadline for a TP forward (#17 Stage 2). A healthy decode
 /// step or chunked prefill completes in well under a second; a wedged
 /// NCCL collective never returns. Generous default so no legitimate step
@@ -349,23 +434,66 @@ impl WorkerPool {
             "tp watchdog: leader forward exceeded deadline — NCCL collective wedged; \
              aborting comm to unblock the leader thread for auto-recovery"
         );
-        match &self.leader_comm {
-            Some(c) => match c.0.abort() {
-                Ok(()) => tracing::error!(
-                    model = %model_id,
-                    "tp watchdog: ncclCommAbort succeeded — wedged collective unblocked; \
-                     failing the step so the model auto-recovers (unload+reload)"
-                ),
-                Err(e) => tracing::error!(
-                    model = %model_id, error = ?e,
-                    "tp watchdog: ncclCommAbort failed — recovery may stall until a process restart"
-                ),
-            },
-            None => tracing::error!(
+        let Some(comm) = self.leader_comm.as_ref().map(|c| c.0.clone()) else {
+            tracing::error!(
                 model = %model_id,
-                "tp watchdog: no cached leader comm handle — cannot abort; recovery will rely \
-                 on a process restart"
-            ),
+                "tp watchdog: no cached leader comm handle — cannot abort; exiting so \
+                 systemd restarts us (#265)"
+            );
+            exit_unrecoverable(model_id, "no leader comm handle to abort");
+        };
+
+        // `ncclCommAbort` can itself block forever — observed on beast
+        // 2026-08-17, where neither the success nor the failure branch below
+        // was ever reached and the daemon sat wedged for 70 minutes with 27
+        // connections queued on its listener (#265). So the abort runs on its
+        // own thread and the deadline below does NOT depend on it returning:
+        // an abort that hangs must not take the supervisor with it.
+        let done = Arc::new(AtomicBool::new(false));
+        let signal = Arc::clone(&done);
+        let owned_model = model_id.to_string();
+        std::thread::Builder::new()
+            .name("tp-watchdog-abort".into())
+            .spawn(move || {
+                match comm.abort() {
+                    Ok(()) => tracing::error!(
+                        model = %owned_model,
+                        "tp watchdog: ncclCommAbort succeeded — wedged collective unblocked; \
+                         failing the step so the model auto-recovers (unload+reload)"
+                    ),
+                    Err(e) => tracing::error!(
+                        model = %owned_model, error = ?e,
+                        "tp watchdog: ncclCommAbort failed — the collective is still wedged"
+                    ),
+                }
+                // Returning at all is the signal, success or failure: a
+                // returned abort leaves the process responsive, so
+                // auto-recovery can run and we must not exit under it.
+                signal.store(true, Ordering::Release);
+            })
+            .map(|_| ())
+            .unwrap_or_else(|e| {
+                // Cannot even spawn a thread — nothing here can recover.
+                tracing::error!(model = %model_id, error = %e, "tp watchdog: abort thread spawn failed");
+                exit_unrecoverable(model_id, "could not spawn the abort thread");
+            });
+
+        // Deadman. Disarmed only by the abort returning.
+        let deadline = tp_abort_deadline();
+        let owned_model = model_id.to_string();
+        if let Err(e) = std::thread::Builder::new()
+            .name("tp-watchdog-deadman".into())
+            .spawn(move || {
+                if wait_for_abort(&done, deadline) {
+                    return; // abort came back; the normal path owns recovery
+                }
+                exit_unrecoverable(
+                    &owned_model,
+                    "ncclCommAbort did not return within the deadline",
+                );
+            })
+        {
+            tracing::error!(model = %model_id, error = %e, "tp watchdog: deadman spawn failed");
         }
     }
 
@@ -1442,3 +1570,73 @@ impl WorkerPool {
         &self.exe
     }
 }
+
+#[cfg(test)]
+mod watchdog_tests {
+    use super::*;
+    use std::time::Duration;
+
+    /// An abort that returns disarms the deadman — the normal path then
+    /// owns recovery and the process must NOT exit.
+    #[test]
+    fn abort_returning_disarms_the_deadman() {
+        let done = Arc::new(AtomicBool::new(false));
+        let signal = Arc::clone(&done);
+        std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(50));
+            signal.store(true, Ordering::Release);
+        });
+        assert!(
+            wait_for_abort(&done, Duration::from_secs(5)),
+            "a returning abort must disarm the deadman"
+        );
+    }
+
+    /// The #265 case: `ncclCommAbort` never returns. The deadline must
+    /// elapse and report unrecoverable, which is what triggers the exit.
+    #[test]
+    fn abort_that_never_returns_trips_the_deadline() {
+        let done = Arc::new(AtomicBool::new(false)); // never set
+        let start = std::time::Instant::now();
+        assert!(
+            !wait_for_abort(&done, Duration::from_millis(200)),
+            "an abort that never returns must trip the deadline"
+        );
+        assert!(
+            start.elapsed() >= Duration::from_millis(200),
+            "must actually wait the deadline rather than failing fast"
+        );
+    }
+
+    /// An abort landing in the last instant still counts as returned —
+    /// exiting on a race would restart a process that had just recovered.
+    #[test]
+    fn abort_landing_at_the_deadline_still_counts() {
+        let done = Arc::new(AtomicBool::new(false));
+        let signal = Arc::clone(&done);
+        std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(180));
+            signal.store(true, Ordering::Release);
+        });
+        assert!(
+            wait_for_abort(&done, Duration::from_millis(200)),
+            "an abort landing at the deadline must not be treated as a hang"
+        );
+    }
+
+    /// The deadline is operator-tunable but never zero — a zero deadline
+    /// would exit the moment any watchdog fired.
+    #[test]
+    fn deadline_defaults_sanely_and_rejects_zero() {
+        assert_eq!(tp_abort_deadline(), Duration::from_secs(60));
+        unsafe { std::env::set_var("NEURON_TP_ABORT_DEADLINE_S", "0") };
+        assert_eq!(
+            tp_abort_deadline(),
+            Duration::from_secs(60),
+            "zero must fall back, not exit instantly on every watchdog"
+        );
+        unsafe { std::env::set_var("NEURON_TP_ABORT_DEADLINE_S", "15") };
+        assert_eq!(tp_abort_deadline(), Duration::from_secs(15));
+        unsafe { std::env::remove_var("NEURON_TP_ABORT_DEADLINE_S") };
+    }
+}

crates/neuron/src/main.rs

@@ -186,6 +186,9 @@ async fn daemon(args: Args) -> Result<()> {
     let registry = HarnessRegistry::from_configs(&cfg.harnesses, &bind_url, &cfg.harness);
     discovery_result.harnesses = registry.names();
     let candle = registry.candle();
+    // A second handle for the shutdown future: it must close admission
+    // before the drain begins, and `candle` is moved into NeuronState.
+    let shutdown_candle = candle.clone();
 
     let health_cache = Arc::new(health::HealthCache::new());
     health_cache
@@ -246,7 +249,7 @@ async fn daemon(args: Args) -> Result<()> {
     }
 
     axum::serve(listener, app)
-        .with_graceful_shutdown(startup::shutdown_signal())
+        .with_graceful_shutdown(startup::shutdown_signal(shutdown_candle))
         .await?;
 
     // Deactivation: serve has returned (graceful shutdown signal

crates/neuron/src/startup.rs

@@ -9,6 +9,7 @@ use crate::activation::ActivationTracker;
 use crate::harness::HarnessRegistry;
 use crate::harness::preflight::PreflightError;
 use cortex_core::harness::ModelSpec;
+use std::sync::Arc;
 use std::time::{Duration, Instant};
 use tokio::signal;
 
@@ -165,7 +166,18 @@ fn preflight_kind(err: &PreflightError) -> &'static str {
 /// Wired into `axum::serve(...).with_graceful_shutdown(shutdown_signal())`
 /// so the HTTP listener stops accepting new connections, lets in-flight
 /// requests drain, and then yields control back to main for cleanup.
-pub async fn shutdown_signal() {
+///
+/// **Before returning, it closes admission and wakes every waiter (#256).**
+/// Order matters: axum's graceful shutdown waits for *all* in-flight
+/// requests, and a request queued behind the TP pool lock is in-flight as
+/// far as axum is concerned. Those waits are unbounded — held by whichever
+/// request is mid-forward — so without this the drain cannot finish and
+/// systemd SIGKILLs at `TimeoutStopSec`. Observed 2026-08-15: SIGTERM at
+/// 13:03:14, one queued request logging "still waiting on pool lock" every
+/// two seconds, `Failed with result 'timeout'` at 13:05:24. The same node
+/// drained in about a second when idle, which is what identified the
+/// queued waiter as the cause.
+pub async fn shutdown_signal(candle: Option<Arc<crate::harness::candle::CandleHarness>>) {
     let ctrl_c = async {
         signal::ctrl_c().await.ok();
     };
@@ -179,6 +191,13 @@ pub async fn shutdown_signal() {
         _ = ctrl_c => tracing::info!("received SIGINT, shutting down"),
         _ = terminate => tracing::info!("received SIGTERM, shutting down"),
     }
+    // Refuse new admissions and abandon the unbounded waits, so the drain
+    // that starts the moment this future resolves has a bounded amount of
+    // work left to wait for.
+    crate::harness::candle::begin_shutdown();
+    if let Some(candle) = candle {
+        candle.close_all_admission().await;
+    }
 }
 
 /// Unload every model currently registered. Called from `main.rs` after

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,
+    ChatCompletionChunk, ChunkChoice, CompletionTokensDetails, HelexaTiming, 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.
@@ -405,6 +485,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -443,6 +524,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 +549,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 +637,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -500,10 +650,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 +675,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 +734,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -603,6 +787,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -649,6 +834,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -667,10 +853,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 +881,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -721,6 +910,7 @@ mod tests {
             prompt_tokens: 42,
             completion_tokens: 5,
             reasoning_tokens: 2,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -755,6 +945,7 @@ mod tests {
             prompt_tokens: 10,
             completion_tokens: 7,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await

crates/neuron/src/wire/openai_responses.rs

@@ -33,9 +33,9 @@
 
 use cortex_core::openai::{ChatCompletionRequest, ChatMessage, MessageContent};
 use cortex_core::responses::{
-    OutputTokensDetails, ResponsesContentPart, ResponsesInput, ResponsesInputElement,
-    ResponsesInputItem, ResponsesMessageContent, ResponsesOutputContent, ResponsesOutputItem,
-    ResponsesRequest, ResponsesResponse, ResponsesUsage, events,
+    InputTokensDetails, OutputTokensDetails, ResponsesContentPart, ResponsesInput,
+    ResponsesInputElement, ResponsesInputItem, ResponsesMessageContent, ResponsesOutputContent,
+    ResponsesOutputItem, ResponsesRequest, ResponsesResponse, ResponsesUsage, events,
 };
 use serde_json::{Value, json};
 use tokio::sync::mpsc;
@@ -156,6 +156,10 @@ pub fn request_to_chat(req: ResponsesRequest) -> Result<ChatCompletionRequest, T
         messages,
         temperature: req.temperature,
         top_p: req.top_p,
+        top_k: req.top_k,
+        seed: req.seed,
+        repetition_penalty: req.repetition_penalty,
+        repeat_last_n: req.repeat_last_n,
         max_tokens: req.max_output_tokens,
         // Responses spells the cap `max_output_tokens`, already
         // resolved onto the legacy field above.
@@ -365,6 +369,12 @@ async fn run_projection(
     // final `response.completed` payload so clients that only read
     // the terminal response still see the calls.
     let mut tool_items: Vec<Value> = Vec::new();
+    // Reasoning item state (#267) and whether the message item's
+    // opening frames have gone out. The message is opened lazily,
+    // because its `output_index` depends on whether a reasoning item
+    // claimed index 0 first.
+    let mut reasoning = ReasoningTracker::default();
+    let mut message_open = false;
 
     while let Some(event) = rx.recv().await {
         match event {
@@ -378,12 +388,22 @@ async fn run_projection(
                 if text.is_empty() {
                     continue;
                 }
+                // Thinking is over the moment visible text starts.
+                if !close_reasoning(&tx, &meta, &mut reasoning).await {
+                    return;
+                }
+                if !message_open {
+                    if !emit_message_open_frames(&tx, &meta, reasoning.message_index()).await {
+                        return;
+                    }
+                    message_open = true;
+                }
                 accumulated.push_str(&text);
                 let frame = ResponseStreamFrame {
                     event_name: events::OUTPUT_TEXT_DELTA,
                     data: json!({
                         "item_id": meta.message_item_id,
-                        "output_index": 0,
+                        "output_index": reasoning.message_index(),
                         "content_index": 0,
                         "delta": text,
                     }),
@@ -392,10 +412,45 @@ async fn run_projection(
                     return;
                 }
             }
-            InferenceEvent::ReasoningDelta(_) => {
-                // No representation in our Responses model yet.
-                // Stage where it'd land: a `response.reasoning_*`
-                // event family alongside `response.output_text.*`.
+            InferenceEvent::ReasoningDelta(text) => {
+                // Stream the think block as its own output item (#267).
+                // This used to be dropped on the floor, which meant a
+                // model that reasoned for minutes emitted no events at
+                // all and clients closed the stream as dead — dsh
+                // reports `pi-ai stream idle timeout after 300000ms`.
+                // Its watchdog resets on any parsed event, so surfacing
+                // reasoning is what keeps a thinking model alive on the
+                // wire; SSE comment keep-alives cannot, because idle
+                // timers count events, not comments.
+                if text.is_empty() {
+                    continue;
+                }
+                if !reasoning.open {
+                    if reasoning.item.is_some() {
+                        // Reasoning resumed after visible text. Our
+                        // model has one reasoning item per response, so
+                        // fold it into the existing summary rather than
+                        // opening a second item out of order.
+                        continue;
+                    }
+                    if !emit_reasoning_open_frames(&tx, &meta).await {
+                        return;
+                    }
+                    reasoning.open = true;
+                }
+                reasoning.text.push_str(&text);
+                let frame = ResponseStreamFrame {
+                    event_name: events::REASONING_SUMMARY_TEXT_DELTA,
+                    data: json!({
+                        "item_id": reasoning_item_id(&meta),
+                        "output_index": 0,
+                        "summary_index": 0,
+                        "delta": text,
+                    }),
+                };
+                if tx.send(frame).await.is_err() {
+                    return;
+                }
             }
             InferenceEvent::ToolCall {
                 index,
@@ -403,9 +458,19 @@ async fn run_projection(
                 name,
                 arguments,
             } => {
-                // The message item opened by the start frames holds
-                // output_index 0; function_call items follow it.
-                let output_index = 1 + index as u64;
+                if !close_reasoning(&tx, &meta, &mut reasoning).await {
+                    return;
+                }
+                if !message_open {
+                    if !emit_message_open_frames(&tx, &meta, reasoning.message_index()).await {
+                        return;
+                    }
+                    message_open = true;
+                }
+                // function_call items follow the message item, which
+                // itself follows any reasoning item — so the base is the
+                // message's index, not a hardcoded 0.
+                let output_index = reasoning.message_index() + 1 + index as u64;
                 let item_id = format!(
                     "fc_{}_{index}",
                     meta.response_id.trim_start_matches("resp_")
@@ -479,6 +544,7 @@ async fn run_projection(
                 prompt_tokens,
                 completion_tokens,
                 reasoning_tokens,
+                cached_tokens,
                 // Responses-side `helexa_timing` surfacing not wired yet;
                 // the bench harness reads timing off the chat path (#85).
                 timing: _,
@@ -495,7 +561,12 @@ async fn run_projection(
                     output_tokens_details: (reasoning_tokens > 0).then_some(OutputTokensDetails {
                         reasoning_tokens: reasoning_tokens as u64,
                     }),
-                    input_tokens_details: None,
+                    // Prefix-cache reuse (#269). Omitted at zero so a
+                    // client sees unchanged JSON when nothing was
+                    // cached.
+                    input_tokens_details: (cached_tokens > 0).then_some(InputTokensDetails {
+                        cached_tokens: cached_tokens as u64,
+                    }),
                 });
             }
         }
@@ -508,18 +579,158 @@ async fn run_projection(
         return;
     }
 
+    // A stream can end while still thinking — a reasoning model that
+    // exhausts max_output_tokens mid-block never emits visible text.
+    // Close the item so the client sees a finished response rather than
+    // a reasoning item left open forever.
+    if !close_reasoning(&tx, &meta, &mut reasoning).await {
+        return;
+    }
+    // The message item is opened lazily, so a response that produced no
+    // text (tool-call-only turns, or the case above) has not announced
+    // it yet. The finish frames reference it, so it must exist first.
+    if !message_open && !emit_message_open_frames(&tx, &meta, reasoning.message_index()).await {
+        return;
+    }
+
     let reason = finish.unwrap_or(FinishReason::Stop);
     let _ = emit_finish_frames(
         &tx,
         &meta,
-        &accumulated,
-        reason,
-        usage.as_ref(),
-        &tool_items,
+        FinishContext {
+            full_text: &accumulated,
+            reason,
+            usage: usage.as_ref(),
+            tool_items: &tool_items,
+            message_index: reasoning.message_index(),
+            reasoning_item: reasoning.item.as_ref(),
+        },
     )
     .await;
 }
 
+/// Streaming state for a reasoning model's think block (#267).
+///
+/// The block is its own output item, ahead of the message, so the
+/// message's `output_index` depends on whether reasoning happened —
+/// which is only known once the first token arrives. This tracks that
+/// decision and the accumulated summary text.
+#[derive(Default)]
+struct ReasoningTracker {
+    /// Frames opening the reasoning item have been emitted.
+    open: bool,
+    /// Everything the model has thought so far, for the `*.done` frames
+    /// and the terminal `completed` payload.
+    text: String,
+    /// The finished reasoning item, replayed inside `response.completed`
+    /// so a client that only reads the terminal response still sees it.
+    item: Option<Value>,
+}
+
+impl ReasoningTracker {
+    /// `output_index` the message item takes: 1 when a reasoning item
+    /// occupies 0, else 0.
+    fn message_index(&self) -> u64 {
+        u64::from(self.open || self.item.is_some())
+    }
+}
+
+fn reasoning_item_id(meta: &ResponseMeta) -> String {
+    format!("rs_{}", meta.response_id.trim_start_matches("resp_"))
+}
+
+/// Open the reasoning item and its summary part, once.
+async fn emit_reasoning_open_frames(
+    tx: &mpsc::Sender<ResponseStreamFrame>,
+    meta: &ResponseMeta,
+) -> bool {
+    let item_id = reasoning_item_id(meta);
+    let frames = [
+        ResponseStreamFrame {
+            event_name: events::OUTPUT_ITEM_ADDED,
+            data: json!({
+                "output_index": 0,
+                "item": {
+                    "type": "reasoning",
+                    "id": item_id,
+                    "summary": [],
+                    "status": "in_progress",
+                },
+            }),
+        },
+        ResponseStreamFrame {
+            event_name: events::REASONING_SUMMARY_PART_ADDED,
+            data: json!({
+                "item_id": item_id,
+                "output_index": 0,
+                "summary_index": 0,
+                "part": { "type": "summary_text", "text": "" },
+            }),
+        },
+    ];
+    for frame in frames {
+        if tx.send(frame).await.is_err() {
+            return false;
+        }
+    }
+    true
+}
+
+/// Close the reasoning item, if one was opened, and record it for the
+/// terminal payload. Idempotent — every path that emits a non-reasoning
+/// item calls this first, and only the first call does anything.
+async fn close_reasoning(
+    tx: &mpsc::Sender<ResponseStreamFrame>,
+    meta: &ResponseMeta,
+    reasoning: &mut ReasoningTracker,
+) -> bool {
+    if !reasoning.open {
+        return true;
+    }
+    reasoning.open = false;
+    let item_id = reasoning_item_id(meta);
+    let full_item = json!({
+        "type": "reasoning",
+        "id": item_id,
+        "summary": [{ "type": "summary_text", "text": reasoning.text }],
+        "status": "completed",
+    });
+    let frames = [
+        ResponseStreamFrame {
+            event_name: events::REASONING_SUMMARY_TEXT_DONE,
+            data: json!({
+                "item_id": item_id,
+                "output_index": 0,
+                "summary_index": 0,
+                "text": reasoning.text,
+            }),
+        },
+        ResponseStreamFrame {
+            event_name: events::REASONING_SUMMARY_PART_DONE,
+            data: json!({
+                "item_id": item_id,
+                "output_index": 0,
+                "summary_index": 0,
+                "part": { "type": "summary_text", "text": reasoning.text },
+            }),
+        },
+        ResponseStreamFrame {
+            event_name: events::OUTPUT_ITEM_DONE,
+            data: json!({
+                "output_index": 0,
+                "item": full_item.clone(),
+            }),
+        },
+    ];
+    reasoning.item = Some(full_item);
+    for frame in frames {
+        if tx.send(frame).await.is_err() {
+            return false;
+        }
+    }
+    true
+}
+
 async fn emit_start_frames(tx: &mpsc::Sender<ResponseStreamFrame>, meta: &ResponseMeta) -> bool {
     let shell = response_shell(meta, "in_progress", &[], None);
     let frames = [
@@ -540,10 +751,31 @@ async fn emit_start_frames(tx: &mpsc::Sender<ResponseStreamFrame>, meta: &Respon
             event_name: events::IN_PROGRESS,
             data: json!({ "response": shell }),
         },
+    ];
+    for frame in frames {
+        if tx.send(frame).await.is_err() {
+            return false;
+        }
+    }
+    true
+}
+
+/// Open the message item, once, at `output_index`.
+///
+/// Deferred rather than emitted with the start frames because the
+/// message is no longer guaranteed to be `output[0]`: a reasoning model
+/// puts its think block in an item ahead of it, and an item's index has
+/// to be right the first time it is announced.
+async fn emit_message_open_frames(
+    tx: &mpsc::Sender<ResponseStreamFrame>,
+    meta: &ResponseMeta,
+    output_index: u64,
+) -> bool {
+    let frames = [
         ResponseStreamFrame {
             event_name: events::OUTPUT_ITEM_ADDED,
             data: json!({
-                "output_index": 0,
+                "output_index": output_index,
                 "item": empty_message_item(&meta.message_item_id),
             }),
         },
@@ -551,7 +783,7 @@ async fn emit_start_frames(tx: &mpsc::Sender<ResponseStreamFrame>, meta: &Respon
             event_name: events::CONTENT_PART_ADDED,
             data: json!({
                 "item_id": meta.message_item_id,
-                "output_index": 0,
+                "output_index": output_index,
                 "content_index": 0,
                 "part": { "type": "output_text", "text": "", "annotations": [] },
             }),
@@ -565,14 +797,34 @@ async fn emit_start_frames(tx: &mpsc::Sender<ResponseStreamFrame>, meta: &Respon
     true
 }
 
+/// Everything the terminal frames need that isn't stream identity.
+/// Grouped rather than passed loose: the reasoning item (#267) pushed
+/// the argument list past the point where positional `&str`/`u64`
+/// parameters are safe to read at a call site.
+struct FinishContext<'a> {
+    full_text: &'a str,
+    reason: FinishReason,
+    usage: Option<&'a ResponsesUsage>,
+    tool_items: &'a [Value],
+    /// Where the message item sits — 1 when a reasoning item took 0.
+    message_index: u64,
+    /// The completed reasoning item, replayed inside `completed`.
+    reasoning_item: Option<&'a Value>,
+}
+
 async fn emit_finish_frames(
     tx: &mpsc::Sender<ResponseStreamFrame>,
     meta: &ResponseMeta,
-    full_text: &str,
-    reason: FinishReason,
-    usage: Option<&ResponsesUsage>,
-    tool_items: &[Value],
+    ctx: FinishContext<'_>,
 ) -> bool {
+    let FinishContext {
+        full_text,
+        reason,
+        usage,
+        tool_items,
+        message_index,
+        reasoning_item,
+    } = ctx;
     let status = finish_to_status(reason);
     let full_part = json!({
         "type": "output_text",
@@ -586,10 +838,14 @@ async fn emit_finish_frames(
         "content": [full_part.clone()],
         "status": status,
     });
-    // Terminal output array: the message item (always present — its
-    // open frames were emitted eagerly at stream start) followed by
-    // any completed function_call items.
-    let mut output_items = Vec::with_capacity(1 + tool_items.len());
+    // Terminal output array, in wire order: the reasoning item (when
+    // the model thought), then the message, then any completed
+    // function_call items. Order matters — a client reconstructing the
+    // response from `completed` alone should see what the stream showed.
+    let mut output_items = Vec::with_capacity(2 + tool_items.len());
+    if let Some(item) = reasoning_item {
+        output_items.push(item.clone());
+    }
     output_items.push(full_item.clone());
     output_items.extend(tool_items.iter().cloned());
     let frames = [
@@ -597,7 +853,7 @@ async fn emit_finish_frames(
             event_name: events::OUTPUT_TEXT_DONE,
             data: json!({
                 "item_id": meta.message_item_id,
-                "output_index": 0,
+                "output_index": message_index,
                 "content_index": 0,
                 "text": full_text,
             }),
@@ -606,7 +862,7 @@ async fn emit_finish_frames(
             event_name: events::CONTENT_PART_DONE,
             data: json!({
                 "item_id": meta.message_item_id,
-                "output_index": 0,
+                "output_index": message_index,
                 "content_index": 0,
                 "part": full_part,
             }),
@@ -614,7 +870,7 @@ async fn emit_finish_frames(
         ResponseStreamFrame {
             event_name: events::OUTPUT_ITEM_DONE,
             data: json!({
-                "output_index": 0,
+                "output_index": message_index,
                 "item": full_item.clone(),
             }),
         },
@@ -756,6 +1012,10 @@ mod tests {
             max_output_tokens: None,
             temperature: None,
             top_p: None,
+            top_k: None,
+            seed: None,
+            repetition_penalty: None,
+            repeat_last_n: None,
             previous_response_id: None,
             extra: Value::Object(Default::default()),
         };
@@ -778,6 +1038,10 @@ mod tests {
             max_output_tokens: None,
             temperature: None,
             top_p: None,
+            top_k: None,
+            seed: None,
+            repetition_penalty: None,
+            repeat_last_n: None,
             previous_response_id: None,
             extra: Value::Object(Default::default()),
         };
@@ -801,6 +1065,10 @@ mod tests {
             max_output_tokens: None,
             temperature: None,
             top_p: None,
+            top_k: None,
+            seed: None,
+            repetition_penalty: None,
+            repeat_last_n: None,
             previous_response_id: Some("resp_prev".into()),
             extra: Value::Object(Default::default()),
         };
@@ -833,6 +1101,10 @@ mod tests {
             max_output_tokens: None,
             temperature: None,
             top_p: None,
+            top_k: None,
+            seed: None,
+            repetition_penalty: None,
+            repeat_last_n: None,
             previous_response_id: None,
             extra: Value::Object(Default::default()),
         };
@@ -863,6 +1135,10 @@ mod tests {
             max_output_tokens: None,
             temperature: None,
             top_p: None,
+            top_k: None,
+            seed: None,
+            repetition_penalty: None,
+            repeat_last_n: None,
             previous_response_id: None,
             extra: Value::Object(Default::default()),
         };
@@ -908,6 +1184,10 @@ mod tests {
             max_output_tokens: None,
             temperature: None,
             top_p: None,
+            top_k: None,
+            seed: None,
+            repetition_penalty: None,
+            repeat_last_n: None,
             previous_response_id: None,
             extra: Value::Object(Default::default()),
         };
@@ -945,6 +1225,10 @@ mod tests {
             max_output_tokens: None,
             temperature: None,
             top_p: None,
+            top_k: None,
+            seed: None,
+            repetition_penalty: None,
+            repeat_last_n: None,
             previous_response_id: None,
             extra: Value::Object(Default::default()),
         };
@@ -971,6 +1255,10 @@ mod tests {
             max_output_tokens: None,
             temperature: None,
             top_p: None,
+            top_k: None,
+            seed: None,
+            repetition_penalty: None,
+            repeat_last_n: None,
             previous_response_id: None,
             extra: Value::Object(Default::default()),
         };
@@ -1095,6 +1383,208 @@ mod tests {
         out
     }
 
+    /// The regression #267 was filed for: a reasoning model must put
+    /// events on the wire while it thinks.
+    ///
+    /// These deltas used to be dropped, so a model that reasoned for
+    /// minutes emitted nothing at all and clients closed the stream as
+    /// dead — dsh reports `pi-ai stream idle timeout after 300000ms`
+    /// and retries, so a task whose thinking outlasts the client's idle
+    /// timeout could never finish. Its watchdog resets on any parsed
+    /// event, which is why emitting these fixes it and why SSE comment
+    /// keep-alives could not: idle timers count events, not comments.
+    /// The Responses surface reports cache reuse too (#269), in its own
+    /// spelling: `input_tokens_details.cached_tokens`.
+    ///
+    /// This is the surface dsh uses, and the one that showed
+    /// `Cache hit 0%` against a 42.1K-token context while neuron was
+    /// reusing thousands of tokens per prefill.
+    #[tokio::test]
+    async fn cached_tokens_reach_the_completed_usage() {
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(8);
+        let out = project_responses_stream(rx, meta());
+
+        tx.send(InferenceEvent::Start).await.unwrap();
+        tx.send(InferenceEvent::TextDelta("ok".into()))
+            .await
+            .unwrap();
+        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 frames = collect(out).await;
+        let usage = &frames.last().unwrap().data["response"]["usage"];
+        assert_eq!(usage["input_tokens_details"]["cached_tokens"], 2068);
+        assert_eq!(usage["input_tokens"], 7413);
+        // Sub-count, not additive — a cache hit must not inflate the total.
+        assert_eq!(usage["total_tokens"], 7425);
+    }
+
+    #[tokio::test]
+    async fn reasoning_projects_a_full_item_lifecycle_before_the_message() {
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(16);
+        let out = project_responses_stream(rx, meta());
+
+        tx.send(InferenceEvent::Start).await.unwrap();
+        tx.send(InferenceEvent::ReasoningDelta("think".into()))
+            .await
+            .unwrap();
+        tx.send(InferenceEvent::ReasoningDelta("ing".into()))
+            .await
+            .unwrap();
+        tx.send(InferenceEvent::TextDelta("answer".into()))
+            .await
+            .unwrap();
+        tx.send(InferenceEvent::Finish {
+            reason: FinishReason::Stop,
+            prompt_tokens: 0,
+            completion_tokens: 0,
+            reasoning_tokens: 2,
+            cached_tokens: 0,
+            timing: None,
+        })
+        .await
+        .unwrap();
+        drop(tx);
+
+        let frames = collect(out).await;
+        let names: Vec<&str> = frames.iter().map(|f| f.event_name).collect();
+        assert_eq!(
+            names,
+            vec![
+                events::CREATED,
+                events::IN_PROGRESS,
+                events::OUTPUT_ITEM_ADDED,
+                events::REASONING_SUMMARY_PART_ADDED,
+                events::REASONING_SUMMARY_TEXT_DELTA,
+                events::REASONING_SUMMARY_TEXT_DELTA,
+                events::REASONING_SUMMARY_TEXT_DONE,
+                events::REASONING_SUMMARY_PART_DONE,
+                events::OUTPUT_ITEM_DONE,
+                events::OUTPUT_ITEM_ADDED,
+                events::CONTENT_PART_ADDED,
+                events::OUTPUT_TEXT_DELTA,
+                events::OUTPUT_TEXT_DONE,
+                events::CONTENT_PART_DONE,
+                events::OUTPUT_ITEM_DONE,
+                events::COMPLETED,
+            ]
+        );
+
+        // The reasoning item owns output_index 0, so the message shifts
+        // to 1 — an item's index has to be right when it is announced,
+        // which is why the message is opened lazily.
+        let reasoning_added = &frames[2];
+        assert_eq!(reasoning_added.data["output_index"], 0);
+        assert_eq!(reasoning_added.data["item"]["type"], "reasoning");
+        let message_added = &frames[9];
+        assert_eq!(message_added.data["output_index"], 1);
+        assert_eq!(message_added.data["item"]["type"], "message");
+        assert_eq!(frames[11].data["output_index"], 1, "text delta index");
+
+        // Accumulated thinking, not just the last fragment.
+        assert_eq!(frames[6].data["text"], "thinking");
+
+        // The terminal payload replays both items, in wire order, so a
+        // client reading only `completed` sees what the stream showed.
+        let output = frames.last().unwrap().data["response"]["output"]
+            .as_array()
+            .expect("output array");
+        assert_eq!(output.len(), 2);
+        assert_eq!(output[0]["type"], "reasoning");
+        assert_eq!(output[0]["summary"][0]["text"], "thinking");
+        assert_eq!(output[1]["type"], "message");
+    }
+
+    /// A model can exhaust its budget mid-thought and never produce
+    /// visible text. The reasoning item must still be closed and the
+    /// message still opened, or the client waits on a response that
+    /// never resolves.
+    #[tokio::test]
+    async fn reasoning_only_stream_still_closes_cleanly() {
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(16);
+        let out = project_responses_stream(rx, meta());
+
+        tx.send(InferenceEvent::Start).await.unwrap();
+        tx.send(InferenceEvent::ReasoningDelta("pondering".into()))
+            .await
+            .unwrap();
+        tx.send(InferenceEvent::Finish {
+            reason: FinishReason::Length,
+            prompt_tokens: 0,
+            completion_tokens: 0,
+            reasoning_tokens: 1,
+            cached_tokens: 0,
+            timing: None,
+        })
+        .await
+        .unwrap();
+        drop(tx);
+
+        let frames = collect(out).await;
+        let names: Vec<&str> = frames.iter().map(|f| f.event_name).collect();
+        assert!(names.contains(&events::REASONING_SUMMARY_PART_DONE));
+        assert!(
+            names.contains(&events::CONTENT_PART_ADDED),
+            "message item must still be opened so the finish frames refer to something"
+        );
+        assert_eq!(names.last(), Some(&events::COMPLETED));
+        let last = frames.last().unwrap();
+        assert_eq!(last.data["response"]["status"], "incomplete");
+    }
+
+    /// Tool calls sit after the message, which sits after any reasoning
+    /// item — so their index is relative to the message, not a
+    /// hardcoded 0.
+    #[tokio::test]
+    async fn tool_call_index_shifts_past_a_reasoning_item() {
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(16);
+        let out = project_responses_stream(rx, meta());
+
+        tx.send(InferenceEvent::Start).await.unwrap();
+        tx.send(InferenceEvent::ReasoningDelta("plan".into()))
+            .await
+            .unwrap();
+        tx.send(InferenceEvent::ToolCall {
+            index: 0,
+            id: "call_1".into(),
+            name: "bash".into(),
+            arguments: "{}".into(),
+        })
+        .await
+        .unwrap();
+        tx.send(InferenceEvent::Finish {
+            reason: FinishReason::ToolCalls,
+            prompt_tokens: 0,
+            completion_tokens: 0,
+            reasoning_tokens: 1,
+            cached_tokens: 0,
+            timing: None,
+        })
+        .await
+        .unwrap();
+        drop(tx);
+
+        let frames = collect(out).await;
+        let call_added = frames
+            .iter()
+            .find(|f| {
+                f.event_name == events::OUTPUT_ITEM_ADDED
+                    && f.data["item"]["type"] == "function_call"
+            })
+            .expect("function_call item announced");
+        // reasoning=0, message=1, first call=2.
+        assert_eq!(call_added.data["output_index"], 2);
+    }
+
     #[tokio::test]
     async fn full_stream_emits_expected_event_sequence() {
         let (tx, rx) = mpsc::channel::<InferenceEvent>(8);
@@ -1112,6 +1602,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -1173,6 +1664,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -1245,6 +1737,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -1274,6 +1767,7 @@ mod tests {
             prompt_tokens: 30,
             completion_tokens: 12,
             reasoning_tokens: 4,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -1305,6 +1799,7 @@ mod tests {
             prompt_tokens: 8,
             completion_tokens: 3,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -1330,6 +1825,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await
@@ -1377,6 +1873,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            cached_tokens: 0,
             timing: None,
         })
         .await

data/neuron.service

@@ -2,6 +2,22 @@
 Description=Neuron — per-node GPU discovery and harness daemon for cortex
 After=network-online.target
 Wants=network-online.target
+# Start-limit bound for the self-restart path (#265). When a TP forward
+# wedges and the NCCL abort that would unblock it also hangs, neuron
+# exits 70 so Restart=on-failure recovers it in seconds instead of the
+# 70-minute outage that needed a human on 2026-08-17.
+#
+# That only helps if the wedge is transient. A deterministic one — a bad
+# build, a sick card, a model that wedges on load — would otherwise
+# restart forever, and a flapping node is WORSE than a down one: cortex
+# marks it healthy between restarts and keeps routing to it, so every
+# client sees intermittent failures instead of one node cleanly out.
+#
+# Four failures in ten minutes stops the unit and leaves it `failed`,
+# where `systemctl status` and an alert can see it. Visibly broken beats
+# invisibly cycling.
+StartLimitIntervalSec=600
+StartLimitBurst=4
 
 [Service]
 Type=simple
@@ -24,6 +40,11 @@ TimeoutStartSec=1800s
 # On stop, neuron drains in-flight requests then unloads every model
 # to release CUDA contexts cleanly. Allow generous time for big-model
 # unloads; systemd will SIGKILL after this bound.
+#
+# Note this bound does NOT apply to the self-restart path (#265): a
+# wedged process exits directly rather than draining, precisely because
+# draining cannot work when the collective is stuck — and waiting this
+# out would add two minutes to an outage that is already total.
 TimeoutStopSec=120s
 KillSignal=SIGTERM
 

helexa.ai/content/docs/2-operating/01-install.md

@@ -19,7 +19,7 @@ gateway.
 
 ```
                  ┌──────────────┐
-   clients ─────▶│    cortex    │   :31313 API, :31314 metrics
+   clients ────▶│    cortex    │   :31313 API, :31314 metrics
                  └──┬────┬────┬─┘
                     │    │    │
                ┌────▼┐ ┌─▼──┐ ┌▼────┐

helexa.ai/content/docs/2-operating/03-neurons.md

@@ -108,6 +108,43 @@ KV cache for several concurrent long contexts is itself substantial.
 `max_per_principal` is what stops one busy client consuming the whole
 model.
 
+### Anonymous callers
+
+Fair share is keyed on the account and key cortex resolves from the
+bearer token. A caller that sends no credential — or one that does not
+resolve, which `require_auth = false` deliberately tolerates — has no
+key to be shared out, so `max_per_principal` cannot bind it.
+
+Anonymous traffic is therefore served from the capacity left over once
+identified traffic is satisfied. It never takes a seat while an
+authenticated request is waiting for one, whoever arrived first, and it
+cannot hold every seat at once:
+
+```toml
+anon_max_in_flight = 7   # unset: max_in_flight - 1, floored at 1
+anon_max_pending = 15    # unset: max_in_flight + max_queue_depth - 1
+```
+
+The defaults hold back one seat and one queue place, which is what
+bounds how long an authenticated request can be made to wait: a request
+already running cannot be preempted, so without a reserved seat an
+anonymous burst arriving during an idle moment locks the model for as
+long as all of it takes to finish. On a single-seat model the floor
+keeps anonymous callers served — priority alone decides who wins there.
+
+Set `anon_max_in_flight = 0` to refuse anonymous traffic outright, or
+`require_auth = true` on the gateway to make attribution universal.
+Anonymous callers still contend with each other, and under sustained
+authenticated load they are refused rather than served slowly — the
+refusal is a retryable `503`, so a client that backs off gets served
+when the load passes.
+
+Watch it with `cortex_model_anon_in_flight` against
+`cortex_model_in_flight` (how much of a model's load is unattributable)
+and `cortex_model_rejections_total{reason="anon_yield"}`. That last
+counter rising while `reason="wait_timeout"` stays flat means the
+reservation is doing its job, not that the model is overloaded.
+
 ## VRAM guards
 
 neuron refuses work it cannot finish rather than crashing partway: