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

docs/user-guide/evaluations.md

@@ -31,6 +31,16 @@ uv sync --extra dev --extra eval-wandb     # Weights & Biases run tracking
 uv sync --extra dev --extra eval-sheets    # Google Sheets results export
 ```
 
+TauBench additionally requires Python 3.12 or newer and the upstream `tau2`
+package. Install the pinned revision explicitly before running that benchmark:
+
+```bash
+uv pip install "tau2 @ git+https://github.com/sierra-research/tau2-bench.git@fc0055dc4e0a316c3f83133267fbd6faaa770992"
+```
+
+OpenJarvis does not install third-party packages automatically when an
+evaluation is imported or run.
+
 !!! note "Python version requirement"
     Python 3.10 requires the `tomli` package for TOML config parsing. `openjarvis` declares it as a conditional dependency, so it is installed automatically.
 

src/openjarvis/evals/datasets/taubench.py

@@ -8,13 +8,12 @@ Reference: https://github.com/sierra-research/tau2-bench
 
 from __future__ import annotations
 
+import json
 import logging
 import os
-import subprocess
-import sys
+from importlib import metadata
 from typing import Iterable, List, Optional
 
-from openjarvis.core.paths import get_cache_dir
 from openjarvis.evals.core.dataset import DatasetProvider
 from openjarvis.evals.core.splits import apply_split
 from openjarvis.evals.core.types import EvalRecord
@@ -22,48 +21,50 @@ from openjarvis.evals.core.types import EvalRecord
 LOGGER = logging.getLogger(__name__)
 
 TAU2_REPO = "https://github.com/sierra-research/tau2-bench.git"
-CACHE_DIR = get_cache_dir() / "tau2-bench"
+# v1.0.1. Keep the full commit SHA here (rather than a movable tag) so every
+# TauBench setup uses the same third-party code.
+TAU2_REVISION = "fc0055dc4e0a316c3f83133267fbd6faaa770992"
+TAU2_INSTALL_SPEC = f"tau2 @ git+{TAU2_REPO}@{TAU2_REVISION}"
 
 DOMAINS = ("airline", "retail", "telecom")
 
 
 def _ensure_tau2() -> None:
-    """Ensure tau2 package is importable; install from cache if needed."""
+    """Ensure the explicitly installed, pinned tau2 package is importable."""
+    try:
+        distribution = metadata.distribution("tau2")
+    except metadata.PackageNotFoundError as exc:
+        raise ImportError(
+            "TauBench requires tau2, which OpenJarvis does not install at "
+            "runtime. Install the pinned dependency explicitly (Python >=3.12): "
+            f'uv pip install "{TAU2_INSTALL_SPEC}"'
+        ) from exc
+
+    try:
+        direct_url_text = distribution.read_text("direct_url.json")
+        direct_url = json.loads(direct_url_text or "")
+        vcs_info = direct_url.get("vcs_info", {})
+        installed_repo = direct_url.get("url")
+        installed_revision = vcs_info.get("commit_id")
+    except (json.JSONDecodeError, AttributeError):
+        installed_repo = None
+        installed_revision = None
+
+    if installed_repo != TAU2_REPO or installed_revision != TAU2_REVISION:
+        raise ImportError(
+            "The installed tau2 package does not match OpenJarvis's pinned "
+            "source revision. Reinstall it explicitly (Python >=3.12): "
+            f'uv pip install --force-reinstall "{TAU2_INSTALL_SPEC}"'
+        )
+
     try:
         import tau2  # noqa: F401
-    except ImportError:
-        # Clone and install from source
-        if not CACHE_DIR.exists():
-            LOGGER.info("Cloning tau2-bench from %s ...", TAU2_REPO)
-            CACHE_DIR.parent.mkdir(parents=True, exist_ok=True)
-            subprocess.run(
-                ["git", "clone", "--depth", "1", TAU2_REPO, str(CACHE_DIR)],
-                check=True,
-                capture_output=True,
-            )
-        LOGGER.info("Installing tau2-bench ...")
-        # Try `python -m pip` first; fall back to `uv pip` for uv-managed venvs
-        # which don't ship pip by default.
-        try:
-            subprocess.run(
-                [sys.executable, "-m", "pip", "install", "-e", str(CACHE_DIR)],
-                check=True,
-                capture_output=True,
-            )
-        except (subprocess.CalledProcessError, FileNotFoundError):
-            subprocess.run(
-                [
-                    "uv",
-                    "pip",
-                    "install",
-                    "--python",
-                    sys.executable,
-                    "-e",
-                    str(CACHE_DIR),
-                ],
-                check=True,
-                capture_output=True,
-            )
+    except ImportError as exc:
+        raise ImportError(
+            "The pinned tau2 package is installed but cannot be imported. "
+            "Reinstall it explicitly (Python >=3.12): "
+            f'uv pip install --force-reinstall "{TAU2_INSTALL_SPEC}"'
+        ) from exc
 
 
 class TauBenchDataset(DatasetProvider):

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/evals/datasets/test_taubench.py

@@ -0,0 +1,98 @@
+"""Tests for the TauBench optional dependency boundary."""
+
+from __future__ import annotations
+
+import builtins
+import sys
+from types import ModuleType
+from unittest.mock import Mock
+
+import pytest
+
+from openjarvis.evals.datasets import taubench
+
+
+def _mock_direct_url(monkeypatch, direct_url):
+    distribution = Mock()
+    distribution.read_text.return_value = direct_url
+    monkeypatch.setattr(
+        taubench.metadata, "distribution", Mock(return_value=distribution)
+    )
+
+
+def test_ensure_tau2_accepts_the_pinned_source_revision(monkeypatch):
+    monkeypatch.setitem(sys.modules, "tau2", ModuleType("tau2"))
+    _mock_direct_url(
+        monkeypatch,
+        (
+            '{"url": "https://github.com/sierra-research/tau2-bench.git", '
+            '"vcs_info": {"vcs": "git", '
+            f'"commit_id": "{taubench.TAU2_REVISION}"}}}}'
+        ),
+    )
+
+    taubench._ensure_tau2()
+
+
+def test_ensure_tau2_requires_explicit_pinned_install(monkeypatch):
+    monkeypatch.setitem(sys.modules, "tau2", None)
+    monkeypatch.setattr(
+        taubench.metadata,
+        "distribution",
+        Mock(side_effect=taubench.metadata.PackageNotFoundError),
+    )
+
+    with pytest.raises(ImportError) as exc_info:
+        taubench._ensure_tau2()
+
+    message = str(exc_info.value)
+    assert "does not install at runtime" in message
+    assert taubench.TAU2_REVISION in message
+    assert "uv pip install" in message
+
+
+@pytest.mark.parametrize(
+    "direct_url",
+    [
+        # Editable install left behind by the previous runtime installer.
+        '{"url": "file:///home/user/.openjarvis/cache/tau2-bench", '
+        '"dir_info": {"editable": true}}',
+        # A git install from an arbitrary upstream revision.
+        '{"url": "https://github.com/sierra-research/tau2-bench.git", '
+        '"vcs_info": {"vcs": "git", "commit_id": "deadbeef"}}',
+        # Registry installs do not carry PEP 610 direct-origin metadata.
+        None,
+    ],
+)
+def test_ensure_tau2_rejects_unpinned_install(monkeypatch, direct_url):
+    _mock_direct_url(monkeypatch, direct_url)
+    original_import = builtins.__import__
+
+    def guarded_import(name, *args, **kwargs):
+        if name == "tau2":
+            raise AssertionError("unverified tau2 package was imported")
+        return original_import(name, *args, **kwargs)
+
+    monkeypatch.setattr(builtins, "__import__", guarded_import)
+
+    with pytest.raises(ImportError) as exc_info:
+        taubench._ensure_tau2()
+
+    message = str(exc_info.value)
+    assert "does not match" in message
+    assert taubench.TAU2_REVISION in message
+    assert "--force-reinstall" in message
+
+
+def test_verify_requirements_reports_install_instruction(monkeypatch):
+    monkeypatch.setitem(sys.modules, "tau2", None)
+    monkeypatch.setattr(
+        taubench.metadata,
+        "distribution",
+        Mock(side_effect=taubench.metadata.PackageNotFoundError),
+    )
+
+    issues = taubench.TauBenchDataset().verify_requirements()
+
+    assert len(issues) == 1
+    assert taubench.TAU2_REVISION in issues[0]

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():