fix(neuron): pass the model id into the worker streaming path

The metrics call added in the previous commit referenced `model_id` in
`stream_inference_via_worker`, which has no such binding: that function
gets the model name from its tracing span, and a span field is not a
value. Caught by the CUDA type-check, which is the only job that
compiles either streaming path — both are `#[cfg(feature = "cuda")]`
and neither builds locally.

Threaded through as an owned `String` (the function runs in a spawned
task) following the `_inner` clone pattern the call site already uses
for the other captured state.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015v1FQqShiRp9rrJEEcgNER

Cargo.lock

@@ -3149,6 +3149,8 @@ dependencies = [
  "image",
  "libc",
  "memmap2",
+ "metrics",
+ "metrics-exporter-prometheus",
  "minijinja",
  "minijinja-contrib",
  "rayon",

asset/monitoring/README.md

@@ -6,10 +6,18 @@ bench UI shows.
 
 ## Topology
 
-- **cortex** runs on `hanzalova.internal` (10.6.0.46) and is the **only**
-  Prometheus target — it exposes every `cortex_*` metric on `:31314`, already
-  labelled by `{node,model}` / `{node,device}` from its neuron poller. neuron
-  has no `/metrics` endpoint.
+- **cortex** runs on `hanzalova.internal` (10.6.0.46) and exposes every
+  `cortex_*` metric on `:31314`, already labelled by `{node,model}` /
+  `{node,device}` from its neuron poller. This is the fleet-shaped view:
+  who is loaded where, how busy, how much VRAM.
+- **neuron** additionally serves `/metrics` on its own API port
+  (`:13131`) on each GPU host. This is the request-shaped view — where a
+  request's time actually went, as histograms rather than polled gauges.
+  It exists because cortex's numbers are a ~10s poll of `/health`
+  republished as gauges, which cannot see inside a one-second forward and
+  carries no distribution. Scrape job: `prometheus-neuron.scrape.yml`.
+  **No firewalld rule needed** — 13131 is already open for the control
+  plane (`data/neuron-firewalld.xml`).
 - **Prometheus + Grafana** run on `golgafrinchans.kosherinata.internal` as
   podman quadlets. Host ports (from the registered band — 9090 is Cockpit,
   3000 was avoided): **Prometheus `:26559`** (container config bind-mounted at

asset/monitoring/prometheus-cortex.scrape.yml

@@ -4,9 +4,12 @@
 # Cockpit on that box):  curl -X POST http://localhost:26559/-/reload
 #
 # cortex runs on hanzalova.internal (10.6.0.46) and exposes ALL fleet metrics
-# on :31314 — it is the only Prometheus target in the stack (neuron has no
-# /metrics; cortex's poller already relabels everything by {node,model} /
-# {node,device}). Reaching it cross-DC (kosherinata -> hanzalova) requires the
+# on :31314; cortex's poller already relabels everything by {node,model} /
+# {node,device}. It is no longer the only target — neuron now serves its own
+# /metrics on :13131 with the per-request phase histograms cortex's ~10s
+# gauge poll cannot express (see prometheus-neuron.scrape.yml); the two are
+# complementary, not redundant.
+# Reaching cortex cross-DC (kosherinata -> hanzalova) requires the
 # firewalld rule in asset/monitoring/firewalld-cortex-metrics.sh, which opens
 # :31314 to this host's mesh IP (10.3.101.4) only.
 #

asset/monitoring/prometheus-neuron.scrape.yml

@@ -0,0 +1,40 @@
+# helexa neuron scrape job — append into the `scrape_configs:` list of
+# /etc/prometheus/prometheus.yml on golgafrinchans.kosherinata.internal,
+# then reload:  curl -X POST http://localhost:26559/-/reload
+#
+# WHY A SECOND TARGET, when cortex already publishes fleet metrics.
+#
+# cortex's numbers come from polling each neuron's /health every ~10s and
+# republishing it as gauges. That is the right shape for routing decisions
+# and the wrong shape for asking where a request's time went: an EMA
+# sampled at 10s cannot see inside a one-second forward, and gauges carry
+# no distribution. The finest split cortex can express is prefill vs
+# decode.
+#
+# neuron serves /metrics on its own API port (13131) — the port is already
+# open in data/neuron-firewalld.xml for the cortex control plane, so unlike
+# the cortex scrape this needs NO new firewalld rule.
+#
+# What is here that cortex cannot give you:
+#   neuron_decode_phase_seconds{model,phase=forward|sample|emit|residual}
+#     Decode time by phase, summed per request, as a real histogram.
+#     The `_sum` series is the point — ratio a phase against the whole to
+#     see where decode goes:
+#       sum(rate(neuron_decode_phase_seconds_sum{phase="forward"}[5m]))
+#         / sum(rate(neuron_decode_phase_seconds_sum[5m]))
+#     `residual` is decode_ms minus the three measured phases, emitted as
+#     its own series so unbracketed time is visible rather than divided
+#     silently among the phases that were measured.
+#   neuron_prefill_seconds / neuron_decode_seconds  (histograms)
+#   neuron_completion_tokens_total  (the per-token denominator)
+#
+# A model whose serving path does not measure phases emits NO phase series
+# at all rather than zeros — absent means "nobody looked", not "free".
+- job_name: neuron
+  static_configs:
+    - targets:
+        - 'beast.hanzalova.internal:13131'
+        - 'benjy.hanzalova.internal:13131'
+        - 'quadbrat.hanzalova.internal:13131'
+      labels:
+        fleet: hanzalova

crates/cortex-core/src/openai.rs

@@ -210,6 +210,19 @@ pub struct HelexaTiming {
     pub prefill_ms: u64,
     pub decode_ms: u64,
     pub prefill_tokens: u64,
+    /// Decode-loop phase split, when the serving path measured it.
+    /// Absent (rather than zero) on paths that do not, so a consumer
+    /// can tell "not measured" from "measured as free". Together these
+    /// should roughly account for `decode_ms`; the residual is time in
+    /// the loop that no bracket covers.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub forward_ms: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sample_ms: Option<u64>,
+    /// Detokenise + hand to the consumer. Includes bounded-channel
+    /// backpressure, so a slow reader is charged here.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub emit_ms: Option<u64>,
 }
 
 /// Sub-counts of `Usage::completion_tokens`.

crates/neuron/Cargo.toml

@@ -47,6 +47,8 @@ cuda-integration = ["cuda"]
 cortex-core.workspace = true
 tokio.workspace = true
 axum.workspace = true
+metrics.workspace = true
+metrics-exporter-prometheus.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 reqwest.workspace = true

crates/neuron/src/api.rs

@@ -10,6 +10,7 @@ use crate::wire::{openai_chat, openai_responses};
 use axum::Router;
 use axum::extract::{Path, State};
 use axum::http::StatusCode;
+use axum::http::header;
 use axum::response::sse::{Event, KeepAlive, Sse};
 use axum::response::{IntoResponse, Json};
 use axum::routing::{get, post};
@@ -43,6 +44,7 @@ pub struct NeuronState {
 /// Build the neuron API router.
 pub fn neuron_routes() -> Router<Arc<NeuronState>> {
     Router::new()
+        .route("/metrics", get(metrics_handler))
         .route("/version", get(version_handler))
         .route("/discovery", get(discovery_handler))
         .route("/health", get(health_handler))
@@ -64,6 +66,29 @@ async fn version_handler() -> Json<cortex_core::build_info::BuildInfo> {
     Json(crate::version::build_info())
 }
 
+/// `GET /metrics` — Prometheus text format.
+///
+/// Served from the API port rather than a second listener, so a scrape
+/// target needs no extra firewall rule. A missing recorder is a 503
+/// rather than an empty 200: an empty scrape looks exactly like a
+/// healthy process with nothing to say, and would leave the fleet
+/// silently unmonitored.
+async fn metrics_handler() -> impl IntoResponse {
+    match crate::metrics::render() {
+        Some(body) => (
+            StatusCode::OK,
+            [(header::CONTENT_TYPE, "text/plain; version=0.0.4")],
+            body,
+        )
+            .into_response(),
+        None => (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "metrics recorder not installed",
+        )
+            .into_response(),
+    }
+}
+
 async fn discovery_handler(State(state): State<Arc<NeuronState>>) -> Json<DiscoveryResponse> {
     Json(state.discovery.clone())
 }

crates/neuron/src/harness/arch/mod.rs

@@ -20,6 +20,7 @@
 //! TP-aware analogues live in `harness/tp/tp_<family>.rs` and follow
 //! the pattern set by `tp_qwen3.rs`.
 
+pub mod phase_probe;
 pub mod qwen3_5;
 pub mod qwen4_exp;
 pub mod snapshot;

crates/neuron/src/harness/arch/phase_probe.rs

@@ -0,0 +1,167 @@
+//! Opt-in phase timing inside a model's forward pass.
+//!
+//! The request-level split (`PhaseTiming` on the wire) says whether a
+//! decode step's time went into the forward, the sampler or the emit.
+//! When the answer is "the forward", this says *where* in the forward —
+//! embedding, a gather, the layer stack, the head.
+//!
+//! # Why this needs a device synchronise, and why it is off by default
+//!
+//! candle's CUDA ops are asynchronous on the stream. An `Instant` taken
+//! around GPU work therefore measures how fast kernels were *queued*,
+//! not how long they ran, and whichever phase happens to contain the
+//! next real synchronisation point silently absorbs everyone else's
+//! cost. The numbers look plausible and attribute the time to the wrong
+//! place — a worse failure than no numbers at all.
+//!
+//! So each mark synchronises the device first. That is only affordable
+//! when the phases being measured are much larger than a sync: at a few
+//! tens of microseconds per sync, a handful of marks is noise against a
+//! step measured in hundreds of milliseconds, and material against one
+//! measured in single milliseconds. Since that ratio is a property of
+//! how slow the model currently is rather than something safe to assume,
+//! the probe stays off unless `NEURON_FORWARD_PHASES` is set.
+
+use candle_core::Device;
+use std::sync::OnceLock;
+use std::time::Instant;
+
+fn enabled() -> bool {
+    static ON: OnceLock<bool> = OnceLock::new();
+    *ON.get_or_init(|| std::env::var("NEURON_FORWARD_PHASES").is_ok())
+}
+
+/// Accumulates named phases within one forward pass.
+///
+/// Construct at the top of the forward, call [`mark`] at each boundary,
+/// and [`emit`] at the end. Disabled, every method is a predictable
+/// branch and nothing is allocated.
+///
+/// [`mark`]: PhaseProbe::mark
+/// [`emit`]: PhaseProbe::emit
+pub struct PhaseProbe {
+    on: bool,
+    device: Device,
+    last: Instant,
+    marks: Vec<(&'static str, f64)>,
+}
+
+impl PhaseProbe {
+    pub fn new(device: &Device) -> Self {
+        Self::with_enabled(device, enabled())
+    }
+
+    /// Constructor with the gate supplied explicitly.
+    ///
+    /// [`new`] reads the environment through a `OnceLock`, which a test
+    /// cannot toggle — so without this the enabled path would be
+    /// unreachable from a test, and a probe that only ever runs
+    /// disabled is not evidence that it records anything.
+    ///
+    /// [`new`]: PhaseProbe::new
+    pub fn with_enabled(device: &Device, on: bool) -> Self {
+        // Start from a synchronised device, or the first phase inherits
+        // whatever was still in flight when the forward was entered.
+        if on {
+            let _ = device.synchronize();
+        }
+        Self {
+            on,
+            device: device.clone(),
+            last: Instant::now(),
+            marks: Vec::new(),
+        }
+    }
+
+    /// Close the phase that ended here and open the next one.
+    pub fn mark(&mut self, name: &'static str) {
+        if !self.on {
+            return;
+        }
+        // A failed synchronise means the elapsed time below is not the
+        // phase's own. Record it as such rather than reporting a number
+        // that looks like a measurement.
+        let synced = self.device.synchronize().is_ok();
+        let ms = self.last.elapsed().as_secs_f64() * 1e3;
+        self.marks
+            .push((if synced { name } else { "UNSYNCED" }, ms));
+        self.last = Instant::now();
+    }
+
+    /// The phases recorded so far, in order.
+    #[cfg(test)]
+    pub fn marks(&self) -> &[(&'static str, f64)] {
+        &self.marks
+    }
+
+    /// Emit one line carrying every phase of this forward.
+    pub fn emit(self, rows: usize) {
+        if !self.on || self.marks.is_empty() {
+            return;
+        }
+        let total: f64 = self.marks.iter().map(|(_, ms)| ms).sum();
+        let detail = self
+            .marks
+            .iter()
+            .map(|(n, ms)| format!("{n}={ms:.3}"))
+            .collect::<Vec<_>>()
+            .join(" ");
+        tracing::info!(rows, total_ms = total, %detail, "forward phases");
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// A disabled probe must record nothing at all — not zero-valued
+    /// phases. Zeros in a log read as "this phase was free", which is
+    /// the same failure the wire-side `Option` exists to prevent.
+    #[test]
+    fn a_disabled_probe_records_nothing() {
+        let mut p = PhaseProbe::with_enabled(&Device::Cpu, false);
+        p.mark("embed");
+        p.mark("layers");
+        assert!(
+            p.marks().is_empty(),
+            "disabled probe recorded {:?}",
+            p.marks()
+        );
+    }
+
+    /// An enabled probe records one phase per mark, in call order, and
+    /// each phase measures only the span since the previous mark rather
+    /// than since the start.
+    #[test]
+    fn phases_are_recorded_in_order_and_do_not_accumulate() {
+        let mut p = PhaseProbe::with_enabled(&Device::Cpu, true);
+        p.mark("embed");
+        std::thread::sleep(std::time::Duration::from_millis(20));
+        p.mark("layers");
+        p.mark("head");
+
+        let names: Vec<_> = p.marks().iter().map(|(n, _)| *n).collect();
+        assert_eq!(names, ["embed", "layers", "head"]);
+
+        let ms = |i: usize| p.marks()[i].1;
+        // "layers" contains the sleep; the marks either side must not.
+        // A probe that timed from the start instead of from the previous
+        // mark would make "head" >= "layers", which is the mistake this
+        // pins.
+        assert!(
+            ms(1) >= 15.0,
+            "layers should carry the sleep, got {:?}",
+            p.marks()
+        );
+        assert!(
+            ms(2) < 10.0,
+            "head must not inherit it, got {:?}",
+            p.marks()
+        );
+        assert!(
+            ms(0) < 10.0,
+            "embed must not inherit it, got {:?}",
+            p.marks()
+        );
+    }
+}

crates/neuron/src/harness/arch/qwen4_exp/model.rs

@@ -45,6 +45,7 @@ use super::config::{Config, TextConfig};
 use super::decoder::DecoderLayer;
 use super::hyper::HyperConnection;
 use super::ple::{MmapNGramTable, NGramHasher, NGramTable, ShardedNGramTable};
+use crate::harness::arch::phase_probe::PhaseProbe;
 use crate::harness::arch::snapshot::{KvCacheSnapshot, PleSnapshot};
 
 /// The hashed n-gram lookup, hoisted to the model because it is a
@@ -255,6 +256,7 @@ impl Qwen4ExpForCausalLM {
 
         // Four identical streams — the residual this architecture
         // carries between layers, not one.
+        let mut probe = PhaseProbe::new(&self.device);
         let x = self.embed_tokens.forward(input_ids)?;
         let mut h = x.repeat((1, 1, self.hc_count))?;
 
@@ -264,6 +266,7 @@ impl Qwen4ExpForCausalLM {
         } else {
             None
         };
+        probe.mark("embed");
 
         // Computed once: it is a function of the ids, and one layer
         // consumes it.
@@ -274,12 +277,17 @@ impl Qwen4ExpForCausalLM {
             }
             None => None,
         };
+        // The n-gram table is mapped from the checkpoint rather than
+        // resident (#310), so this phase is a host page-fault path, not
+        // device work: it is where a cold or evicted mapping shows up.
+        probe.mark("ngram");
 
         for (i, layer) in self.layers.iter_mut().enumerate() {
             h = layer
                 .forward(&h, ngram.as_ref(), mask.as_ref(), &cos, &sin, offset)
                 .with_context(|| format!("layer {i}"))?;
         }
+        probe.mark("layers");
 
         // The mixer's hc_norm is the only normalisation before the head.
         let x = self.mixer.collapse(&h)?;
@@ -292,7 +300,10 @@ impl Qwen4ExpForCausalLM {
         // them all and [`Self::logits_all_positions`] exists to be
         // compared against it.
         let l = x.dim(1)?;
-        Ok((self.lm_head.forward(&x.i((.., l - 1.., ..))?)?, h))
+        let logits = self.lm_head.forward(&x.i((.., l - 1.., ..))?)?;
+        probe.mark("head");
+        probe.emit(seq_len);
+        Ok((logits, h))
     }
 
     /// Capture every piece of per-request state at one token boundary:

crates/neuron/src/harness/arch/qwen4_exp/ple.rs

@@ -685,10 +685,35 @@ impl PleBlock {
             Tensor::cat(&[&pad, &prepended], 2)?
         });
 
-        // Pad both sides by the full reach, then keep the left-aligned
-        // window: output t sees inputs t - j*dilation only.
-        let out = prepended.conv1d(&self.conv1d_weight, reach, 1, self.dilation, wide)?;
-        let out = candle_nn::ops::silu(&out.narrow(2, 0, prep_len)?)?;
+        // Written as `kernel_size` shifted broadcast multiplies rather
+        // than a `conv1d` with `groups = wide`. candle decomposes a
+        // grouped convolution into one single-group call per group plus
+        // a `groups`-way concat, so the depthwise form costs O(wide)
+        // kernel launches — ~40k of them for the 10240-wide stream here,
+        // which dominates the whole forward pass. This form is O(kernel
+        // _size) launches and computes the same sum:
+        //
+        //     out[t] = sum_j w[.., j] * x[t - (kernel_size - 1 - j) * dilation]
+        //
+        // with positions left of the sequence reading as zero. Tap
+        // `kernel_size - 1` is the current position, so it always
+        // contributes and seeds the accumulator.
+        let tap = |j: usize| -> candle_core::Result<Tensor> {
+            self.conv1d_weight.narrow(2, j, 1)?.reshape((1, wide, 1))
+        };
+        let mut acc = prepended.broadcast_mul(&tap(self.kernel_size - 1)?)?;
+        for j in 0..self.kernel_size - 1 {
+            let lag = (self.kernel_size - 1 - j) * self.dilation;
+            // Every position this tap reads is left of the sequence, so
+            // it contributes only the zeros the padding would have held.
+            if lag >= prep_len {
+                continue;
+            }
+            let pad = Tensor::zeros((b, wide, lag), prepended.dtype(), prepended.device())?;
+            let shifted = Tensor::cat(&[&pad, &prepended.narrow(2, 0, prep_len - lag)?], 2)?;
+            acc = (acc + shifted.broadcast_mul(&tap(j)?)?)?;
+        }
+        let out = candle_nn::ops::silu(&acc)?;
         out.narrow(2, prep_len - l, l)?
             .transpose(1, 2)?
             .contiguous()
@@ -1052,6 +1077,59 @@ mod tests {
         assert_eq!(ple.left_context(), dilation);
     }
 
+    /// Which tap reaches how far back, at the shipped `kernel_size`.
+    ///
+    /// Tap `j` must read position `t - (kernel_size - 1 - j) * dilation`,
+    /// so the weights run far-to-near and the *last* one is the current
+    /// position. Every other conv test here uses `kernel_size = 2`, where
+    /// that mapping and its reverse agree on the only non-zero lag — so
+    /// none of them can see a transposed tap order. The shipped config is
+    /// `kernel_size = 4`, where the four lags are 9, 6, 3 and 0: a single
+    /// impulse with four distinct weights lands each tap in its own
+    /// position, and reversing the order swaps the 9 and 3 responses.
+    #[test]
+    fn conv_taps_run_far_to_near_across_the_full_kernel() {
+        let dev = Device::Cpu;
+        let (wide, dilation, kernel_size) = (1, 3, 4);
+        // Far to near: w[0] reaches 9 back, w[3] is the current position.
+        let conv_w =
+            Tensor::from_vec(vec![1.0f32, 2.0, 3.0, 4.0], (wide, 1, kernel_size), &dev).unwrap();
+        let mut ple = block(1, 1, kernel_size, dilation, eye(1), eye(1), conv_w);
+
+        let mut impulse = vec![0.0f32; 10];
+        impulse[0] = 1.0;
+        let x = Tensor::from_vec(impulse, (1, 10, wide), &dev).unwrap();
+        let got: Vec<f32> = ple
+            .short_conv(&x)
+            .unwrap()
+            .flatten_all()
+            .unwrap()
+            .to_vec1()
+            .unwrap();
+
+        let silu = |v: f32| v / (1.0 + (-v).exp());
+        // Weight 4 at lag 0, 3 at lag 3, 2 at lag 6, 1 at lag 9.
+        let want = [
+            silu(4.0),
+            0.0,
+            0.0,
+            silu(3.0),
+            0.0,
+            0.0,
+            silu(2.0),
+            0.0,
+            0.0,
+            silu(1.0),
+        ];
+        for (i, (g, w)) in got.iter().zip(want.iter()).enumerate() {
+            assert!(
+                (g - w).abs() < 1e-5,
+                "position {i}: got {got:?} want {want:?}"
+            );
+        }
+        assert_eq!(ple.left_context(), (kernel_size - 1) * dilation);
+    }
+
     /// Decoding one token at a time must give the same answer as running
     /// the whole sequence at once — the property the rolling state
     /// exists for, and the one that breaks silently when the cached

crates/neuron/src/harness/candle.rs

@@ -25,6 +25,11 @@ use cortex_core::openai::{
     ChatCompletionChunk, ChatCompletionRequest, ChatCompletionResponse, ChatMessage, MessageContent,
 };
 
+// `PhaseTiming` is constructed only on the two cuda-gated streaming
+// paths (device-worker and TP), so importing it unconditionally warns
+// on a CPU-only build and clippy runs with `-D warnings`.
+#[cfg(feature = "cuda")]
+use crate::wire::PhaseTiming;
 use crate::wire::{
     FinishReason, FinishTiming, InferenceEvent, ReasoningTokenPair, ToolCallTokenPair,
     detect_reasoning_token_pair, detect_tool_call_token_pair, openai_chat as wire_chat,
@@ -3505,11 +3510,13 @@ impl CandleHarness {
                 let reasoning_tokens_inner = loaded.reasoning_tokens.clone();
                 let tool_call_tokens_inner = loaded.tool_call_tokens.clone();
                 let tool_schemas_inner = tool_schemas.clone();
+                let model_id_inner = model_id.clone();
                 tokio::spawn(
                     async move {
                         let _admit = admit;
                         let _inference_guard = loaded_for_task.inference_lock.lock().await;
                         match stream_inference_via_worker(
+                            model_id_inner,
                             worker,
                             handle,
                             tokenizer,
@@ -5322,6 +5329,9 @@ impl CandleHarness {
                 // read them; populated at the prefill→decode boundary inside.
                 let mut prefill_ms_measured: u32 = 0;
                 let mut decode_start: Option<std::time::Instant> = None;
+                let mut phase_forward = std::time::Duration::ZERO;
+                let mut phase_sample = std::time::Duration::ZERO;
+                let mut phase_emit = std::time::Duration::ZERO;
 
                 // Hoisted out of `'work` so the completion path below
                 // can report it (#269) — the reuse count is decided
@@ -5482,6 +5492,12 @@ impl CandleHarness {
                         };
                     // Decode-phase timer for the Finish prefill/decode split (#85).
                     decode_start = Some(std::time::Instant::now());
+                    // Phase split (see `PhaseTiming`). Instrumented here
+                    // as well as on the single-GPU path so a healthy
+                    // model is a control for a slow one: both archs share
+                    // this loop, this sampler and a 248,320 vocabulary,
+                    // so a phase that is cheap here and expensive there
+                    // is a property of the arch, not of the loop.
 
                     if Some(next_token) == eos_id {
                         finish_reason = FinishReason::Stop;
@@ -5568,6 +5584,7 @@ impl CandleHarness {
                         }
 
                         for index in 0..max_new.saturating_sub(1) {
+                            let t_forward = std::time::Instant::now();
                             let logits_vec = match pool
                                 .generate_step(
                                     &model_id,
@@ -5583,6 +5600,8 @@ impl CandleHarness {
                                     break 'work;
                                 }
                             };
+                            phase_forward += t_forward.elapsed();
+                            let t_sample = std::time::Instant::now();
                             let logits = match Tensor::new(logits_vec.as_slice(), &Device::Cpu) {
                                 Ok(t) => t,
                                 Err(e) => {
@@ -5611,6 +5630,11 @@ impl CandleHarness {
                                     break 'work;
                                 }
                             };
+                            phase_sample += t_sample.elapsed();
+                            // NB: this per-step VRAM query is a second
+                            // channel round-trip per token. It is charged
+                            // to neither phase, so it lands in the
+                            // residual between the three and `decode_ms`.
                             // Always await the query (even when the
                             // trace! is filtered out by RUST_LOG): the
                             // channel hop is ~tens of µs, comparable to
@@ -5630,6 +5654,7 @@ impl CandleHarness {
                                 finish_reason = FinishReason::Stop;
                                 break;
                             }
+                            let t_emit = std::time::Instant::now();
                             all_tokens.push(next_token);
                             match handle_tool_call_marker(
                                 next_token,
@@ -5711,6 +5736,7 @@ impl CandleHarness {
                                     "TP stream: decode_stream step failed"
                                 ),
                             }
+                            phase_emit += t_emit.elapsed();
                         }
                     }
                 }
@@ -5775,8 +5801,24 @@ impl CandleHarness {
                         decode_ms = decode_start
                             .map(|d| d.elapsed().as_millis() as u32)
                             .unwrap_or(0),
+                        forward_ms = phase_forward.as_millis(),
+                        sample_ms = phase_sample.as_millis(),
+                        emit_ms = phase_emit.as_millis(),
                         "chat_completion (stream): done"
                     );
+                    crate::metrics::record_finish(
+                        &model_id,
+                        prefill_ms_measured,
+                        decode_start
+                            .map(|d| d.elapsed().as_millis() as u32)
+                            .unwrap_or(0),
+                        all_tokens.len() as u32,
+                        Some((
+                            phase_forward.as_millis() as u32,
+                            phase_sample.as_millis() as u32,
+                            phase_emit.as_millis() as u32,
+                        )),
+                    );
                     let _ = tx
                         .send(InferenceEvent::Finish {
                             reason: finish_reason,
@@ -5790,6 +5832,11 @@ impl CandleHarness {
                                     .map(|d| d.elapsed().as_millis() as u32)
                                     .unwrap_or(0),
                                 prefill_tokens: prompt_len as u32,
+                                phases: Some(PhaseTiming {
+                                    forward_ms: phase_forward.as_millis() as u32,
+                                    sample_ms: phase_sample.as_millis() as u32,
+                                    emit_ms: phase_emit.as_millis() as u32,
+                                }),
                             }),
                         })
                         .await;
@@ -7236,6 +7283,9 @@ async fn run_inference_via_worker(
 #[cfg(feature = "cuda")]
 #[allow(clippy::too_many_arguments)]
 async fn stream_inference_via_worker(
+    // Owned: this runs in a spawned task, and the only other source of
+    // the model name here is the tracing span, which is not a value.
+    model_id: String,
     worker: Arc<super::device_worker::DeviceWorkerHandle>,
     handle: super::device_worker::ArchHandle,
     tokenizer: Tokenizer,
@@ -7369,6 +7419,13 @@ async fn stream_inference_via_worker(
     };
     // Decode-phase timer for the Finish prefill/decode split (#85).
     let decode_start = std::time::Instant::now();
+    // Phase split of the decode loop. Each boundary below is already a
+    // synchronisation point -- the worker hands back CPU-side logits --
+    // so this adds no device sync and measures elapsed work rather than
+    // kernel-launch time.
+    let mut phase_forward = std::time::Duration::ZERO;
+    let mut phase_sample = std::time::Duration::ZERO;
+    let mut phase_emit = std::time::Duration::ZERO;
 
     // Per-token routing. `tokenizers::DecodeStream` carries five
     // generic parameters (`M, N, PT, PP, D`) which makes naming
@@ -7475,10 +7532,14 @@ async fn stream_inference_via_worker(
     }
 
     for index in 0..max_new.saturating_sub(1) {
+        let t_forward = std::time::Instant::now();
         let logits_vec = worker
             .forward_logits(handle, vec![next_token], prompt_len + index)
             .await
             .map_err(|e| anyhow::anyhow!("decode step {index}: {e:#}"))?;
+        phase_forward += t_forward.elapsed();
+
+        let t_sample = std::time::Instant::now();
         let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
         next_token = match sample_with_penalty(
             &logits,
@@ -7498,11 +7559,15 @@ async fn stream_inference_via_worker(
                 return Err(e);
             }
         };
+        phase_sample += t_sample.elapsed();
         if Some(next_token) == eos_id {
             finish_reason = FinishReason::Stop;
             break;
         }
-        if !route_token!(next_token) {
+        let t_emit = std::time::Instant::now();
+        let routed = route_token!(next_token);
+        phase_emit += t_emit.elapsed();
+        if !routed {
             return Ok(finish_reason.as_openai_str().to_string());
         }
     }
@@ -7524,8 +7589,22 @@ async fn stream_inference_via_worker(
         finish_reason = ?finish_reason,
         prefill_ms = prefill_elapsed.as_millis(),
         decode_ms = decode_start.elapsed().as_millis(),
+        forward_ms = phase_forward.as_millis(),
+        sample_ms = phase_sample.as_millis(),
+        emit_ms = phase_emit.as_millis(),
         "chat_completion (stream): done"
     );
+    crate::metrics::record_finish(
+        &model_id,
+        prefill_elapsed.as_millis() as u32,
+        decode_start.elapsed().as_millis() as u32,
+        all_tokens.len() as u32,
+        Some((
+            phase_forward.as_millis() as u32,
+            phase_sample.as_millis() as u32,
+            phase_emit.as_millis() as u32,
+        )),
+    );
     let _ = tx
         .send(InferenceEvent::Finish {
             reason: finish_reason,
@@ -7537,6 +7616,11 @@ async fn stream_inference_via_worker(
                 prefill_ms: prefill_elapsed.as_millis() as u32,
                 decode_ms: decode_start.elapsed().as_millis() as u32,
                 prefill_tokens: prefill_prompt_len as u32,
+                phases: Some(PhaseTiming {
+                    forward_ms: phase_forward.as_millis() as u32,
+                    sample_ms: phase_sample.as_millis() as u32,
+                    emit_ms: phase_emit.as_millis() as u32,
+                }),
             }),
         })
         .await;
@@ -7757,6 +7841,7 @@ fn run_inference_streaming(
             prefill_ms: prefill_elapsed.as_millis() as u32,
             decode_ms: decode_start.elapsed().as_millis() as u32,
             prefill_tokens: prompt_tokens.len() as u32,
+            phases: None,
         }),
     });
     Ok(())

crates/neuron/src/harness/engine.rs

@@ -663,6 +663,7 @@ async fn finish_slot(slot: &mut Slot, reason: FinishReason, rate: &PrefillRateEm
                 prefill_ms: slot.prefill_ms,
                 decode_ms: slot.decode_start.elapsed().as_millis() as u32,
                 prefill_tokens: slot.prefill_tokens,
+                phases: None,
             },
         })
         .await;
@@ -1151,6 +1152,7 @@ mod tests {
                     prefill_ms: 6000,
                     decode_ms: 900_000,
                     prefill_tokens: 7413,
+                    phases: None,
                 },
             })
             .await

crates/neuron/src/lib.rs

@@ -5,6 +5,7 @@ pub mod cuda;
 pub mod discovery;
 pub mod harness;
 pub mod health;
+pub mod metrics;
 pub mod startup;
 pub mod version;
 pub mod wire;

crates/neuron/src/main.rs

@@ -220,6 +220,12 @@ async fn daemon(args: Args) -> Result<()> {
     // host look down to anything probing `/health` during pre-warm.
     // The pre-warm task runs in the background instead — `/health`
     // surfaces its progress via the activation field.
+    // Before the listener binds, so the very first scrape after a
+    // restart finds a recorder rather than a 503.
+    if let Err(e) = neuron::metrics::install() {
+        tracing::warn!(error = %e, "metrics recorder not installed; /metrics will 503");
+    }
+
     let app = api::neuron_routes().with_state(Arc::clone(&state));
     let addr: std::net::SocketAddr = format!("0.0.0.0:{port}").parse()?;
     let listener = tokio::net::TcpListener::bind(addr).await?;

crates/neuron/src/metrics.rs

@@ -0,0 +1,173 @@
+//! Prometheus metrics for neuron.
+//!
+//! neuron had no `/metrics` surface at all: every signal it produced
+//! reached Prometheus only by cortex polling `/health` every ~10 s and
+//! republishing it as gauges. That is enough for routing decisions and
+//! useless for asking where a request's time went — an EMA sampled at
+//! 10 s cannot see inside a one-second forward, and the finest split
+//! that existed anywhere was prefill versus decode.
+//!
+//! Served from the main API port rather than a second listener: neuron
+//! already runs an axum server, and a separate port is one more thing
+//! to open in the firewall for no benefit.
+
+use anyhow::Result;
+use metrics_exporter_prometheus::{Matcher, PrometheusBuilder, PrometheusHandle};
+use std::sync::OnceLock;
+
+static HANDLE: OnceLock<PrometheusHandle> = OnceLock::new();
+
+/// Install the process-wide recorder. Idempotent in effect: a second
+/// call leaves the first handle in place.
+pub fn install() -> Result<()> {
+    let handle = with_buckets(PrometheusBuilder::new())?
+        .install_recorder()
+        .map_err(|e| anyhow::anyhow!("failed to install Prometheus recorder: {e}"))?;
+    describe();
+    let _ = HANDLE.set(handle);
+    Ok(())
+}
+
+/// Render the current metrics, or `None` if the recorder was never
+/// installed. A `None` here must surface as an error rather than an
+/// empty body: an empty scrape is indistinguishable from a healthy
+/// process with nothing to report.
+pub fn render() -> Option<String> {
+    HANDLE.get().map(|h| h.render())
+}
+
+/// Explicit buckets, or `metrics-exporter-prometheus` renders every
+/// histogram as a **summary** — `{quantile=...}` series, no `_bucket`,
+/// no aggregation across hosts, and every quantile decaying to `0`
+/// when idle so "no traffic" looks like "instant". The same trap
+/// cortex's exporter documents; the fix has to be repeated per metric.
+fn with_buckets(builder: PrometheusBuilder) -> Result<PrometheusBuilder> {
+    // Per-request phase sums: microseconds for an emit on a fast model,
+    // minutes for the forward on a long turn.
+    let phase_seconds = &[
+        0.001, 0.005, 0.025, 0.1, 0.25, 1.0, 2.5, 5.0, 15.0, 30.0, 60.0, 300.0,
+    ];
+    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,
+    ];
+    builder
+        .set_buckets_for_metric(
+            Matcher::Full("neuron_decode_phase_seconds".into()),
+            phase_seconds,
+        )?
+        .set_buckets_for_metric(
+            Matcher::Full("neuron_prefill_seconds".into()),
+            seconds_short,
+        )?
+        .set_buckets_for_metric(Matcher::Full("neuron_decode_seconds".into()), seconds_long)
+        .map_err(|e| anyhow::anyhow!("failed to configure histogram buckets: {e}"))
+}
+
+fn describe() {
+    metrics::describe_histogram!(
+        "neuron_decode_phase_seconds",
+        "Decode-loop time by phase, summed per request. The `_sum` \
+         series is the point: ratio one phase against the total to see \
+         where decode goes, without needing per-step resolution."
+    );
+    metrics::describe_histogram!("neuron_prefill_seconds", "Prefill wall-clock per request");
+    metrics::describe_histogram!("neuron_decode_seconds", "Decode wall-clock per request");
+    metrics::describe_counter!(
+        "neuron_completion_tokens_total",
+        "Tokens generated, the denominator for per-token phase cost"
+    );
+}
+
+/// Record one finished inference.
+///
+/// `phases` is optional because not every serving path measures them,
+/// and a path that does not must emit **nothing** rather than zeros —
+/// a zero sample pulls the histogram's mean down and reads as "this
+/// phase is free here", which is the opposite of "nobody looked".
+///
+/// The residual is emitted as its own phase. Whatever the brackets
+/// failed to cover is then visible as a labelled series instead of
+/// being silently divided among the phases that were measured.
+pub fn record_finish(
+    model: &str,
+    prefill_ms: u32,
+    decode_ms: u32,
+    completion_tokens: u32,
+    phases: Option<(u32, u32, u32)>,
+) {
+    let m = model.to_string();
+    metrics::histogram!("neuron_prefill_seconds", "model" => m.clone())
+        .record(prefill_ms as f64 / 1e3);
+    metrics::histogram!("neuron_decode_seconds", "model" => m.clone())
+        .record(decode_ms as f64 / 1e3);
+    metrics::counter!("neuron_completion_tokens_total", "model" => m.clone())
+        .increment(completion_tokens as u64);
+
+    let Some((forward_ms, sample_ms, emit_ms)) = phases else {
+        return;
+    };
+    for (phase, ms) in [
+        ("forward", forward_ms),
+        ("sample", sample_ms),
+        ("emit", emit_ms),
+    ] {
+        metrics::histogram!("neuron_decode_phase_seconds", "model" => m.clone(), "phase" => phase)
+            .record(ms as f64 / 1e3);
+    }
+    // Saturating: the phases are measured inside the decode window, so
+    // they cannot legitimately exceed it, but clock granularity can
+    // round them a millisecond over and an underflow here would report
+    // an enormous residual.
+    let residual =
+        decode_ms.saturating_sub(forward_ms.saturating_add(sample_ms).saturating_add(emit_ms));
+    metrics::histogram!("neuron_decode_phase_seconds", "model" => m, "phase" => "residual")
+        .record(residual as f64 / 1e3);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// An unmeasured path must contribute no phase samples at all.
+    /// Zeros would drag the histogram's mean toward zero and read as
+    /// "this phase costs nothing on this model".
+    #[test]
+    fn phases_absent_means_no_phase_series() {
+        let handle = with_buckets(PrometheusBuilder::new())
+            .expect("buckets")
+            .install_recorder()
+            .expect("recorder");
+        describe();
+
+        record_finish("measured", 100, 1000, 64, Some((900, 50, 10)));
+        record_finish("unmeasured", 100, 1000, 64, None);
+        let out = handle.render();
+
+        assert!(
+            out.contains("neuron_decode_phase_seconds") && out.contains(r#"model="measured""#),
+            "measured model must emit phases:\n{out}"
+        );
+        for line in out
+            .lines()
+            .filter(|l| l.contains("neuron_decode_phase_seconds"))
+        {
+            assert!(
+                !line.contains(r#"model="unmeasured""#),
+                "unmeasured model must emit no phase series, got: {line}"
+            );
+        }
+        // Buckets, not a summary: `histogram_quantile()` needs `_bucket`.
+        assert!(
+            out.contains("neuron_decode_phase_seconds_bucket"),
+            "must export as a histogram, not a summary:\n{out}"
+        );
+        // 1000 - (900 + 50 + 10) = 40 ms of residual, as its own series.
+        assert!(
+            out.contains(r#"phase="residual""#),
+            "residual must be its own series:\n{out}"
+        );
+    }
+}

crates/neuron/src/wire/event.rs

@@ -121,6 +121,41 @@ pub struct FinishTiming {
     /// the elapsed `prefill_ms` drops while this stays the full prompt
     /// length, so a high implied rate is itself the cache-hit signal.
     pub prefill_tokens: u32,
+    /// How the decode loop's wall-clock divides, when the serving path
+    /// measured it. `None` means "this path does not report phases",
+    /// which is deliberately distinct from "measured, and they were
+    /// zero" — a phase split that cannot express its own absence
+    /// invites reading an unmeasured path as an idle one.
+    pub phases: Option<PhaseTiming>,
+}
+
+/// Where a decode loop's wall-clock actually went, summed over every
+/// step of the request.
+///
+/// These are sums across steps, not a breakdown of one step, and each
+/// boundary is a point the loop already synchronises at: the forward
+/// hands back CPU-side logits, so no device synchronisation is added
+/// and the numbers are elapsed work rather than kernel-launch time.
+///
+/// The three should roughly account for `decode_ms`; a large residual
+/// is itself a finding (it is time inside the loop that none of the
+/// brackets cover).
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct PhaseTiming {
+    /// Submitting the step to the device worker and getting logits
+    /// back: the channel hop, the model forward, and the device-to-host
+    /// copy of the logits row. For a large vocabulary that copy is not
+    /// negligible -- 248,320 f32 is ~993 KB per step.
+    pub forward_ms: u32,
+    /// Wrapping the logits in a CPU tensor and sampling: repeat/presence
+    /// penalties plus the `LogitsProcessor`. Scales with vocabulary,
+    /// not with model size, and top-p sorts the distribution.
+    pub sample_ms: u32,
+    /// Detokenising the sampled token and handing the delta to the
+    /// consumer. **Includes backpressure**: the send awaits a bounded
+    /// channel, so a slow reader is charged here. This is emit cost,
+    /// not detokenisation cost, and the two are not separated.
+    pub emit_ms: u32,
 }
 
 /// Why a stream stopped. Stays small on purpose — anything that

crates/neuron/src/wire/mod.rs

@@ -22,6 +22,6 @@ pub mod openai_chat;
 pub mod openai_responses;
 
 pub use event::{
-    FinishReason, FinishTiming, InferenceEvent, ReasoningTokenPair, ToolCallTokenPair,
+    FinishReason, FinishTiming, InferenceEvent, PhaseTiming, ReasoningTokenPair, ToolCallTokenPair,
     detect_reasoning_token_pair, detect_tool_call_token_pair,
 };

crates/neuron/src/wire/openai_chat.rs

@@ -444,6 +444,9 @@ fn usage_chunk(
                 prefill_ms: t.prefill_ms as u64,
                 decode_ms: t.decode_ms as u64,
                 prefill_tokens: t.prefill_tokens as u64,
+                forward_ms: t.phases.map(|p| p.forward_ms as u64),
+                sample_ms: t.phases.map(|p| p.sample_ms as u64),
+                emit_ms: t.phases.map(|p| p.emit_ms as u64),
             }),
         }),
         extra: serde_json::Value::Object(Default::default()),
@@ -596,6 +599,9 @@ pub async fn collect_chat_completion(
                 prefill_ms: t.prefill_ms as u64,
                 decode_ms: t.decode_ms as u64,
                 prefill_tokens: t.prefill_tokens as u64,
+                forward_ms: t.phases.map(|p| p.forward_ms as u64),
+                sample_ms: t.phases.map(|p| p.sample_ms as u64),
+                emit_ms: t.phases.map(|p| p.emit_ms as u64),
             }),
         }),
         extra: serde_json::Value::Object(Default::default()),
@@ -685,6 +691,7 @@ mod tests {
         );
     }
     use super::*;
+    use crate::wire::event::PhaseTiming;
 
     // ── Non-streaming collector (#285) ────────────────────────────
 
@@ -799,6 +806,7 @@ mod tests {
                 prefill_ms: 12,
                 decode_ms: 34,
                 prefill_tokens: 100,
+                phases: None,
             }),
         })
         .await
