fix: add configurable llm_call_timeout to bound stalled LLM calls

raven/agent/loop/main.py

@@ -6,7 +6,7 @@
 import json
 import re
 import time
-from contextlib import AsyncExitStack
+from contextlib import AsyncExitStack, aclosing
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
@@ -1295,27 +1295,36 @@ async def _llm_call_stream(
         tool_call_slots: list[dict[str, Any]] = []
         final_usage: dict[str, Any] | None = None
 
-        async for delta in self.provider.chat_stream(
-            messages=messages,
-            tools=tools,
-            model=model,
-        ):
-            reasoning_delta = getattr(delta, "reasoning_content", None)
-            if reasoning_delta:
-                reasoning_buf.append(reasoning_delta)
-                if on_reasoning_delta is not None:
-                    await on_reasoning_delta(reasoning_delta)
-            if delta.content:
-                content_buf.append(delta.content)
-                if on_token_delta is not None:
-                    await on_token_delta(delta.content)
-            if delta.tool_call_delta:
-                _merge_tool_call_fragments(
-                    tool_call_slots,
-                    delta.tool_call_delta,
-                )
-            if delta.usage is not None:
-                final_usage = delta.usage
+        # aclosing() guarantees the async generator (and its underlying stream)
+        # is closed when a TimeoutError from the per-chunk idle cap unwinds the
+        # loop, so a stalled stream terminates with a structured error instead
+        # of hanging or leaking the connection. The stream path has no retry
+        # (see docstring), so the error surfaces as the turn's response.
+        try:
+            async with aclosing(self.provider.chat_stream(messages=messages, tools=tools, model=model)) as stream:
+                async for delta in stream:
+                    reasoning_delta = getattr(delta, "reasoning_content", None)
+                    if reasoning_delta:
+                        reasoning_buf.append(reasoning_delta)
+                        if on_reasoning_delta is not None:
+                            await on_reasoning_delta(reasoning_delta)
+                    if delta.content:
+                        content_buf.append(delta.content)
+                        if on_token_delta is not None:
+                            await on_token_delta(delta.content)
+                    if delta.tool_call_delta:
+                        _merge_tool_call_fragments(
+                            tool_call_slots,
+                            delta.tool_call_delta,
+                        )
+                    if delta.usage is not None:
+                        final_usage = delta.usage
+        except TimeoutError:
+            return LLMResponse(
+                content="".join(content_buf),
+                finish_reason="error",
+                error_classification=self.provider.classify_error(TimeoutError()),
+            )
 
         tool_calls = _finalize_tool_calls(tool_call_slots)
         finish_reason = "tool_calls" if tool_calls else "stop"

raven/cli/_helpers.py

@@ -135,6 +135,7 @@ def make_provider(config: Config):
         temperature=defaults.temperature,
         max_tokens=defaults.max_tokens,
         reasoning_effort=defaults.reasoning_effort,
+        timeout=defaults.llm_call_timeout,
     )
     return provider
 
@@ -155,6 +156,7 @@ def make_lazy_provider(config: Config):
             temperature=defaults.temperature,
             max_tokens=defaults.max_tokens,
             reasoning_effort=defaults.reasoning_effort,
+            timeout=defaults.llm_call_timeout,
         ),
     )
     provider.prewarm()

raven/config/schema.py

@@ -251,6 +251,10 @@ class AgentDefaults(Base):
     max_tokens: int = 8192
     context_window_tokens: int = 65_536
     temperature: float = 0.1
+    # Per-call wall-clock cap (seconds) for every LLM request (main loop and
+    # sub-agents). Bounds a stalled backend that trickles bytes without ever
+    # finishing, which an httpx per-read timeout never catches.
+    llm_call_timeout: int = 600
     max_tool_iterations: int = 40
     # Cap on subagent VMs running at once (excess spawns queue). ge=1: a
     # 0/negative cap would deadlock every subagent (Semaphore(0)).

raven/providers/azure_openai_provider.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import uuid
 from typing import Any
 from urllib.parse import urljoin
