Merge pull request 'feat(helexa-bench): percentiles + prefill/decode split in store & report (#86)' (#101) from feat/86-bench-percentiles into main

Bench reported only a single median per cell, hiding tail latency and
unable to record the server-measured prefill/decode split now emitted
on `usage.helexa_timing` (#85).

- scenario: parse `usage.helexa_timing` into ScenarioMetrics
  (prefill_ms, decode_ms, prefill_tokens).
- store: persist the three columns (additive PRAGMA-guarded migration
  via ensure_columns, so pre-#85 DBs backfill as NULL); aggregate now
  emits p50/p95/p99 for TTFT and total (nearest-rank) plus a
  prefill-tok/s median derived from the split.
- report: markdown gains prefill tok/s, TTFT p95, total p95 columns;
  JSON gains p95/p99 + prefill_ms/decode_ms/prefill_tps medians.

Tests: nearest-rank percentile, idempotent backfill migration, and a
report cell asserting percentiles + prefill split.

Part of the Performance observability epic (#83). Stacked on #85.

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

crates/cortex-core/src/openai.rs

@@ -116,6 +116,23 @@ pub struct Usage {
     /// prompt caching lands (#11); `None` until then.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub prompt_tokens_details: Option<PromptTokensDetails>,
+    /// helexa extension (non-OpenAI): server-measured prefill/decode
+    /// timing, so the bench harness can compute true prefill vs decode
+    /// tok/s instead of inferring both from client-side SSE arrival
+    /// (#85). Additive and optional — standard OpenAI clients ignore
+    /// it; cortex forwards usage verbatim so it survives proxying.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub helexa_timing: Option<HelexaTiming>,
+}
+
+/// helexa extension carried on [`Usage::helexa_timing`]. Mirrors
+/// neuron's internal `FinishTiming`. All fields are server-measured;
+/// `prefill_tokens` is the prefill-rate denominator.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct HelexaTiming {
+    pub prefill_ms: u64,
+    pub decode_ms: u64,
+    pub prefill_tokens: u64,
 }
 
 /// Sub-counts of `Usage::completion_tokens`.

crates/cortex-core/src/responses.rs

@@ -66,14 +66,48 @@ pub struct ResponsesRequest {
     pub extra: Value,
 }
 
-/// `input` is either a single string or an array of typed items.
+/// `input` is either a single string or an array of items.
 /// `#[serde(untagged)]` so the wire shape `"input": "hi"` and
 /// `"input": [{...}]` both deserialize.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(untagged)]
 pub enum ResponsesInput {
     Text(String),
-    Items(Vec<ResponsesInputItem>),
+    Items(Vec<ResponsesInputElement>),
+}
+
+/// One element of an `input` array.
+///
+/// OpenAI's Responses API accepts three shapes here, and real clients
+/// use all of them — most notably agent-zero (via litellm), which
+/// sends the bare "easy message" form. We must tolerate every shape,
+/// because `input` is an `#[serde(untagged)]` array: a single element
+/// that matches no variant fails the *entire* request with a 422
+/// (`did not match any variant of untagged enum ResponsesInput`).
+///
+/// 1. [`Self::Typed`] — an item carrying an explicit `"type"`
+///    discriminant (`message`, `function_call`, `function_call_output`,
+///    `reasoning`).
+/// 2. [`Self::EasyMessage`] — a bare `{role, content}` with **no**
+///    `type` field. This is OpenAI's `EasyInputMessage` and what
+///    litellm emits for every turn. `content` is optional so an
+///    assistant turn carrying only tool calls (`content: null`) still
+///    parses.
+/// 3. [`Self::Other`] — anything else, captured as raw JSON and
+///    dropped during translation. This is the forward-compat escape
+///    hatch that mirrors [`ResponsesRequest::extra`] at the item
+///    level: an unmodeled item type can never again reject the whole
+///    request.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum ResponsesInputElement {
+    Typed(ResponsesInputItem),
+    EasyMessage {
+        role: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        content: Option<ResponsesMessageContent>,
+    },
+    Other(Value),
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -91,8 +125,11 @@ pub enum ResponsesInputItem {
         name: String,
         arguments: String,
     },
-    /// User is feeding a tool result back into the model.
-    FunctionCallOutput { call_id: String, output: String },
+    /// User is feeding a tool result back into the model. `output`
+    /// is a `Value` because OpenAI allows it to be either a plain
+    /// string or an array of content parts; the translator renders
+    /// either form to text rather than losing the tool result.
+    FunctionCallOutput { call_id: String, output: Value },
     /// Reasoning items emitted by o-series models. Accepted but
     /// not forwarded to the model — neuron's candle path doesn't
     /// surface reasoning separately yet.
@@ -132,6 +169,11 @@ pub enum ResponsesContentPart {
         #[serde(default, skip_serializing_if = "Vec::is_empty")]
         annotations: Vec<Value>,
     },
+    /// Any content-part type we don't model (e.g. `refusal`, audio).
+    /// Captured as a unit so an unknown part can't reject the whole
+    /// request; dropped during translation.
+    #[serde(other)]
+    Unknown,
 }
 
 // ── Response (non-streaming) ─────────────────────────────────────────
@@ -277,20 +319,116 @@ mod tests {
             ResponsesInput::Items(items) => {
                 assert_eq!(items.len(), 1);
                 match &items[0] {
-                    ResponsesInputItem::Message { role, content } => {
+                    ResponsesInputElement::Typed(ResponsesInputItem::Message { role, content }) => {
                         assert_eq!(role, "user");
                         match content {
                             ResponsesMessageContent::Text(t) => assert_eq!(t, "hi"),
                             other => panic!("expected Text content, got {other:?}"),
                         }
                     }
-                    other => panic!("expected Message item, got {other:?}"),
+                    other => panic!("expected typed Message item, got {other:?}"),
                 }
             }
             other => panic!("expected Items, got {other:?}"),
         }
     }
 
+    #[test]
+    fn deserialises_bare_easy_message_without_type() {
+        // The shape agent-zero (via litellm) actually sends: `input`
+        // items are bare `{role, content}` with NO `type` field. This
+        // is the exact payload that was returning 422.
+        let raw = r#"{
+            "model": "Qwen/Qwen3.6-27B",
+            "store": true,
+            "tools": [{"type": "function", "name": "x", "description": "d", "parameters": {}}],
+            "input": [
+                {"role": "system", "content": "you are helpful"},
+                {"role": "assistant", "content": "{\"tool_name\":\"response\"}"},
+                {"role": "user", "content": "hi"}
+            ]
+        }"#;
+        let req: ResponsesRequest = serde_json::from_str(raw).unwrap();
+        let items = match req.input {
+            ResponsesInput::Items(i) => i,
+            other => panic!("expected Items, got {other:?}"),
+        };
+        assert_eq!(items.len(), 3);
+        for el in &items {
+            assert!(
+                matches!(el, ResponsesInputElement::EasyMessage { .. }),
+                "expected EasyMessage, got {el:?}"
+            );
+        }
+        // `tools` / `store` ride through `extra`, not `input`.
+        assert!(req.extra.get("tools").is_some());
+        assert_eq!(req.extra.get("store"), Some(&Value::Bool(true)));
+    }
+
+    #[test]
+    fn tolerates_null_content_and_unknown_item_types() {
+        // An assistant turn carrying only tool calls has `content: null`;
+        // and a future/unmodeled item type must not 422 the request.
+        let raw = r#"{
+            "model": "m",
+            "input": [
+                {"role": "assistant", "content": null},
+                {"type": "item_reference", "id": "abc"},
+                {"type": "function_call_output", "call_id": "c1",
+                 "output": [{"type": "output_text", "text": "result"}]},
+                {"role": "user", "content": "go"}
+            ]
+        }"#;
+        let req: ResponsesRequest = serde_json::from_str(raw).unwrap();
+        let items = match req.input {
+            ResponsesInput::Items(i) => i,
+            other => panic!("expected Items, got {other:?}"),
+        };
+        assert_eq!(items.len(), 4);
+        assert!(matches!(
+            &items[0],
+            ResponsesInputElement::EasyMessage { content: None, .. }
+        ));
+        assert!(matches!(&items[1], ResponsesInputElement::Other(_)));
+        assert!(matches!(
+            &items[2],
+            ResponsesInputElement::Typed(ResponsesInputItem::FunctionCallOutput { .. })
+        ));
+        assert!(matches!(
+            &items[3],
+            ResponsesInputElement::EasyMessage { .. }
+        ));
+    }
+
+    #[test]
+    fn tolerates_unknown_content_part_type() {
+        // A `refusal` (or any unmodeled) content part must parse, not 422.
+        let raw = r#"{
+            "model": "m",
+            "input": [
+                {"role": "assistant", "content": [
+                    {"type": "refusal", "refusal": "no"},
+                    {"type": "output_text", "text": "ok"}
+                ]}
+            ]
+        }"#;
+        let req: ResponsesRequest = serde_json::from_str(raw).unwrap();
+        let items = match req.input {
+            ResponsesInput::Items(i) => i,
+            other => panic!("expected Items, got {other:?}"),
+        };
+        let parts = match &items[0] {
+            ResponsesInputElement::EasyMessage {
+                content: Some(ResponsesMessageContent::Parts(p)),
+                ..
+            } => p,
+            other => panic!("expected EasyMessage with Parts, got {other:?}"),
+        };
+        assert_eq!(parts.len(), 2);
+        assert!(matches!(&parts[0], ResponsesContentPart::Unknown));
+        assert!(matches!(&parts[1], ResponsesContentPart::OutputText { .. }));
+    }
+
     #[test]
     fn deserialises_input_with_image() {
         let raw = r#"{
@@ -308,10 +446,10 @@ mod tests {
             other => panic!("expected Items, got {other:?}"),
         };
         let parts = match &items[0] {
-            ResponsesInputItem::Message {
+            ResponsesInputElement::Typed(ResponsesInputItem::Message {
                 content: ResponsesMessageContent::Parts(p),
                 ..
-            } => p,
+            }) => p,
             other => panic!("expected Parts, got {other:?}"),
         };
         assert_eq!(parts.len(), 2);

crates/cortex-core/src/translate.rs

@@ -400,6 +400,7 @@ pub fn openai_to_anthropic(resp: ChatCompletionResponse) -> MessagesResponse {
         total_tokens: 0,
         completion_tokens_details: None,
         prompt_tokens_details: None,
+        helexa_timing: None,
     });
 
     MessagesResponse {
@@ -772,6 +773,7 @@ mod stream_tests {
             total_tokens: 267,
             completion_tokens_details: None,
             prompt_tokens_details: None,
+            helexa_timing: None,
         });
         t.on_chunk(&usage_chunk);
         let fin = t.finish();