@@ -813,6 +821,65 @@ mod tests {
         assert_eq!((t.prefill_ms, t.decode_ms, t.prefill_tokens), (12, 34, 100));
     }
 
+    /// The phase split must survive to the wire, and an unmeasured path
+    /// must serialise as **absent** rather than as zeros.
+    ///
+    /// That distinction is the reason the field is an `Option` at all:
+    /// a consumer reading `forward_ms: 0` off a path that never
+    /// measured would conclude the forward is free. The serialised
+    /// form is asserted, not just the struct, because
+    /// `skip_serializing_if` is what actually enforces it and a plain
+    /// field comparison cannot see it.
+    #[tokio::test]
+    async fn phase_timing_reaches_the_wire_and_absence_is_not_zero() {
+        async fn timing_json(phases: Option<PhaseTiming>) -> serde_json::Value {
+            let (tx, rx) = mpsc::channel(8);
+            tx.send(InferenceEvent::Finish {
+                reason: FinishReason::Stop,
+                prompt_tokens: 10,
+                completion_tokens: 5,
+                reasoning_tokens: 0,
+                cached_tokens: 0,
+                timing: Some(FinishTiming {
+                    prefill_ms: 12,
+                    decode_ms: 340,
+                    prefill_tokens: 10,
+                    phases,
+                }),
+            })
+            .await
+            .expect("send");
+            drop(tx);
+            let r = collect_chat_completion(rx, "id".into(), 1, "m".into())
+                .await
+                .expect("collected");
+            serde_json::to_value(r.usage.expect("usage").helexa_timing.expect("timing"))
+                .expect("serialise")
+        }
+
+        let measured = timing_json(Some(PhaseTiming {
+            forward_ms: 300,
+            sample_ms: 25,
+            emit_ms: 4,
+        }))
+        .await;
+        assert_eq!(measured["forward_ms"], 300);
+        assert_eq!(measured["sample_ms"], 25);
+        assert_eq!(measured["emit_ms"], 4);
+
+        let unmeasured = timing_json(None).await;
+        for f in ["forward_ms", "sample_ms", "emit_ms"] {
+            assert!(
+                unmeasured.get(f).is_none(),
+                "unmeasured {f} must be absent, not {:?} -- a zero here reads as \
+                 'this phase is free' rather than 'nobody measured it'",
+                unmeasured.get(f)
+            );
+        }
+        // The prefill/decode split is unconditional and must be unaffected.
+        assert_eq!(unmeasured["decode_ms"], 340);
+    }
+
     /// A dropped engine slot ends the channel with no Finish. Returning
     /// a 200 with partial text would hide a fatal worker fault.
     #[tokio::test]
