From a08282c3e97f2622b1b2f0a0c7c702b1f951a84d Mon Sep 17 00:00:00 2001 From: Jon Saad-Falcon <41205309+jonsaadfalcon@users.noreply.github.com> Date: Sun, 31 May 2026 16:09:36 -0700 Subject: [PATCH] fix(server): stream tool_calls instead of agent filler on tool requests (#460) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- src/openjarvis/server/models.py | 3 + src/openjarvis/server/routes.py | 130 +++++++++++++++++++++++++++++--- tests/server/test_routes.py | 98 ++++++++++++++++++++++++ 3 files changed, 220 insertions(+), 11 deletions(-) diff --git a/src/openjarvis/server/models.py b/src/openjarvis/server/models.py index be67f394..ccdf8e09 100644 --- a/src/openjarvis/server/models.py +++ b/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): diff --git a/src/openjarvis/server/routes.py b/src/openjarvis/server/routes.py index 59e891c0..49498199 100644 --- a/src/openjarvis/server/routes.py +++ b/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( diff --git a/tests/server/test_routes.py b/tests/server/test_routes.py index 93e34216..24d6bfda 100644 --- a/tests/server/test_routes.py +++ b/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",