crates/helexa-bench/src/report.rs

@@ -9,22 +9,26 @@ use anyhow::Result;
 pub fn render_markdown(rows: &[ReportRow]) -> String {
     let mut out = String::new();
     out.push_str(
-        "| engine | model | prompt tok | TTFT (s) | decode tok/s | total (s) | build | n |\n",
+        "| engine | model | prompt tok | prefill tok/s | TTFT (s) | TTFT p95 | \
+         decode tok/s | total (s) | total p95 | build | n |\n",
     );
-    out.push_str("|---|---|---:|---:|---:|---:|---|---:|\n");
+    out.push_str("|---|---|---:|---:|---:|---:|---:|---:|---:|---|---:|\n");
     for r in rows {
         let ptok = r
             .prompt_tokens
             .map(|t| t.to_string())
             .unwrap_or_else(|| format!("~{}", r.prompt_size_approx));
         out.push_str(&format!(
-            "| {} | {} | {} | {} | {} | {} | `{}` | {} |\n",
+            "| {} | {} | {} | {} | {} | {} | {} | {} | {} | `{}` | {} |\n",
             r.target_name,
             r.model_id,
             ptok,
+            fmt_opt(r.prefill_tps_median, 1),
             fmt_opt(r.ttft_s_median, 3),
+            fmt_opt(r.ttft_s_p95, 3),
             fmt_opt(r.decode_tps_median, 1),
             fmt_opt(r.total_s_median, 3),
+            fmt_opt(r.total_s_p95, 3),
             r.git_sha,
             r.samples,
         ));
@@ -43,8 +47,15 @@ pub fn render_json(rows: &[ReportRow]) -> Result<String> {
                 "prompt_size_approx": r.prompt_size_approx,
                 "prompt_tokens": r.prompt_tokens,
                 "ttft_s_median": r.ttft_s_median,
+                "ttft_s_p95": r.ttft_s_p95,
+                "ttft_s_p99": r.ttft_s_p99,
                 "decode_tps_median": r.decode_tps_median,
                 "total_s_median": r.total_s_median,
+                "total_s_p95": r.total_s_p95,
+                "total_s_p99": r.total_s_p99,
+                "prefill_ms_median": r.prefill_ms_median,
+                "decode_ms_median": r.decode_ms_median,
+                "prefill_tps_median": r.prefill_tps_median,
                 "git_sha": r.git_sha,
                 "samples": r.samples,
                 "gpu": r.gpu,
@@ -77,14 +88,24 @@ mod tests {
             ttft_s_median: Some(0.123),
             decode_tps_median: Some(45.6),
             total_s_median: Some(1.234),
+            ttft_s_p95: Some(0.222),
+            ttft_s_p99: Some(0.250),
+            total_s_p95: Some(1.5),
+            total_s_p99: Some(1.6),
+            prefill_ms_median: Some(120.0),
+            decode_ms_median: Some(1100.0),
+            prefill_tps_median: Some(1066.7),
             samples: 5,
             gpu: Some("2× RTX 5090".into()),
         }];
         let md = render_markdown(&rows);
         assert!(md.contains("| engine |"));
+        assert!(md.contains("prefill tok/s"));
         assert!(md.contains("beast"));
         assert!(md.contains("`30d50d6`"));
         assert!(md.contains("0.123"));
+        // p95 column rendered.
+        assert!(md.contains("0.222"));
     }
 
     #[test]
@@ -99,6 +120,13 @@ mod tests {
             ttft_s_median: Some(0.1),
             decode_tps_median: None,
             total_s_median: Some(0.5),
+            ttft_s_p95: Some(0.1),
+            ttft_s_p99: Some(0.1),
+            total_s_p95: Some(0.5),
+            total_s_p99: Some(0.5),
+            prefill_ms_median: None,
+            decode_ms_median: None,
+            prefill_tps_median: None,
             samples: 1,
             gpu: None,
         }];

crates/helexa-bench/src/scenario.rs

@@ -62,6 +62,16 @@ pub struct ScenarioMetrics {
     pub prompt_tokens: Option<u64>,
     /// Completion tokens: from `usage` when present, else content-chunk count.
     pub completion_tokens: u64,
+    /// Server-measured prefill duration (ms), from the `usage.helexa_timing`
+    /// extension (#85). `None` when the server didn't emit it (external
+    /// engines, non-instrumented paths). The honest prefill-phase number,
+    /// distinct from client-observed `ttft_s` which also includes request
+    /// setup + first-byte network latency.
+    pub prefill_ms: Option<u64>,
+    /// Server-measured decode duration (ms), from `usage.helexa_timing`.
+    pub decode_ms: Option<u64>,
+    /// Tokens submitted to prefill — the denominator for prefill tok/s.
+    pub prefill_tokens: Option<u64>,
 }
 
 #[async_trait]
@@ -160,6 +170,9 @@ async fn stream_and_measure(
     let mut chunk_count: u64 = 0;
     let mut prompt_tokens: Option<u64> = None;
     let mut completion_tokens: Option<u64> = None;
+    let mut prefill_ms: Option<u64> = None;
+    let mut decode_ms: Option<u64> = None;
+    let mut prefill_tokens: Option<u64> = None;
 
     while let Some(event) = stream.next().await {
         let event = event.context("reading SSE stream")?;
@@ -188,6 +201,11 @@ async fn stream_and_measure(
         if let Some(usage) = chunk.usage {
             prompt_tokens = Some(usage.prompt_tokens);
             completion_tokens = Some(usage.completion_tokens);
+            if let Some(t) = usage.helexa_timing {
+                prefill_ms = Some(t.prefill_ms);
+                decode_ms = Some(t.decode_ms);
+                prefill_tokens = Some(t.prefill_tokens);
+            }
         }
     }
     let end = Instant::now();
@@ -212,6 +230,9 @@ async fn stream_and_measure(
         total_s: (end - start).as_secs_f64(),
         prompt_tokens,
         completion_tokens: tokens,
+        prefill_ms,
+        decode_ms,
+        prefill_tokens,
     })
 }
 

