Assistant streams its answer token by token

ai/client.py streams the completion and puts the chunks back together
(Assembler), covering OpenAI's piecewise tool-call arguments, Ollama's whole
calls, OpenRouter's comment lines, unnumbered calls and in-stream errors.
The turn emits a delta event per piece and a message per round as before,
and the chat grows the last answer as deltas arrive.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WgBfkm4rCmtd1T4vHWjEPx

backend/fluksio/ai/agent.py

@@ -119,8 +119,10 @@ async def run_turn(
 ) -> AsyncIterator[dict[str, Any]]:
     """One user turn, yielding progress events as it goes.
 
-    A round is appended to *messages* only once every tool in it has answered,
-    so a turn cut off part-way leaves a transcript the provider still accepts.
+    A round's text arrives as ``delta`` events while the model writes it, and
+    once more whole as ``message`` when the round is done. A round is appended
+    to *messages* only once every tool in it has answered, so a turn cut off
+    part-way leaves a transcript the provider still accepts.
     """
     messages.append({"role": "user", "content": text})
     tools = await tool_specs()
@@ -128,18 +130,22 @@ async def run_turn(
     token = caller_authorization.set(authorization)
     try:
         for _ in range(config.max_rounds):
+            choice: dict[str, Any] = {}
             try:
-                response = await client.chat_completion(
+                async for part in client.stream_completion(
                     api_base=config.api_base,
                     api_key=config.api_key,
                     model=config.model,
                     messages=[system, *messages],
                     tools=tools,
-                )
+                ):
+                    if isinstance(part, str):
+                        yield {"type": "delta", "content": part}
+                    else:
+                        choice = part
             except (client.ProviderError, httpx.HTTPError) as exc:
                 yield {"type": "error", "detail": f"The provider answered {exc}"}
                 return
-            choice = response["choices"][0]["message"]
             content = choice.get("content") or ""
             calls = choice.get("tool_calls") or []
             assistant: dict[str, Any] = {"role": "assistant", "content": content}

backend/fluksio/ai/client.py

@@ -4,10 +4,14 @@ Chat completions with function tools is the dialect every hosted provider and
 every local server speaks — OpenAI, Anthropic's compatibility endpoint,
 OpenRouter, Ollama, vLLM — so one client covers them without a provider SDK.
 The admin types the URL; the model never chooses where a request goes.
+
+The call is streamed, so the answer reaches the browser as it is written.
 """
 
 from __future__ import annotations
 
+import json
+from collections.abc import AsyncIterator
 from typing import Any
 from urllib.parse import urlparse
 
@@ -39,7 +43,69 @@ def _detail(response: httpx.Response) -> str:
     return f"{response.status_code}: {text[:300]}"
 
 
-async def chat_completion(
+class Assembler:
+    """The streamed chunks put back together into the message a plain call returns.
+
+    The providers stream the same ``data:`` lines and differ only in the
+    margins: OpenRouter interleaves ``:`` comment lines, a thinking model's
+    reasoning arrives in a field of its own and is left out, OpenAI and
+    Anthropic send a tool call's arguments in pieces where Ollama sends the call
+    whole, and some leave out the call's ``index``.
+    """
+
+    def __init__(self) -> None:
+        self.content = ""
+        self.calls: list[dict[str, Any]] = []
+
+    def feed(self, line: str) -> str:
+        """One line of the stream; returns the text it adds to the answer."""
+        if not line.startswith("data:"):
+            return ""
+        data = line[5:].strip()
+        if not data or data == "[DONE]":
+            return ""
+        try:
+            chunk = json.loads(data)
+        except ValueError as exc:
+            raise ProviderError(f"sent {data[:300]}") from exc
+        error = chunk.get("error")
+        if error:
+            if isinstance(error, dict):
+                error = error.get("message", error)
+            raise ProviderError(str(error)[:300])
+        choices = chunk.get("choices") or [{}]
+        delta = choices[0].get("delta") or {}
+        for piece in delta.get("tool_calls") or []:
+            self._merge(piece)
+        text: str = delta.get("content") or ""
+        self.content += text
+        return text
+
+    def _merge(self, piece: dict[str, Any]) -> None:
+        index = piece.get("index", len(self.calls))
+        while len(self.calls) <= index:
+            self.calls.append(
+                {
+                    "id": "",
+                    "type": "function",
+                    "function": {"name": "", "arguments": ""},
+                }
+            )
+        call = self.calls[index]
+        call["id"] = piece.get("id") or call["id"]
+        function = piece.get("function") or {}
+        call["function"]["name"] = function.get("name") or call["function"]["name"]
+        call["function"]["arguments"] += function.get("arguments") or ""
+
+    def message(self) -> dict[str, Any]:
+        return {
+            "role": "assistant",
+            "content": self.content or None,
+            "tool_calls": self.calls or None,
+        }
+
+
+async def stream_completion(
     *,
     api_base: str,
     api_key: str,
@@ -47,21 +113,33 @@ async def chat_completion(
     messages: list[dict[str, Any]],
     tools: list[dict[str, Any]],
     timeout: float = 120.0,
-) -> dict[str, Any]:
-    """``POST {api_base}/chat/completions``, the whole answer at once."""
-    body: dict[str, Any] = {"model": model, "messages": messages}
+) -> AsyncIterator[str | dict[str, Any]]:
+    """``POST {api_base}/chat/completions`` streamed.
+
+    Yields each piece of the answer's text as it arrives, then the whole
+    message. The timeout is per chunk, so a long answer is not cut off.
+    """
+    body: dict[str, Any] = {"model": model, "messages": messages, "stream": True}
     if tools:
         body["tools"] = tools
-    async with httpx.AsyncClient(timeout=timeout) as http:
-        response = await http.post(
+    assembler = Assembler()
+    async with (
+        httpx.AsyncClient(timeout=timeout) as http,
+        http.stream(
+            "POST",
             api_base.rstrip("/") + "/chat/completions",
             json=body,
             headers=_headers(api_key),
-        )
-    if response.is_error:
-        raise ProviderError(_detail(response))
-    result: dict[str, Any] = response.json()
-    return result
+        ) as response,
+    ):
+        if response.is_error:
+            await response.aread()
+            raise ProviderError(_detail(response))
+        async for line in response.aiter_lines():
+            text = assembler.feed(line)
+            if text:
+                yield text
+    yield assembler.message()
 
 
 async def list_models(api_base: str, api_key: str, timeout: float = 15.0) -> list[str]:

backend/fluksio/api/routes/ai.py

@@ -259,8 +259,9 @@ def chat(
 ) -> StreamingResponse:
     """One turn, streamed as server-sent events.
 
-    ``message``, ``tool_call`` and ``tool_result`` events as the model works,
-    ``error`` when something stops it, and ``done`` last either way.
+    ``delta`` as the model writes, ``message`` with a round's whole text,
+    ``tool_call`` and ``tool_result`` as it works, ``error`` when something
+    stops it, and ``done`` last either way.
     """
     config = ai_config.load()
     if config is None or not config.ready:

backend/tests/test_ai.py

@@ -1,14 +1,14 @@
 """The assistant: who may set it up, and that it acts as the person chatting."""
 
 import json
-from collections.abc import Iterator
+from collections.abc import AsyncIterator, Iterator
 from pathlib import Path
 from typing import Any
 
 import pytest
 from fastapi.testclient import TestClient
 
-from fluksio.ai import agent
+from fluksio.ai import agent, client
 from fluksio.core.config import settings
 
 
@@ -94,12 +94,16 @@ def test_a_turn_runs_the_tools_as_the_person_chatting(
         ]
     )
 
-    async def fake_completion(**kwargs: Any) -> dict[str, Any]:
+    async def fake_completion(**kwargs: Any) -> AsyncIterator[str | dict[str, Any]]:
         seen.append(kwargs["messages"])
         assert any(t["function"]["name"] == "save_flow" for t in kwargs["tools"])
-        return {"choices": [{"message": next(answers)}]}
+        answer = next(answers)
+        if answer["content"]:
+            yield "Counted "
+            yield "them."
+        yield answer
 
-    monkeypatch.setattr(agent.client, "chat_completion", fake_completion)
+    monkeypatch.setattr(agent.client, "stream_completion", fake_completion)
 
     with client.stream(
         "POST",
@@ -114,6 +118,8 @@ def test_a_turn_runs_the_tools_as_the_person_chatting(
         "start",
         "tool_call",
         "tool_result",
+        "delta",
+        "delta",
         "message",
         "done",
     ]
@@ -136,3 +142,111 @@ def test_a_turn_runs_the_tools_as_the_person_chatting(
     }
     client.delete("/api/v1/ai/chat", headers=superuser_token_headers)
     assert client.get("/api/v1/ai/chat", headers=superuser_token_headers).json() == []
+
+
+# One answer per provider as it streams it. Ollama's text is recorded from a
+# local server, the rest follows the shape each endpoint documents.
+STREAMS = {
+    "ollama": """
+data: {"id":"chatcmpl-888","object":"chat.completion.chunk","model":"gemma3:4b-it-qat","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":"Hello"},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-888","object":"chat.completion.chunk","model":"gemma3:4b-it-qat","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":" there"},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-888","object":"chat.completion.chunk","model":"gemma3:4b-it-qat","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":"stop"}]}
+
+data: [DONE]
+""",
+    "ollama-tools": """
+data: {"id":"chatcmpl-12","object":"chat.completion.chunk","model":"qwen3","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":"","reasoning":"The user wants"},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-12","object":"chat.completion.chunk","model":"qwen3","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":"","tool_calls":[{"id":"call_k2","index":0,"type":"function","function":{"name":"get_flow","arguments":"{\\"name\\":\\"demo\\"}"}}]},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-12","object":"chat.completion.chunk","model":"qwen3","system_fingerprint":"fp_ollama","choices":[{"index":0,"delta":{"role":"assistant","content":""},"finish_reason":"tool_calls"}]}
+
+data: [DONE]
+""",
+    "openai": """
+data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"role":"assistant","content":null,"tool_calls":[{"index":0,"id":"call_a","type":"function","function":{"name":"get_flow","arguments":""}}],"refusal":null},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"name\\""}}]},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"tool_calls":[{"index":1,"id":"call_b","type":"function","function":{"name":"list_flows","arguments":""}}]},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":":\\"demo\\"}"}}]},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{"tool_calls":[{"index":1,"function":{"arguments":"{}"}}]},"finish_reason":null}]}
+
+data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}
+
+data: {"id":"chatcmpl-9","object":"chat.completion.chunk","model":"gpt-4.1","choices":[],"usage":{"prompt_tokens":9,"completion_tokens":12,"total_tokens":21}}
+
+data: [DONE]
+""",
+    "anthropic": """
+data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"role":"assistant","content":"Reading "},"finish_reason":null}]}
+
+data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"content":"it."},"finish_reason":null}]}
+
+data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"toolu_01","type":"function","function":{"name":"get_flow","arguments":""}}]},"finish_reason":null}]}
+
+data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\\"name\\": \\"de"}}]},"finish_reason":null}]}
+
+data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"mo\\"}"}}]},"finish_reason":null}]}
+
+data: {"id":"msg_01","object":"chat.completion.chunk","model":"claude-sonnet-4-5","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}
+
+data: [DONE]
+""",
+    "openrouter": """
+: OPENROUTER PROCESSING
+
+data: {"id":"gen-1","provider":"Google","model":"google/gemini-2.5-flash","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"","reasoning":"Short answer."},"finish_reason":null}]}
+
+: OPENROUTER PROCESSING
+
+data: {"id":"gen-1","provider":"Google","model":"google/gemini-2.5-flash","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":"Hello"},"finish_reason":null}]}
+
+data: {"id":"gen-1","provider":"Google","model":"google/gemini-2.5-flash","object":"chat.completion.chunk","choices":[{"index":0,"delta":{"role":"assistant","content":" there"},"finish_reason":"stop"}],"usage":{"prompt_tokens":4,"completion_tokens":2,"total_tokens":6}}
+
+data: [DONE]
+""",
+    # Gemini's compatibility endpoint sends each call whole and unnumbered.
+    "gemini": """
+data: {"choices":[{"delta":{"role":"assistant","tool_calls":[{"id":"c1","type":"function","function":{"name":"get_flow","arguments":"{\\"name\\":\\"demo\\"}"}},{"id":"c2","type":"function","function":{"name":"list_flows","arguments":"{}"}}]},"finish_reason":"tool_calls","index":0}],"object":"chat.completion.chunk"}
+
+data: [DONE]
+""",
+}
+
+GET_DEMO = {"name": "get_flow", "arguments": '{"name":"demo"}'}
+LIST = {"name": "list_flows", "arguments": "{}"}
+
+
+@pytest.mark.parametrize(
+    ("provider", "content", "calls"),
+    [
+        ("ollama", "Hello there", []),
+        ("ollama-tools", None, [GET_DEMO]),
+        ("openai", None, [GET_DEMO, LIST]),
+        ("anthropic", "Reading it.", [{**GET_DEMO, "arguments": '{"name": "demo"}'}]),
+        ("openrouter", "Hello there", []),
+        ("gemini", None, [GET_DEMO, LIST]),
+    ],
+)
+def test_a_streamed_answer_comes_back_whole(
+    provider: str, content: str | None, calls: list[dict[str, str]]
+) -> None:
+    assembler = client.Assembler()
+    pieces = [assembler.feed(line) for line in STREAMS[provider].splitlines()]
+    message = assembler.message()
+    assert "".join(pieces) == (content or "")
+    assert message["content"] == content
+    assert [c["function"] for c in message["tool_calls"] or []] == calls
+    assert all(c["id"] and c["type"] == "function" for c in message["tool_calls"] or [])
+
+
+def test_an_error_inside_the_stream_stops_it() -> None:
+    assembler = client.Assembler()
+    line = 'data: {"error":{"code":502,"message":"Provider disconnected"},"choices":[{"index":0,"delta":{"content":""},"finish_reason":"error"}]}'
+    with pytest.raises(client.ProviderError, match="Provider disconnected"):
+        assembler.feed(line)

docs/code/agents.md

@@ -54,7 +54,8 @@ A superuser configures an OpenAI-compatible provider under
 the dashboard's Assistant entry for signed-in users.
 
 The assistant uses the same tools with your own permissions. Draft edits
-appear in the editor, and each answer renders as markdown with a tool trace.
+appear in the editor. Each answer streams in as the model writes it and
+renders as markdown with a tool trace.
 Conversations are stored per user on the instance; the bin button clears
 yours.
 `MCP_ENABLED` is only needed for external clients, so the built-in assistant

frontend/src/client/sdk.gen.ts

@@ -115,8 +115,9 @@ export class AiService {
      * Chat
      * One turn, streamed as server-sent events.
      *
-     * ``message``, ``tool_call`` and ``tool_result`` events as the model works,
-     * ``error`` when something stops it, and ``done`` last either way.
+     * ``delta`` as the model writes, ``message`` with a round's whole text,
+     * ``tool_call`` and ``tool_result`` as it works, ``error`` when something
+     * stops it, and ``done`` last either way.
      * @param data The data for the request.
      * @param data.requestBody
      * @returns unknown Successful Response

frontend/src/components/Chat/ChatPanel.tsx

@@ -175,10 +175,12 @@ export function ChatPanel() {
       .finally(() => setLoaded(true))
   }, [open, loaded, available])
 
-  // biome-ignore lint/correctness/useExhaustiveDependencies: scroll when the list grows, not on every render.
+  const tail = shown[shown.length - 1]
+  const tailLength = tail && "text" in tail ? tail.text.length : 0
+  // biome-ignore lint/correctness/useExhaustiveDependencies: scroll when the list or a streamed answer grows, not on every render.
   useEffect(() => {
     bottom.current?.scrollIntoView({ block: "end" })
-  }, [shown.length, waiting])
+  }, [shown.length, tailLength, waiting])
 
   const send = async () => {
     const text = draft.trim()
@@ -190,11 +192,18 @@ export function ChatPanel() {
     controller.current = abort
     const onEvent = (event: ChatEvent) => {
       switch (event.type) {
+        // A round's text grows in the last item and is settled by `message`;
+        // a tool call or the next turn always comes between two rounds.
+        case "delta":
         case "message":
-          setItems((prev) => [
-            ...prev,
-            { kind: "assistant", text: event.content },
-          ])
+          setItems((prev) => {
+            const last = prev[prev.length - 1]
+            if (last?.kind !== "assistant")
+              return [...prev, { kind: "assistant", text: event.content }]
+            const text =
+              event.type === "delta" ? last.text + event.content : event.content
+            return [...prev.slice(0, -1), { kind: "assistant", text }]
+          })
           break
         case "tool_call":
           setItems((prev) => [

frontend/src/components/Chat/stream.ts

@@ -3,6 +3,7 @@ import { apiToken } from "@/lib/portal"
 
 export type ChatEvent =
   | { type: "start" }
+  | { type: "delta"; content: string }
   | { type: "message"; content: string }
   | { type: "tool_call"; name: string; arguments: Record<string, unknown> }
   | { type: "tool_result"; name: string; ok: boolean; error?: string }