From 34cab2f5bd778eaa44c735458f11296a9b9eaf9a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Kiene?= <46886660+fkiene@users.noreply.github.com> Date: Sun, 26 Jul 2026 14:05:10 +0200 Subject: [PATCH 1/2] fix(sub): emit Anthropic SSE pings on quiet reroute streams MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude Code aborts after ~180s with no downstream events ("Response stalled mid-stream"). Sub backends often go silent while reasoning or while tool args are buffered, which showed up as incomplete_stream with no terminal frame at exactly three minutes. Emit event: ping every 15s once message_start is out — on total upstream silence and on upstream noise that produces no client-visible events. Same shape as claude-code-proxy's historical tool-buffer keepalive. Signed-off-by: François Kiene <46886660+fkiene@users.noreply.github.com> --- CHANGELOG.md | 10 ++ crates/llmtrim-cli/src/reroute/sse.rs | 84 ++++++++++++++- crates/llmtrim-cli/src/serve.rs | 145 +++++++++++++++++--------- 3 files changed, 190 insertions(+), 49 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 05e670b..b536cce 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/crates/llmtrim-cli/src/reroute/sse.rs b/crates/llmtrim-cli/src/reroute/sse.rs index d64b318..592ebc3 100644 --- a/crates/llmtrim-cli/src/reroute/sse.rs +++ b/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 [`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); + } } diff --git a/crates/llmtrim-cli/src/serve.rs b/crates/llmtrim-cli/src/serve.rs index 220e0a6..49aed7f 100644 --- a/crates/llmtrim-cli/src/serve.rs +++ b/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, @@ -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::from(out.into_bytes())), st, @@ -2802,56 +2808,101 @@ mod imp { let bytes = Bytes::from(out.into_bytes()); return Some((Ok::(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; - } - }, + } } } }); From 46a7ee3e176148644e6dd530a3fd794e976fd38a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fran=C3=A7ois=20Kiene?= <46886660+fkiene@users.noreply.github.com> Date: Sun, 26 Jul 2026 14:07:54 +0200 Subject: [PATCH 2/2] docs(sub): fix emit_ping intra-doc link for -D warnings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: François Kiene <46886660+fkiene@users.noreply.github.com> --- crates/llmtrim-cli/src/reroute/sse.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/llmtrim-cli/src/reroute/sse.rs b/crates/llmtrim-cli/src/reroute/sse.rs index 592ebc3..b36730a 100644 --- a/crates/llmtrim-cli/src/reroute/sse.rs +++ b/crates/llmtrim-cli/src/reroute/sse.rs @@ -296,8 +296,8 @@ 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 [`emit_ping`] — pings before `message_start` or after close are not useful and can - /// confuse Claude Code's stream parser. + /// 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 }