docs(readme): drop the voice points section

README.md

@@ -1,8 +1,10 @@
 # beaver-agent
 
-My real beaver-gateway setup (paired with the protocol from beaver.kotikot.com).
+My real [beaver-gateway](https://git.kotikot.com/beaver/beaver-gateway) setup: the Beavering Manager, a personal agent that lives on a home server, reads an Obsidian vault, talks through Telegram and deep chats in the vault, remembers people and promises, and has hands (Komodo, Home Assistant, T3 Code threads, calendars, Firefly, Telegram history through beavergram). The narrative and the docs are on [the site](https://beaver.kotikot.com/docs); this README is the operator's part: how to run it.
 
-Uses Claude Code and Raycast subscriptions for agents, has some MCPs set up. You can use this as-is or modify to match your needs: `config.py` assembles the gateway from the `beaver_agent/` package.
+Uses Claude Code and Raycast subscriptions for agents, has some MCPs set up. You can use this as-is or modify to match your needs: `config.py` assembles the gateway from the `beaver_agent/` package. It is opinion, not core - every piece is meant to be rewritten for your own vault.
+
+Seeding memory from an existing vault (old chats into digests, diary days and thoughts into notes about people) is one run of `scripts/seed_memory.py`; `--dry-run` first.
 
 ## Requirements
 

beaver_agent/agents.py

@@ -21,6 +21,8 @@ from beaver_agent.policy import (
     vault_zones,
 )
 from beaver_agent.vault import TZ, VAULT
+from beaver_agent.voice import AGENT as VOICE
+from beaver_agent.voice import PROMPT as VOICE_PROMPT
 
 if TYPE_CHECKING:
     from beaver_gateway.agents.base import BaseAgent
@@ -106,6 +108,18 @@ def triage(name: str, model: str, effort: str) -> ClaudeAgent:
     )
 
 
