fix(neuron): stream reasoning as reasoning_content instead of dropping it

A reasoning model's think block was discarded token by token when
`include_thinking` was false (the default): the projection hit `continue`
and emitted nothing at all. The goal was right — model-internal scratchpad
should not appear in the UI of a client that never asked for it — but the
method silenced the stream for the entire reasoning phase.

On a hard prompt that phase runs for minutes, and for all of it not one SSE
event went out. Clients read that as a dead connection. The DeepSeek Harness
reports

    Failure reason: pi-ai stream idle timeout after 300000ms

and retries, so any task whose reasoning outlasts the client's idle timeout
could never complete however healthy the model was. SSE comment keep-alives
do not rescue it: idle timers reset on events, not comments.

`reasoning_content` is not in the OpenAI spec but is the de-facto slot —
DeepSeek's own API emits it, as do vLLM and SGLang. We were already on both
sides of that convention and only implemented one: `helexa-acp`'s OpenAI
provider reads `choice.delta.reasoning_content`, and `helexa-bench` times a
stream's first reasoning token from it, while neuron emitted the field
nowhere.

Emitting it keeps the original containment guarantee intact — it is not
`content`, so an unaware client ignores the unknown field and still shows
clean output — while the stream stays visibly alive and reasoning-aware
clients get the field they are already looking for.

The existing test asserting the drop is replaced by one asserting both
halves: reasoning arrives as `reasoning_content`, and never as `content`.
`include_thinking_off_drops_reasoning_even_with_markers` is renamed to
`..._keeps_reasoning_out_of_content`, which is what it actually checked —
it passed unchanged because containment is unaltered.

`include_thinking = true` is untouched: it still re-wraps with the model's
literal markers as content, for helexa-acp's ThinkParser.

The non-streaming projection still drops reasoning; it has no idle timeout
to trip, so it is left alone rather than changed on spec.

Refs #267

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

crates/cortex-gateway/src/state.rs

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

crates/neuron/src/api.rs

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

crates/neuron/src/wire/openai_chat.rs

