fix(server): stream tool_calls instead of agent filler on tool requests (#460)

When a client streams (stream:true) with explicit `tools`, the server
routed to the agent stream bridge, which ignored request_body.tools, ran
the agent's own tool loop, and word-split filler content into fake token
deltas — dropping the caller's tool_calls. This is the streaming analog
of #414 (whose non-streaming fix was #454).

Now stream+tools bypasses the agent and streams the model's raw
function-calling decision via engine.stream_full(), emitting OpenAI-shape
tool_calls deltas and a tool_calls finish_reason. Adds tool_calls to
DeltaMessage and removes the now-dead _handle_agent_stream.

Verified end-to-end on Ollama (qwen3.5:4b) plus a unit regression test.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/openjarvis/server/models.py

@@ -82,6 +82,9 @@ class ChatCompletionResponse(BaseModel):
 class DeltaMessage(BaseModel):
     role: Optional[str] = None
     content: Optional[str] = None
+    # Streaming tool_calls (OpenAI delta shape, with `index`). Present only
+    # on streamed raw function-calling responses (stream:true + tools).
+    tool_calls: Optional[List[Dict[str, Any]]] = None
 
 
 class StreamChoice(BaseModel):

src/openjarvis/server/routes.py

@@ -138,13 +138,19 @@ async def chat_completions(request_body: ChatCompletionRequest, request: Request
             )
 
     if request_body.stream:
-        bus = getattr(request.app.state, "bus", None)
-        # Use the agent stream bridge only when tools are present (the
-        # bridge runs agent.run() synchronously and word-splits the result,
-        # so it can't stream tokens in real-time).  For plain chat, stream
-        # directly from the engine for true token-by-token output.
-        if agent is not None and bus is not None and request_body.tools:
-            return await _handle_agent_stream(agent, bus, model, request_body)
+        # When the client passes `tools`, stream the model's raw
+        # OpenAI-compat function-calling decision directly from the engine
+        # (bypassing the agent) — the streaming mirror of the non-streaming
+        # #454 fix.  Routing tools through the agent stream bridge ignored
+        # `request_body.tools`, ran the agent's own tool loop, and
+        # word-split generic filler content into fake token deltas, so the
+        # caller's tool_calls were dropped entirely (the streaming analog of
+        # #414).  For plain chat (no tools), stream token-by-token directly
+        # from the engine for true real-time output.
+        if request_body.tools:
+            return await _handle_stream_tools(
+                engine, model, request_body, complexity_info
+            )
         return await _handle_stream(engine, model, request_body, complexity_info)
 
     # Non-streaming: use agent if available, otherwise direct engine call.
@@ -306,11 +312,113 @@ def _handle_agent(
     )
 
 
-async def _handle_agent_stream(agent, bus, model, req):
-    """Stream agent response with EventBus events via SSE."""
-    from openjarvis.server.stream_bridge import create_agent_stream
+async def _handle_stream_tools(
+    engine,
+    model: str,
+    req: ChatCompletionRequest,
+    complexity_info=None,
+):
+    """Stream a raw OpenAI-compat function-calling response via SSE.
 
-    return await create_agent_stream(agent, bus, model, req)
+    Used when the client passes `tools` together with `stream:true`.  Sources
+    tool_calls from ``engine.stream_full()`` (which forwards the tools to the
+    backend and parses tool_calls out of the streamed response) and emits them
+    as SSE deltas, bypassing the agent entirely.  This is the streaming mirror
+    of the non-streaming ``_handle_direct`` tool path.
+
+    Engines without a tool-aware ``stream_full`` override fall back to the
+    base-class default (content tokens + a ``stop`` finish_reason, no
+    tool_calls) — identical to the prior plain-stream behaviour, so this never
+    regresses non-tool-capable engines.
+    """
+    from openjarvis.server.cloud_router import is_cloud_model
+
+    messages = _to_messages(req.messages)
+    chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
+    use_cloud = is_cloud_model(model)
+
+    async def generate():
+        # Send the role chunk first (OpenAI convention).
+        first_chunk = ChatCompletionChunk(
+            id=chunk_id,
+            model=model,
+            choices=[StreamChoice(delta=DeltaMessage(role="assistant"))],
+        )
+        yield f"data: {first_chunk.model_dump_json()}\n\n"
+
+        finish_reason = "stop"
+        try:
+            async for sc in engine.stream_full(
+                messages,
+                model=model,
+                temperature=req.temperature,
+                max_tokens=req.max_tokens,
+                tools=req.tools,
+            ):
+                if sc.content:
+                    content_chunk = ChatCompletionChunk(
+                        id=chunk_id,
+                        model=model,
+                        choices=[StreamChoice(delta=DeltaMessage(content=sc.content))],
+                    )
+                    yield f"data: {content_chunk.model_dump_json()}\n\n"
+                if sc.tool_calls:
+                    tc_chunk = ChatCompletionChunk(
+                        id=chunk_id,
+                        model=model,
+                        choices=[
+                            StreamChoice(delta=DeltaMessage(tool_calls=sc.tool_calls))
+                        ],
+                    )
+                    yield f"data: {tc_chunk.model_dump_json()}\n\n"
+                if sc.finish_reason:
+                    finish_reason = sc.finish_reason
+        except Exception as exc:
+            import logging
+
+            logging.getLogger("openjarvis.server").error(
+                "Tool stream error: %s",
+                exc,
+                exc_info=True,
+            )
+            error_chunk = ChatCompletionChunk(
+                id=chunk_id,
+                model=model,
+                choices=[
+                    StreamChoice(
+                        delta=DeltaMessage(
+                            content=f"\n\nError during generation: {exc}",
+                        ),
+                        finish_reason="stop",
+                    )
+                ],
+            )
+            yield f"data: {error_chunk.model_dump_json()}\n\n"
+            yield "data: [DONE]\n\n"
+            return
+
+        import json as _json
+
+        finish_data = ChatCompletionChunk(
+            id=chunk_id,
+            model=model,
+            choices=[StreamChoice(delta=DeltaMessage(), finish_reason=finish_reason)],
+        )
+        finish_dict = _json.loads(finish_data.model_dump_json())
+        # Tag the finish chunk with the engine label, matching _handle_stream
+        # so UI/telemetry consumers see the same field on the tools path.
+        finish_dict.setdefault("telemetry", {})
+        finish_dict["telemetry"]["engine"] = "cloud" if use_cloud else "ollama"
+        if complexity_info is not None:
+            finish_dict["complexity"] = complexity_info.model_dump()
+        yield f"data: {_json.dumps(finish_dict)}\n\n"
+        yield "data: [DONE]\n\n"
+
+    return StreamingResponse(
+        generate(),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "Connection": "keep-alive"},
+    )
 
 
 async def _handle_stream(

tests/server/test_routes.py

@@ -308,6 +308,104 @@ class TestChatCompletions:
                     content += delta_content
         assert content == "Hello world"
 
+    def test_streaming_with_tools_emits_tool_calls_and_bypasses_agent(self):
+        """Regression for the streaming analog of #414.
+
+        When the client streams (`stream:true`) WITH explicit `tools`, the
+        response must carry the model's real tool_calls (sourced from
+        engine.stream_full) and a finish_reason of "tool_calls" — NOT route
+        through the agent bridge, which ignores request_body.tools, runs the
+        agent's own tool loop, and word-splits generic filler content,
+        dropping the tool_calls the caller asked for.
+        """
+        from openjarvis.core.events import EventBus
+        from openjarvis.engine._stubs import StreamChunk
+
+        engine = _make_engine()
+
+        async def mock_stream_full(
+            messages,
+            *,
+            model,
+            temperature=0.7,
+            max_tokens=1024,
+            **kwargs,
+        ):
+            # Ollama-shape: a complete tool_call arrives in a single chunk
+            # carrying finish_reason="tool_calls".
+            yield StreamChunk(
+                tool_calls=[
+                    {
+                        "index": 0,
+                        "id": "call_1",
+                        "type": "function",
+                        "function": {
+                            "name": "get_weather",
+                            "arguments": '{"city": "Paris"}',
+                        },
+                    }
+                ],
+                finish_reason="tool_calls",
+            )
+
+        engine.stream_full = mock_stream_full
+        # bus present + agent registered == the exact live condition under
+        # which the pre-fix code routed to the (broken) agent stream bridge.
+        agent = _make_agent(content="GENERIC AGENT FILLER")
+        app = create_app(engine, "test-model", agent=agent, bus=EventBus())
+        client = TestClient(app)
+
+        resp = client.post(
+            "/v1/chat/completions",
+            json={
+                "model": "test-model",
+                "messages": [
+                    {"role": "user", "content": "Weather in Paris? Use get_weather."}
+                ],
+                "tools": [
+                    {
+                        "type": "function",
+                        "function": {
+                            "name": "get_weather",
+                            "parameters": {
+                                "type": "object",
+                                "properties": {"city": {"type": "string"}},
+                                "required": ["city"],
+                            },
+                        },
+                    }
+                ],
+                "stream": True,
+            },
+        )
+        assert resp.status_code == 200
+        assert "text/event-stream" in resp.headers.get("content-type", "")
+
+        tool_call_names: list[str] = []
+        finish_reasons: list[str] = []
+        collected_content = ""
+        for line in resp.text.strip().split("\n"):
+            if not line.startswith("data:") or "[DONE]" in line:
+                continue
+            data = json.loads(line[5:].strip())
+            choice = data.get("choices", [{}])[0]
+            delta = choice.get("delta", {})
+            for tc in delta.get("tool_calls") or []:
+                tool_call_names.append(tc["function"]["name"])
+            if delta.get("content"):
+                collected_content += delta["content"]
+            if choice.get("finish_reason"):
+                finish_reasons.append(choice["finish_reason"])
+
+        # The real tool_call must be streamed through to the client.
+        assert "get_weather" in tool_call_names
+        # finish_reason must signal tool_calls, not a plain stop.
+        assert "tool_calls" in finish_reasons
+        # The agent's filler must NOT have been streamed...
+        assert "GENERIC AGENT FILLER" not in collected_content
+        # ...and the agent must not have been invoked at all.
+        assert not agent.run.called
+
     def test_finish_reason_default(self, client):
         resp = client.post(
             "/v1/chat/completions",