@@ -146,8 +147,10 @@ async def chat(
         )
 
         try:
-            async with httpx.AsyncClient(timeout=60.0, verify=True) as client:
-                response = await client.post(url, headers=headers, json=payload)
+            async with httpx.AsyncClient(timeout=self.generation.timeout, verify=True) as client:
+                response = await asyncio.wait_for(
+                    client.post(url, headers=headers, json=payload), self.generation.timeout
+                )
                 if response.status_code != 200:
                     return LLMResponse(
                         content=f"Azure OpenAI API Error {response.status_code}: {response.text}",
@@ -161,6 +164,7 @@ async def chat(
             return LLMResponse(
                 content=f"Error calling Azure OpenAI: {repr(e)}",
                 finish_reason="error",
+                error_classification=self.classify_error(e),
             )
 
     def _parse_response(self, response: dict[str, Any]) -> LLMResponse:

raven/providers/base.py

@@ -107,6 +107,7 @@ class GenerationSettings:
     temperature: float = 0.7
     max_tokens: int = 4096
     reasoning_effort: str | None = None
+    timeout: float = 600.0
 
 
 class LLMProvider(ABC):
@@ -355,11 +356,17 @@ def has(*needles: str) -> bool:
         ):
             return ErrorClassification("server", retryable=True, should_fallback=True)
 
-        # Timeout / connection → retry + fallback.
-        if {"timeout", "apitimeouterror", "apiconnectionerror"} & names or has(
-            "timeout",
-            "timed out",
-            "connection",
+        # Timeout / connection → retry + fallback. isinstance covers the builtin
+        # TimeoutError raised by asyncio.wait_for (its class name "timeouterror"
+        # and empty str() match neither the name set nor the substrings below).
+        if (
+            isinstance(exc, TimeoutError)
+            or {"timeout", "apitimeouterror", "apiconnectionerror"} & names
+            or has(
+                "timeout",
+                "timed out",
+                "connection",
+            )
         ):
             return ErrorClassification("network", retryable=True, should_fallback=True)
 

raven/providers/custom_provider.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import uuid
 from typing import Any
 
@@ -45,9 +46,15 @@ async def chat(
         if tools:
             kwargs.update(tools=tools, tool_choice=tool_choice or "auto")
         try:
-            return self._parse(await self._client.chat.completions.create(**kwargs))
+            return self._parse(
+                await asyncio.wait_for(self._client.chat.completions.create(**kwargs), self.generation.timeout)
+            )
         except Exception as e:
-            return LLMResponse(content=f"Error: {e}", finish_reason="error")
+            return LLMResponse(
+                content=f"Error: {e}",
+                finish_reason="error",
+                error_classification=self.classify_error(e),
+            )
 
     def _parse(self, response: Any) -> LLMResponse:
         choice = response.choices[0]

raven/providers/litellm_provider.py

@@ -1,5 +1,6 @@
 """LiteLLM provider implementation for multi-provider support."""
 
+import asyncio
 import hashlib
 import os
 import secrets
@@ -291,12 +292,11 @@ async def chat(
             "messages": self._sanitize_messages(self._sanitize_empty_content(messages), extra_keys=extra_msg_keys),
             "max_tokens": max_tokens,
             "temperature": temperature,
-            # Hard cap on a single LLM call to prevent the agent from hanging
-            # when an upstream endpoint half-closes or never sends FIN. LiteLLM's
-            # default is 6000s — too long for an interactive agent loop. 600s
-            # covers slow reasoning-heavy generations while still failing fast
-            # on stuck connections, so chat_with_retry can retry or give up.
-            "timeout": 600,
+            # Per-read httpx cap forwarded to the underlying client. This alone
+            # cannot bound a backend that trickles bytes forever (the read timer
+            # resets on every chunk), so the awaited call is also wrapped in an
+            # asyncio.wait_for wall-clock cap below.
+            "timeout": self.generation.timeout,
         }
 
         # Apply model-specific overrides (e.g. kimi-k2.5 temperature)
@@ -327,7 +327,7 @@ async def chat(
             kwargs["tool_choice"] = tool_choice or "auto"
 
         try:
-            response = await acompletion(**kwargs)
+            response = await asyncio.wait_for(acompletion(**kwargs), self.generation.timeout)
             return self._parse_response(response)
         except Exception as e:
             # Return error as content for graceful handling, but classify the
@@ -379,6 +379,7 @@ async def chat_stream(
             # when usage is explicitly requested; without it the stream carries
             # no token counts and downstream cost / context tracking sees zero.
             "stream_options": {"include_usage": True},
+            "timeout": self.generation.timeout,
         }
 
         self._apply_model_overrides(model, kwargs)
@@ -398,11 +399,26 @@ async def chat_stream(
             kwargs["tools"] = tools
             kwargs["tool_choice"] = tool_choice or "auto"
 
-        response = await acompletion(**kwargs)
-        async for chunk in response:
-            delta = self._normalize_stream_chunk(chunk)
-            if delta is not None:
-                yield delta
+        response = await asyncio.wait_for(acompletion(**kwargs), self.generation.timeout)
+        # Per-chunk idle cap: the timer resets on every chunk, so a long but
+        # steadily-progressing generation is fine while a mid-stream stall (no
+        # bytes for `timeout` seconds) raises TimeoutError instead of hanging.
+        # aclose() in finally closes the underlying HTTP stream deterministically
+        # on that timeout, mirroring the `async with` cleanup on the other paths.
+        stream = response.__aiter__()
+        try:
+            while True:
+                try:
+                    chunk = await asyncio.wait_for(stream.__anext__(), self.generation.timeout)
+                except StopAsyncIteration:
+                    break
+                delta = self._normalize_stream_chunk(chunk)
+                if delta is not None:
+                    yield delta
+        finally:
+            aclose = getattr(stream, "aclose", None)
+            if aclose is not None:
+                await aclose()
 
     def _normalize_stream_chunk(self, chunk: Any) -> StreamDelta | None:
         """Normalize a raw provider chunk into a StreamDelta.

raven/providers/openai_codex_provider.py

@@ -68,14 +68,19 @@ async def chat(
 
         url = DEFAULT_CODEX_URL
 
+        timeout = self.generation.timeout
         try:
             try:
-                content, tool_calls, finish_reason = await _request_codex(url, headers, body, verify=True)
+                content, tool_calls, finish_reason = await _request_codex(
+                    url, headers, body, verify=True, timeout=timeout
+                )
             except Exception as e:
                 if "CERTIFICATE_VERIFY_FAILED" not in str(e):
                     raise
                 logger.warning("SSL certificate verification failed for Codex API; retrying with verify=False")
-                content, tool_calls, finish_reason = await _request_codex(url, headers, body, verify=False)
+                content, tool_calls, finish_reason = await _request_codex(
+                    url, headers, body, verify=False, timeout=timeout
+                )
             return LLMResponse(
                 content=content,
                 tool_calls=tool_calls,
@@ -85,6 +90,7 @@ async def chat(
             return LLMResponse(
                 content=f"Error calling Codex: {str(e)}",
                 finish_reason="error",
+                error_classification=self.classify_error(e),
             )
 
     def get_default_model(self) -> str:
@@ -114,13 +120,14 @@ async def _request_codex(
     headers: dict[str, str],
     body: dict[str, Any],
     verify: bool,
+    timeout: float,
 ) -> tuple[str, list[ToolCallRequest], str]:
-    async with httpx.AsyncClient(timeout=60.0, verify=verify) as client:
+    async with httpx.AsyncClient(timeout=timeout, verify=verify) as client:
         async with client.stream("POST", url, headers=headers, json=body) as response:
             if response.status_code != 200:
                 text = await response.aread()
                 raise RuntimeError(_friendly_error(response.status_code, text.decode("utf-8", "ignore")))
-            return await _consume_sse(response)
+            return await _consume_sse(response, timeout)
 
 
 def _convert_tools(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
@@ -236,9 +243,17 @@ def _prompt_cache_key(messages: list[dict[str, Any]]) -> str:
     return hashlib.sha256(raw.encode("utf-8")).hexdigest()
 
 
-async def _iter_sse(response: httpx.Response) -> AsyncGenerator[dict[str, Any], None]:
+async def _iter_sse(response: httpx.Response, timeout: float) -> AsyncGenerator[dict[str, Any], None]:
     buffer: list[str] = []
-    async for line in response.aiter_lines():
+    # Per-event idle cap: aiter_lines resets httpx's read timer on every byte,
+    # so a trickle/keepalive stall never trips it. wait_for on each line bounds
+    # the silence between SSE lines without penalizing a long, progressing run.
+    lines = response.aiter_lines()
+    while True:
+        try:
+            line = await asyncio.wait_for(lines.__anext__(), timeout)
+        except StopAsyncIteration:
+            break
         if line == "":
             if buffer:
                 data_lines = [ln[5:].strip() for ln in buffer if ln.startswith("data:")]
@@ -256,13 +271,13 @@ async def _iter_sse(response: httpx.Response) -> AsyncGenerator[dict[str, Any],
         buffer.append(line)
 
 
-async def _consume_sse(response: httpx.Response) -> tuple[str, list[ToolCallRequest], str]:
+async def _consume_sse(response: httpx.Response, timeout: float) -> tuple[str, list[ToolCallRequest], str]:
     content = ""
     tool_calls: list[ToolCallRequest] = []
     tool_call_buffers: dict[str, dict[str, Any]] = {}
     finish_reason = "stop"
 
-    async for event in _iter_sse(response):
+    async for event in _iter_sse(response, timeout):
         event_type = event.get("type")
         if event_type == "response.output_item.added":
             item = event.get("item") or {}

raven/providers/per_model_provider.py

@@ -30,6 +30,12 @@ def __init__(self, models: "Sequence[ModelEndpoint]", fallback: LLMProvider):
             if m.model
         }
         self._default = next(iter(self._by_model), None) or fallback.get_default_model()
+        # Per-model sub-providers are built here without generation settings;
+        # inherit the fallback's (already configured from AgentDefaults) and push
+        # them down so routed calls honor temperature / max_tokens / timeout.
+        self.generation = fallback.generation
+        for sub in self._by_model.values():
+            sub.generation = self.generation
 
     def _pick(self, model: str | None) -> LLMProvider:
         return self._by_model.get(model or "", self._fallback)

tests/test_agent_loop_stream.py

@@ -12,7 +12,7 @@
 from typing import Any
 
 from raven.agent.loop import AgentLoop
-from raven.providers.base import LLMResponse, StreamDelta
+from raven.providers.base import LLMProvider, LLMResponse, StreamDelta
 
 
 class _FakeProvider:
@@ -217,6 +217,34 @@ async def on_delta(_text: str) -> None:
 # ---------------------------------------------------------------------------
 
 
+async def test_llm_call_stream_timeout_returns_structured_error() -> None:
+    """A mid-stream stall (TimeoutError from the per-chunk idle cap) terminates
+    with a structured, retryable error response instead of propagating and
+    crashing the turn. Already-streamed content is preserved on the response."""
+
+    class _TimeoutStreamProvider:
+        classify_error = LLMProvider.classify_error
+
+        async def chat_stream(self, **_kwargs: Any):
+            yield StreamDelta(content="partial")
+            raise TimeoutError
+
+    call = _bind_helper(_TimeoutStreamProvider())
+    seen: list[str] = []
+
+    async def on_delta(text: str) -> None:
+        seen.append(text)
+
+    response = await call(messages=[], tools=None, model="m", on_token_delta=on_delta)
+
+    assert response.finish_reason == "error"
+    assert response.error_classification is not None
+    assert response.error_classification.category == "network"
+    assert response.error_classification.retryable is True
+    assert response.content == "partial"
+    assert seen == ["partial"]
+
+
 async def test_llm_call_stream_empty_stream_yields_empty_content() -> None:
     """Provider yields zero chunks → response.content == '' + finish_reason='stop'."""
     provider = _FakeProvider([])