fix(server): preserve grounded agent stream content (#736)

* fix(server): preserve grounded agent stream content

* test(server): cover active grounded stream path

* fix(server): retain agent stream bridge

src/openjarvis/server/stream_bridge.py

@@ -246,62 +246,15 @@ class AgentStreamBridge:
                     {"results": tool_results_data},
                 )
 
-            # Stream content using real LLM token streaming via
-            # engine.stream_full() when the engine is available.
+            # ``agent.run()`` already produced the authoritative, grounded
+            # response.  Do not call the engine again here: a second inference
+            # would not have the agent's system prompt, tool transcript, or
+            # other internal context and could therefore contradict the
+            # result reported by the agent events.  Replay the final content
+            # in chunks so the OpenAI-compatible streaming response stays
+            # consistent with the completed agent run.
             content = agent_result.content or ""
-            engine = getattr(self._agent, "_engine", None)
-            used_real_streaming = False
-
-            if engine is not None and hasattr(engine, "stream_full") and content:
-                # Re-stream using the engine for real token delivery.
-                # Build the same messages the agent used for its final turn.
-                try:
-                    from openjarvis.core.types import Message as MsgType
-                    from openjarvis.core.types import Role as RoleType
-
-                    replay_messages = []
-                    for m in self._request.messages:
-                        role = (
-                            RoleType(m.role)
-                            if m.role in {r.value for r in RoleType}
-                            else RoleType.USER
-                        )
-                        replay_messages.append(
-                            MsgType(
-                                role=role,
-                                content=m.content or "",
-                                name=m.name,
-                                tool_call_id=m.tool_call_id,
-                            )
-                        )
-
-                    async for sc in engine.stream_full(
-                        replay_messages,
-                        model=self._model,
-                    ):
-                        if sc.content:
-                            chunk = ChatCompletionChunk(
-                                id=self._chunk_id,
-                                model=self._model,
-                                choices=[
-                                    StreamChoice(
-                                        delta=DeltaMessage(content=sc.content),
-                                    )
-                                ],
-                            )
-                            yield f"data: {chunk.model_dump_json()}\n\n"
-                    used_real_streaming = True
-                except Exception as stream_exc:
-                    import logging as _logging
-
-                    _logger = _logging.getLogger("openjarvis.server")
-                    _logger.warning(
-                        "Real streaming failed, falling back to word replay: %s",
-                        stream_exc,
-                    )
-
-            # Fallback: word-by-word replay if real streaming was not used
-            if not used_real_streaming and content:
+            if content:
                 words = content.split(" ")
                 for i, word in enumerate(words):
                     token = word if i == 0 else " " + word

tests/server/test_routes.py

@@ -847,6 +847,47 @@ class TestIdentityPromptInjection:
         assert len(system_msgs) == 1
         assert system_msgs[0].content == "Be terse."
 
+    def test_stream_uses_grounded_agent_result_without_replay(self):
+        """Regression for #734: web streaming emits the agent's final answer."""
+        from openjarvis.core.events import EventBus
+
+        captured: list = []
+        engine = _make_capturing_engine(captured)
+        agent = _make_agent(content="My name is Jarvis Prime.")
+        agent._tools = [object()]
+        agent._engine = engine
+        client = TestClient(
+            create_app(
+                engine,
+                "test-model",
+                agent=agent,
+                bus=EventBus(),
+                config=_identity_config(),
+            )
+        )
+
+        resp = client.post(
+            "/v1/chat/completions",
+            json={
+                "model": "test-model",
+                "messages": [{"role": "user", "content": "who are you?"}],
+                "stream": True,
+            },
+        )
+
+        assert resp.status_code == 200
+        streamed_content = ""
+        for line in resp.text.splitlines():
+            if not line.startswith("data: {"):
+                continue
+            payload = json.loads(line.removeprefix("data: "))
+            choices = payload.get("choices", [])
+            if choices and choices[0]["delta"].get("content"):
+                streamed_content += choices[0]["delta"]["content"]
+        assert streamed_content == "My name is Jarvis Prime."
+        assert captured == []
+        agent.run.assert_called_once()
+
     def test_direct_injects_identity_when_absent(self):
         captured: list = []
         engine = _make_capturing_engine(captured)

tests/server/test_stream_bridge.py

@@ -1,6 +1,68 @@
-import json
+"""Regression tests for streaming completed agent responses."""
 
-from openjarvis.server.stream_bridge import AgentStreamBridge
+from __future__ import annotations
+
+import asyncio
+import json
+from unittest.mock import MagicMock
+
+import pytest
+
+pytest.importorskip("fastapi")
+
+from openjarvis.agents._stubs import AgentResult  # noqa: E402
+from openjarvis.core.events import EventBus  # noqa: E402
+from openjarvis.core.types import ToolResult  # noqa: E402
+from openjarvis.server.models import ChatCompletionRequest  # noqa: E402
+from openjarvis.server.stream_bridge import AgentStreamBridge  # noqa: E402
+
+
+def _streamed_content(events: list[str]) -> str:
+    """Join assistant content from OpenAI-compatible data chunks."""
+    content = []
+    for event in events:
+        if not event.startswith("data: {"):
+            continue
+        payload = json.loads(event.removeprefix("data: ").strip())
+        choices = payload.get("choices")
+        if choices and choices[0]["delta"].get("content"):
+            content.append(choices[0]["delta"]["content"])
+    return "".join(content)
+
+
+def test_stream_replays_grounded_agent_result_without_second_inference():
+    grounded_content = "My name is Jarvis. The tool reports 72 degrees."
+    agent = MagicMock()
+    agent._model = "configured-model"
+    agent.run.return_value = AgentResult(
+        content=grounded_content,
+        tool_results=[
+            ToolResult(tool_name="weather", content="72 degrees", success=True)
+        ],
+        metadata={"prompt_tokens": 10, "completion_tokens": 12, "total_tokens": 22},
+    )
+
+    async def ungrounded_replay(*args, **kwargs):
+        raise AssertionError("stream_full must not run after agent.run")
+        yield  # pragma: no cover
+
+    agent._engine.stream_full = ungrounded_replay
+    request = ChatCompletionRequest(
+        model="requested-model",
+        messages=[{"role": "user", "content": "Who are you, and what's outside?"}],
+        stream=True,
+    )
+    bridge = AgentStreamBridge(agent, EventBus(), request.model, request)
+
+    async def collect_events() -> list[str]:
+        return [event async for event in bridge.stream()]
+
+    events = asyncio.run(collect_events())
+
+    assert _streamed_content(events) == grounded_content
+    assert any(event.startswith("event: tool_results\n") for event in events)
+    agent.run.assert_called_once()
+    assert agent._model == "configured-model"
 
 
 def test_tool_call_start_serializes_arguments_for_sse_without_mutating_event():