crates/helexa-bench/src/store.rs

@@ -51,6 +51,11 @@ pub struct RunRecord {
     pub decode_tps: Option<f64>,
     pub total_s: Option<f64>,
     pub completion_tokens: Option<u64>,
+    // server-measured prefill/decode split (#85), null on engines/paths
+    // that don't emit `usage.helexa_timing`.
+    pub prefill_ms: Option<u64>,
+    pub decode_ms: Option<u64>,
+    pub prefill_tokens: Option<u64>,
     // outcome
     pub ok: bool,
     pub error: Option<String>,
@@ -123,6 +128,9 @@ impl Store {
                 decode_tps           REAL,
                 total_s              REAL,
                 completion_tokens    INTEGER,
+                prefill_ms           INTEGER,
+                decode_ms            INTEGER,
+                prefill_tokens       INTEGER,
                 ok                   INTEGER NOT NULL,
                 error                TEXT
             );
@@ -133,6 +141,39 @@ impl Store {
             "#,
         )
         .context("initialising sqlite schema")?;
+        // Additive migrations for DBs created before a column existed.
+        // `CREATE TABLE IF NOT EXISTS` above only seeds fresh DBs; existing
+        // ones need the columns backfilled (as NULL) so older rows coexist
+        // with new metrics. There is no migration framework — each entry is
+        // an idempotent "add if missing".
+        Self::ensure_columns(
+            conn,
+            "runs",
+            &[
+                ("prefill_ms", "INTEGER"),
+                ("decode_ms", "INTEGER"),
+                ("prefill_tokens", "INTEGER"),
+            ],
+        )?;
+        Ok(())
+    }
+
+    /// Add any of `columns` that the table is missing (`ALTER TABLE ADD
+    /// COLUMN`). Idempotent: existing columns are read from
+    /// `PRAGMA table_info` and skipped, so this is safe to run on every open.
+    fn ensure_columns(conn: &Connection, table: &str, columns: &[(&str, &str)]) -> Result<()> {
+        let mut existing = std::collections::HashSet::new();
+        let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
+        let names = stmt.query_map([], |row| row.get::<_, String>(1))?;
+        for name in names {
+            existing.insert(name?);
+        }
+        for (name, ty) in columns {
+            if !existing.contains(*name) {
+                conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {name} {ty};"))
+                    .with_context(|| format!("adding column {table}.{name}"))?;
+            }
+        }
         Ok(())
     }
 
@@ -166,6 +207,7 @@ impl Store {
                 model_id, harness, capabilities_json, devices_json,
                 scenario_id, prompt_size_approx, prompt_tokens_actual, max_tokens,
                 ttft_s, decode_tps, total_s, completion_tokens,
+                prefill_ms, decode_ms, prefill_tokens,
                 ok, error
             ) VALUES (
                 ?1, ?2, ?3, ?4,
@@ -176,7 +218,8 @@ impl Store {
                 ?20, ?21, ?22, ?23,
                 ?24, ?25, ?26, ?27,
                 ?28, ?29, ?30, ?31,
-                ?32, ?33
+                ?32, ?33, ?34,
+                ?35, ?36
             )",
             params![
                 r.ts,
@@ -210,6 +253,9 @@ impl Store {
                 r.decode_tps,
                 r.total_s,
                 r.completion_tokens,
+                r.prefill_ms,
+                r.decode_ms,
+                r.prefill_tokens,
                 r.ok as i64,
                 r.error,
             ],
@@ -224,7 +270,8 @@ impl Store {
         // successful run, then median that SHA's samples.
         let mut stmt = self.conn.prepare(
             "SELECT target_name, model_id, scenario_id, prompt_size_approx, git_sha,
-                    ttft_s, decode_tps, total_s, prompt_tokens_actual, gpus_json
+                    ttft_s, decode_tps, total_s, prompt_tokens_actual, gpus_json,
+                    prefill_ms, decode_ms, prefill_tokens
              FROM runs
              WHERE ok=1
              ORDER BY target_name, model_id, scenario_id, id",
@@ -241,6 +288,9 @@ impl Store {
                 total_s: row.get(7)?,
                 prompt_tokens_actual: row.get(8)?,
                 gpus_json: row.get(9)?,
+                prefill_ms: row.get(10)?,
+                decode_ms: row.get(11)?,
+                prefill_tokens: row.get(12)?,
             })
         })?;
         let raws: Vec<RawRow> = rows.collect::<rusqlite::Result<_>>()?;
@@ -379,7 +429,7 @@ impl Store {
             "SELECT id, ts, target_name, hostname, git_sha, build_timestamp, package_version,
                     model_id, harness, scenario_id, prompt_size_approx, prompt_tokens_actual,
                     max_tokens, ttft_s, decode_tps, total_s, completion_tokens, ok, error,
-                    gpus_json
+                    gpus_json, prefill_ms, decode_ms, prefill_tokens
              FROM runs",
         );
         let mut conds: Vec<String> = Vec::new();
@@ -435,6 +485,9 @@ impl Store {
                     completion_tokens: r.get(16)?,
                     ok: r.get::<_, i64>(17)? != 0,
                     error: r.get(18)?,
+                    prefill_ms: r.get(20)?,
+                    decode_ms: r.get(21)?,
+                    prefill_tokens: r.get(22)?,
                 })
             })?
             .collect::<rusqlite::Result<_>>()?;
@@ -554,6 +607,9 @@ pub struct RunRow {
     pub decode_tps: Option<f64>,
     pub total_s: Option<f64>,
     pub completion_tokens: Option<u64>,
+    pub prefill_ms: Option<u64>,
+    pub decode_ms: Option<u64>,
+    pub prefill_tokens: Option<u64>,
     pub ok: bool,
     pub error: Option<String>,
 }
@@ -569,6 +625,9 @@ struct RawRow {
     total_s: Option<f64>,
     prompt_tokens_actual: Option<u64>,
     gpus_json: Option<String>,
+    prefill_ms: Option<u64>,
+    decode_ms: Option<u64>,
+    prefill_tokens: Option<u64>,
 }
 
 /// An aggregated cell ready for the report table.