+def voice(name: str, model: str) -> ClaudeAgent:
+    """Колонки по квартире: чужие люди, поэтому ни vault, ни инструментов."""
+    return ClaudeAgent(
+        name=name,
+        model=model,
+        cwd=Path("/tmp"),  # noqa: S108
+        system_prompt=VOICE_PROMPT,
+        kinds=("branch",),
+        options=ClaudeOptions(tools=(), disallowed_tools=TRIAGE_DISALLOWED),
+    )
+
+
 def raycast(name: str, model: str, effort: str) -> RaycastAgent:
     """Быстрые ответы через Raycast: веб и чтение страниц, vault через obsidian-fs."""
     return RaycastAgent(
@@ -131,6 +145,7 @@ agents: list[BaseAgent] = [
     distiller("beaver-distiller", "claude-opus-5", "medium"),
     curator("beaver-curator", "claude-opus-5", "high"),
     triage("beaver-triage", "claude-sonnet-5", "low"),
+    voice(VOICE, "claude-haiku-4-5-20251001"),
     raycast("beaver-gemini-pro-high", "google-gemini-3.1-pro", "high"),
     raycast("beaver-gemini-pro-low", "google-gemini-3.1-pro", "low"),
     raycast("beaver-gemini-flash-high", "google-gemini-3.5-flash", "high"),

beaver_agent/frontends.py

@@ -16,6 +16,8 @@ from beaver_gateway.frontends.turn_record import slugify
 
 from beaver_agent import texts
 from beaver_agent.vault import ATTACHMENTS, BEAVER, CHATS, TZ, VAULT
+from beaver_agent.voice import AGENT as VOICE
+from beaver_agent.voice import VoiceFrontend
 
 if TYPE_CHECKING:
     from pathlib import Path
@@ -62,4 +64,5 @@ frontends: list[Frontend] = [
     MarkdownFrontend(
         vault_path=CHATS, default_agent=DEEP, log_all_chats=True, chat_path=chat_path
     ),
+    VoiceFrontend(agent=VOICE),
 ]

beaver_agent/jobs/__init__.py

@@ -11,12 +11,14 @@ from beaver_agent.jobs.memory import curate, memory
 from beaver_agent.jobs.rotation import rotate
 from beaver_agent.jobs.t3code import t3code_event
 from beaver_agent.jobs.vibegram import vibegram
+from beaver_agent.jobs.voice import rotate_voice
 
 jobs = [
     Job("ротация", rotate, cron="0 * * * *"),
     Job("вайбграм", vibegram, cron="*/10 * * * *", critical=False),
     Job("закрытие", close_idle, cron="20 4 * * *", critical=False),
     Job("память", memory, cron="30 4 * * 0", critical=False),
+    Job("голосовой", rotate_voice, cron="5 4 * * *", critical=False),
     Job("куратор", curate, cron="0 2,8,12,16,20 * * *", critical=False),
     Job("deploy", deploy, webhook=True),
     Job("komodo", komodo_alert, webhook=True, dedupe=False),

beaver_agent/jobs/voice.py

@@ -0,0 +1,14 @@
+"""Раз в ночь: голосовая ветка начинается заново в том же топике."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from beaver_agent.voice import rotate
+
+if TYPE_CHECKING:
+    from beaver_gateway.jobs.scheduler import JobRun
+
+
+async def rotate_voice(run: JobRun) -> None:
+    await rotate(run.conversations)

beaver_agent/memory/recall.py

@@ -198,6 +198,8 @@ class Recall:
     tz: str = "Europe/Warsaw"
     diary_scan: int = 400
     """Сколько последних дней дневника просматривать на упоминания."""
+    mute: frozenset[str] = frozenset()
+    """Агенты, которым справку не показывают: чужие люди, чужие уши."""
     _people: People | None = field(default=None, init=False, repr=False)
     _diary_cache: dict[str, tuple[float, list[str]]] = field(
         default_factory=dict, init=False, repr=False
@@ -211,6 +213,8 @@ class Recall:
         return self._people
 
     def block(self, ctx: RecallContext) -> str | None:
+        if ctx.agent in self.mute:
+            return None
         today = ctx.now.astimezone(ZoneInfo(self.tz)).date()
         lines = [self.person_line(p) for p in mentions(ctx.text, self.people.all())]
         if ctx.kind == "master":
@@ -306,8 +310,12 @@ class ReplyLog:
     dir: Path
     tz: str = "Europe/Warsaw"
     max_chars: int = 600
+    mute: frozenset[str] = frozenset()
+    """Агенты, чьи разговоры в лог реплик не идут."""
 
     def write(self, said: UserSaid) -> None:
+        if said.agent in self.mute:
+            return
         text = said.text
         if text.startswith("[сид:"):
             # спавн ветки с текстом: строка сида едет над самим сообщением

beaver_agent/memory/watch.py

@@ -18,6 +18,7 @@ from beaver_agent.vault import (
     TZ,
     VAULT,
 )
+from beaver_agent.voice import AGENT as VOICE
 
 watch = VaultWatch(
     VAULT,
@@ -52,8 +53,11 @@ recall = Recall(
     diary_dir=DIARY,
     boards_dir=BOARDS,
     tz=TZ,
+    mute=frozenset({VOICE}),
 )
-replies = ReplyLog(REPLIES, tz=TZ)
+"""Голосовому агенту справки нет: за колонкой может стоять кто угодно."""
+
+replies = ReplyLog(REPLIES, tz=TZ, mute=frozenset({VOICE}))
 
 CURATOR_WATCH = [
     Watched("хендауты", DAYS, "????-??-??.md"),

beaver_agent/voice/__init__.py

@@ -0,0 +1,17 @@
+"""Голосовые точки по квартире: своя ветка, свой дешёвый агент, свои ключи."""
+
+from beaver_agent.voice.frontend import VoiceFrontend
+from beaver_agent.voice.texts import KEY, PROMPT, TITLE
+from beaver_agent.voice.thread import AGENT, FRONTEND, current, open_thread, rotate
+
+__all__ = [
+    "AGENT",
+    "FRONTEND",
+    "KEY",
+    "PROMPT",
+    "TITLE",
+    "VoiceFrontend",
+    "current",
+    "open_thread",
+    "rotate",
+]

beaver_agent/voice/frontend.py

@@ -0,0 +1,253 @@
+"""Окно голосовых точек: ESP32 шлёт json на /voice, отвечает дешёвый агент."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from typing import TYPE_CHECKING, Any
+from uuid import uuid4
+
+from beaver_gateway.agents.claude import ClaudeAgent
+from beaver_gateway.conversations.turns import content_of
+from beaver_gateway.frontends.accumulate import StreamAccumulator
+from beaver_gateway.frontends.base import Frontend
+from beaver_gateway.frontends.bearer import require_token
+from beaver_gateway.frontends.sse import (
+    KEEPALIVE,
+    SSE_HEADERS,
+    events_with_heartbeat,
+    sse_pack,
+)
+from beaver_gateway.security import audit
+from fastapi import FastAPI, HTTPException, Request, status
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from beaver_agent.voice import texts
+from beaver_agent.voice.thread import FRONTEND, current, open_thread
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator, Mapping
+
+    from beaver_gateway.frontends.base import GatewayRuntime
+    from beaver_gateway.storage.models import Conversation
+
+_log = logging.getLogger("beaver_agent.voice")
+
+FIELDS = ("room", "device", "content")
+
+
+class VoiceFrontend(Frontend):
+    """Одна ветка на всю квартиру, свой агент, свои ключи.
+
+    Тёрн идёт мимо очереди, а значит и мимо конверта: справка по vault в
+    этот разговор не попадает даже случайно. В телеграм-ветку вопрос
+    кладётся отдельной репликой, ответ пишется черновиком, как обычно.
+    """
+
+    name = FRONTEND
+    kinds = ("branch",)
+    path = "/voice"
+    scope = "voice"
+
+    def __init__(self, *, agent: str, timeout: float = 90.0) -> None:
+        self.agent = agent
+        self.timeout = timeout
+        """Сколько ждать ответа; дальше 504, а тёрн договаривает в телеграм."""
+        self._runtime: GatewayRuntime | None = None
+        self._app: FastAPI | None = None
+        self._lock = asyncio.Lock()
+        self._late: set[asyncio.Task[Any]] = set()
+
+    def configure(self, runtime: GatewayRuntime) -> None:
+        if runtime.conversations is None or runtime.bus is None:
+            msg = "VoiceFrontend: нужны runtime.conversations и runtime.bus"
+            raise RuntimeError(msg)
+        agent = runtime.agents.get(self.agent)
+        if not isinstance(agent, ClaudeAgent) or not agent.serves("branch"):
+            msg = f"VoiceFrontend: агент {self.agent!r} не обслуживает ветки"
+            raise RuntimeError(msg)
+        self._runtime = runtime
+        self._app = self._build_app(runtime)
+
+    def app(self) -> FastAPI | None:
+        return self._app
+
+    def _build_app(self, runtime: GatewayRuntime) -> FastAPI:
+        app = FastAPI(title="beaver-agent / голосовой")
+
+        @app.get("/healthz")
+        async def healthz() -> dict[str, str]:
+            return {"status": "ok", "frontend": self.name}
+
+        @app.post("/")
+        async def ask(request: Request) -> Any:
+            token = await require_token(request, runtime, scope=self.scope)
+            body = await _body(request)
+            room, device, content = (_field(body, name) for name in FIELDS)
+            conv = await self.thread()
+            await audit.log(
+                runtime,
+                actor=f"token:{token}",
+                kind="voice_ask",
+                agent_name=conv.agent_name,
+                conversation=conv.external_id,
+                room=room,
+                device=device,
+                request_id=body.get("request_id"),
+            )
+            _log.info("голос: %s/%s -> %s", room, device, conv.external_id)
+            await runtime.conversations.say(
+                conv, texts.ASKED.format(room=room, device=device, text=content)
+            )
+            prompt = await self._prompt(
+                conv, texts.HEARD.format(room=room, device=device, text=content)
+            )
+            if bool(body.get("stream")):
+                return StreamingResponse(
+                    self._sse(conv, prompt, body),
+                    media_type="text/event-stream",
+                    headers=SSE_HEADERS,
+                )
+            return JSONResponse(await self._answer(conv, prompt, body))
+
+        @app.exception_handler(HTTPException)
+        async def http_error(_request: Request, exc: HTTPException) -> JSONResponse:
+            return JSONResponse(
+                status_code=exc.status_code,
+                content={"error": exc.detail},
+                headers=exc.headers,
+            )
+
+        return app
+
+    async def thread(self) -> Conversation:
+        conversations = self.runtime.conversations
+        async with self._lock:
+            conv = await current(conversations)
+            if conv is not None:
+                return conv
+            return await open_thread(conversations, agent=self.agent)
+
+    @property
+    def runtime(self) -> GatewayRuntime:
+        if self._runtime is None:
+            msg = "VoiceFrontend: configure() не вызывали"
+            raise RuntimeError(msg)
+        return self._runtime
+
+    async def _prompt(self, conv: Conversation, text: str) -> str:
+        seed = await self.runtime.conversations.pending_seed(conv)
+        return f"{seed}\n\n{text}" if seed else text
+
+    async def _answer(
+        self, conv: Conversation, prompt: str, body: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        turn_id = _turn_id()
+        task = asyncio.create_task(self._run(conv, prompt, turn_id))
+        self._late.add(task)
+        task.add_done_callback(self._late.discard)
+        done, _ = await asyncio.wait({task}, timeout=self.timeout)
+        if not done:
+            _log.warning("голос: %s не уложился в %.0f c", turn_id, self.timeout)
+            raise HTTPException(
+                status.HTTP_504_GATEWAY_TIMEOUT, "агент не успел; ответ уйдёт в ветку"
+            )
+        return _body_out(body, reply=task.result(), conv=conv)
+
+    async def _run(self, conv: Conversation, prompt: str, turn_id: str) -> str:
+        reply, _capture = await self.runtime.conversations.run_text_turn(
+            conv, prompt, origin="user", turn_id=turn_id
+        )
+        self._published(conv, reply=reply, turn_id=turn_id)
+        return reply
+
+    async def _sse(
+        self, conv: Conversation, prompt: str, body: Mapping[str, Any]
+    ) -> AsyncIterator[bytes]:
+        turn_id = _turn_id()
+        acc = StreamAccumulator()
+        model = self.runtime.agents[conv.agent_name].model
+        try:
+            events = self.runtime.conversations.turn(
+                conv,
+                messages=[{"role": "user", "content": content_of(prompt, None)}],
+                origin="user",
+                turn_id=turn_id,
+            )
+            async for event in events_with_heartbeat(events):
+                if event is None:
+                    yield KEEPALIVE
+                    continue
+                acc.feed(event)
+                chunk = _text_delta(event)
+                if chunk:
+                    yield sse_pack("delta", {"text": chunk})
+        except Exception as exc:  # noqa: BLE001
+            _log.exception("голос: тёрн %s упал", turn_id)
+            yield sse_pack("error", {"error": str(exc)})
+            return
+        reply = _text_of(acc.finalize(model=model))
+        self._published(conv, reply=reply, turn_id=turn_id)
+        yield sse_pack("done", _body_out(body, reply=reply, conv=conv))
+
+    def _published(self, conv: Conversation, *, reply: str, turn_id: str) -> None:
+        self.runtime.bus.publish(
+            "reply",
+            conversation_id=conv.external_id,
+            turn_id=turn_id,
+            source=self.name,
+            text=reply,
+        )
+
+
+async def _body(request: Request) -> Mapping[str, Any]:
+    try:
+        data = await request.json()
+    except json.JSONDecodeError as exc:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, f"кривой json: {exc}") from exc
+    if not isinstance(data, dict):
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "тело должно быть объектом")
+    return data
+
+
+def _field(body: Mapping[str, Any], name: str) -> str:
+    value = body.get(name)
+    if not isinstance(value, str) or not value.strip():
+        raise HTTPException(
+            status.HTTP_400_BAD_REQUEST, f"нужно непустое строковое поле {name!r}"
+        )
+    return value.strip()
+
+
+def _body_out(
+    body: Mapping[str, Any], *, reply: str, conv: Conversation
+) -> dict[str, Any]:
+    return {
+        "request_id": body.get("request_id"),
+        "room": body.get("room"),
+        "device": body.get("device"),
+        "reply": reply,
+        "conversation": conv.external_id,
+    }
+
+
+def _turn_id() -> str:
+    return f"turn_{uuid4().hex[:12]}"
+
+
+def _text_delta(event: Any) -> str:
+    if getattr(event, "type", "") != "content_block_delta":
+        return ""
+    delta = getattr(event, "delta", None)
+    if getattr(delta, "type", "") != "text_delta":
+        return ""
+    return getattr(delta, "text", "")
+
+
+def _text_of(message: Any) -> str:
+    return "\n\n".join(
+        getattr(b, "text", "")
+        for b in message.content
+        if getattr(b, "type", "") == "text"
+    ).strip()

beaver_agent/voice/texts.py

@@ -0,0 +1,38 @@
+"""Что голосовой агент читает как промпт и что видно в его телеграм-ветке."""
+
+from __future__ import annotations
+
+TITLE = "🎤 Голосовой"
+"""Имя ветки; телеграм заводит топик под этим именем."""
+
+KEY = "квартира"
+"""Ключ единственного треда: одна ветка на всю квартиру."""
+
+PROMPT = """\
+Ты - голос из колонки в квартире. Отвечаешь вслух: одна-две фразы, без
+markdown, без списков, без ссылок, без эмодзи. Не знаешь - так и говори,
+коротко.
+
+Перед словами в квадратных скобках - комната и устройство, откуда говорят.
+С тобой говорят гости и домашние, кто именно - неизвестно.
+
+Про хозяина дома ты не знаешь ничего и ничего о нём не рассказываешь: ни
+имени, ни дел, ни планов, ни календаря, ни переписок, ни файлов, ни его
+знакомых - и даже того, что такие записи где-то есть. Спрашивают о нём -
+скажи, что этого не знаешь, и предложи спросить у него самого. Придумывать
+вместо ответа нельзя.
+
+Инструментов у тебя нет: ни файлов, ни поиска, ни календаря, ни умного дома.
+Всё, что ты умеешь, - разговаривать.
+
+Иногда в этой же ветке пишет сам хозяин, из телеграма. Ему отвечаешь так же
+коротко, и правила про него не отменяются.
+"""
+
+ASKED = "🎤 {room} · {device}\n{text}"
+"""Вопрос, как он показывается в телеграм-ветке."""
+
+HEARD = "[{room} · {device}] {text}"
+"""Вопрос, как его видит модель."""
+
+ROTATED = "🎤 Новый день - ветка начата заново, прошлого разговора агент не помнит."

beaver_agent/voice/thread.py

@@ -0,0 +1,62 @@
+"""Единственная ветка голосовых точек: как её открыть и как сменить за ночь."""
+
+from __future__ import annotations
+
+import logging
+from typing import TYPE_CHECKING
+
+from beaver_agent.voice.texts import KEY, ROTATED, TITLE
+
+if TYPE_CHECKING:
+    from beaver_gateway.conversations.service import Conversations
+    from beaver_gateway.storage.models import Conversation
+
+AGENT = "beaver-voice"
+"""Кто сидит в голосовой ветке; ему не видно ни справки, ни vault."""
+
+FRONTEND = "voice"
+TELEGRAM = "telegram"
+
+_log = logging.getLogger("beaver_agent.voice")
+
+
+async def current(conversations: Conversations) -> Conversation | None:
+    conv = await conversations.find_bound(frontend=FRONTEND, external_id=KEY)
+    return conv if conv is not None and conv.status == "open" else None
+
+
+async def open_thread(
+    conversations: Conversations, *, agent: str, window: str | None = None
+) -> Conversation:
+    """Завести ветку; без ``window`` окно ей открывает телеграм - новым топиком."""
+    conv = await conversations.spawn(
+        kind="branch",
+        agent=agent,
+        title=TITLE,
+        origin=FRONTEND,
+        binding=(TELEGRAM, window) if window else None,
+    )
+    await conversations.bind(conv, frontend=FRONTEND, external_id=KEY)
+    _log.info("голосовая ветка %s (%s)", conv.external_id, agent)
+    return conv
+
+
+async def window_of(conversations: Conversations, conv: Conversation) -> str | None:
+    for binding in await conversations.bindings(conv):
+        if binding.frontend == TELEGRAM and binding.visible:
+            return binding.external_id
+    return None
+
+
+async def rotate(conversations: Conversations) -> Conversation | None:
+    """Ночью - новая ветка в том же топике; молчавшую за сутки не трогаем."""
+    old = await current(conversations)
+    if old is None or old.last_user_activity_at is None:
+        return None
+    new = await open_thread(
+        conversations, agent=old.agent_name, window=await window_of(conversations, old)
+    )
+    await conversations.close(old)
+    await conversations.say(new, ROTATED)
+    _log.info("голосовая ветка: %s -> %s", old.external_id, new.external_id)
+    return new

caddy/site.caddy

@@ -1,5 +1,5 @@
 # Роуты beaver. Один апстрим, префиксы не срезаются: gateway сам живёт под
-# /anthropic, /mcp, /md, /api, /admin, а / ведёт в админку.
+# /anthropic, /mcp, /md, /api, /admin, /voice, а / ведёт в админку.
 # Домен и апстрим - из env сервера (domains.env / .env), tls - забота
 # серверного Caddyfile.
 {$BEAVER_DOMAIN} {

config.py

@@ -3,7 +3,8 @@
 - vault.py - пути и зона агента; prompts.py - как из гранул собирается промпт роли;
 - skills.py - наборы скиллов по окнам; policy.py - что можно трогать в vault;
 - hands/ - MCP-руки и кому какая выдана; agents.py - роли и их экземпляры;
-- frontends.py - окна; texts.py - всё, что gateway говорит, по-русски;
+- frontends.py - окна; voice/ - голосовые точки по квартире и их ветка;
+- texts.py - всё, что gateway говорит, по-русски;
 - memory/ - справка, реплики, хендауты, вотчер, куратор; jobs/ - кроны и вебхуки.
 """
 

tests/test_voice.py

@@ -0,0 +1,294 @@
+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": "ало клод как дела",
+}
+
+
+@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=ASK, 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=ASK, headers=VOICE)).json()
+        again = (
+            await http.post("/", json={**ASK, "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=ASK, 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=ASK, headers=VOICE),
+            http.post("/", json={**ASK, "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_a_streaming_ask_gets_the_answer_in_pieces() -> None:
+    conversations = FakeConversations()
+    frontend, _ = stand(conversations)
+    frames: list[tuple[str, dict[str, Any]]] = []
+    async with (
+        client(frontend) as http,
+        http.stream("POST", "/", json={**ASK, "stream": True}, 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"]