docs/Tuning-Guide.md

@@ -135,18 +135,23 @@ a local model can be held to less without affecting the frontier roles.
 
 ### `llm.request_timeout_s` — default 600
 
-How long one model request may take before it is abandoned. It applies to
-every request, on both routes.
+How long a model may go silent before its request is abandoned. On the chat
+route, which every local model takes, replies are streamed (#174): this is the
+longest wait for the next piece of a reply, including the wait for the first
+one while the model reads its prompt. A slow model that keeps writing is never
+cut off. On the passthrough route, which only hosted Anthropic models take,
+it is still the whole reply.
 
-- **Too low:** a slow model's long answer is cut off, and the driver's run
-  fails on work that was going well. At 120 s, the old fixed value, a dense
-  27B model on the RTX 4090 timed out writing one module in one turn.
-- **Too high:** a hung connection holds the milestone for that long before
-  anything notices.
+- **Too low:** a model reading a long prompt is cut off before it answers, and
+  the run fails on work that was going well.
+- **Too high:** a hung model holds the run for that long before anything
+  notices.
 - **Deciding:** this is a safety net, not a budget. `max_output_tokens` and
-  `max_turns` bound the work; set this above the slowest single answer you
-  expect, which is the model's output ceiling divided by its generation
-  speed.
+  `max_turns` bound the work. Set it above the longest the model takes to
+  read its largest prompt (its context window divided by its prompt speed),
+  and, for hosted Anthropic models, above their slowest whole answer.
+  Abandoning a streamed request also frees the inference server; abandoning
+  a plain one, measured on the 4090, left it finishing the reply anyway.
 
 ### `tuning.driver.max_turns` — default 40
 

nisse-core/src/nisse_core/models/interruptible.py

@@ -1,7 +1,7 @@
 """A model call that a stop can break off (L2).
 
 A ``Cancel`` takes effect at the next tool boundary, but one model call can
-run for many minutes (``request_timeout_s``), longer than a service manager
+run for many minutes, longer than a service manager
 waits before it kills the process. ``Interruptible`` wraps the client and
 knows when a call is in flight; ``interrupt``, called from the signal
 handler, raises ``Stopped`` inside that call so it ends at once.

nisse-core/src/nisse_core/models/llm.py

@@ -42,7 +42,7 @@ RATE_LIMIT_RETRY_S = 20.0
 # The proxy, or what's behind it, briefly gone: a 502/503/504 or a dropped
 # connection. Outages delay, never fail (L5): waited out with backoff before
 # the work is stopped. Not a 500 (it may be the request) and not a timeout
-# (a hung model has had its request_timeout_s already).
+# (a model silent for request_timeout_s is hung, not slow).
 OUTAGE_STATUSES = frozenset({502, 503, 504})
 OUTAGE_BACKOFF_S = (5.0, 15.0, 45.0)
 
@@ -53,10 +53,14 @@ OUTAGE_BACKOFF_S = (5.0, 15.0, 45.0)
 # was observed at 4,064 against a 4,096 limit.
 DEFAULT_MAX_OUTPUT_TOKENS = 16384
 
-# How long one request may take before it is abandoned (llm.request_timeout_s).
-# A safety net for a hung connection, not a budget: a dense local model
-# writing a whole module in one turn was measured taking over 120 s, and
-# timing it out failed work that was going well.
+# How long a model may go silent before its request is abandoned
+# (llm.request_timeout_s). A safety net for a hung connection, not a budget.
+# The chat route streams its replies (#174), so this bounds the wait for the
+# next piece of a reply, not the whole reply: a slow model that keeps writing
+# is never cut off, while one that stops is. Abandoning a stream also frees
+# the inference server, measured; abandoning a plain request left the 4090
+# finishing it for whoever came next. The passthrough route, for hosted
+# models only, still waits for the whole reply.
 DEFAULT_REQUEST_TIMEOUT_S = 600.0
 
 
@@ -212,10 +216,14 @@ class LiteLLMClient:
         # nothing to check against. A caller that knows the model's limits
         # (the context budget) passes a tighter one.
         body["max_tokens"] = max_tokens or self._max_tokens
+        # Streamed, so the timeout is on silence, not on the whole reply
+        # (DEFAULT_REQUEST_TIMEOUT_S); usage then comes in the last piece.
+        body["stream"] = True
+        body["stream_options"] = {"include_usage": True}
         self.last_exchange = exchange = {"route": "chat", "request": body}
         self.last_rate_limit_wait_s = 0.0  # not the last call's, if this one raises
-        response, self.last_rate_limit_wait_s = _post_with_rate_limit_retry(
-            self._http, "/chat/completions", body, self._sleep
+        response, self.last_rate_limit_wait_s = _send(
+            self._http, "/chat/completions", body, self._sleep, _assemble_chat
         )
         _keep_response(exchange, response)
         if response.status_code != 200:
@@ -353,7 +361,7 @@ class AnthropicPassthroughClient:
             body["tool_choice"] = {"type": "any"}  # Anthropic's name for it
         self.last_exchange = exchange = {"route": "messages", "request": body}
         self.last_rate_limit_wait_s = 0.0  # not the last call's, if this one raises
-        response, self.last_rate_limit_wait_s = _post_with_rate_limit_retry(
+        response, self.last_rate_limit_wait_s = _send(
             self._http, "/v1/messages", body, self._sleep
         )
         _keep_response(exchange, response)
@@ -415,22 +423,42 @@ class AnthropicPassthroughClient:
         self._http.close()
 
 
-def _post_with_rate_limit_retry(
-    http: httpx.Client, path: str, body: dict[str, Any], sleep: Any
+def _send(
+    http: httpx.Client,
+    path: str,
+    body: dict[str, Any],
+    sleep: Any,
+    assemble: Any = None,
 ) -> tuple[httpx.Response, float]:
     """The response, and how long was spent waiting: out a rate limit (once)
-    or an outage (with backoff), not on the model."""
+    or an outage (with backoff), not on the model.
+
+    A streamed reply is put back together by ``assemble`` into the response
+    the server would have sent unstreamed, so everything after reads one
+    shape. A server that answers a streamed request with plain JSON is
+    taken as it is."""
     waited = 0.0
     rate_limited = False
     outages = iter(OUTAGE_BACKOFF_S)
     while True:
         try:
-            response = http.post(path, json=body)
+            with http.stream("POST", path, json=body) as response:
+                streamed = response.headers.get("content-type", "").startswith(
+                    "text/event-stream"
+                )
+                if response.status_code == 200 and streamed and assemble:
+                    data = assemble(response.iter_lines())
+                    return _assembled(response, data), waited
+                response.read()
         except httpx.TimeoutException as exc:
-            raise UpstreamUnavailable(f"LLM request failed: {exc}") from exc
+            raise UpstreamUnavailable(
+                f"LLM request failed: the model said nothing for "
+                f"{http.timeout.read:g} s ({exc})"
+            ) from exc
         except httpx.HTTPError as exc:
-            # Could not reach the proxy at all: nothing about the request was
-            # wrong, so callers holding work in progress should keep it.
+            # Could not reach the proxy, or lost it mid-reply: nothing about
+            # the request was wrong, so callers holding work in progress
+            # should keep it.
             pause = next(outages, None)
             if pause is None:
                 raise UpstreamUnavailable(f"LLM request failed: {exc}") from exc
@@ -453,6 +481,88 @@ def _post_with_rate_limit_retry(
         return response, waited
 
 
+def _assembled(streamed: httpx.Response, data: dict[str, Any]) -> httpx.Response:
+    headers = {}
+    if cost := streamed.headers.get("x-litellm-response-cost"):
+        headers["x-litellm-response-cost"] = cost
+    return httpx.Response(200, json=data, headers=headers, request=streamed.request)
+
+
+def _assemble_chat(lines: Any) -> dict[str, Any]:
+    """A streamed chat completion, put back together as the one response it
+    stands for: the text, the reasoning, each tool call's arguments joined
+    from their pieces, the finish reason and the usage."""
+    model = ""
+    text: list[str] = []
+    reasoning: list[str] = []
+    calls: dict[int, dict[str, Any]] = {}
+    finish = None
+    usage: dict[str, Any] = {}
+    done = False
+    for line in lines:
+        if not line.startswith("data:"):
+            continue  # blank lines, comments, keep-alives
+        payload = line[len("data:") :].strip()
+        if payload == "[DONE]":
+            done = True
+            break
+        try:
+            chunk = json.loads(payload)
+        except json.JSONDecodeError as exc:
+            raise LLMError(
+                f"malformed piece of a streamed reply: {payload[:200]!r}"
+            ) from exc
+        if not isinstance(chunk, dict):
+            raise LLMError(f"malformed piece of a streamed reply: {payload[:200]!r}")
+        if "error" in chunk:
+            # The proxy failing mid-reply: upstream trouble, as a non-200
+            # would have been before the reply began.
+            said = json.dumps(chunk["error"])[:300]
+            if _looks_like_budget_refusal(said):
+                raise BudgetExhausted(f"the proxy refused on budget grounds: {said}")
+            raise UpstreamUnavailable(f"the reply broke off with an error: {said}")
+        model = chunk.get("model") or model
+        if isinstance(chunk.get("usage"), dict):
+            usage = chunk["usage"]
+        for choice in chunk.get("choices") or []:
+            delta = choice.get("delta") or {}
+            if isinstance(delta.get("content"), str):
+                text.append(delta["content"])
+            if isinstance(delta.get("reasoning_content"), str):
+                reasoning.append(delta["reasoning_content"])
+            for piece in delta.get("tool_calls") or []:
+                call = calls.setdefault(
+                    int(piece.get("index") or 0),
+                    {
+                        "id": "",
+                        "type": "function",
+                        "function": {"name": "", "arguments": ""},
+                    },
+                )
+                call["id"] = call["id"] or piece.get("id") or ""
+                function = piece.get("function") or {}
+                # The name comes once, the arguments in pieces.
+                call["function"]["name"] = (
+                    call["function"]["name"] or function.get("name") or ""
+                )
+                call["function"]["arguments"] += function.get("arguments") or ""
+            if choice.get("finish_reason"):
+                finish = choice["finish_reason"]
+    if not done and finish is None:
+        # Cut off without a word: the connection, not the model.
+        raise UpstreamUnavailable("the streamed reply stopped before it finished")
+    message: dict[str, Any] = {"role": "assistant", "content": "".join(text) or None}
+    if reasoning:
+        message["reasoning_content"] = "".join(reasoning)
+    if calls:
+        message["tool_calls"] = [calls[i] for i in sorted(calls)]
+    return {
+        "model": model,
+        "choices": [{"index": 0, "message": message, "finish_reason": finish}],
+        "usage": usage,
+    }
+
+
 # Model families we have a native renderer adapter for. L9: delegations take
 # the vendor's passthrough route because a normalising middlebox is lossy
 # exactly where a long-running delegation lives.
@@ -467,9 +577,17 @@ def _keep_response(exchange: dict[str, Any], response: httpx.Response) -> None:
 
 
 def _response_cost(response: httpx.Response) -> float:
+    """LiteLLM's price for the call: a header on a plain reply, the usage's
+    ``cost`` on a streamed one (sent last, once it's known)."""
+    found: Any = response.headers.get("x-litellm-response-cost")
+    if not found:
+        try:
+            found = (response.json().get("usage") or {}).get("cost")
+        except (ValueError, AttributeError):
+            found = None
     try:
-        return float(response.headers.get("x-litellm-response-cost") or 0.0)
-    except ValueError:
+        return float(found or 0.0)
+    except (TypeError, ValueError):
         return 0.0
 
 

nisse-core/tests/test_llm.py

@@ -602,3 +602,223 @@ def test_a_budget_refusal_is_not_waited_on() -> None:
     with pytest.raises(BudgetExhausted):
         make_client(handler).complete(EVENTS, tools=None, model="local/qwen")
     assert len(calls) == 1
+
+
+# --- streamed replies on the chat route (#174) -----------------------------------
+
+
+def sse(*chunks: dict, done: bool = True) -> httpx.Response:
+    lines = "".join(f"data: {json.dumps(c)}\n\n" for c in chunks)
+    return httpx.Response(
+        200,
+        headers={"content-type": "text/event-stream"},
+        content=lines + ("data: [DONE]\n\n" if done else ""),
+    )
+
+
+def piece(delta: dict | None = None, finish: str | None = None, **extra) -> dict:
+    choice: dict = {"index": 0, "delta": delta or {}}
+    if finish:
+        choice["finish_reason"] = finish
+    return {"model": "qwen-real", "choices": [choice], **extra}
+
+
+class Failing(httpx.SyncByteStream):
+    """A reply that sends its first piece, then fails."""
+
+    def __init__(self, first: bytes, failure: Exception) -> None:
+        self._first, self._failure = first, failure
+
+    def __iter__(self):
+        yield self._first
+        raise self._failure
+
+
+def test_the_chat_route_asks_for_a_stream_with_usage() -> None:
+    seen: dict = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.update(json.loads(request.content))
+        return sse(piece({"content": "hi"}, "stop"))
+
+    make_client(handler).complete(EVENTS, tools=None, model="local/qwen")
+    assert seen["stream"] is True
+    assert seen["stream_options"] == {"include_usage": True}
+
+
+def test_a_streamed_reply_is_put_back_together() -> None:
+    usage = {"prompt_tokens": 7, "completion_tokens": 3, "cost": 0.25}
+    usage["prompt_tokens_details"] = {"cached_tokens": 4}
+    client = make_client(
+        lambda request: sse(
+            piece({"role": "assistant", "reasoning_content": "think"}),
+            piece({"reasoning_content": "ing"}),
+            piece({"content": "hel"}),
+            piece({"content": "lo"}),
+            piece(finish="stop"),
+            {"model": "qwen-real", "choices": [], "usage": usage},
+        )
+    )
+    [event] = client.complete(EVENTS, tools=None, model="local/qwen")
+    assert event == AssistantText(text="hello", model="qwen-real")
+    assert client.last_usage == {"input": 7, "output": 3, "cached": 4}
+    assert client.last_cost == 0.25
+    assert client.last_model == "qwen-real"
+    # The call record keeps the reply as one response, reasoning included.
+    message = client.last_exchange["response"]["choices"][0]["message"]
+    assert message["reasoning_content"] == "thinking"
+    assert message["content"] == "hello"
+
+
+def test_streamed_tool_calls_are_joined_from_their_pieces() -> None:
+    def call(index: int, **function) -> dict:
+        return {"tool_calls": [{"index": index, "function": function}]}
+
+    client = make_client(
+        lambda request: sse(
+            piece(call(0, name="write_file", arguments='{"path":')),
+            piece(call(0, arguments=' "a.txt"}')),
+            piece(call(1, name="read_file", arguments="{")),
+            piece(call(1, arguments='"path": "b"}')),
+            piece(finish="tool_calls"),
+        )
+    )
+    first, second = client.complete(EVENTS, tools=None, model="local/qwen")
+    assert isinstance(first, ToolCall) and isinstance(second, ToolCall)
+    assert (first.tool, first.arguments) == ("write_file", {"path": "a.txt"})
+    assert (second.tool, second.arguments) == ("read_file", {"path": "b"})
+
+
+def test_a_streamed_reply_cut_off_at_the_limit_is_truncated() -> None:
+    client = make_client(lambda request: sse(piece({"content": "half"}, "length")))
+    with pytest.raises(TruncatedResponse):
+        client.complete(EVENTS, tools=None, model="local/qwen")
+
+
+def test_a_stream_that_stops_before_it_finishes_is_upstream_trouble() -> None:
+    client = make_client(lambda request: sse(piece({"content": "half"}), done=False))
+    with pytest.raises(UpstreamUnavailable, match="stopped before it finished"):
+        client.complete(EVENTS, tools=None, model="local/qwen")
+
+
+def test_an_error_mid_stream_is_upstream_trouble() -> None:
+    broke = make_client(lambda request: sse({"error": {"message": "backend died"}}))
+    with pytest.raises(UpstreamUnavailable, match="backend died"):
+        broke.complete(EVENTS, tools=None, model="local/qwen")
+    spent = make_client(
+        lambda request: sse({"error": {"message": "Budget has been exceeded"}})
+    )
+    with pytest.raises(BudgetExhausted):
+        spent.complete(EVENTS, tools=None, model="local/qwen")
+
+
+def test_a_model_gone_quiet_mid_reply_times_out_once() -> None:
+    calls: list[int] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append(1)
+        first = f"data: {json.dumps(piece({'content': 'hel'}))}\n\n".encode()
+        return httpx.Response(
+            200,
+            headers={"content-type": "text/event-stream"},
+            stream=Failing(first, httpx.ReadTimeout("quiet")),
+        )
+
+    with pytest.raises(UpstreamUnavailable, match="said nothing for 600 s"):
+        make_client(handler).complete(EVENTS, tools=None, model="local/qwen")
+    assert len(calls) == 1  # a hung model is not asked again
+
+
+def test_a_connection_lost_mid_reply_is_retried() -> None:
+    calls: list[int] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append(1)
+        if len(calls) == 1:
+            first = f"data: {json.dumps(piece({'content': 'hel'}))}\n\n".encode()
+            return httpx.Response(
+                200,
+                headers={"content-type": "text/event-stream"},
+                stream=Failing(first, httpx.RemoteProtocolError("peer closed")),
+            )
+        return sse(piece({"content": "hello"}, "stop"))
+
+    [event] = make_client(handler).complete(EVENTS, tools=None, model="local/qwen")
+    assert event == AssistantText(text="hello", model="qwen-real") and len(calls) == 2
+
+
+def test_the_passthrough_still_asks_for_the_whole_reply() -> None:
+    seen: dict = {}
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        seen.update(json.loads(request.content))
+        return httpx.Response(
+            200,
+            json={"model": "m", "content": [{"type": "text", "text": "hi"}]},
+        )
+
+    AnthropicPassthroughClient(
+        "https://litellm.test", "tok", transport=httpx.MockTransport(handler)
+    ).complete(EVENTS, tools=None, model="claude")
+    assert "stream" not in seen
+
+
+class _Server:
+    """A real HTTP server that streams a reply in pieces, ``gap`` seconds
+    apart: the timeout is httpx's own, so only a real socket shows what it
+    measures."""
+
+    def __init__(self, pieces: int, gap: float) -> None:
+        import threading
+        from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+        class Handler(BaseHTTPRequestHandler):
+            def do_POST(self) -> None:
+                self.rfile.read(int(self.headers["content-length"]))
+                self.send_response(200)
+                self.send_header("content-type", "text/event-stream")
+                self.end_headers()
+                import time
+
+                try:
+                    for n in range(pieces):
+                        time.sleep(gap)
+                        chunk = piece({"content": str(n)})
+                        self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
+                        self.wfile.flush()
+                    self.wfile.write(b"data: [DONE]\n\n")
+                except (BrokenPipeError, ConnectionResetError):
+                    pass  # the client gave up, as it should
+
+            def log_message(self, *args: Any) -> None:
+                pass
+
+        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        self.url = f"http://127.0.0.1:{self.server.server_address[1]}"
+        threading.Thread(target=self.server.serve_forever, daemon=True).start()
+
+    def close(self) -> None:
+        self.server.shutdown()
+        self.server.server_close()
+
+
+def test_a_slow_reply_that_keeps_coming_outlasts_the_timeout() -> None:
+    # Six pieces 0.1 s apart take longer than the 0.35 s timeout, but no gap
+    # does: a model still writing is never cut off (#174).
+    server = _Server(pieces=6, gap=0.1)
+    try:
+        client = LiteLLMClient(server.url, "tok", timeout_s=0.35)
+        [event] = client.complete(EVENTS, tools=None, model="local/qwen")
+    finally:
+        server.close()
+    assert isinstance(event, AssistantText) and event.text == "012345"
+
+
+def test_a_reply_that_goes_quiet_times_out() -> None:
+    server = _Server(pieces=2, gap=1.0)
+    try:
+        client = LiteLLMClient(server.url, "tok", timeout_s=0.2)
+        with pytest.raises(UpstreamUnavailable, match="said nothing"):
+            client.complete(EVENTS, tools=None, model="local/qwen")
+    finally:
+        server.close()

nisse-dev.example.yaml

@@ -40,8 +40,9 @@ llm:                                  # L9: LiteLLM is the only LLM destination
   # Ceiling on each response. The model's own max_output_tokens in LiteLLM
   # caps it further, which is where to hold a local model to less.
   max_output_tokens: 16384
-  # How long one request may take before it is abandoned. A safety net for a
-  # hung connection; slow local models can take minutes over a long answer.
+  # How long the model may go silent before its request is abandoned. Local
+  # models' replies are streamed, so a slow one that keeps writing is never
+  # cut off; this bounds reading a long prompt, or a hung model.
   request_timeout_s: 600
 
 roles:                                # L9: role -> LiteLLM model alias