fix: keep Claude streams alive during long thinking and recover from stalls

- Request summarized thinking when streaming to adaptive models, so the
  connection is not idle (and cut by proxies) while the model thinks
- Abort streams with no data for 90s and retry with lower thinking effort
  when nothing visible has been produced yet
- Enable eager_input_streaming for tool input, with fallback on rejection

Co-authored-by: cowagent <cow@cowagent.ai>

models/claudeapi/claude_api_bot.py

@@ -2,6 +2,8 @@
 
 import json
 import re
+import socket
+import threading
 import time
 from typing import Optional
 
@@ -66,6 +68,92 @@ CACHE_TTL_1H = "1h"
 MAX_CACHE_BREAKPOINTS = 4
 UNCACHEABLE_BLOCK_TYPES = ("thinking", "redacted_thinking")
 
+# Without it the API buffers a tool argument until the whole value is complete,
+# so a large `write` streams nothing but a ping every 30s for minutes, and
+# networks that drop quiet connections throw the whole generation away.
+EAGER_INPUT_STREAMING = "eager_input_streaming"
+
+# With thinking summaries and eager tool input, a healthy stream is never quiet
+# for long. Some connections go silent mid-generation yet stay open for half an
+# hour, outliving the socket read timeout, so a stream with no event for this
+# long is cut and retried.
+STREAM_STALL_SECONDS = 90
+
+
+# While thinking the API may send nothing at all, so a long think looks exactly
+# like a dead connection, and some networks cut it at that point. Thinking less
+# is what makes the retry finish.
+_LOWER_EFFORT = {"max": "medium", "xhigh": "medium", "high": "medium", "medium": "low"}
+_MIN_THINKING_BUDGET = 1024
+
+
+def _lower_thinking(params: dict) -> Optional[dict]:
+    """Return params asking for less thinking, or None when it cannot go lower."""
+    thinking = params.get("thinking") or {}
+    if thinking.get("type") == "enabled":
+        budget = thinking.get("budget_tokens") or 0
+        if budget <= _MIN_THINKING_BUDGET:
+            return None
+        return dict(params, thinking=dict(thinking, budget_tokens=max(_MIN_THINKING_BUDGET, budget // 2)))
+    if not thinking:
+        return None
+    output_config = params.get("output_config") or {}
+    lower = _LOWER_EFFORT.get(output_config.get("effort") or "high")
+    if not lower:
+        return None
+    return dict(params, output_config=dict(output_config, effort=lower))
+
+
+def _abort_response(response) -> None:
+    """Unblock a reader stuck in recv() on this response's socket."""
+    raw = getattr(response, "raw", None)
+    getters = (
+        lambda: raw._fp.fp.raw._sock,
+        lambda: raw._connection.sock,
+        lambda: raw.connection.sock,
+    )
+    for get_sock in getters:
+        try:
+            sock = get_sock()
+        except Exception:
+            continue
+        if sock is not None:
+            try:
+                sock.shutdown(socket.SHUT_RDWR)
+            except OSError:
+                pass
+            break
+    try:
+        response.close()
+    except Exception:
+        pass
+
+
+class _StallWatchdog:
+    """Cuts a streaming response that has received no event for ``limit`` seconds."""
+
+    def __init__(self, response, limit: Optional[float] = None):
+        self.response = response
+        self.limit = STREAM_STALL_SECONDS if limit is None else limit
+        self.interval = min(5.0, self.limit / 3)
+        self.tripped = False
+        self._last = time.monotonic()
+        self._stop = threading.Event()
+        threading.Thread(target=self._run, daemon=True).start()
+
+    def beat(self) -> None:
+        self._last = time.monotonic()
+
+    def stop(self) -> None:
+        self._stop.set()
+
+    def _run(self) -> None:
+        while not self._stop.wait(self.interval):
+            if time.monotonic() - self._last > self.limit:
+                self.tripped = True
+                _abort_response(self.response)
+                return
+
 
 # OpenAI对话模型API (可用)
 class ClaudeAPIBot(Bot, OpenAIImage):
@@ -419,6 +507,9 @@ class ClaudeAPIBot(Bot, OpenAIImage):
             request_params["system"] = system_prompt
 
         if tools:
+            if stream:
+                tools = [dict(t, **{EAGER_INPUT_STREAMING: True}) if isinstance(t, dict) and "input_schema" in t else t
+                         for t in tools]
             request_params["tools"] = tools
             # Agent turns resend the same long prefix on every step of the tool
             # loop, so cache it; one-off calls without tools are left alone.
@@ -442,6 +533,11 @@ class ClaudeAPIBot(Bot, OpenAIImage):
         )
         if thinking_params:
             request_params["thinking"] = thinking_params
+        elif stream and (actual_model or "").lower().startswith(ADAPTIVE_THINKING_MODELS):
+            # These models think even when the field is omitted, but then stream
+            # nothing until the thinking ends; summaries keep the connection busy
+            # so networks that cut idle streams do not drop the request.
+            request_params["thinking"] = {"type": "adaptive", "display": "summarized"}
 
         try:
             if stream:
@@ -660,6 +756,33 @@ class ClaudeAPIBot(Bot, OpenAIImage):
         return formatted_response
 
     def _handle_stream_response(self, request_params):
+        """Stream a response; if it stalls while the model is still thinking, retry with less effort.
+
+        A request that stalls during silent thinking stalls again when resent
+        unchanged, so the agent-level retry alone would fail the same way.
+        """
+        while True:
+            visible = False
+            stalled_chunk = None
+            for chunk in self._stream_once(request_params):
+                if chunk.get("error") and chunk.get("stalled") and not visible:
+                    stalled_chunk = chunk
+                    break
+                delta = ((chunk.get("choices") or [{}])[0] or {}).get("delta") or {}
+                if delta.get("content") or delta.get("tool_calls"):
+                    visible = True
+                yield chunk
+            if stalled_chunk is None:
+                return
+            lowered = _lower_thinking(request_params)
+            if lowered is None:
+                yield stalled_chunk
+                return
+            logger.warning(f"[Claude] Stream went silent while thinking; retrying with "
+                           f"{lowered.get('output_config', {}).get('effort') or lowered.get('thinking')}")
+            request_params = lowered
+
+    def _stream_once(self, request_params):
         """Handle streaming Claude API response using HTTP requests"""
         # Prepare headers
         headers = {
@@ -682,6 +805,7 @@ class ClaudeAPIBot(Bot, OpenAIImage):
         usage_output_tokens = 0
         usage_cache_write = 0
         usage_cache_read = 0
+        watchdog = None
 
         try:
             # Make streaming HTTP request
@@ -695,6 +819,22 @@ class ClaudeAPIBot(Bot, OpenAIImage):
                 timeout=180
             )
 
+            if response.status_code == 400 and EAGER_INPUT_STREAMING in response.text:
+                # Some Anthropic-compatible gateways reject the field; drop it and resend.
+                logger.warning("[Claude] Endpoint rejected eager_input_streaming, retrying without it")
+                request_params["tools"] = [
+                    {k: v for k, v in t.items() if k != EAGER_INPUT_STREAMING} if isinstance(t, dict) else t
+                    for t in request_params.get("tools") or []
+                ]
+                response = requests.post(
+                    f"{self.api_base}/messages",
+                    headers=headers,
+                    json=request_params,
+                    proxies=proxies,
+                    stream=True,
+                    timeout=180
+                )
+
             if response.status_code != 200:
                 error_text = response.text
                 try:
@@ -711,10 +851,12 @@ class ClaudeAPIBot(Bot, OpenAIImage):
                 return
 
             # Process streaming response
+            watchdog = _StallWatchdog(response)
             for line in response.iter_lines():
                 if line:
                     line = line.decode('utf-8')
                     if line.startswith('data: '):
+                        watchdog.beat()
                         line = line[6:]  # Remove 'data: ' prefix
                         if line == '[DONE]':
                             break
@@ -839,17 +981,29 @@ class ClaudeAPIBot(Bot, OpenAIImage):
                         except json.JSONDecodeError:
                             continue
 
-        except requests.RequestException as e:
-            logger.error(f"Claude streaming request error: {e}")
-            yield {
-                "error": True,
-                "message": f"Connection error: {str(e)}",
-                "status_code": 0
-            }
+            if watchdog.tripped:
+                raise requests.ConnectionError("stream ended after the stall watchdog fired")
+
         except Exception as e:
-            logger.error(f"Claude streaming error: {e}")
+            stalled = watchdog is not None and watchdog.tripped
+            if stalled:
+                message = f"Connection error: stream stalled (no data for {STREAM_STALL_SECONDS}s), aborted for retry"
+                logger.error(f"Claude {message}")
+                status_code = 0
+            elif isinstance(e, requests.RequestException):
+                logger.error(f"Claude streaming request error: {e}")
+                message = f"Connection error: {str(e)}"
+                status_code = 0
+            else:
+                logger.error(f"Claude streaming error: {e}")
+                message = str(e)
+                status_code = 500
             yield {
                 "error": True,
-                "message": str(e),
-                "status_code": 500
+                "message": message,
+                "status_code": status_code,
+                "stalled": stalled,
             }
+        finally:
+            if watchdog is not None:
+                watchdog.stop()

tests/test_claude_eager_tool_streaming.py

@@ -0,0 +1,44 @@
+# encoding:utf-8
+import os
+import sys
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+WRITE_TOOL = {
+    "name": "write",
+    "description": "Write a file",
+    "input_schema": {"type": "object", "properties": {"path": {"type": "string"}}},
+}
+
+
+def _bot(monkeypatch, captured, stream):
+    from config import conf
+    from models.claudeapi.claude_api_bot import ClaudeAPIBot
+
+    bot = ClaudeAPIBot.__new__(ClaudeAPIBot)
+    monkeypatch.setitem(conf(), "model", "claude-opus-5-5")
+    monkeypatch.setitem(conf(), "character_desc", "")
+    handler = "_handle_stream_response" if stream else "_handle_sync_response"
+    monkeypatch.setattr(bot, handler, lambda request_params: captured.setdefault("request", request_params) or {})
+    return bot
+
+
+def test_streaming_tool_call_enables_eager_input_streaming(monkeypatch):
+    captured = {}
+    server_tool = {"type": "web_search_20250305", "name": "web_search"}
+    tools = [dict(WRITE_TOOL), server_tool]
+    _bot(monkeypatch, captured, stream=True).call_with_tools(
+        messages=[{"role": "user", "content": "hi"}], tools=tools, stream=True)
+
+    sent = captured["request"]["tools"]
+    assert sent[0]["eager_input_streaming"] is True
+    assert "eager_input_streaming" not in sent[1]
+    assert "eager_input_streaming" not in tools[0]
+
+
+def test_sync_tool_call_leaves_tools_untouched(monkeypatch):
+    captured = {}
+    _bot(monkeypatch, captured, stream=False).call_with_tools(
+        messages=[{"role": "user", "content": "hi"}], tools=[dict(WRITE_TOOL)], stream=False)
+
+    assert "eager_input_streaming" not in captured["request"]["tools"][0]

tests/test_claude_stream_stall.py

@@ -0,0 +1,158 @@
+# encoding:utf-8
+import json
+import os
+import socket
+import sys
+import threading
+import time
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+THINKING = [
+    {"type": "message_start", "message": {"usage": {"input_tokens": 1}}},
+    {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}},
+    {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "hmm"}},
+]
+TEXT_THEN_SILENCE = [
+    {"type": "message_start", "message": {"usage": {"input_tokens": 1}}},
+    {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+    {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "hi"}},
+]
+COMPLETE = [
+    {"type": "message_start", "message": {"usage": {"input_tokens": 1}}},
+    {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+    {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "done"}},
+    {"type": "content_block_stop", "index": 0},
+    {"type": "message_delta", "delta": {"stop_reason": "end_turn"}, "usage": {"output_tokens": 2}},
+    {"type": "message_stop"},
+]
+
+
+def _read_body(conn):
+    data = b""
+    while b"\r\n\r\n" not in data:
+        data += conn.recv(65536)
+    head, body = data.split(b"\r\n\r\n", 1)
+    length = next(int(line.split(b":")[1]) for line in head.split(b"\r\n") if line.lower().startswith(b"content-length"))
+    while len(body) < length:
+        body += conn.recv(65536)
+    return json.loads(body)
+
+
+def _scripted_server(responses):
+    """Serve each (events, hang) in turn: send the events, then stay silent if hang, else close."""
+    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    srv.bind(("127.0.0.1", 0))
+    srv.listen(len(responses))
+    release = threading.Event()
+    bodies = []
+
+    def serve():
+        silent = []
+        for events, hang in responses:
+            conn, _ = srv.accept()
+            bodies.append(_read_body(conn))
+            conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n"
+                         b"Transfer-Encoding: chunked\r\n\r\n")
+            for event in events:
+                payload = f"data: {json.dumps(event)}\n\n".encode()
+                conn.sendall(f"{len(payload):x}\r\n".encode() + payload + b"\r\n")
+            if hang:
+                silent.append(conn)
+                continue
+            conn.sendall(b"0\r\n\r\n")
+            conn.close()
+        release.wait(30)
+        for conn in silent:
+            conn.close()
+        srv.close()
+
+    threading.Thread(target=serve, daemon=True).start()
+    return srv.getsockname()[1], release, bodies
+
+
+def _bot(monkeypatch, port):
+    from config import conf
+    from models.claudeapi import claude_api_bot
+    from models.claudeapi.claude_api_bot import ClaudeAPIBot
+
+    monkeypatch.setattr(claude_api_bot, "STREAM_STALL_SECONDS", 1.5)
+    monkeypatch.setitem(conf(), "claude_api_base", f"http://127.0.0.1:{port}/v1")
+    monkeypatch.setitem(conf(), "claude_api_key", "test")
+    monkeypatch.setitem(conf(), "proxy", "")
+    monkeypatch.setenv("NO_PROXY", "*")
+    return ClaudeAPIBot.__new__(ClaudeAPIBot)
+
+
+def _run(bot, params, release):
+    started = time.monotonic()
+    try:
+        chunks = list(bot._handle_stream_response(params))
+    finally:
+        release.set()
+    return chunks, time.monotonic() - started
+
+
+def test_silent_stream_is_cut_and_reported_as_retryable(monkeypatch):
+    port, release, _ = _scripted_server([(TEXT_THEN_SILENCE, True)])
+    chunks, elapsed = _run(_bot(monkeypatch, port), {"model": "m", "messages": []}, release)
+
+    assert elapsed < 10
+    errors = [c for c in chunks if c.get("error")]
+    assert len(errors) == 1
+    assert errors[0]["status_code"] == 0
+    assert "connection" in errors[0]["message"].lower()
+    assert "stalled" in errors[0]["message"]
+
+
+def test_stall_while_thinking_retries_with_lower_effort(monkeypatch):
+    port, release, bodies = _scripted_server([(THINKING, True), (COMPLETE, False)])
+    params = {"model": "m", "messages": [], "thinking": {"type": "adaptive", "display": "summarized"},
+              "output_config": {"effort": "high"}}
+    chunks, _ = _run(_bot(monkeypatch, port), params, release)
+
+    assert not [c for c in chunks if c.get("error")]
+    assert "done" in "".join(c["choices"][0]["delta"].get("content", "") for c in chunks if c.get("choices"))
+    assert [b["output_config"]["effort"] for b in bodies] == ["high", "medium"]
+
+
+def test_stall_after_visible_output_is_not_retried_inside_the_bot(monkeypatch):
+    port, release, bodies = _scripted_server([(TEXT_THEN_SILENCE, True)])
+    params = {"model": "m", "messages": [], "thinking": {"type": "adaptive", "display": "summarized"},
+              "output_config": {"effort": "high"}}
+    chunks, _ = _run(_bot(monkeypatch, port), params, release)
+
+    assert len(bodies) == 1
+    assert [c for c in chunks if c.get("error")]
+
+
+def _captured_request(monkeypatch, stream):
+    from config import conf
+    from models.claudeapi.claude_api_bot import ClaudeAPIBot
+
+    captured = {}
+    bot = ClaudeAPIBot.__new__(ClaudeAPIBot)
+    monkeypatch.setitem(conf(), "model", "claude-opus-5-5")
+    monkeypatch.setitem(conf(), "character_desc", "")
+    handler = "_handle_stream_response" if stream else "_handle_sync_response"
+    monkeypatch.setattr(bot, handler, lambda request_params: captured.setdefault("request", request_params) or {})
+    bot.call_with_tools(messages=[{"role": "user", "content": "hi"}], tools=[], stream=stream,
+                        thinking={"type": "disabled"})
+    return captured["request"]
+
+
+def test_streaming_adaptive_model_always_asks_for_thinking_summaries(monkeypatch):
+    assert _captured_request(monkeypatch, stream=True)["thinking"] == {"type": "adaptive", "display": "summarized"}
+    assert "thinking" not in _captured_request(monkeypatch, stream=False)
+
+
+def test_lower_thinking_steps_down_and_stops():
+    from models.claudeapi.claude_api_bot import _lower_thinking
+
+    adaptive = {"thinking": {"type": "adaptive"}, "output_config": {"effort": "xhigh"}}
+    assert _lower_thinking(adaptive)["output_config"]["effort"] == "medium"
+    assert _lower_thinking({"thinking": {"type": "adaptive"}})["output_config"]["effort"] == "medium"
+    assert _lower_thinking({"thinking": {"type": "adaptive"}, "output_config": {"effort": "low"}}) is None
+    assert _lower_thinking({"thinking": {"type": "enabled", "budget_tokens": 8000}})["thinking"]["budget_tokens"] == 4000
+    assert _lower_thinking({"thinking": {"type": "enabled", "budget_tokens": 1024}}) is None
+    assert _lower_thinking({"model": "m"}) is None