@@ -583,6 +642,19 @@ pub struct ReportRow {
     pub ttft_s_median: Option<f64>,
     pub decode_tps_median: Option<f64>,
     pub total_s_median: Option<f64>,
+    /// Latency tail percentiles — where batch-1 pain actually shows up, and
+    /// invisible behind a bare median. p95/p99 nearest-rank; with few
+    /// samples they collapse toward the max (honest, not interpolated).
+    pub ttft_s_p95: Option<f64>,
+    pub ttft_s_p99: Option<f64>,
+    pub total_s_p95: Option<f64>,
+    pub total_s_p99: Option<f64>,
+    /// Server-measured prefill/decode split (#85). `prefill_tps_median` is
+    /// the true prompt-encoding rate (prefill_tokens / prefill_ms),
+    /// complementing `decode_tps_median` (the generation rate).
+    pub prefill_ms_median: Option<f64>,
+    pub decode_ms_median: Option<f64>,
+    pub prefill_tps_median: Option<f64>,
     pub samples: usize,
     /// Public-facing resource name (the host's GPU(s)), e.g. "2× RTX 5090".
     pub gpu: Option<String>,
@@ -611,6 +683,11 @@ fn aggregate(raws: Vec<RawRow>) -> Vec<ReportRow> {
         let latest_sha = rows.last().map(|r| r.git_sha.clone()).unwrap_or_default();
         let cell: Vec<&RawRow> = rows.iter().filter(|r| r.git_sha == latest_sha).collect();
         let prompt_size_approx = cell.first().map(|r| r.prompt_size_approx).unwrap_or(0);
+        // Per-row prefill tok/s, derived from the server-measured split.
+        let prefill_tps = |r: &&RawRow| match (r.prefill_tokens, r.prefill_ms) {
+            (Some(tok), Some(ms)) if ms > 0 => Some(tok as f64 * 1000.0 / ms as f64),
+            _ => None,
+        };
         out.push(ReportRow {
             target_name,
             model_id,
@@ -621,6 +698,13 @@ fn aggregate(raws: Vec<RawRow>) -> Vec<ReportRow> {
             ttft_s_median: median(cell.iter().filter_map(|r| r.ttft_s)),
             decode_tps_median: median(cell.iter().filter_map(|r| r.decode_tps)),
             total_s_median: median(cell.iter().filter_map(|r| r.total_s)),
+            ttft_s_p95: percentile(cell.iter().filter_map(|r| r.ttft_s), 95.0),
+            ttft_s_p99: percentile(cell.iter().filter_map(|r| r.ttft_s), 99.0),
+            total_s_p95: percentile(cell.iter().filter_map(|r| r.total_s), 95.0),
+            total_s_p99: percentile(cell.iter().filter_map(|r| r.total_s), 99.0),
+            prefill_ms_median: median(cell.iter().filter_map(|r| r.prefill_ms.map(|m| m as f64))),
+            decode_ms_median: median(cell.iter().filter_map(|r| r.decode_ms.map(|m| m as f64))),
+            prefill_tps_median: median(cell.iter().filter_map(prefill_tps)),
             samples: cell.len(),
             gpu: cell
                 .iter()
@@ -680,6 +764,22 @@ fn median(values: impl Iterator<Item = f64>) -> Option<f64> {
     Some((v[lo] + v[hi]) / 2.0)
 }
 
+/// Nearest-rank percentile (`p` in 0..=100). Chosen over interpolation
+/// because bench cells hold only a handful of samples: with n=5, p95/p99
+/// resolve to the max, which honestly says "this is the worst we saw"
+/// rather than inventing a value between samples we never observed.
+fn percentile(values: impl Iterator<Item = f64>, p: f64) -> Option<f64> {
+    let mut v: Vec<f64> = values.collect();
+    if v.is_empty() {
+        return None;
+    }
+    v.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
+    // rank = ceil(p/100 * n), clamped to [1, n]; index is rank-1.
+    let rank = (p / 100.0 * v.len() as f64).ceil() as usize;
+    let idx = rank.clamp(1, v.len()) - 1;
+    Some(v[idx])
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -717,6 +817,9 @@ mod tests {
             decode_tps: Some(50.0),
             total_s: Some(1.0),
             completion_tokens: Some(50),
+            prefill_ms: Some(200),
+            decode_ms: Some(1000),
+            prefill_tokens: Some(130),
             ok,
             error: if ok { None } else { Some("boom".into()) },
         }
@@ -755,6 +858,66 @@ mod tests {
         assert!((rows[0].ttft_s_median.unwrap() - 0.3).abs() < 1e-9);
     }
 
+    #[test]
+    fn report_surfaces_percentiles_and_prefill_split() {
+        let s = Store::open_in_memory().unwrap();
+        // Five samples on one cell with spread TTFT so percentiles differ
+        // from the median, plus a server-measured prefill/decode split.
+        for (i, ttft) in [0.10, 0.12, 0.14, 0.16, 0.50].iter().enumerate() {
+            let mut r = rec("beast", "sha", "m", "chat:128", true);
+            r.ttft_s = Some(*ttft);
+            r.total_s = Some(ttft + 1.0);
+            r.prefill_ms = Some(200 + i as u64);
+            r.prefill_tokens = Some(400);
+            s.insert_run(&r).unwrap();
+        }
+        let rows = s.report_rows().unwrap();
+        assert_eq!(rows.len(), 1);
+        let row = &rows[0];
+        assert_eq!(row.samples, 5);
+        // p50 is the middle value; p95/p99 (nearest-rank, n=5) hit the max.
+        assert!((row.ttft_s_median.unwrap() - 0.14).abs() < 1e-9);
+        assert!((row.ttft_s_p95.unwrap() - 0.50).abs() < 1e-9);
+        assert!((row.ttft_s_p99.unwrap() - 0.50).abs() < 1e-9);
+        // prefill tok/s = 400 tok / ~0.2 s ≈ 2000 tok/s.
+        assert!(row.prefill_tps_median.unwrap() > 1900.0);
+        assert!(row.prefill_ms_median.is_some());
+    }
+
+    #[test]
+    fn percentile_nearest_rank() {
+        let vals = || [1.0, 2.0, 3.0, 4.0, 5.0].into_iter();
+        assert_eq!(percentile(vals(), 50.0), Some(3.0));
+        assert_eq!(percentile(vals(), 95.0), Some(5.0));
+        assert_eq!(percentile(vals(), 99.0), Some(5.0));
+        assert_eq!(percentile(std::iter::empty(), 95.0), None);
+    }
+
+    #[test]
+    fn migration_is_idempotent_and_backfills() {
+        // A DB whose `runs` table predates the prefill columns: create the
+        // pre-#85 shape, insert a row, then run ensure_columns twice.
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute_batch(
+            "CREATE TABLE runs (id INTEGER PRIMARY KEY, ttft_s REAL);
+             INSERT INTO runs (ttft_s) VALUES (0.1);",
+        )
+        .unwrap();
+        for _ in 0..2 {
+            Store::ensure_columns(
+                &conn,
+                "runs",
+                &[("prefill_ms", "INTEGER"), ("decode_ms", "INTEGER")],
+            )
+            .unwrap();
+        }
+        // Columns now exist and the old row reads them back as NULL.
+        let got: Option<i64> = conn
+            .query_row("SELECT prefill_ms FROM runs", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(got, None);
+    }
+
     #[test]
     fn gpu_label_formats() {
         let two = r#"[{"name":"NVIDIA GeForce RTX 5090"},{"name":"NVIDIA GeForce RTX 5090"}]"#;

crates/helexa-bench/src/sweep.rs

@@ -9,7 +9,7 @@
 
 use crate::client::TargetClient;
 use crate::config::{BenchConfig, TargetConfig, TargetKind};
-use crate::scenario::{RunCtx, build_scenarios};
+use crate::scenario::{RunCtx, ScenarioMetrics, build_scenarios};
 use crate::store::{RunRecord, Store};
 use anyhow::Result;
 use cortex_core::build_info::BuildInfo;
@@ -187,18 +187,11 @@ impl Sweeper {
         prompt_size: u32,
         result: Result<&crate::scenario::ScenarioMetrics, &str>,
     ) -> RunRecord {
-        let (ok, error, ttft, decode, total, prompt_tokens, completion) = match result {
-            Ok(m) => (
-                true,
-                None,
-                Some(m.ttft_s),
-                m.decode_tps,
-                Some(m.total_s),
-                m.prompt_tokens,
-                Some(m.completion_tokens),
-            ),
-            Err(e) => (false, Some(e.to_string()), None, None, None, None, None),
+        let (m, error): (Option<&ScenarioMetrics>, Option<String>) = match result {
+            Ok(m) => (Some(m), None),
+            Err(e) => (None, Some(e.to_string())),
         };
+        let ok = m.is_some();
 
         RunRecord {
             ts: chrono::Utc::now().to_rfc3339(),
@@ -230,12 +223,15 @@ impl Sweeper {
                 .unwrap_or_else(|_| "[]".to_string()),
             scenario_id: scenario_id.to_string(),
             prompt_size_approx: prompt_size,
-            prompt_tokens_actual: prompt_tokens,
+            prompt_tokens_actual: m.and_then(|m| m.prompt_tokens),
             max_tokens: self.cfg.scenarios.max_tokens,
-            ttft_s: ttft,
-            decode_tps: decode,
-            total_s: total,
-            completion_tokens: completion,
+            ttft_s: m.map(|m| m.ttft_s),
+            decode_tps: m.and_then(|m| m.decode_tps),
+            total_s: m.map(|m| m.total_s),
+            completion_tokens: m.map(|m| m.completion_tokens),
+            prefill_ms: m.and_then(|m| m.prefill_ms),
+            decode_ms: m.and_then(|m| m.decode_ms),
+            prefill_tokens: m.and_then(|m| m.prefill_tokens),
             ok,
             error,
         }

crates/helexa-bench/tests/api.rs

@@ -46,6 +46,9 @@ fn rec(
         decode_tps: if ok { Some(30.0) } else { None },
         total_s: if ok { Some(2.0) } else { None },
         completion_tokens: if ok { Some(60) } else { None },
+        prefill_ms: if ok { Some(150) } else { None },
+        decode_ms: if ok { Some(1800) } else { None },
+        prefill_tokens: if ok { Some(130) } else { None },
         ok,
         error: if ok { None } else { Some("boom".into()) },
     }

crates/neuron/src/api.rs

@@ -16,7 +16,7 @@ use cortex_core::discovery::{DiscoveryResponse, HealthResponse};
 use cortex_core::entitlements::{HEADER_ACCOUNT_ID, HEADER_KEY_ID};
 use cortex_core::harness::ModelSpec;
 use cortex_core::openai::{ChatCompletionRequest, MessageContent};
-use cortex_core::responses::{ResponsesRequest, ResponsesUsage};
+use cortex_core::responses::{OutputTokensDetails, ResponsesRequest, ResponsesUsage};
 use futures::stream::{self, StreamExt};
 use serde_json::{Value, json};
 use std::convert::Infallible;
@@ -418,8 +418,14 @@ async fn responses(
                     input_tokens: u.prompt_tokens,
                     output_tokens: u.completion_tokens,
                     total_tokens: u.prompt_tokens + u.completion_tokens,
-                    // Non-streaming reasoning accounting deferred (#64).
-                    output_tokens_details: None,
+                    // Carry the reasoning sub-count through from the chat
+                    // usage — the non-streaming path now splits off the
+                    // `<think>` span and counts it (see `split_off_reasoning`).
+                    output_tokens_details: u.completion_tokens_details.as_ref().map(|d| {
+                        OutputTokensDetails {
+                            reasoning_tokens: d.reasoning_tokens,
+                        }
+                    }),
                     input_tokens_details: None,
                 });
                 let meta = openai_responses::ResponseMeta {

crates/neuron/src/harness/candle.rs

@@ -23,11 +23,11 @@ use candle_transformers::models::qwen3_moe as qwen3_moe_dense;
 use cortex_core::harness::{Harness, HarnessHealth, ModelInfo, ModelSpec};
 use cortex_core::openai::{
     ChatCompletionChoice, ChatCompletionChunk, ChatCompletionRequest, ChatCompletionResponse,
-    ChatMessage, MessageContent, Usage,
+    ChatMessage, CompletionTokensDetails, MessageContent, Usage,
 };
 
 use crate::wire::{
-    FinishReason, InferenceEvent, ReasoningTokenPair, ToolCallTokenPair,
+    FinishReason, FinishTiming, InferenceEvent, ReasoningTokenPair, ToolCallTokenPair,
     detect_reasoning_token_pair, detect_tool_call_token_pair, openai_chat as wire_chat,
 };
 use std::collections::HashMap;
@@ -784,6 +784,39 @@ impl ModelArch {
     }
 }
 
+/// Split a non-streaming completion's generated tokens into the
+/// visible answer and the leading reasoning span.
+///
+/// Reasoning models (Qwen3 `<think>`, DeepSeek-R1, …) emit their
+/// chain-of-thought *before* the answer, and the chat template injects
+/// the **opening** marker into the prompt — so the generated tokens look
+/// like `…reasoning… </think> …answer…` with no opening marker present
+/// in the output. The streaming path drops reasoning as
+/// [`InferenceEvent::ReasoningDelta`]; the non-streaming path has to do
+/// the equivalent post-hoc or the chain-of-thought leaks into the
+/// assistant `content` (which broke agent-zero v2.0, whose parser
+/// expected the bare JSON answer, not a `<think>` preamble).
+///
+/// Returns `(content_ids, reasoning_token_count)`. Strategy: if the model
+/// declares a reasoning marker pair and its **close** token appears in
+/// `generated_ids`, everything up to and including the last close token is
+/// reasoning and only the tail is the answer. Otherwise (non-reasoning
+/// model, thinking disabled, or a generation truncated mid-reasoning) the
+/// tokens are returned unchanged. Splitting on the token id — not a
+/// decoded `</think>` string — keeps this robust against tokenizer
+/// byte-fallback and special-token handling.
+fn split_off_reasoning<'a>(
+    generated_ids: &'a [u32],
+    reasoning: Option<&ReasoningTokenPair>,
+) -> (&'a [u32], u64) {
+    if let Some(pair) = reasoning
+        && let Some(idx) = generated_ids.iter().rposition(|&t| t == pair.close_id)
+    {
+        return (&generated_ids[idx + 1..], (idx + 1) as u64);
+    }
+    (generated_ids, 0)
+}
+
 /// Squeeze any leading singleton dims off the logits tensor so the
 /// caller gets a rank-1 `[vocab_size]` slice ready for sampling. Bails
 /// on a non-singleton leading dim (would mean a batched forward, which
@@ -2454,21 +2487,36 @@ impl CandleHarness {
                 )));
             };
 
+            // Strip the leading `<think>` span so the chain-of-thought
+            // doesn't leak into `content` (the streaming path drops it
+            // as ReasoningDelta; this is the non-streaming equivalent).
+            let (content_ids, reasoning_tokens) =
+                split_off_reasoning(&generated_ids, loaded.reasoning_tokens.as_ref());
             let completion_text = loaded
                 .tokenizer
-                .decode(&generated_ids, true)
+                .decode(content_ids, true)
                 .map_err(|e| InferenceError::Other(anyhow::anyhow!("detokenize: {e}")))?;
+            // The first answer token after `</think>` is usually a
+            // newline pair; trim it so `content` starts at the answer.
+            let completion_text = if reasoning_tokens > 0 {
+                completion_text.trim_start().to_string()
+            } else {
+                completion_text
+            };
 
             let usage = Usage {
                 prompt_tokens: prompt_len as u64,
                 completion_tokens: generated_ids.len() as u64,
                 total_tokens: (prompt_len + generated_ids.len()) as u64,
-                // Reasoning accounting is streaming-only: the
-                // non-streaming path doesn't track `in_reasoning`
-                // (would require post-hoc <think> span parsing).
-                // Deferred — see #64.
-                completion_tokens_details: None,
+                // `reasoning_tokens` is an additive sub-count of
+                // `completion_tokens` (which still counts every
+                // generated token, reasoning included).
+                completion_tokens_details: (reasoning_tokens > 0)
+                    .then_some(CompletionTokensDetails { reasoning_tokens }),
                 prompt_tokens_details: None,
+                // Non-streaming path: prefill/decode split is only
+                // surfaced on the streaming Finish event today (#85).
+                helexa_timing: None,
             };
 
             tracing::info!(
@@ -3957,6 +4005,11 @@ impl CandleHarness {
                 // call — promotes the terminal finish_reason to ToolCalls
                 // so Anthropic clients see stop_reason: tool_use.
                 let mut emitted_tool_call = false;
+                // Prefill/decode split timers (#85). Declared outside 'work
+                // so the terminal Finish — built after the block exits — can
+                // 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;
 
                 'work: {
                     // Prefix-cache decision (#11): vision requests
@@ -4084,14 +4137,16 @@ impl CandleHarness {
                             break 'work;
                         }
                     };
+                    let prefill_elapsed = prefill_start.elapsed();
+                    prefill_ms_measured = prefill_elapsed.as_millis() as u32;
                     tp_for_task
                         .prefill_rate
-                        .record(prompt_len, prefill_start.elapsed());
+                        .record(prompt_len, prefill_elapsed);
                     let (post_prefill_vram_free_mb, _) = tp_for_task.query_vram().await;
                     tracing::info!(
                         model = %model_id,
                         prompt_len,
-                        prefill_ms = prefill_start.elapsed().as_millis(),
+                        prefill_ms = prefill_elapsed.as_millis(),
                         vram_free_mb = post_prefill_vram_free_mb,
                         "TP chat_completion (stream): prefill complete"
                     );
@@ -4116,6 +4171,8 @@ impl CandleHarness {
                                 break 'work;
                             }
                         };
+                    // Decode-phase timer for the Finish prefill/decode split (#85).
+                    decode_start = Some(std::time::Instant::now());
 
                     if Some(next_token) == eos_id {
                         finish_reason = FinishReason::Stop;
@@ -4393,6 +4450,13 @@ impl CandleHarness {
                             prompt_tokens: prompt_len as u32,
                             completion_tokens: all_tokens.len() as u32,
                             reasoning_tokens: reasoning_token_count,
+                            timing: Some(FinishTiming {
+                                prefill_ms: prefill_ms_measured,
+                                decode_ms: decode_start
+                                    .map(|d| d.elapsed().as_millis() as u32)
+                                    .unwrap_or(0),
+                                prefill_tokens: prompt_len as u32,
+                            }),
                         })
                         .await;
                 }
@@ -4722,19 +4786,31 @@ async fn chat_completion_tp_inner(
     }
     drop(pool);
 
+    // Strip the leading `<think>` span (see `split_off_reasoning` and the
+    // single-GPU path) so the chain-of-thought doesn't leak into `content`.
+    let (content_ids, reasoning_tokens) =
+        split_off_reasoning(&generated, tp.reasoning_tokens.as_ref());
     let completion_text = tp
         .tokenizer
-        .decode(&generated, true)
+        .decode(content_ids, true)
         .map_err(|e| InferenceError::Other(anyhow::anyhow!("detokenize: {e}")))?;
+    let completion_text = if reasoning_tokens > 0 {
+        completion_text.trim_start().to_string()
+    } else {
+        completion_text
+    };
 
     let usage = Usage {
         prompt_tokens: prompt_len as u64,
         completion_tokens: generated.len() as u64,
         total_tokens: (prompt_len + generated.len()) as u64,
-        // Reasoning accounting is streaming-only (non-streaming TP path
-        // doesn't track `in_reasoning`). Deferred — see #64.
-        completion_tokens_details: None,
+        // `reasoning_tokens` is an additive sub-count of `completion_tokens`.
+        completion_tokens_details: (reasoning_tokens > 0)
+            .then_some(CompletionTokensDetails { reasoning_tokens }),
         prompt_tokens_details: None,
+        // Non-streaming path: prefill/decode split is only surfaced on
+        // the streaming Finish event today (#85).
+        helexa_timing: None,
     };
 
     tracing::info!(
@@ -6064,7 +6140,8 @@ async fn stream_inference_via_worker(
             }
         }
     };
-    prefill_rate.record(prefill_prompt_len, prefill_start.elapsed());
+    let prefill_elapsed = prefill_start.elapsed();
+    prefill_rate.record(prefill_prompt_len, prefill_elapsed);
     let logits = Tensor::new(logits_vec.as_slice(), &Device::Cpu)?;
     let mut next_token = match sample_with_penalty(&logits, &all_tokens, &mut logits_processor) {
         Ok(t) => t,
@@ -6077,6 +6154,8 @@ async fn stream_inference_via_worker(
             return Err(e);
         }
     };
+    // Decode-phase timer for the Finish prefill/decode split (#85).
+    let decode_start = std::time::Instant::now();
 
     // Per-token routing. `tokenizers::DecodeStream` carries five
     // generic parameters (`M, N, PT, PP, D`) which makes naming
@@ -6221,6 +6300,11 @@ async fn stream_inference_via_worker(
             prompt_tokens: prompt_tokens.len() as u32,
             completion_tokens: all_tokens.len() as u32,
             reasoning_tokens: reasoning_token_count,
+            timing: Some(FinishTiming {
+                prefill_ms: prefill_elapsed.as_millis() as u32,
+                decode_ms: decode_start.elapsed().as_millis() as u32,
+                prefill_tokens: prefill_prompt_len as u32,
+            }),
         })
         .await;
 
@@ -6355,6 +6439,10 @@ fn run_inference_streaming(
     // See `inference_tp_stream`: promotes finish_reason to ToolCalls.
     let mut emitted_tool_call = false;
 
+    // Time prefill and decode separately so the Finish event can carry
+    // a server-measured prefill/decode split (#85) instead of leaving
+    // the client to infer both from SSE chunk arrival.
+    let prefill_start = std::time::Instant::now();
     let reused = restore_or_clear_local(arch, prefix_cache, prompt_tokens)?;
     // Two-stage prefill around the retokenization-stable snapshot
     // boundary — see `run_inference_via_worker`.
@@ -6373,6 +6461,8 @@ fn run_inference_streaming(
         None => chunked_prefill_local(arch, device, prompt_tokens, reused)?,
     };
     let mut next_token = sample_with_penalty(&logits, &all_tokens, &mut logits_processor)?;
+    let prefill_elapsed = prefill_start.elapsed();
+    let decode_start = std::time::Instant::now();
 
     // Per-token routing block, used at both the prefill-sample
     // tail and the decode loop. Macros are ugly but Rust's
@@ -6481,6 +6571,11 @@ fn run_inference_streaming(
         prompt_tokens: prompt_tokens.len() as u32,
         completion_tokens: all_tokens.len() as u32,
         reasoning_tokens: reasoning_token_count,
+        timing: Some(FinishTiming {
+            prefill_ms: prefill_elapsed.as_millis() as u32,
+            decode_ms: decode_start.elapsed().as_millis() as u32,
+            prefill_tokens: prompt_tokens.len() as u32,
+        }),
     });
     Ok(())
 }
@@ -6505,6 +6600,60 @@ mod tests {
 
     const IM_START: u32 = 999;
 
+    fn think_pair() -> ReasoningTokenPair {
+        ReasoningTokenPair {
+            open_id: 100,
+            close_id: 200,
+            open_text: "<think>".into(),
+            close_text: "</think>".into(),
+        }
+    }
+
+    #[test]
+    fn split_off_reasoning_strips_up_to_close_marker() {
+        // [reasoning_a, reasoning_b, </think>, answer_x, answer_y]
+        let ids = [10, 11, 200, 42, 43];
+        let (content, reasoning) = split_off_reasoning(&ids, Some(&think_pair()));
+        assert_eq!(content, &[42, 43]);
+        assert_eq!(reasoning, 3); // two reasoning tokens + the close marker
+    }
+
+    #[test]
+    fn split_off_reasoning_no_close_marker_returns_all() {
+        // Thinking disabled / model never closed the span: return as-is.
+        let ids = [42, 43, 44];
+        let (content, reasoning) = split_off_reasoning(&ids, Some(&think_pair()));
+        assert_eq!(content, &ids);
+        assert_eq!(reasoning, 0);
+    }
+
+    #[test]
+    fn split_off_reasoning_no_marker_pair_is_noop() {
+        let ids = [1, 2, 3];
+        let (content, reasoning) = split_off_reasoning(&ids, None);
+        assert_eq!(content, &ids);
+        assert_eq!(reasoning, 0);
+    }
+
+    #[test]
+    fn split_off_reasoning_close_at_end_yields_empty_content() {
+        // All reasoning, answer truncated to nothing after the marker.
+        let ids = [10, 11, 200];
+        let (content, reasoning) = split_off_reasoning(&ids, Some(&think_pair()));
+        assert!(content.is_empty());
+        assert_eq!(reasoning, 3);
+    }
+
+    #[test]
+    fn split_off_reasoning_splits_on_last_close_marker() {
+        // Defensive: if the model emits its own <think></think> pair plus
+        // the prompt-injected one, split on the LAST close marker.
+        let ids = [200, 10, 200, 42];
+        let (content, reasoning) = split_off_reasoning(&ids, Some(&think_pair()));
+        assert_eq!(content, &[42]);
+        assert_eq!(reasoning, 3);
+    }
+
     #[test]
     fn stable_snapshot_cut_lands_after_last_im_start() {
         // ChatML shape: [im_start, "system", ..., im_start, "user",

crates/neuron/src/wire/event.rs

@@ -84,9 +84,38 @@ pub enum InferenceEvent {
         /// `output_tokens_details.reasoning_tokens` (responses).
         /// Zero for non-reasoning models.
         reasoning_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
+        /// this as a `helexa_timing` extension on the OpenAI `usage`
+        /// object so the bench harness can compute true prefill vs
+        /// decode tok/s instead of inferring both from client-side
+        /// SSE arrival (#85).
+        timing: Option<FinishTiming>,
     },
 }
 
+/// Server-measured timing for one completed inference, attached to
+/// [`InferenceEvent::Finish`]. The whole point is to separate the two
+/// phases the client cannot tell apart from chunk-arrival timing:
+/// prefill (tokenize + prompt forward pass, ending at the first
+/// sampled token) and decode (every subsequent token through EOS /
+/// `max_tokens`).
+#[derive(Debug, Clone, Copy)]
+pub struct FinishTiming {
+    /// Wall-clock of the prefill phase in milliseconds: from the start
+    /// of the prompt forward pass(es) to the first sampled token.
+    pub prefill_ms: u32,
+    /// Wall-clock of the decode phase in milliseconds: from the first
+    /// sampled token to stream end.
+    pub decode_ms: u32,
+    /// Prompt tokens submitted to the prefill forward pass — the
+    /// denominator for prefill tok/s. With prefix-KV-cache hits (#11)
+    /// 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,
+}
+
 /// Why a stream stopped. Stays small on purpose — anything that
 /// doesn't map cleanly to one of these collapses to [`Stop`].
 ///

crates/neuron/src/wire/mod.rs

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

crates/neuron/src/wire/openai_chat.rs

@@ -26,11 +26,13 @@
 //! producer blocks on its own send. The bounded channels
 //! propagate without us writing any logic.
 
-use cortex_core::openai::{ChatCompletionChunk, ChunkChoice, CompletionTokensDetails, Usage};
+use cortex_core::openai::{
+    ChatCompletionChunk, ChunkChoice, CompletionTokensDetails, HelexaTiming, Usage,
+};
 use serde_json::json;
 use tokio::sync::mpsc;
 
-use super::event::{FinishReason, InferenceEvent, ReasoningTokenPair};
+use super::event::{FinishReason, FinishTiming, InferenceEvent, ReasoningTokenPair};
 
 /// Output channel buffer size. Mirrors the input side's bound; one
 /// event maps to at most one chunk, so equal capacity keeps the
@@ -193,12 +195,14 @@ pub fn project_chat_stream_with(
                     prompt_tokens,
                     completion_tokens,
                     reasoning_tokens,
+                    timing,
                 } => {
                     // The finish_reason chunk, then an OpenAI-style
                     // usage-only chunk (`choices: []`, `usage` populated).
                     // Clients (opencode) read this to track context size;
                     // cortex's Anthropic translator also picks `usage` up
-                    // for its `message_delta`.
+                    // for its `message_delta`. `timing` rides along as the
+                    // `helexa_timing` usage extension for the bench harness (#85).
                     vec![
                         final_chunk(&id, created, &model_id, reason),
                         usage_chunk(
@@ -208,6 +212,7 @@ pub fn project_chat_stream_with(
                             prompt_tokens,
                             completion_tokens,
                             reasoning_tokens,
+                            timing,
                         ),
                     ]
                 }
@@ -334,6 +339,7 @@ fn usage_chunk(
     prompt_tokens: u32,
     completion_tokens: u32,
     reasoning_tokens: u32,
+    timing: Option<FinishTiming>,
 ) -> ChatCompletionChunk {
     ChatCompletionChunk {
         id: id.into(),
@@ -351,6 +357,14 @@ fn usage_chunk(
                 reasoning_tokens: reasoning_tokens as u64,
             }),
             prompt_tokens_details: None,
+            // 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.
+            helexa_timing: timing.map(|t| HelexaTiming {
+                prefill_ms: t.prefill_ms as u64,
+                decode_ms: t.decode_ms as u64,
+                prefill_tokens: t.prefill_tokens as u64,
+            }),
         }),
         extra: serde_json::Value::Object(Default::default()),
     }
@@ -391,6 +405,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -413,6 +428,45 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn finish_timing_surfaces_on_usage_chunk() {
+        // O1 (#85) wire contract: a Finish carrying FinishTiming must
+        // surface as `usage.helexa_timing` on the trailing usage chunk,
+        // which is what the bench harness reads to compute true prefill
+        // vs decode tok/s. Absent timing must leave it None.
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
+        let out_rx = project_chat_stream(rx, "id-1".into(), 1700, "m".into());
+
+        tx.send(InferenceEvent::Start).await.unwrap();
+        tx.send(InferenceEvent::Finish {
+            reason: FinishReason::Stop,
+            prompt_tokens: 128,
+            completion_tokens: 64,
+            reasoning_tokens: 0,
+            timing: Some(FinishTiming {
+                prefill_ms: 200,
+                decode_ms: 1500,
+                prefill_tokens: 128,
+            }),
+        })
+        .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 timing = usage
+            .helexa_timing
+            .as_ref()
+            .expect("helexa_timing populated when Finish carried timing");
+        assert_eq!(timing.prefill_ms, 200);
+        assert_eq!(timing.decode_ms, 1500);
+        assert_eq!(timing.prefill_tokens, 128);
+    }
+
     #[tokio::test]
     async fn empty_text_delta_is_dropped() {
         let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
@@ -434,6 +488,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -496,6 +551,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -547,6 +603,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -592,6 +649,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -635,6 +693,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -662,6 +721,7 @@ mod tests {
             prompt_tokens: 42,
             completion_tokens: 5,
             reasoning_tokens: 2,
+            timing: None,
         })
         .await
         .unwrap();
@@ -695,6 +755,7 @@ mod tests {
             prompt_tokens: 10,
             completion_tokens: 7,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();

crates/neuron/src/wire/openai_responses.rs

@@ -29,9 +29,9 @@
 
 use cortex_core::openai::{ChatCompletionRequest, ChatMessage, MessageContent};
 use cortex_core::responses::{
-    OutputTokensDetails, ResponsesContentPart, ResponsesInput, ResponsesInputItem,
-    ResponsesMessageContent, ResponsesOutputContent, ResponsesOutputItem, ResponsesRequest,
-    ResponsesResponse, ResponsesUsage, events,
+    OutputTokensDetails, ResponsesContentPart, ResponsesInput, ResponsesInputElement,
+    ResponsesInputItem, ResponsesMessageContent, ResponsesOutputContent, ResponsesOutputItem,
+    ResponsesRequest, ResponsesResponse, ResponsesUsage, events,
 };
 use serde_json::{Value, json};
 use tokio::sync::mpsc;
@@ -109,8 +109,26 @@ pub fn request_to_chat(req: ResponsesRequest) -> Result<ChatCompletionRequest, T
             });
         }
         ResponsesInput::Items(items) => {
-            for item in items {
-                if let Some(msg) = input_item_to_chat(item) {
+            for element in items {
+                let msg = match element {
+                    ResponsesInputElement::Typed(item) => input_item_to_chat(item),
+                    // Bare `{role, content}` (OpenAI EasyInputMessage —
+                    // what litellm/agent-zero emit). `content: null`
+                    // (e.g. an assistant turn carrying only tool calls)
+                    // collapses to an empty string so the turn is kept.
+                    ResponsesInputElement::EasyMessage { role, content } => Some(ChatMessage {
+                        role,
+                        content: content
+                            .map(message_content_to_chat)
+                            .unwrap_or_else(|| MessageContent::Text(String::new())),
+                        extra: Value::Object(Default::default()),
+                    }),
+                    // Forward-compat: an item shape we don't model.
+                    // Dropped rather than rejected (see
+                    // `ResponsesInputElement::Other`).
+                    ResponsesInputElement::Other(_) => None,
+                };
+                if let Some(msg) = msg {
                     messages.push(msg);
                 }
             }
@@ -159,11 +177,18 @@ fn input_item_to_chat(item: ResponsesInputItem) -> Option<ChatMessage> {
             })
         }
         ResponsesInputItem::FunctionCallOutput { call_id, output } => {
+            // `output` is either a plain string or an array of content
+            // parts. Render a string as-is; anything else to compact
+            // JSON so the tool result text reaches the model intact.
+            let output_text = match output {
+                Value::String(s) => s,
+                other => other.to_string(),
+            };
             let mut extra = serde_json::Map::new();
             extra.insert("tool_call_id".into(), Value::String(call_id));
             Some(ChatMessage {
                 role: "tool".into(),
-                content: MessageContent::Text(output),
+                content: MessageContent::Text(output_text),
                 extra: Value::Object(extra),
             })
         }
@@ -192,7 +217,9 @@ fn message_content_to_chat(content: ResponsesMessageContent) -> MessageContent {
                     .filter_map(|p| match p {
                         ResponsesContentPart::InputText { text }
                         | ResponsesContentPart::OutputText { text, .. } => Some(text),
-                        ResponsesContentPart::InputImage { .. } => None,
+                        ResponsesContentPart::InputImage { .. } | ResponsesContentPart::Unknown => {
+                            None
+                        }
                     })
                     .collect::<Vec<_>>()
                     .join("\n\n");
@@ -211,6 +238,7 @@ fn message_content_to_chat(content: ResponsesMessageContent) -> MessageContent {
                             "image_url": { "url": image_url },
                         }));
                     }
