309 lines
11 KiB
Python
309 lines
11 KiB
Python
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)
|