🪲 BUG-#17: Stream via SSE for custom response_paths configs

pycodeloop/providers/generic.py

@@ -146,6 +146,13 @@ class GenericProvider(Provider):
     `message_content` rename per-message keys). `params` are extra
     static fields merged into every request body (e.g. `temperature`,
     `max_tokens`, vendor-specific flags).
+
+    Streaming (OpenAI-style SSE) still works with `response_paths` —
+    the wire format is unchanged, only the *non-streaming* response's
+    JSON key paths differ. `response_shape: "anthropic"` is the
+    exception: that's a genuinely different SSE envelope `_stream()`
+    doesn't understand, so streaming is skipped for it in favor of one
+    blocking, fully-buffered `on_delta` call.
     """
 
     name = "generic"
@@ -164,6 +171,7 @@ def __init__(
         repetition_min_period: int = _REPETITION_MIN_PERIOD,
         repetition_max_period: int = _REPETITION_MAX_PERIOD,
         repetition_repeats: int = _REPETITION_REPEATS,
+        supports_openai_sse: bool = True,
         **kwargs,
     ) -> None:
         super().__init__(model=model, api_key=api_key, **kwargs)
@@ -178,6 +186,7 @@ def __init__(
         self.repetition_max_period = repetition_max_period
         self.repetition_repeats = repetition_repeats
         self._uses_default_parser = response_parser is None
+        self._supports_openai_sse = supports_openai_sse
         self._config_path: Path | None = None
 
     @classmethod
@@ -199,8 +208,9 @@ def _build_from_json(cls, path: str | Path) -> GenericProvider:
         if not api_key and data.get("api_key_env"):
             api_key = os.environ.get(data["api_key_env"])
 
+        response_shape = data.get("response_shape")
         response_parser = None
-        if data.get("response_shape") == "anthropic":
+        if response_shape == "anthropic":
             response_parser = anthropic_response
         elif "response_paths" in data:
             response_parser = response_parser_from_paths(
@@ -221,6 +231,7 @@ def _build_from_json(cls, path: str | Path) -> GenericProvider:
             request_builder=request_builder,
             response_parser=response_parser,
             timeout=data.get("timeout", 60.0),
+            supports_openai_sse=response_shape != "anthropic",
         )
 
     def reload(self) -> None:
@@ -242,6 +253,7 @@ def reload(self) -> None:
         self.response_parser = fresh.response_parser
         self.timeout = fresh.timeout
         self._uses_default_parser = fresh._uses_default_parser
+        self._supports_openai_sse = fresh._supports_openai_sse
 
     @staticmethod
     def _default_request(
@@ -289,7 +301,7 @@ def complete(
         body = self.request_builder(system_prompt, messages, tools, self.model)
         known_tools = {tool["name"] for tool in tools}
 
-        if on_delta is not None and self._uses_default_parser:
+        if on_delta is not None and self._supports_openai_sse:
             return self._stream(body, on_delta, known_tools)
 
         with self._open(body) as response:

tests/providers/test_generic.py

@@ -488,6 +488,65 @@ def test_custom_tool_call_paths(self):
         self.assertEqual(result.tool_calls[0].name, "read_file")
         self.assertEqual(result.tool_calls[0].arguments, {"path": "a.py"})
 
+    def test_response_paths_config_still_streams(self):
+        path = self._write_config(
+            {
+                "url": "http://fake/answer",
+                "model": "my-model",
+                "response_paths": {"text": "result.answer"},
+            }
+        )
+        provider = GenericProvider.from_json(path)
+
+        chunks = [
+            {"choices": [{"delta": {"content": "hel"}}]},
+            {"choices": [{"delta": {"content": "lo"}}]},
+            {"choices": [{"delta": {}, "finish_reason": "stop"}]},
+        ]
+        sse_body = (
+            "".join(f"data: {json.dumps(c)}\n" for c in chunks)
+            + "data: [DONE]\n"
+        ).encode()
+
+        deltas = []
+        with mock.patch(
+            "pycodeloop.providers.generic.urllib.request.urlopen",
+            return_value=_FakeResponse(sse_body),
+        ):
+            result = provider.complete("sys", [], [], on_delta=deltas.append)
+
+        self.assertEqual(deltas, ["hel", "lo"])
+        self.assertEqual(result.text, "hello")
+
+    def test_anthropic_response_shape_falls_back_to_a_single_on_delta_call(
+        self,
+    ):
+        path = self._write_config(
+            {
+                "url": "http://fake/answer",
+                "model": "my-model",
+                "response_shape": "anthropic",
+            }
+        )
+        provider = GenericProvider.from_json(path)
+
+        response_body = json.dumps(
+            {
+                "content": [{"type": "text", "text": "hello"}],
+                "usage": {"input_tokens": 3, "output_tokens": 1},
+            }
+        ).encode()
+
+        deltas = []
+        with mock.patch(
+            "pycodeloop.providers.generic.urllib.request.urlopen",
+            return_value=_FakeResponse(response_body),
+        ):
+            result = provider.complete("sys", [], [], on_delta=deltas.append)
+
+        self.assertEqual(deltas, ["hello"])
+        self.assertEqual(result.text, "hello")
+
 
 class TestGetProviderJsonDispatch(GenericProviderTestCase):
     def test_get_provider_loads_json_config_by_path(self):