@@ -898,6 +965,7 @@ mod tests {
                 prefill_ms: 200,
                 decode_ms: 1500,
                 prefill_tokens: 128,
+                phases: None,
             }),
         })
         .await

crates/neuron/tests/api.rs

@@ -99,6 +99,49 @@ async fn test_version_endpoint() {
     assert!(!body.git_sha.is_empty());
 }
 
+#[tokio::test]
+async fn test_metrics_endpoint_serves_prometheus_text() {
+    // The recorder is process-wide and other tests in this binary may
+    // have installed it already; either way the route must answer, and
+    // must never answer 200-with-empty-body (indistinguishable from a
+    // healthy process reporting nothing).
+    let _ = neuron::metrics::install();
+    neuron::metrics::record_finish("probe-model", 100, 1000, 64, Some((900, 50, 10)));
+    let url = spawn_neuron(fake_discovery()).await;
+
+    let resp = reqwest::Client::new()
+        .get(format!("{url}/metrics"))
+        .send()
+        .await
+        .expect("request should succeed");
+
+    assert_eq!(resp.status(), 200);
+    let ct = resp
+        .headers()
+        .get(reqwest::header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or_default()
+        .to_string();
+    assert!(
+        ct.starts_with("text/plain"),
+        "Prometheus needs a text/plain scrape, got {ct:?}"
+    );
+
+    let body = resp.text().await.expect("body");
+    // The exporter renders only metrics that carry samples -- a
+    // `describe_*` on its own produces nothing -- so the scrape has to
+    // be checked after a recorded request, which is also the path that
+    // actually matters: record here, readable there.
+    assert!(
+        body.contains("neuron_decode_phase_seconds"),
+        "a recorded request must appear in the scrape:\n{body}"
+    );
+    assert!(
+        body.contains(r#"phase="forward""#) && body.contains(r#"phase="residual""#),
+        "phase labels must survive to the scrape:\n{body}"
+    );
+}
+
 #[tokio::test]
 async fn test_health_endpoint() {
     let url = spawn_neuron(fake_discovery()).await;