fix(sub): emit Anthropic SSE pings on quiet reroute streams

CHANGELOG.md

@@ -6,6 +6,16 @@ All notable changes to this project are documented here. The format follows
 
 ## [Unreleased]
 
+### Fixed
+
+- **Sub streams no longer die to Claude Code's ~180s stall watchdog.** Quiet Codex/Grok/Kimi
+  turns (long reasoning, buffered tool-arg generation) used to go silent on the Anthropic SSE
+  side, so Claude Code aborted with `Response stalled mid-stream` after exactly three minutes
+  (`incomplete_stream` / `no terminal frame` in `serve.log`). The live reroute stream now emits
+  Anthropic `event: ping` keepalives every 15s once `message_start` has been sent — both on total
+  upstream silence and when upstream is still sending chunks that produce no client-visible
+  events. Same shape as claude-code-proxy's historical tool-buffer keepalive.
+
 ## [0.11.10] - 2026-07-26
 
 ### Fixed

crates/llmtrim-cli/src/reroute/sse.rs

@@ -6,11 +6,25 @@
 //! content-block index bookkeeping in one place — the reducers never compute Anthropic indices,
 //! they just say "a text block started / a delta arrived / it stopped" and the encoder assigns and
 //! tracks the index. This is deliberately the *only* place that emits `message_start`,
-//! `content_block_*`, `message_delta`, and `message_stop`, so the wire contract can't drift between
-//! providers.
+//! `content_block_*`, `message_delta`, `message_stop`, and mid-stream `ping` keepalives, so the
+//! wire contract can't drift between providers.
+//!
+//! Claude Code aborts a live stream after ~180s with no downstream events ("Response stalled
+//! mid-stream"). Sub backends often go quiet while reasoning or while we buffer tool-call
+//! arguments, so the serve path emits Anthropic `ping` frames on
+//! [`KEEPALIVE_INTERVAL`] once `message_start` has been sent.
+
+use std::time::Duration;
 
 use serde_json::{Value, json};
 
+/// How often a quiet sub stream should emit an Anthropic SSE `ping` to Claude Code.
+///
+/// Matches claude-code-proxy's historical 15s keepalive (well under Claude Code's ~180s idle
+/// watchdog). Used by the live reroute stream in `serve` for both total upstream silence and
+/// upstream noise that produces no client-visible ReduceEvents (buffered tool args).
+pub const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15);
+
 /// Why the model stopped, mapped onto Anthropic `stop_reason`.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum StopReason {
@@ -280,6 +294,25 @@ impl AnthropicSseEncoder {
         }
     }
 
