🪲 BUG-#79: Preserve partial stream text on mid-response timeout

docs/examples/provider.example.json

@@ -5,7 +5,7 @@
   "headers": {
     "X-Org": "acme"
   },
-  "timeout": 60,
+  "timeout": 180,
 
   "_comment": "response_paths is optional. Omit it entirely if the API already returns the OpenAI chat-completions shape (choices[0].message.content, usage.prompt_tokens, ...). Keep it only to remap a different shape, like the example below.",
   "response_paths": {

pycodeloop/providers/generic.py

@@ -137,7 +137,7 @@ class GenericProvider(Provider):
           "model": "my-model",
           "api_key_env": "MY_API_KEY",
           "headers": {"X-Custom": "value"},
-          "timeout": 60,
+          "timeout": 180,
           "context_window": 4096,
           "response_paths": {
             "text": "choices.0.message.content",
@@ -191,7 +191,7 @@ def __init__(
         auth_prefix: str = "Bearer ",
         request_builder: RequestBuilder | None = None,
         response_parser: ResponseParser | None = None,
-        timeout: float = 60.0,
+        timeout: float = 180.0,
         repetition_min_period: int = _REPETITION_MIN_PERIOD,
         repetition_max_period: int = _REPETITION_MAX_PERIOD,
         repetition_repeats: int = _REPETITION_REPEATS,
@@ -258,7 +258,7 @@ def _build_from_json(cls, path: str | Path) -> GenericProvider:
             auth_prefix=data.get("auth_prefix", "Bearer "),
             request_builder=request_builder,
             response_parser=response_parser,
-            timeout=data.get("timeout", 60.0),
+            timeout=data.get("timeout", 180.0),
             context_window=data.get("context_window"),
             supports_openai_sse=response_shape != "anthropic",
             include_usage_in_stream=data.get("include_usage_in_stream", True),
@@ -406,82 +406,90 @@ def _stream(
         saw_terminal_marker = False
         usage = Usage()
 
-        with self._open(body, config) as response:
-            for raw_line in response:
-                if cancel_event is not None and cancel_event.is_set():
-                    stop_reason = "cancelled"
-                    saw_terminal_marker = True
-                    break
-                line = raw_line.decode().strip()
-                if not line or not line.startswith("data: "):
-                    continue
-                payload = line[len("data: ") :]
-                if payload == "[DONE]":
-                    saw_terminal_marker = True
-                    break
-                try:
-                    chunk = json.loads(payload)
-                except json.JSONDecodeError:
-                    if stop_reason is None:
-                        stop_reason = "malformed_stream"
-                    break
-
-                if chunk.get("usage"):
-                    usage = Usage(
-                        input_tokens=chunk["usage"].get("prompt_tokens", 0),
-                        output_tokens=chunk["usage"].get(
-                            "completion_tokens", 0
-                        ),
-                    )
-
-                choices = chunk.get("choices") or []
-                if not choices:
-                    continue
-                choice = choices[0]
-                delta = choice.get("delta") or {}
-
-                if delta.get("content"):
-                    candidate = text + delta["content"]
-                    if _is_repeating(
-                        candidate,
-                        self.repetition_min_period,
-                        self.repetition_max_period,
-                        self.repetition_repeats,
-                    ):
-                        stop_reason = "repetition"
+        try:
+            with self._open(body, config) as response:
+                for raw_line in response:
+                    if cancel_event is not None and cancel_event.is_set():
+                        stop_reason = "cancelled"
+                        saw_terminal_marker = True
+                        break
+                    line = raw_line.decode().strip()
+                    if not line or not line.startswith("data: "):
+                        continue
+                    payload = line[len("data: ") :]
+                    if payload == "[DONE]":
+                        saw_terminal_marker = True
                         break
-                    text = candidate
-                    on_delta(delta["content"])
-
-                for tc in delta.get("tool_calls") or []:
-                    index = tc.get("index", 0)
-                    acc = pending.setdefault(
-                        index,
-                        {
-                            "id": None,
-                            "name": None,
-                            "arguments": "",
-                            "extra": {},
-                        },
-                    )
-                    if tc.get("id"):
-                        acc["id"] = tc["id"]
-                    function = tc.get("function") or {}
-                    if function.get("name"):
-                        acc["name"] = function["name"]
-                    if function.get("arguments"):
-                        acc["arguments"] += function["arguments"]
-                    acc["extra"].update(
-                        {
-                            k: v
-                            for k, v in tc.items()
-                            if k not in ("index", "id", "type", "function")
-                        }
-                    )
-
-                if choice.get("finish_reason"):
-                    stop_reason = choice["finish_reason"]
-                    saw_terminal_marker = True
+                    try:
+                        chunk = json.loads(payload)
+                    except json.JSONDecodeError:
+                        if stop_reason is None:
+                            stop_reason = "malformed_stream"
+                        break
+
+                    if chunk.get("usage"):
+                        usage = Usage(
+                            input_tokens=chunk["usage"].get(
+                                "prompt_tokens", 0
+                            ),
+                            output_tokens=chunk["usage"].get(
+                                "completion_tokens", 0
+                            ),
+                        )
+
+                    choices = chunk.get("choices") or []
+                    if not choices:
+                        continue
+                    choice = choices[0]
+                    delta = choice.get("delta") or {}
+
+                    if delta.get("content"):
+                        candidate = text + delta["content"]
+                        if _is_repeating(
+                            candidate,
+                            self.repetition_min_period,
+                            self.repetition_max_period,
+                            self.repetition_repeats,
+                        ):
+                            stop_reason = "repetition"
+                            break
+                        text = candidate
+                        on_delta(delta["content"])
+
+                    for tc in delta.get("tool_calls") or []:
+                        index = tc.get("index", 0)
+                        acc = pending.setdefault(
+                            index,
+                            {
+                                "id": None,
+                                "name": None,
+                                "arguments": "",
+                                "extra": {},
+                            },
+                        )
+                        if tc.get("id"):
+                            acc["id"] = tc["id"]
+                        function = tc.get("function") or {}
+                        if function.get("name"):
+                            acc["name"] = function["name"]
+                        if function.get("arguments"):
+                            acc["arguments"] += function["arguments"]
+                        acc["extra"].update(
+                            {
+                                k: v
+                                for k, v in tc.items()
+                                if k not in ("index", "id", "type", "function")
+                            }
+                        )
+
+                    if choice.get("finish_reason"):
+                        stop_reason = choice["finish_reason"]
+                        saw_terminal_marker = True
+        except (TimeoutError, ConnectionError, urllib.error.URLError):
+            if not text and not pending:
+                raise
+            stop_reason = "connection_lost"
+            saw_terminal_marker = False
 
         if stop_reason is None:
             stop_reason = "stop" if saw_terminal_marker else "connection_lost"

tests/providers/test_generic.py

@@ -21,6 +21,37 @@ def __exit__(self, *exc):
         return False
 
 
+class _TimeoutAfterLinesResponse(io.BytesIO):
+    """Like `_FakeResponse`, but raises `TimeoutError` once iteration
+    passes `raise_after` lines — simulates the model going quiet
+    mid-stream (long reasoning) and the socket's read timeout firing
+    before any terminal marker or `finish_reason` arrives."""
+
+    def __init__(self, data: bytes, raise_after: int):
+        super().__init__(data)
+        self._raise_after = raise_after
+        self._yielded = 0
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        self.close()
+        return False
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        if self._yielded >= self._raise_after:
+            raise TimeoutError("timed out")
+        line = super().readline()
+        if not line:
+            raise StopIteration
+        self._yielded += 1
+        return line
+
+
 class GenericProviderTestCase(unittest.TestCase):
     def setUp(self):
         tmpdir = tempfile.TemporaryDirectory()
@@ -356,6 +387,49 @@ def test_streaming_flags_a_connection_dropped_mid_response(self):
         self.assertEqual(result.text, "cut off mid")
         self.assertEqual(result.stop_reason, "connection_lost")
 
+    def test_streaming_returns_partial_text_on_mid_stream_timeout(self):
+        """Regression: a `TimeoutError` raised mid-read (model silent for
+        longer than the socket timeout while "thinking") used to
+        propagate out of `_stream()` uncaught, discarding whatever text
+        had already streamed in and losing the assistant's turn
+        entirely instead of returning it as a partial response."""
+        path = self._write_config(
+            {"url": "http://fake/v1/chat/completions", "model": "my-model"}
+        )
+        provider = GenericProvider.from_json(path)
+
+        chunks = [
+            {"choices": [{"delta": {"content": "partial "}}]},
+            {"choices": [{"delta": {"content": "answer"}}]},
+        ]
+        sse_body = "".join(f"data: {json.dumps(c)}\n" for c in chunks).encode()
+
+        with mock.patch(
+            "pycodeloop.providers.generic.urllib.request.urlopen",
+            return_value=_TimeoutAfterLinesResponse(sse_body, raise_after=1),
+        ):
+            result = provider.complete("sys", [], [], on_delta=lambda _: None)
+
+        self.assertEqual(result.text, "partial ")
+        self.assertEqual(result.stop_reason, "connection_lost")
+
+    def test_streaming_reraises_timeout_when_nothing_was_streamed_yet(self):
+        """A timeout before any content or tool-call delta arrived means
+        nothing was generated to preserve — the exception should still
+        propagate so `Agent._complete()`'s existing retry logic kicks
+        in, instead of being swallowed into an empty response."""
+        path = self._write_config(
+            {"url": "http://fake/v1/chat/completions", "model": "my-model"}
+        )
+        provider = GenericProvider.from_json(path)
+
+        with mock.patch(
+            "pycodeloop.providers.generic.urllib.request.urlopen",
+            return_value=_TimeoutAfterLinesResponse(b"", raise_after=0),
+        ):
+            with self.assertRaises(TimeoutError):
+                provider.complete("sys", [], [], on_delta=lambda _: None)
+
     def test_streaming_stops_promptly_when_cancel_event_is_set(self):
         """Regression: cancel_event was accepted nowhere in the streaming
         read loop, so pressing Esc/Cancel mid-response did nothing until