+                    ResponsesContentPart::Unknown => {}
                 }
             }
             MessageContent::Parts(out)
@@ -309,6 +337,9 @@ async fn run_projection(
                 prompt_tokens,
                 completion_tokens,
                 reasoning_tokens,
+                // Responses-side `helexa_timing` surfacing not wired yet;
+                // the bench harness reads timing off the chat path (#85).
+                timing: _,
             } => {
                 finish = Some(reason);
                 // Surface usage on the streaming `response.completed`
@@ -535,6 +566,18 @@ mod tests {
     use super::*;
     use cortex_core::openai::MessageContent;
 
+    /// Wrap typed items as `input` elements. Most translator tests
+    /// exercise the typed path; the bare easy-message and unknown-item
+    /// paths have dedicated tests below.
+    fn typed_items(items: Vec<ResponsesInputItem>) -> ResponsesInput {
+        ResponsesInput::Items(
+            items
+                .into_iter()
+                .map(ResponsesInputElement::Typed)
+                .collect(),
+        )
+    }
+
     fn meta() -> ResponseMeta {
         ResponseMeta {
             response_id: "resp_1".into(),
@@ -614,7 +657,7 @@ mod tests {
     fn translates_input_items_to_chat_messages() {
         let req = ResponsesRequest {
             model: "m".into(),
-            input: ResponsesInput::Items(vec![
+            input: typed_items(vec![
                 ResponsesInputItem::Message {
                     role: "user".into(),
                     content: ResponsesMessageContent::Text("first".into()),
@@ -646,7 +689,7 @@ mod tests {
     fn image_input_translates_to_chat_parts_array() {
         let req = ResponsesRequest {
             model: "m".into(),
-            input: ResponsesInput::Items(vec![ResponsesInputItem::Message {
+            input: typed_items(vec![ResponsesInputItem::Message {
                 role: "user".into(),
                 content: ResponsesMessageContent::Parts(vec![
                     ResponsesContentPart::InputText {
@@ -687,7 +730,7 @@ mod tests {
         // it's dropped — but it must not break translation.
         let req = ResponsesRequest {
             model: "m".into(),
-            input: ResponsesInput::Items(vec![ResponsesInputItem::Message {
+            input: typed_items(vec![ResponsesInputItem::Message {
                 role: "user".into(),
                 content: ResponsesMessageContent::Parts(vec![
                     ResponsesContentPart::InputText {
@@ -729,7 +772,7 @@ mod tests {
     fn text_only_parts_collapse_to_string() {
         let req = ResponsesRequest {
             model: "m".into(),
-            input: ResponsesInput::Items(vec![ResponsesInputItem::Message {
+            input: typed_items(vec![ResponsesInputItem::Message {
                 role: "user".into(),
                 content: ResponsesMessageContent::Parts(vec![
                     ResponsesContentPart::InputText {
@@ -759,7 +802,7 @@ mod tests {
     fn reasoning_items_are_silently_dropped() {
         let req = ResponsesRequest {
             model: "m".into(),
-            input: ResponsesInput::Items(vec![
+            input: typed_items(vec![
                 ResponsesInputItem::Reasoning { content: vec![] },
                 ResponsesInputItem::Message {
                     role: "user".into(),
@@ -779,6 +822,74 @@ mod tests {
         assert_eq!(chat.messages[0].role, "user");
     }
 
+    #[test]
+    fn bare_easy_messages_translate_like_typed_messages() {
+        // The agent-zero / litellm shape: bare `{role, content}` items
+        // with no `type`. Deserialize from raw JSON (not hand-built)
+        // so this exercises the real parse path end to end.
+        let raw = r#"{
+            "model": "Qwen/Qwen3.6-27B",
+            "store": true,
+            "input": [
+                {"role": "system", "content": "be terse"},
+                {"role": "assistant", "content": "{\"tool_name\":\"response\"}"},
+                {"role": "user", "content": "alpha"}
+            ]
+        }"#;
+        let req: ResponsesRequest = serde_json::from_str(raw).unwrap();
+        let chat = request_to_chat(req).unwrap();
+        let roles: Vec<&str> = chat.messages.iter().map(|m| m.role.as_str()).collect();
+        assert_eq!(roles, vec!["system", "assistant", "user"]);
+        assert!(matches!(
+            &chat.messages[2].content,
+            MessageContent::Text(t) if t == "alpha"
+        ));
+    }
+
+    #[test]
+    fn null_content_and_unknown_items_survive_translation() {
+        // An assistant turn with `content: null` is kept (empty text);
+        // an unmodeled item type is dropped, not rejected.
+        let raw = r#"{
+            "model": "m",
+            "input": [
+                {"role": "assistant", "content": null},
+                {"type": "item_reference", "id": "x"},
+                {"role": "user", "content": "go"}
+            ]
+        }"#;
+        let req: ResponsesRequest = serde_json::from_str(raw).unwrap();
+        let chat = request_to_chat(req).unwrap();
+        // assistant(null) kept, item_reference dropped, user kept.
+        let roles: Vec<&str> = chat.messages.iter().map(|m| m.role.as_str()).collect();
+        assert_eq!(roles, vec!["assistant", "user"]);
+        assert!(matches!(
+            &chat.messages[0].content,
+            MessageContent::Text(t) if t.is_empty()
+        ));
+    }
+
+    #[test]
+    fn function_call_output_array_renders_to_text() {
+        // OpenAI allows `function_call_output.output` to be an array of
+        // content parts; the tool result must reach the model as text.
+        let raw = r#"{
+            "model": "m",
+            "input": [
+                {"type": "function_call_output", "call_id": "c1",
+                 "output": [{"type": "output_text", "text": "42"}]}
+            ]
+        }"#;
+        let req: ResponsesRequest = serde_json::from_str(raw).unwrap();
+        let chat = request_to_chat(req).unwrap();
+        assert_eq!(chat.messages.len(), 1);
+        assert_eq!(chat.messages[0].role, "tool");
+        match &chat.messages[0].content {
+            MessageContent::Text(t) => assert!(t.contains("42"), "got {t:?}"),
+            other => panic!("expected text, got {other:?}"),
+        }
+    }
+
     // ── streaming projector ─────────────────────────────────────────
 
     async fn collect(mut rx: mpsc::Receiver<ResponseStreamFrame>) -> Vec<ResponseStreamFrame> {
@@ -806,6 +917,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -856,6 +968,7 @@ mod tests {
             prompt_tokens: 30,
             completion_tokens: 12,
             reasoning_tokens: 4,
+            timing: None,
         })
         .await
         .unwrap();
@@ -886,6 +999,7 @@ mod tests {
             prompt_tokens: 8,
             completion_tokens: 3,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -910,6 +1024,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();
@@ -956,6 +1071,7 @@ mod tests {
             prompt_tokens: 0,
             completion_tokens: 0,
             reasoning_tokens: 0,
+            timing: None,
         })
         .await
         .unwrap();