+    /// True once `message_start` has been sent and neither `message_stop` nor an `error` frame has
+    /// closed the stream. The serve path uses this to decide whether a quiet period should emit a
+    /// keepalive [`Self::emit_ping`] — pings before `message_start` or after close are not useful
+    /// and can confuse Claude Code's stream parser.
+    pub fn is_open(&self) -> bool {
+        self.started && !self.stopped
+    }
+
+    /// Append an Anthropic SSE `ping` frame if the message is open. No-op before `message_start` or
+    /// after the stream has been closed by `Finish`/`Error`/`finish_if_open`.
+    ///
+    /// Claude Code's idle-stream watchdog treats any SSE event (including `ping`) as activity, so
+    /// this keeps a long-thinking or tool-arg-buffering sub turn from being killed at ~180s.
+    pub fn emit_ping(&mut self, out: &mut String) {
+        if self.is_open() {
+            frame(out, "ping", &json!({"type": "ping"}));
+        }
+    }
+
     fn emit_message_start(&mut self, out: &mut String) {
         self.started = true;
         let data = json!({
@@ -680,4 +713,51 @@ mod tests {
         assert_eq!(got.len(), 1);
         assert_eq!(got[0]["ok"], true);
     }
+
+    #[test]
+    fn emit_ping_only_while_message_is_open() {
+        let mut enc = AnthropicSseEncoder::new("m");
+        let mut out = String::new();
+
+        // Before message_start: no-op.
+        enc.emit_ping(&mut out);
+        assert!(out.is_empty(), "ping before message_start must be a no-op");
+        assert!(!enc.is_open());
+
+        enc.encode(&ReduceEvent::TextStart, &mut out);
+        assert!(enc.is_open());
+        let len_after_start = out.len();
+        enc.emit_ping(&mut out);
+        let ping = &out[len_after_start..];
+        assert!(
+            ping.contains("event: ping") && ping.contains(r#""type":"ping""#),
+            "open stream must emit Anthropic ping frames: {ping:?}"
+        );
+
+        enc.encode(
+            &ReduceEvent::Finish {
+                stop_reason: StopReason::EndTurn,
+                usage: Usage::default(),
+                response_id: None,
+                continuation_eligible: false,
+            },
+            &mut out,
+        );
+        assert!(!enc.is_open());
+        let len_after_finish = out.len();
+        enc.emit_ping(&mut out);
+        assert_eq!(
+            out.len(),
+            len_after_finish,
+            "ping after message_stop must be a no-op"
+        );
+    }
+
+    #[test]
+    fn keepalive_interval_is_under_claude_code_stall_watchdog() {
+        // Document the invariant the serve path relies on: pings must land well before the
+        // ~180s client abort. 15s matches claude-code-proxy's historical keepalive.
+        assert_eq!(KEEPALIVE_INTERVAL.as_secs(), 15);
+        assert!(KEEPALIVE_INTERVAL.as_secs() * 4 < 180);
+    }
 }

crates/llmtrim-cli/src/serve.rs

@@ -2740,6 +2740,10 @@ mod imp {
                 finalize: Finalize,
                 phase: Phase,
                 prelude: String,
+                /// Wall clock of the last client-visible SSE emission (content *or* keepalive
+                /// ping). Drives the 15s Anthropic `ping` so Claude Code's ~180s idle-stream
+                /// watchdog does not kill a quiet reasoning / buffered-tool-arg turn.
+                last_emit: std::time::Instant,
                 // For continuation recording on terminal Finish during live stream
                 provider: crate::reroute::SubProvider,
                 logical_body: Option<Value>,
@@ -2753,6 +2757,7 @@ mod imp {
                 finalize,
                 phase: Phase::Prelude,
                 prelude,
+                last_emit: std::time::Instant::now(),
                 provider: info.provider,
                 logical_body: info.logical_body.clone(),
                 session_id: info.session_id.clone(),
@@ -2772,6 +2777,7 @@ mod imp {
                             if let Ok(mut buf) = st.acc.lock() {
                                 buf.extend_from_slice(out.as_bytes());
                             }
+                            st.last_emit = std::time::Instant::now();
                             return Some((
                                 Ok::<Bytes, std::io::Error>(Bytes::from(out.into_bytes())),
                                 st,
@@ -2802,56 +2808,101 @@ mod imp {
                             let bytes = Bytes::from(out.into_bytes());
                             return Some((Ok::<Bytes, std::io::Error>(bytes), st));
                         }
-                        Phase::Body => match st.inner.next().await {
-                            Some(Ok(frame)) => {
-                                let Ok(chunk) = frame.into_data() else {
+                        Phase::Body => {
+                            // Two quiet cases Claude Code's ~180s stall watchdog kills without help:
+                            // 1) upstream totally silent (long reasoning, hung provider)
+                            // 2) upstream still sending chunks that produce no client-visible
+                            //    ReduceEvents (buffered tool-arg deltas, provider keepalives)
+                            // `select!` covers (1); the empty-`out` branch covers (2). Both emit
+                            // Anthropic `event: ping` on KEEPALIVE_INTERVAL once message_start has
+                            // been sent (encoder.is_open).
+                            let sleep_for = crate::reroute::sse::KEEPALIVE_INTERVAL
+                                .saturating_sub(st.last_emit.elapsed());
+                            tokio::select! {
+                                frame = st.inner.next() => match frame {
+                                    Some(Ok(frame)) => {
+                                        let Ok(chunk) = frame.into_data() else {
+                                            continue;
+                                        };
+                                        let mut out = String::new();
+                                        for ev in st.reducer.push(&chunk) {
+                                            record_codex_continuation(
+                                                &ev,
+                                                &mut st.reducer,
+                                                st.provider,
+                                                st.logical_body.as_ref(),
+                                                st.session_id.as_deref(),
+                                            );
+                                            st.encoder.encode(&ev, &mut out);
+                                        }
+                                        // Upstream noise with no client-visible events: still ping
+                                        // if the idle budget has elapsed (buffered tool args).
+                                        if out.is_empty()
+                                            && st.encoder.is_open()
+                                            && st.last_emit.elapsed()
+                                                >= crate::reroute::sse::KEEPALIVE_INTERVAL
+                                        {
+                                            st.encoder.emit_ping(&mut out);
+                                        }
+                                        if out.is_empty() {
+                                            continue;
+                                        }
+                                        if let Ok(mut buf) = st.acc.lock() {
+                                            buf.extend_from_slice(out.as_bytes());
+                                        }
+                                        st.last_emit = std::time::Instant::now();
+                                        let bytes = Bytes::from(out.into_bytes());
+                                        return Some((Ok(bytes), st));
+                                    }
+                                    Some(Err(_)) => {
+                                        // Transport error mid-stream: surface it as an Anthropic
+                                        // `error` frame so the client can tell a dropped provider
+                                        // connection from a clean end-of-turn, instead of a
+                                        // silently truncated answer.
+                                        let mut out = String::new();
+                                        st.encoder.encode(
+                                            &ReduceEvent::Error {
+                                                message: "llmtrim: upstream stream error"
+                                                    .to_string(),
+                                            },
+                                            &mut out,
+                                        );
+                                        st.phase = Phase::Flush;
+                                        if out.is_empty() {
+                                            continue;
+                                        }
+                                        if let Ok(mut buf) = st.acc.lock() {
+                                            buf.extend_from_slice(out.as_bytes());
+                                        }
+                                        return Some((Ok(Bytes::from(out.into_bytes())), st));
+                                    }
+                                    None => {
+                                        st.phase = Phase::Flush;
+                                        continue;
+                                    }
+                                },
+                                _ = tokio::time::sleep(sleep_for) => {
+                                    if st.encoder.is_open()
+                                        && st.last_emit.elapsed()
+                                            >= crate::reroute::sse::KEEPALIVE_INTERVAL
+                                    {
+                                        let mut out = String::new();
+                                        st.encoder.emit_ping(&mut out);
+                                        if !out.is_empty() {
+                                            if let Ok(mut buf) = st.acc.lock() {
+                                                buf.extend_from_slice(out.as_bytes());
+                                            }
+                                            st.last_emit = std::time::Instant::now();
+                                            return Some((
+                                                Ok(Bytes::from(out.into_bytes())),
+                                                st,
+                                            ));
+                                        }
+                                    }
                                     continue;
-                                };
-                                let mut out = String::new();
-                                for ev in st.reducer.push(&chunk) {
-                                    record_codex_continuation(
-                                        &ev,
-                                        &mut st.reducer,
-                                        st.provider,
-                                        st.logical_body.as_ref(),
-                                        st.session_id.as_deref(),
-                                    );
-                                    st.encoder.encode(&ev, &mut out);
                                 }
-                                if out.is_empty() {
-                                    continue;
-                                }
-                                if let Ok(mut buf) = st.acc.lock() {
-                                    buf.extend_from_slice(out.as_bytes());
-                                }
-                                let bytes = Bytes::from(out.into_bytes());
-                                return Some((Ok(bytes), st));
                             }
-                            Some(Err(_)) => {
-                                // Transport error mid-stream: surface it as an Anthropic `error`
-                                // frame so the client can tell a dropped provider connection from a
-                                // clean end-of-turn, instead of a silently truncated answer.
-                                let mut out = String::new();
-                                st.encoder.encode(
-                                    &ReduceEvent::Error {
-                                        message: "llmtrim: upstream stream error".to_string(),
-                                    },
-                                    &mut out,
-                                );
-                                st.phase = Phase::Flush;
-                                if out.is_empty() {
-                                    continue;
-                                }
-                                if let Ok(mut buf) = st.acc.lock() {
-                                    buf.extend_from_slice(out.as_bytes());
-                                }
-                                return Some((Ok(Bytes::from(out.into_bytes())), st));
-                            }
-                            None => {
-                                st.phase = Phase::Flush;
-                                continue;
-                            }
-                        },
+                        }
                     }
                 }
             });