make openrouter default to provider supporting tool use

consolidate streaming

routstr/upstream/base.py

@@ -766,7 +766,9 @@ class BaseUpstreamProvider:
                     except Exception:
                         pass
 
-            def _process_event(raw_event: bytes) -> Iterator[bytes]:
+            def _process_event(
+                raw_event: bytes, final: bool = False
+            ) -> Iterator[bytes]:
                 """Process one complete SSE event block (lines up to a blank line).
 
                 Handles arbitrary upstream framing across every supported
@@ -872,6 +874,12 @@ class BaseUpstreamProvider:
                         return
                     yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n"
                 else:
+                    if final:
+                        # Final flush of a truncated tail: the upstream closed
+                        # mid-event, so ``data`` is incomplete JSON. Emitting it
+                        # as a ``data:`` frame would hand the client invalid
+                        # JSON (the "unexpected token" parse error). Drop it.
+                        return
                     # Non-JSON data payload (partial fragment already reassembled
                     # by buffering, or a provider control string). Re-prefix each
                     # line so multi-line ``data`` stays valid SSE framing - a bare
@@ -890,7 +898,13 @@ class BaseUpstreamProvider:
                 # boundary-independent for every provider.
                 buffer = b""
                 async for chunk in response.aiter_bytes():
-                    buffer += chunk.replace(b"\r\n", b"\n")
+                    # Normalize the *joined* buffer, not each chunk in
+                    # isolation: a CRLF event delimiter can straddle two
+                    # ``aiter_bytes`` chunks (``...\r`` then ``\n...``). A
+                    # per-chunk replace would leave a stray ``\r`` and the
+                    # ``\n\n`` split would miss the delimiter, merging two
+                    # events into one frame and breaking SSE clients.
+                    buffer = (buffer + chunk).replace(b"\r\n", b"\n")
                     while b"\n\n" in buffer:
                         raw_event, buffer = buffer.split(b"\n\n", 1)
                         for out in _process_event(raw_event):
@@ -898,7 +912,7 @@ class BaseUpstreamProvider:
 
                 # Flush any trailing event that lacked a final blank line.
                 if buffer.strip():
-                    for out in _process_event(buffer):
+                    for out in _process_event(buffer, final=True):
                         yield out
 
                 async with create_session() as session:
@@ -1204,7 +1218,9 @@ class BaseUpstreamProvider:
                     except Exception:
                         pass
 
-            def _process_event(raw_event: bytes) -> Iterator[bytes]:
+            def _process_event(
+                raw_event: bytes, final: bool = False
+            ) -> Iterator[bytes]:
                 """Process one complete SSE event block for the Responses API.
 
                 Buffers full events (delimited by a blank line) so parsing is
@@ -1274,6 +1290,11 @@ class BaseUpstreamProvider:
 
                     yield prefix + b"data: " + json.dumps(obj).encode() + b"\n\n"
                 else:
