voiceagent/tests/test_deep_agent_stream.py

import asyncio
from contextlib import aclosing
from types import SimpleNamespace

import pytest

from voice_assistant.agent.deep_agent import DeepAgentsAdapter


def make_adapter(events, timeout=0.05):
    adapter = object.__new__(DeepAgentsAdapter)
    adapter._closed = False
    adapter._stream_event_timeout = timeout
    adapter._graph = SimpleNamespace(astream=lambda *args, **kwargs: events)
    return adapter


def event(content, node="model", tool_calls=None):
    return SimpleNamespace(content=content, tool_call_chunks=tool_calls or []), {"langgraph_node": node}


@pytest.mark.parametrize(
    ("content", "expected"),
    [
        ("Olá ", ["Olá "]),
        ([{"type": "text", "text": "Olá "}, {"type": "text", "text": "Lara."}], ["Olá ", "Lara."]),
        (["Olá", " ", {"type": "text", "text": "Lara."}], ["Olá", " ", "Lara."]),
        ([{"type": "reasoning", "text": "Interno"}, {"type": "image", "text": "Ignorar"}, {"type": "text", "text": "Resposta"}], ["Resposta"]),
        ([{"type": "text"}, {"type": "text", "text": None}, {"type": "text", "text": ""}], []),
    ],
)
def test_stream_extracts_only_text_blocks(content, expected):
    async def scenario():
        async def source():
            yield event(content)
        assert [part async for part in make_adapter(source()).stream("session", 1, "Olá")] == expected
    asyncio.run(scenario())


def test_stream_filters_non_model_events_and_tool_calls():
    async def scenario():
        async def source():
            yield event("Resultado interno", node="tools")
            yield event("Chamada interna", tool_calls=[{"id": "call-1"}])
            yield event("Resposta final")
        assert [part async for part in make_adapter(source()).stream("session", 1, "Olá")] == ["Resposta final"]
    asyncio.run(scenario())


@pytest.mark.parametrize("finish", ["normal", "early_close", "cancel", "error", "timeout"])
def test_stream_closes_owned_generator(finish):
    async def scenario():
        closed = asyncio.Event()
        waiting = asyncio.Event()

        async def source():
            try:
                yield event("Olá")
                if finish == "error":
                    raise ValueError("provider failure")
                if finish in ("cancel", "timeout"):
                    waiting.set()
                    await asyncio.Event().wait()
            finally:
                closed.set()

        stream = make_adapter(source(), timeout=0.1).stream("session", 1, "Olá")
        assert await anext(stream) == "Olá"
        if finish == "normal":
            with pytest.raises(StopAsyncIteration):
                await anext(stream)
        elif finish == "early_close":
            await stream.aclose()
        elif finish == "cancel":
            task = asyncio.create_task(anext(stream))
            await asyncio.wait_for(waiting.wait(), 1)
            task.cancel()
            with pytest.raises(asyncio.CancelledError):
                await task
        elif finish == "error":
            with pytest.raises(ValueError, match="provider failure"):
                await anext(stream)
        else:
            with pytest.raises(TimeoutError, match="next stream event"):
                await anext(stream)
        assert closed.is_set()
    asyncio.run(scenario())


def test_timeout_does_not_cancel_consumer_between_fragments():
    async def scenario():
        async def source():
            yield event("Primeiro ")
            yield event("segundo")
        async with aclosing(make_adapter(source(), timeout=0.01).stream("session", 1, "Olá")) as stream:
            assert await anext(stream) == "Primeiro "
            await asyncio.sleep(0.04)
            assert await anext(stream) == "segundo"
    asyncio.run(scenario())


def test_timeout_is_renewed_for_each_next_event():
    async def scenario():
        async def source():
            for token in ("a", "b", "c", "d"):
                await asyncio.sleep(0.02)
                yield event(token)
        adapter = make_adapter(source(), timeout=0.06)
        assert "".join([part async for part in adapter.stream("session", 1, "Olá")]) == "abcd"
    asyncio.run(scenario())


@pytest.mark.parametrize("timeout", [0, -1, float("inf"), float("nan"), True, "60", None])
def test_invalid_stream_timeout_is_rejected(timeout):
    with pytest.raises(ValueError, match="finite and positive"):
        DeepAgentsAdapter(graph=SimpleNamespace(), timeout_seconds=timeout)


def test_adapter_streams_real_deepagents_graph_with_fake_model():
    from deepagents import create_deep_agent
    from langchain_core.language_models import FakeListChatModel
    from langgraph.checkpoint.memory import MemorySaver

    class Model(FakeListChatModel):
        def bind_tools(self, tools, *, tool_choice=None, **kwargs):
            return self

    async def scenario():
        adapter = object.__new__(DeepAgentsAdapter)
        adapter._closed = False
        adapter._stream_event_timeout = 2.0
        adapter._graph = create_deep_agent(
            model=Model(responses=["Olá Lara."]), tools=[], checkpointer=MemorySaver(),
        )
        response = await adapter.process("session", 1, "Diga olá")
        assert response == "Olá Lara."
    asyncio.run(scenario())


def test_process_rejects_empty_text_response_and_closes_stream():
    async def scenario():
        closed = asyncio.Event()

        async def source():
            try:
                yield event([{"type": "reasoning", "text": "Interno"}])
            finally:
                closed.set()

        with pytest.raises(RuntimeError, match="empty streaming response"):
            await make_adapter(source()).process("session", 1, "Olá")
        assert closed.is_set()
    asyncio.run(scenario())


def test_process_propagates_provider_failure_and_closes_stream():
    async def scenario():
        closed = asyncio.Event()

        async def source():
            try:
                yield event("Texto parcial")
                raise ValueError("provider failure")
            finally:
                closed.set()

        with pytest.raises(ValueError, match="provider failure"):
            await make_adapter(source()).process("session", 1, "Olá")
        assert closed.is_set()
    asyncio.run(scenario())