AetherTable/python/tests/test_empty_stream_honesty.py

"""Empty upstream answers must never masquerade as a completed agent turn.

Contract under test (audit note): when an SSE body carries only
``data: [DONE]`` (or otherwise reassembles to content=None / no tool_calls),
``complete_with_tools`` previously returned that blank message and
``EngineToolAgent.run_turn`` reported ``{"status": "COMPLETED",
"narration": ""}`` — masking an upstream failure as a successful empty turn.

New contract:
- ``complete_with_tools`` raises ``RuntimeError("LLM_UPSTREAM_EMPTY: ...")``
  when the response carries neither assistant content nor tool_calls.
- ``EngineToolAgent.run_turn`` surfaces that as an honest non-COMPLETED
  status ("LLM_UPSTREAM_EMPTY") — never COMPLETED with empty narration.
- Well-formed responses (content or tool_calls) are unchanged.
- No unit test touches the network: httpx.AsyncClient is monkeypatched.
"""

import asyncio
import json

from vtt_orchestrator.agents.tool_agent import EngineToolAgent
from vtt_orchestrator.routing import llm_client as llm_client_module
from vtt_orchestrator.routing.llm_client import LLMConfig, LLMStreamingGateway


class _FakeResponse:
    def __init__(self, text: str, content_type: str = "application/json",
                 status_code: int = 200, json_body: dict = None):
        self._text = text
        self._json_body = json_body
        self.status_code = status_code
        self.headers = {"content-type": content_type}

    @property
    def text(self) -> str:
        return self._text

    def raise_for_status(self) -> None:
        if self.status_code >= 400:
            raise RuntimeError(f"http {self.status_code}")

    def json(self) -> dict:
        if self._json_body is not None:
            return self._json_body
        raise ValueError(f"Expecting value ... body starts {self._text[:40]!r}")


def _install_responses(monkeypatch, responses) -> type:
    class _PatchedClient:
        _responses = list(responses)

        def __init__(self, *args, **kwargs):
            pass

        async def __aenter__(self):
            return self

        async def __aexit__(self, *exc):
            return False

        async def post(self, url, headers=None, json=None, **kwargs):
            return type(self)._responses.pop(0)

    monkeypatch.setattr(llm_client_module.httpx, "AsyncClient", _PatchedClient)
    return _PatchedClient


def _live_gateway(monkeypatch, tmp_path) -> LLMStreamingGateway:
    monkeypatch.setenv("LLM_API", "http://fake-llm.test/v1")
    monkeypatch.setenv("LLM_KEY", "test-key")
    monkeypatch.setenv("LLM_MODEL", "test-model")
    for var in ("OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GEMINI_API_KEY", "OLLAMA_BASE_URL"):
        monkeypatch.delenv(var, raising=False)
    monkeypatch.setattr(llm_client_module, "LLM_LOG_PATH", tmp_path / "llm.jsonl")
    return LLMStreamingGateway(LLMConfig())


DONE_ONLY_SSE = "data: [DONE]\n\n"