@@ -139,11 +139,35 @@ pub fn project_chat_stream_with(
                 }
                 InferenceEvent::ReasoningDelta(text) => {
                     if !config.include_thinking {
-                        // Default path — reasoning has no slot in
-                        // chat completions, so it's dropped. Naïve
-                        // clients (Zed commit-message generator,
-                        // any vanilla OpenAI client) get clean
-                        // output.
+                        // Default path. This used to `continue` —
+                        // dropping the token and emitting nothing at
+                        // all — to keep model-internal scratchpad out
+                        // of naïve clients' UIs. That goal was right;
+                        // the method silenced the stream.
+                        //
+                        // A reasoning model can think for minutes on a
+                        // hard prompt, and for that whole span not one
+                        // SSE event went out. Clients time that out: the
+                        // DeepSeek Harness reports `pi-ai stream idle
+                        // timeout after 300000ms` and retries, so a task
+                        // that reasons for over five minutes could never
+                        // complete no matter how healthy the model was.
+                        // SSE comment keep-alives do not help — idle
+                        // timers reset on events, not comments.
+                        //
+                        // Emitting `reasoning_content` instead keeps the
+                        // original guarantee (it is not `content`, so
+                        // unaware clients still show clean output) while
+                        // the stream stays visibly alive, and hands
+                        // reasoning-aware clients the field they are
+                        // already looking for.
+                        if text.is_empty() {
+                            continue;
+                        }
+                        let chunk = reasoning_chunk(&id, created, &model_id, &text);
+                        if tx.send(chunk).await.is_err() {
+                            return;
+                        }
                         continue;
                     }
                     let Some(markers) = config.reasoning_markers.as_ref() else {
@@ -246,6 +270,35 @@ fn role_chunk(id: &str, created: u64, model_id: &str) -> ChatCompletionChunk {
     }
 }
 
+/// A reasoning delta, in the field reasoning-aware clients read.
+///
+/// `reasoning_content` is not in the OpenAI spec, but it is the
+/// de-facto slot: DeepSeek's API emits it, and so do vLLM and SGLang
+/// for reasoning models. Our own consumers already expect it —
+/// `helexa-acp`'s OpenAI provider reads `choice.delta.reasoning_content`,
+/// and `helexa-bench` times a stream's first reasoning token from it.
+///
+/// Crucially it is NOT `content`, so a client that has never heard of
+/// reasoning ignores the unknown field and sees clean output — the
+/// property the old drop-it-entirely default was protecting, kept
+/// without the silence that came with it.
+fn reasoning_chunk(id: &str, created: u64, model_id: &str, text: &str) -> ChatCompletionChunk {
+    ChatCompletionChunk {
+        id: id.into(),
+        object: "chat.completion.chunk".into(),
+        created,
+        model: model_id.into(),
+        choices: vec![ChunkChoice {
+            index: 0,
+            delta: json!({ "reasoning_content": text }),
+            finish_reason: None,
+            extra: serde_json::Value::Object(Default::default()),
+        }],
+        usage: None,
+        extra: serde_json::Value::Object(Default::default()),
+    }
+}
+
 fn content_chunk(id: &str, created: u64, model_id: &str, text: &str) -> ChatCompletionChunk {
     ChatCompletionChunk {
         id: id.into(),
@@ -500,10 +553,24 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn reasoning_delta_is_dropped_in_chat_projection() {
+    /// Reasoning is streamed as `reasoning_content`, never as `content`.
+    ///
+    /// Both halves matter. Keeping it out of `content` is what stops
+    /// scratchpad material appearing in the UI of a client that has
+    /// never heard of reasoning — the guarantee the previous
+    /// drop-everything default provided.
+    ///
+    /// But emitting *something* is what keeps the connection alive. A
+    /// model can think for minutes; while it did, this projection sent
+    /// no events at all, and clients closed the stream as dead. The
+    /// DeepSeek Harness reports `pi-ai stream idle timeout after
+    /// 300000ms`, so any task whose reasoning outlasts the client's
+    /// idle timeout could never complete. SSE comment keep-alives do
+    /// not help, because idle timers reset on events, not comments.
+    async fn reasoning_delta_is_streamed_as_reasoning_content() {
         let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
         let out_rx = project_chat_stream(rx, "id".into(), 1, "m".into());
-        tx.send(InferenceEvent::ReasoningDelta("<think>".into()))
+        tx.send(InferenceEvent::ReasoningDelta("pondering".into()))
             .await
             .unwrap();
         tx.send(InferenceEvent::TextDelta("real".into()))
@@ -511,8 +578,27 @@ mod tests {
             .unwrap();
         drop(tx);
         let out = collect(out_rx).await;
-        assert_eq!(out.len(), 1);
-        assert_eq!(out[0].choices[0].delta["content"], "real");
+        assert_eq!(out.len(), 2, "reasoning must produce a chunk, not silence");
+        assert_eq!(out[0].choices[0].delta["reasoning_content"], "pondering");
+        assert!(
+            out[0].choices[0].delta.get("content").is_none(),
+            "reasoning must never arrive as content"
+        );
+        assert_eq!(out[1].choices[0].delta["content"], "real");
+    }
+
+    /// An empty reasoning delta is still nothing to say — it must not
+    /// manufacture a chunk, matching the text path.
+    #[tokio::test]
+    async fn empty_reasoning_delta_is_dropped() {
+        let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
+        let out_rx = project_chat_stream(rx, "id".into(), 1, "m".into());
+        tx.send(InferenceEvent::ReasoningDelta(String::new()))
+            .await
+            .unwrap();
+        drop(tx);
+        let out = collect(out_rx).await;
+        assert!(out.is_empty(), "empty reasoning deltas produce no chunks");
     }
 
     fn pair() -> ReasoningTokenPair {
@@ -667,10 +753,12 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn include_thinking_off_drops_reasoning_even_with_markers() {
-        // Default behaviour even when markers happen to be
-        // configured. The flag is the gate, not the marker
-        // presence.
+    /// With the flag off, reasoning stays out of `content` whether or
+    /// not markers are configured — the flag is the gate, not the
+    /// marker presence. It now travels as `reasoning_content` instead
+    /// of being discarded, so this asserts the containment guarantee
+    /// only: nothing model-internal reaches a naive client's `content`.
+    async fn include_thinking_off_keeps_reasoning_out_of_content() {
         let (tx, rx) = mpsc::channel::<InferenceEvent>(4);
         let out_rx = project_chat_stream_with(
             rx,