+                    if final:
+                        # Final flush of a truncated tail: upstream closed
+                        # mid-event, so ``data`` is incomplete JSON. Dropping it
+                        # avoids handing the client an invalid ``data:`` frame.
+                        return
                     # Re-prefix each line so multi-line ``data`` stays valid SSE
                     # framing for the client.
                     body = b"".join(
@@ -1286,14 +1307,20 @@ class BaseUpstreamProvider:
                 # delimiter so parsing is independent of byte boundaries.
                 buffer = b""
                 async for chunk in response.aiter_bytes():
-                    buffer += chunk.replace(b"\r\n", b"\n")
+                    # Normalize the *joined* buffer, not each chunk in
+                    # isolation: a CRLF event delimiter can straddle two
+                    # ``aiter_bytes`` chunks (``...\r`` then ``\n...``). A
+                    # per-chunk replace would leave a stray ``\r`` and the
+                    # ``\n\n`` split would miss the delimiter, merging two
+                    # events into one frame and breaking SSE clients.
+                    buffer = (buffer + chunk).replace(b"\r\n", b"\n")
                     while b"\n\n" in buffer:
                         raw_event, buffer = buffer.split(b"\n\n", 1)
                         for out in _process_event(raw_event):
                             yield out
 
                 if buffer.strip():
-                    for out in _process_event(buffer):
+                    for out in _process_event(buffer, final=True):
                         yield out
 
                 # Always emit a cost-bearing data chunk

routstr/upstream/openrouter.py

@@ -1,3 +1,4 @@
+import json
 from typing import TYPE_CHECKING
 
 import httpx
@@ -18,6 +19,38 @@ class OpenRouterUpstreamProvider(BaseUpstreamProvider):
     supports_anthropic_messages = True
     litellm_provider_prefix = "openrouter/"
 
+    def prepare_request_body(
+        self, body: bytes | None, model_obj: Model
+    ) -> bytes | None:
+        """Set provider.require_parameters on tool-use requests.
+
+        Without it OpenRouter can route a tool call to an endpoint that doesn't
+        support function calling and 404 with "No endpoints found that support
+        tool use". We leave a client-supplied value untouched.
+        """
+        body = super().prepare_request_body(body, model_obj)
+        if not body:
+            return body
+
+        try:
+            data = json.loads(body)
+        except json.JSONDecodeError:
+            return body
+
+        if not isinstance(data, dict) or not data.get("tools"):
+            return body
+
+        provider = data.get("provider")
+        if not isinstance(provider, dict):
+            provider = {}
+
+        if "require_parameters" in provider:
+            return body
+
+        provider["require_parameters"] = True
+        data["provider"] = provider
+        return json.dumps(data).encode()
+
     def _apply_provider_field(self, response_json: object) -> None:
         """Stamp the ``provider`` field for OpenRouter responses.
 

tests/unit/test_require_parameters.py

@@ -0,0 +1,95 @@
+import json
+import os
+
+os.environ.setdefault("UPSTREAM_BASE_URL", "http://test")
+os.environ.setdefault("UPSTREAM_API_KEY", "test")
+os.environ.setdefault("LIGHTNING_ADDRESS", "test@stm.to")
+
+from routstr.upstream import GenericUpstreamProvider
+from routstr.upstream.openrouter import OpenRouterUpstreamProvider
+
+
+def _model(model_id: str = "openai/gpt-4o"):  # type: ignore[no-untyped-def]
+    from routstr.payment.models import Architecture, Model, Pricing
+
+    return Model(
+        id=model_id,
+        name=model_id,
+        created=0,
+        description="",
+        context_length=128000,
+        architecture=Architecture(
+            modality="text->text",
+            input_modalities=["text"],
+            output_modalities=["text"],
+            tokenizer="GPT",
+            instruct_type=None,
+        ),
+        pricing=Pricing(prompt=0.0, completion=0.0),
+    )
+
+
+def _tool_body() -> dict:
+    return {
+        "model": "openai/gpt-4o",
+        "messages": [{"role": "user", "content": "What's the weather?"}],
+        "tools": [
+            {
+                "type": "function",
+                "function": {"name": "get_weather", "parameters": {}},
+            }
+        ],
+    }
+
+
+def _prepare(provider, body: dict) -> dict:  # type: ignore[no-untyped-def]
+    out = provider.prepare_request_body(json.dumps(body).encode(), _model())
+    assert out is not None
+    return json.loads(out)
+
+
+def test_injects_require_parameters_for_tool_request() -> None:
+    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), _tool_body())
+    assert data["provider"]["require_parameters"] is True
+
+
+def test_generic_provider_on_openrouter_url_is_left_alone() -> None:
+    # Only OpenRouterUpstreamProvider injects; a generic provider pointed at the
+    # same base URL doesn't.
+    provider = GenericUpstreamProvider(base_url="https://openrouter.ai/api/v1")
+    data = _prepare(provider, _tool_body())
+    assert "provider" not in data
+
+
+def test_no_injection_without_tools() -> None:
+    body = {"model": "openai/gpt-4o", "messages": [{"role": "user", "content": "hi"}]}
+    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), body)
+    assert "provider" not in data
+
+
+def test_empty_tools_list_does_not_inject() -> None:
+    body = _tool_body()
+    body["tools"] = []
+    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), body)
+    assert "provider" not in data
+
+
+def test_direct_provider_does_not_inject() -> None:
+    provider = GenericUpstreamProvider(base_url="https://api.openai.com/v1")
+    data = _prepare(provider, _tool_body())
+    assert "provider" not in data
+
+
+def test_keeps_client_set_require_parameters() -> None:
+    body = _tool_body()
+    body["provider"] = {"require_parameters": False}
+    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), body)
+    assert data["provider"]["require_parameters"] is False
+
+
+def test_preserves_other_provider_fields() -> None:
+    body = _tool_body()
+    body["provider"] = {"order": ["openai", "azure"]}
+    data = _prepare(OpenRouterUpstreamProvider(api_key="test"), body)
+    assert data["provider"]["order"] == ["openai", "azure"]
+    assert data["provider"]["require_parameters"] is True

tests/unit/test_streaming_sse_providers.py

@@ -337,3 +337,75 @@ async def test_multiline_non_json_data_each_line_prefixed() -> None:
             continue
         assert line.startswith(b"data: "), f"bare line leaked to client: {line!r}"
     assert b"data: line one" in blob and b"data: line two" in blob
+
+
+@pytest.mark.asyncio
+async def test_crlf_delimiter_split_across_chunk_boundary() -> None:
+    """CRLF event delimiter straddling two TCP reads must not merge events.
+
+    Regression: a per-chunk ``replace(b"\\r\\n", b"\\n")`` left a stray ``\\r``
+    when a ``\\r\\n`` of the ``\\r\\n\\r\\n`` delimiter landed at the very end of
+    one ``aiter_bytes`` chunk and the matching ``\\n`` opened the next. The
+    ``\\n\\n`` split then missed the boundary, glued two events into one frame
+    with two ``data:`` lines, and the client's ``JSON.parse`` threw on the
+    concatenated payload (the "unexpected token"/"Extra data" crash).
+    """
+    e1 = b'data: {"id":"x","choices":[{"delta":{"content":"a"}}]}'
+    e2 = b'data: {"id":"x","choices":[{"delta":{"content":"b"}}]}'
+    chunks = [
+        e1 + b"\r\n\r",  # delimiter cut mid-CRLF
+        b"\n" + e2 + b"\r\n\r\n",
+        b"data: [DONE]\r\n\r\n",
+    ]
+    out = await _drive(chunks)
+
+    # Client-accurate check: a real SSE client concatenates all ``data:`` lines
+    # *within one event* (events are ``\n\n``-delimited) before parsing. A
+    # merged frame would surface here as two objects glued into one payload,
+    # which ``_assert_clean`` (per-line) would miss.
+    blob = b"".join(out)
+    contents: list[str] = []
+    for event in blob.split(b"\n\n"):
+        datas = [
+            ln[len(b"data: ") :]
+            for ln in event.split(b"\n")
+            if ln.startswith(b"data: ")
+        ]
+        if not datas:
+            continue
+        payload = b"".join(datas)
+        if payload.strip() == b"[DONE]":
+            continue
+        obj = json.loads(payload)  # raises if two events were merged into one
+        for c in obj.get("choices", []):
+            if "delta" in c:
+                contents.append(c["delta"]["content"])
+    assert contents == ["a", "b"]
+
+
+@pytest.mark.asyncio
+async def test_truncated_json_tail_on_connection_close() -> None:
+    """A stream that drops mid-event must not emit the partial JSON downstream.
+
+    Regression: the end-of-stream flush ran ``_process_event`` on the leftover
+    buffer unconditionally. When the upstream connection closed mid-event the
+    leftover was incomplete JSON, which fell through to the raw-forward path and
+    handed the client a ``data: {partial`` frame -> ``Unterminated string`` parse
+    error. The truncated tail must be dropped instead.
+    """
+    chunks = [
+        b'data: {"id":"x","choices":[{"delta":{"content":"ok"}}]}\n\n',
+        b'data: {"id":"x","choices":[{"delta":{"con',  # connection dies here
+    ]
+    out = await _drive(chunks)
+    objs = _assert_clean(out)  # raises if the partial tail leaked as a data frame
+    contents = [
+        c["delta"]["content"]
+        for o in objs
+        for c in o.get("choices", [])
+        if "delta" in c
+    ]
+    # The one complete chunk is delivered; the truncated fragment is dropped
+    # entirely (no second delta), and _assert_clean above guarantees nothing
+    # non-JSON ever reached the client.
+    assert contents == ["ok"]