class TestCompleteWithToolsEmptyStream:
    def test_done_only_sse_raises_not_blank_message(self, monkeypatch, tmp_path):
        gw = _live_gateway(monkeypatch, tmp_path)
        _install_responses(
            monkeypatch,
            [_FakeResponse(DONE_ONLY_SSE, content_type="text/event-stream")],
        )

        try:
            message = asyncio.run(
                gw.complete_with_tools([{"role": "user", "content": "hi"}], tools=[])
            )
        except RuntimeError as exc:
            assert "LLM_UPSTREAM_EMPTY" in str(exc)
        else:
            raise AssertionError(
                f"empty SSE stream must not yield a usable message: {message!r}"
            )

    def test_json_body_with_no_content_and_no_tools_also_raises(self, monkeypatch, tmp_path):
        gw = _live_gateway(monkeypatch, tmp_path)
        body = {"choices": [{"index": 0, "message": {"role": "assistant", "content": None},
                             "finish_reason": "stop"}]}
        _install_responses(
            monkeypatch,
            [_FakeResponse(json.dumps(body), content_type="application/json",
                           json_body=body)],
        )

        try:
            asyncio.run(gw.complete_with_tools([], tools=[]))
        except RuntimeError as exc:
            assert "LLM_UPSTREAM_EMPTY" in str(exc)
        else:
            raise AssertionError("blank JSON assistant turn must fail honestly")

    def test_whitespace_only_content_counts_as_empty(self, monkeypatch, tmp_path):
        gw = _live_gateway(monkeypatch, tmp_path)
        _install_responses(
            monkeypatch,
            [_FakeResponse(
                "data: " + json.dumps(
                    {"choices": [{"index": 0, "delta": {"role": "assistant", "content": "   "}}]}
                ) + "\n\ndata: [DONE]\n\n",
                content_type="text/event-stream",
            )],
        )

        try:
            asyncio.run(gw.complete_with_tools([], tools=[]))
        except RuntimeError as exc:
            assert "LLM_UPSTREAM_EMPTY" in str(exc)
        else:
            raise AssertionError("whitespace-only narration must not read as success")

    def test_failure_is_logged_to_jsonl(self, monkeypatch, tmp_path):
        log = tmp_path / "llm.jsonl"
        monkeypatch.setattr(llm_client_module, "LLM_LOG_PATH", log)
        gw = _live_gateway(monkeypatch, tmp_path)
        _install_responses(
            monkeypatch,
            [_FakeResponse(DONE_ONLY_SSE, content_type="text/event-stream")],
        )

        try:
            asyncio.run(gw.complete_with_tools([], tools=[]))
        except RuntimeError:
            pass
        records = [json.loads(line) for line in log.read_text().splitlines() if line.strip()]
        assert records and "LLM_UPSTREAM_EMPTY" in records[-1].get("error", "")

    def test_real_content_still_returns_normally(self, monkeypatch, tmp_path):
        gw = _live_gateway(monkeypatch, tmp_path)
        _install_responses(
            monkeypatch,
            [_FakeResponse(
                "data: " + json.dumps(
                    {"choices": [{"index": 0, "delta": {"content": "The goblin falls."}}]}
                ) + "\n\ndata: [DONE]\n\n",
                content_type="text/event-stream",
            )],
        )

        message = asyncio.run(gw.complete_with_tools([], tools=[]))
        assert message["content"] == "The goblin falls."


class TestAgentTurnEmptyStreamHonesty:
    def test_run_turn_never_reports_completed_on_empty_stream(self, monkeypatch, tmp_path):
        gw = _live_gateway(monkeypatch, tmp_path)
        _install_responses(
            monkeypatch,
            [_FakeResponse(DONE_ONLY_SSE, content_type="text/event-stream")],
        )
        agent = EngineToolAgent(gw)

        result = asyncio.run(agent.run_turn("I attack the goblin", session_id="s1"))

        assert result.get("status") != "COMPLETED"
        assert result.get("status") == "LLM_UPSTREAM_EMPTY"
        assert "LLM_UPSTREAM_EMPTY" in result.get("detail", "")
        # The masking failure mode this regression pins down:
        assert not (result.get("status") == "COMPLETED" and result.get("narration") == "")

    def test_completed_turn_with_narration_unchanged(self, monkeypatch, tmp_path):
        gw = _live_gateway(monkeypatch, tmp_path)
        _install_responses(
            monkeypatch,
            [_FakeResponse(
                "data: " + json.dumps(
                    {"choices": [{"index": 0, "delta": {"content": "You strike true."}}]}
                ) + "\n\ndata: [DONE]\n\n",
                content_type="text/event-stream",
            )],
        )
        agent = EngineToolAgent(gw)

        result = asyncio.run(agent.run_turn("I attack", session_id="s1"))

        assert result["status"] == "COMPLETED"
        assert result["narration"] == "You strike true."

    def test_no_key_still_reports_unavailable(self, monkeypatch, tmp_path):
        monkeypatch.delenv("LLM_KEY", raising=False)
        monkeypatch.delenv("OPENAI_API_KEY", raising=False)
        monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
        monkeypatch.delenv("GEMINI_API_KEY", raising=False)
        monkeypatch.setattr(llm_client_module, "LLM_LOG_PATH", tmp_path / "llm.jsonl")
        agent = EngineToolAgent(LLMStreamingGateway(LLMConfig()))

        result = asyncio.run(agent.run_turn("I attack", session_id="s1"))

        assert result["status"] == "UNAVAILABLE"