import asyncio import json from dataclasses import dataclass, field from datetime import UTC, datetime from typing import Any, cast import httpx import pytest from beaver_gateway.app import AgentRegistry, McpRegistry from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions from beaver_gateway.conversations.envelope import RecallContext from beaver_gateway.events.stream import ( build_content_block_stop, build_message_delta, build_message_start, build_message_stop, build_text_block_start, build_text_delta, ) from beaver_gateway.conversations.texts import UserSaid from beaver_gateway.frontends.base import GatewayRuntime from beaver_gateway.security.auth import TokenStore, scopes_with from beaver_agent.memory.recall import Recall, ReplyLog from beaver_agent.voice import AGENT, KEY, TITLE, VoiceFrontend, rotate from beaver_agent.voice.texts import ROTATED VOICE = {"Authorization": "Bearer voice-key"} OPS = {"Authorization": "Bearer ops-key"} ASK = { "room": "andreii", "device": "esp32-1", "request_id": "3fd0d6ed-a81b-489c-a1d2-3b97ba84892b", "content": "ало клод как дела", } PLAIN = {**ASK, "stream": False} """То же, но одним JSON: без поля фронтенд стримит.""" @dataclass class Conv: external_id: str = "conv-1" agent_name: str = AGENT kind: str = "branch" title: str = TITLE status: str = "open" last_user_activity_at: datetime | None = None flags: dict[str, Any] = field(default_factory=dict) @dataclass class Binding: frontend: str external_id: str visible: bool = True class FakeConversations: """Ровно те методы сервиса, которыми пользуется голосовое окно.""" def __init__(self, bound: Conv | None = None) -> None: self.conv = bound self.bindings_of: dict[str, list[Binding]] = {} self.said: list[tuple[str, str]] = [] self.prompts: list[str] = [] self.bound: list[tuple[str, str, str]] = [] self.closed: list[str] = [] self.spawned: list[dict[str, Any]] = [] self.reply = "живу, слушаю" self.spawn_hook = None async def find_bound(self, *, frontend: str, external_id: str) -> Conv | None: if self.conv is None or frontend != "voice" or external_id != KEY: return None return self.conv async def spawn(self, **kwargs: Any) -> Conv: self.spawned.append(kwargs) conv = Conv(external_id=f"conv-{len(self.spawned) + 1}") self.conv = conv binding = kwargs.get("binding") if binding: self.bindings_of[conv.external_id] = [Binding(*binding)] return conv async def bind(self, conv: Conv, *, frontend: str, external_id: str) -> None: self.bound.append((conv.external_id, frontend, external_id)) async def bindings(self, conv: Conv) -> list[Binding]: return self.bindings_of.get(conv.external_id, []) async def close(self, conv: Conv) -> None: conv.status = "closed" self.closed.append(conv.external_id) async def say(self, conv: Conv, text: str) -> None: self.said.append((conv.external_id, text)) async def pending_seed(self, conv: Conv) -> str | None: return conv.flags.pop("seed", None) async def run_text_turn(self, conv: Conv, prompt: str, **_: Any) -> tuple[str, Any]: self.prompts.append(prompt) return self.reply, None async def turn(self, conv: Conv, *, messages: list[Any], **_: Any) -> Any: self.prompts.append(messages[0]["content"]) yield build_message_start(message_id="m", model="haiku") yield build_text_block_start(0) for chunk in ("живу, ", "слушаю"): yield build_text_delta(0, chunk) yield build_content_block_stop(0) yield build_message_delta(stop_reason="end_turn") yield build_message_stop() class Bus: def __init__(self) -> None: self.events: list[dict[str, Any]] = [] def publish(self, type_: str, **data: Any) -> dict[str, Any]: event = {"type": type_, **data} self.events.append(event) return event def stand(conversations: FakeConversations) -> tuple[VoiceFrontend, GatewayRuntime]: agent = ClaudeAgent( name=AGENT, model="claude-haiku-4-5-20251001", cwd=".", system_prompt="голос", kinds=("branch",), options=ClaudeOptions(tools=()), ) frontend = VoiceFrontend(agent=AGENT) runtime = GatewayRuntime( agents=AgentRegistry([agent]), mcps=McpRegistry([]), backends={}, token_store=TokenStore( bootstrap={"voice": "voice-key", "ops": "ops-key"}, bootstrap_scopes={"voice": "voice", "ops": "api"}, ), db=cast("Any", None), frontends=(frontend,), conversations=conversations, bus=Bus(), scopes=scopes_with(["voice"]), ) frontend.configure(runtime) return frontend, runtime def client(frontend: VoiceFrontend) -> httpx.AsyncClient: return httpx.AsyncClient( transport=httpx.ASGITransport(app=cast("Any", frontend.app())), base_url="http://t", ) async def test_the_payload_is_answered_and_its_fields_come_back() -> None: conversations = FakeConversations() frontend, _ = stand(conversations) async with client(frontend) as http: response = await http.post("/", json=PLAIN, headers=VOICE) assert response.status_code == 200 assert response.json() == { "request_id": ASK["request_id"], "room": "andreii", "device": "esp32-1", "reply": "живу, слушаю", "conversation": "conv-2", } assert conversations.prompts == ["[andreii · esp32-1] ало клод как дела"] async def test_the_first_ask_opens_the_branch_and_the_next_one_keeps_it() -> None: conversations = FakeConversations() frontend, _ = stand(conversations) async with client(frontend) as http: first = (await http.post("/", json=PLAIN, headers=VOICE)).json() again = ( await http.post("/", json={**PLAIN, "content": "а время?"}, headers=VOICE) ).json() assert first["conversation"] == again["conversation"] assert len(conversations.spawned) == 1 spawn = conversations.spawned[0] assert spawn["kind"] == "branch" assert spawn["agent"] == AGENT assert spawn["title"] == TITLE assert spawn["binding"] is None assert conversations.bound == [(first["conversation"], "voice", KEY)] async def test_the_question_lands_in_the_branch_first() -> None: conversations = FakeConversations() frontend, _ = stand(conversations) async with client(frontend) as http: await http.post("/", json=PLAIN, headers=VOICE) assert conversations.said == [("conv-2", "🎤 andreii · esp32-1\nало клод как дела")] async def test_two_rooms_at_once_open_one_branch() -> None: conversations = FakeConversations() frontend, _ = stand(conversations) async with client(frontend) as http: await asyncio.gather( http.post("/", json=PLAIN, headers=VOICE), http.post("/", json={**PLAIN, "room": "кухня"}, headers=VOICE), ) assert len(conversations.spawned) == 1 async def test_only_a_voice_key_opens_the_route() -> None: frontend, _ = stand(FakeConversations()) async with client(frontend) as http: assert (await http.post("/", json=ASK)).status_code == 401 assert (await http.post("/", json=ASK, headers=OPS)).status_code == 403 async def test_a_body_without_the_words_is_a_400() -> None: frontend, _ = stand(FakeConversations()) async with client(frontend) as http: for missing in ("room", "device", "content"): body = {k: v for k, v in ASK.items() if k != missing} response = await http.post("/", json=body, headers=VOICE) assert response.status_code == 400 assert missing in response.json()["error"] async def test_the_night_keeps_the_topic_and_starts_the_branch_over() -> None: old = Conv(last_user_activity_at=datetime.now(UTC)) conversations = FakeConversations(bound=old) conversations.bindings_of["conv-1"] = [Binding("telegram", "42")] new = await rotate(cast("Any", conversations)) assert new is not None assert conversations.spawned[0]["binding"] == ("telegram", "42") assert conversations.bound == [(new.external_id, "voice", KEY)] assert conversations.closed == ["conv-1"] assert conversations.said == [(new.external_id, ROTATED)] async def test_a_branch_nobody_spoke_in_is_left_alone() -> None: conversations = FakeConversations(bound=Conv()) assert await rotate(cast("Any", conversations)) is None assert conversations.spawned == [] def test_the_voice_agent_gets_no_recall_and_no_reply_log(tmp_path) -> None: recall = Recall( vault=tmp_path, people_dir=tmp_path / "люди", notes_dir=tmp_path / "записки", diary_dir=tmp_path / "дни", boards_dir=tmp_path / "доски", mute=frozenset({AGENT}), ) now = datetime.now(UTC) assert ( recall.block(RecallContext(text="Прохор", kind="branch", now=now, agent=AGENT)) is None ) log = ReplyLog(tmp_path / "реплики", mute=frozenset({AGENT})) log.write( UserSaid( conversation_id="c", kind="branch", title=TITLE, text="что там у Прохор", at=now, agent=AGENT, ) ) assert not (tmp_path / "реплики").exists() async def test_the_answer_streams_when_the_body_does_not_say_otherwise() -> None: conversations = FakeConversations() frontend, _ = stand(conversations) frames: list[tuple[str, dict[str, Any]]] = [] async with ( client(frontend) as http, http.stream("POST", "/", json=ASK, headers=VOICE) as sse, ): assert sse.status_code == 200 assert sse.headers["content-type"].startswith("text/event-stream") buffer = "".join([chunk async for chunk in sse.aiter_text()]) for block in buffer.split("\n\n"): lines = [line for line in block.splitlines() if line] if len(lines) == 2 and lines[0].startswith("event: "): frames.append((lines[0][7:], json.loads(lines[1][6:]))) assert [name for name, _ in frames] == ["delta", "delta", "done"] assert [f["text"] for _, f in frames[:2]] == ["живу, ", "слушаю"] assert frames[-1][1]["reply"] == "живу, слушаю" assert frames[-1][1]["request_id"] == ASK["request_id"] async def test_the_flag_is_read_as_a_word_not_as_a_truthy_string() -> None: frontend, _ = stand(FakeConversations()) cases = ((True, True), ("true", True), (False, False), ("false", False), (0, False)) async with client(frontend) as http: for value, streams in cases: response = await http.post( "/", json={**ASK, "stream": value}, headers=VOICE ) kind = response.headers["content-type"] assert kind.startswith("text/event-stream") is streams, (value, kind)