"""Conversations, threads and the one place a turn runs (§3.1, §3.4, §8.2-8.3).

The service owns the rows (:class:`Conversation`, :class:`ConversationBinding`),
the per-conversation queue workers, the fork/spawn/read/inject/say/schedule
API that frontends, jobs and the in-process MCP tools all call, and the
restart recovery: a conversation with ``running_turn`` set at startup gets
its open ``tool_use`` closed in the transcript and an inject saying the
turn was cut.

Routing is deterministic (§0.1): a turn started by a user message streams
back to whoever asked; a turn started by an inject streams nowhere - its
events still hit the bus flagged ``origin="inject"`` so the panel can show
activity, but the only way it speaks is ``say``.
"""

from __future__ import annotations

import asyncio
import contextlib
import inspect
import json
import logging
import re
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING, Any, cast

from claude_agent_sdk import (
    AssistantMessage,
    RateLimitEvent,
    ResultMessage,
    StreamEvent,
    ToolResultBlock,
    ToolUseBlock,
    UserMessage,
    fork_session_via_store,
    project_key_for_directory,
)
from sqlmodel import col, select

from beaver_gateway.core.conversation_store import load_messages
from beaver_gateway.core.injects import InjectQueue, inject_header
from beaver_gateway.core.kinds import KINDS, Kind, as_kind
from beaver_gateway.core.transcript import (
    messages_from_entries,
    render_messages,
    strip_tool_entries,
    text_of,
    window_entries,
)
from beaver_gateway.core.turn_capture import TurnCapture
from beaver_gateway.frontends._accumulate import StreamAccumulator
from beaver_gateway.storage.models import (
    Conversation,
    ConversationBinding,
    ConversationMessage,
    InjectQueueItem,
    RateLimit,
    Usage,
)

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence

    from claude_agent_sdk import SessionStore

    from beaver_gateway.agents.claude import ClaudeAgent
    from beaver_gateway.backends.claude_sdk import ClaudeSdkBackend
    from beaver_gateway.core.bus import EventBus
    from beaver_gateway.core.envelope import Envelope
    from beaver_gateway.core.events import MessageStreamEvent
    from beaver_gateway.core.injects import Priority
    from beaver_gateway.core.registry import AgentRegistry
    from beaver_gateway.core.rotation import HandoutContext
    from beaver_gateway.core.scheduler import Scheduler
    from beaver_gateway.core.sessions import SessionPool
    from beaver_gateway.frontends.base import Frontend
    from beaver_gateway.storage.db import Database

__all__ = [
    "KINDS",
    "SEEDS",
    "ConversationTexts",
    "Conversations",
    "ForkResult",
    "SeedContext",
]

_log = logging.getLogger("beaver_gateway.core.conversations")

SEEDS = ("clean", "morning", "copy", "brief")
_STATUSES = ("open", "merged", "closed", "archived")
_DEFAULT_MERGE_PROMPT = (
    "Ветка закрывается. Напиши слив для мастера: что решили, что сделано, "
    "что не сделано и почему, открытые вопросы. Идентификаторы и ссылки - "
    "дословно. Коротко, прошедшее время."
)
_RELATIVE = re.compile(r"^\+(\d+)\s*([smhd])$")
_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400}


@dataclass(frozen=True, slots=True)
class SeedContext:
    kind: Kind
    seed: str
    agent: str
    parent: Conversation | None
    text: str | None
    title: str | None


@dataclass(frozen=True, slots=True)
class ConversationTexts:
    """Texts the gateway cannot invent for a setup.

    What a merge asks for and what a seed says; ``seed`` may return the body
    for any seed mode (the morning handout lives in the vault the gateway
    knows nothing about).
    """

    merge_prompt: str = _DEFAULT_MERGE_PROMPT
    interrupted: str = "прервано"
    answered: str = "Пользователь ответил: {answer}"
    unanswered: str = (
        "Пользователь не ответил за {minutes} мин. Вопрос ему показан текстом; "
        "заверши тёрн сейчас, ответ придёт следующим сообщением."
    )
    seed: Callable[[SeedContext], Awaitable[str | None] | str | None] | None = None
    handout: Callable[[HandoutContext], Awaitable[str] | str] | str = (
        "Этот мастер закрывается ({reason}). Напиши хендаут за {day}: справку "
        "на утро, не задание - прошедшее время, без повелительного наклонения."
    )
    new_day: str = "Новый день: мастер сменился, хендаут за {day} записан."


@dataclass(frozen=True, slots=True)
class ForkResult:
    conversation: Conversation
    text: str
    capture: TurnCapture


@dataclass
class _Runner:
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    wake: asyncio.Event = field(default_factory=asyncio.Event)
    task: asyncio.Task[None] | None = None
    turn_id: str | None = None
    origin: str | None = None
    text: str | None = None
    started_at: datetime | None = None
    tools: dict[str, dict[str, Any]] = field(default_factory=dict)
    """Tool calls of the running turn, in order; ``describe`` hands them to a
    panel that subscribed mid-turn."""

    def snapshot(self) -> dict[str, Any] | None:
        if self.turn_id is None:
            return None
        return {
            "id": self.turn_id,
            "origin": self.origin,
            "text": self.text,
            "started_at": _iso(self.started_at),
            "tools": list(self.tools.values()),
        }


@dataclass(frozen=True, slots=True)
class _Question:
    conversation_id: str
    turn_id: str | None
    questions: list[dict[str, Any]]
    answer: asyncio.Future[str]


class Conversations:
    def __init__(
        self,
        *,
        db: Database,
        agents: AgentRegistry,
        backends: dict[str, Any],
        bus: EventBus,
        pool: SessionPool,
        store: SessionStore,
        texts: ConversationTexts | None = None,
        frontends: Sequence[Frontend] = (),
        normal_window: float = 3600.0,
        idle_days: Sequence[int] = (2,),
        idle_interval: float = 3600.0,
        question_timeout: float = 600.0,
        envelope: Envelope | None = None,
    ) -> None:
        self._db = db
        self._agents = agents
        self._backends = backends
        self._bus = bus
        self._pool = pool
        self._store = store
        self._texts = texts or ConversationTexts()
        self._frontends = [f for f in frontends if f.name]
        self._normal_window = normal_window
        self._idle_days = tuple(sorted(idle_days))
        self._idle_interval = idle_interval
        self._question_timeout = question_timeout
        self._envelope = envelope
        self.scheduler: Scheduler | None = None
        self._questions: dict[str, _Question] = {}
        self._queue = InjectQueue(db)
        self._runners: dict[int, _Runner] = {}
        self._tasks: set[asyncio.Task[None]] = set()
        self._idle_task: asyncio.Task[None] | None = None

    @property
    def queue(self) -> InjectQueue:
        return self._queue

    @property
    def bus(self) -> EventBus:
        return self._bus

    @property
    def pool(self) -> SessionPool:
        return self._pool

    # ---- rows ----------------------------------------------------------

    async def create(
        self,
        *,
        kind: Kind,
        agent: str,
        parent: Conversation | None = None,
        title: str | None = None,
        origin: str = "api",
        session_id: str | None = None,
        flags: dict[str, Any] | None = None,
    ) -> Conversation:
        if kind not in KINDS:
            msg = f"unknown conversation kind {kind!r}"
            raise ValueError(msg)
        if not self._claude_agent(agent).serves(kind):
            msg = f"agent {agent!r} does not serve kind {kind!r}"
            raise ValueError(msg)
        now = datetime.now(UTC)
        row = Conversation(
            frontend=origin,
            external_id=str(uuid.uuid4()),
            agent_name=agent,
            kind=kind,
            parent_id=parent.id if parent is not None else None,
            title=title,
            session_id=session_id,
            flags=dict(flags or {}),
            last_activity_at=now,
        )
        async with self._db.session() as session:
            session.add(row)
            await session.commit()
            await session.refresh(row)
        self._bus.publish("conversation.created", **self.public(row))
        return row

    async def get(self, public_id: str) -> Conversation | None:
        async with self._db.session() as session:
            result = await session.exec(
                select(Conversation).where(Conversation.external_id == public_id)
            )
            return result.first()

    async def get_row(self, row_id: int) -> Conversation | None:
        async with self._db.session() as session:
            return await session.get(Conversation, row_id)

    async def find(
        self,
        *,
        status: str | None = None,
        kind: str | None = None,
        parent: Conversation | None = None,
        limit: int = 200,
    ) -> list[Conversation]:
        stmt = select(Conversation).order_by(col(Conversation.id).desc()).limit(limit)
        if status is not None:
            stmt = stmt.where(Conversation.status == status)
        if kind is not None:
            stmt = stmt.where(Conversation.kind == kind)
        if parent is not None:
            stmt = stmt.where(Conversation.parent_id == parent.id)
        async with self._db.session() as session:
            return list((await session.exec(stmt)).all())

    async def bindings(self, conv: Conversation) -> list[ConversationBinding]:
        async with self._db.session() as session:
            result = await session.exec(
                select(ConversationBinding)
                .where(ConversationBinding.conversation_id == conv.id)
                .order_by(col(ConversationBinding.id))
            )
            return list(result.all())

    async def bind(
        self,
        conv: Conversation,
        *,
        frontend: str,
        external_id: str,
        visible: bool = True,
    ) -> ConversationBinding:
        if conv.kind not in self.frontend(frontend).kinds:
            msg = f"frontend {frontend!r} does not show kind {conv.kind!r}"
            raise ValueError(msg)
        async with self._db.session() as session:
            existing = list(
                (
                    await session.exec(
                        select(ConversationBinding).where(
                            ConversationBinding.conversation_id == conv.id,
                            ConversationBinding.frontend == frontend,
                        )
                    )
                ).all()
            )
            row = next((b for b in existing if b.external_id == external_id), None)
            if visible:
                for other in existing:
                    if other is not row and other.visible:
                        other.visible = False
                        session.add(other)
                same_window = await session.exec(
                    select(ConversationBinding).where(
                        ConversationBinding.frontend == frontend,
                        ConversationBinding.external_id == external_id,
                        ConversationBinding.conversation_id != conv.id,
                        col(ConversationBinding.visible).is_(True),
                    )
                )
                for other in same_window.all():
                    other.visible = False
                    session.add(other)
            if row is None:
                row = ConversationBinding(
                    conversation_id=cast("int", conv.id),
                    frontend=frontend,
                    external_id=external_id,
                    visible=visible,
                )
            else:
                row.visible = visible
            session.add(row)
            await session.commit()
            await session.refresh(row)
        self._bus.publish(
            "conversation.bound",
            conversation_id=conv.external_id,
            frontend=frontend,
            external_id=external_id,
            visible=visible,
        )
        return row

    async def find_bound(
        self, *, frontend: str, external_id: str
    ) -> Conversation | None:
        async with self._db.session() as session:
            result = await session.exec(
                select(Conversation)
                .join(
                    ConversationBinding,
                    col(ConversationBinding.conversation_id) == col(Conversation.id),
                )
                .where(
                    ConversationBinding.frontend == frontend,
                    ConversationBinding.external_id == external_id,
                    col(ConversationBinding.visible).is_(True),
                )
                .order_by(col(Conversation.id).desc())
            )
            return result.first()

    async def last_binding(
        self, *, frontend: str, kind: str
    ) -> ConversationBinding | None:
        """The window ``frontend`` last used for a conversation of ``kind``.

        A frontend whose window for the master outlives the master itself
        (the Telegram General topic) finds it here after a rotation.
        """
        async with self._db.session() as session:
            result = await session.exec(
                select(ConversationBinding)
                .join(
                    Conversation,
                    col(Conversation.id) == col(ConversationBinding.conversation_id),
                )
                .where(
                    ConversationBinding.frontend == frontend, Conversation.kind == kind
                )
                .order_by(col(ConversationBinding.id).desc())
            )
            return result.first()

    async def set_flags(
        self, conv: Conversation, flags: dict[str, Any]
    ) -> Conversation:
        async def apply(row: Conversation) -> None:
            row.flags = {**row.flags, **flags}

        return await self._update(conv, apply)

    async def set_status(self, conv: Conversation, status: str) -> Conversation:
        if status not in _STATUSES:
            msg = f"unknown status {status!r}"
            raise ValueError(msg)

        async def apply(row: Conversation) -> None:
            row.status = status

        return await self._update(conv, apply)

    async def set_title(self, conv: Conversation, title: str) -> Conversation:
        async def apply(row: Conversation) -> None:
            row.title = title

        return await self._update(conv, apply)

    async def touch_user(self, conv: Conversation) -> Conversation:
        async def apply(row: Conversation) -> None:
            row.last_user_activity_at = datetime.now(UTC)

        return await self._update(conv, apply)

    async def _update(
        self, conv: Conversation, apply: Callable[[Conversation], Awaitable[None]]
    ) -> Conversation:
        async with self._db.session() as session:
            row = await session.get(Conversation, conv.id)
            if row is None:
                msg = f"conversation {conv.external_id} vanished"
                raise LookupError(msg)
            await apply(row)
            row.updated_at = datetime.now(UTC)
            session.add(row)
            await session.commit()
            await session.refresh(row)
        self._bus.publish("conversation.updated", **self.public(row))
        return row

    def public(self, conv: Conversation) -> dict[str, Any]:
        return {
            "id": conv.external_id,
            "kind": conv.kind,
            "agent": conv.agent_name,
            "title": conv.title,
            "status": conv.status,
            "parent_row": conv.parent_id,
            "session_id": conv.session_id,
            "running_turn": conv.running_turn,
            "pending_question": conv.pending_question,
            "flags": conv.flags,
            "origin": conv.frontend,
            "created_at": _iso(conv.created_at),
            "last_user_activity_at": _iso(conv.last_user_activity_at),
            "last_activity_at": _iso(conv.last_activity_at),
        }

    async def describe(self, conv: Conversation) -> dict[str, Any]:
        out = self.public(conv)
        out["title"] = await self.implied_title(conv)
        parent = await self.get_row(conv.parent_id) if conv.parent_id else None
        out["parent"] = parent.external_id if parent is not None else None
        out["bindings"] = [
            {"frontend": b.frontend, "external_id": b.external_id, "visible": b.visible}
            for b in await self.bindings(conv)
        ]
        live = self._pool.get(conv.external_id)
        out["live"] = live is not None
        out["busy"] = live.busy if live is not None else False
        runner = self._runners.get(cast("int", conv.id))
        out["turn"] = runner.snapshot() if runner is not None else None
        pending = self.pending_question(conv.external_id)
        out["question"] = (
            {"id": pending[0], "questions": pending[1]} if pending else None
        )
        return out

    async def rate_limits(self, *, limit: int = 100) -> list[RateLimit]:
        async with self._db.session() as session:
            result = await session.exec(
                select(RateLimit).order_by(col(RateLimit.id).desc()).limit(limit)
            )
            return list(result.all())

    async def context_tokens(self, conv: Conversation) -> int:
        """Size of the context the last turn ran with, from its usage row."""
        async with self._db.session() as session:
            row = (
                await session.exec(
                    select(Usage)
                    .where(Usage.conversation_id == conv.external_id)
                    .order_by(col(Usage.id).desc())
                    .limit(1)
                )
            ).first()
        if row is None:
            return 0
        return row.input_tokens + row.cache_read_tokens + row.cache_creation_tokens

    async def usage_tokens(self, since: datetime) -> int:
        async with self._db.session() as session:
            rows = (
                await session.exec(
                    select(Usage).where(
                        col(Usage.ts) >= since.astimezone(UTC).replace(tzinfo=None)
                    )
                )
            ).all()
        return sum(
            r.input_tokens + r.output_tokens + r.cache_creation_tokens for r in rows
        )

    async def busy(self, conv: Conversation) -> bool:
        """A turn is running, a question is open or a message waits to run."""
        row = await self.get_row(cast("int", conv.id)) or conv
        if row.running_turn or row.pending_question:
            return True
        live = self._pool.get(row.external_id)
        if live is not None and live.busy:
            return True
        pending = await self._queue.pending(cast("int", row.id))
        return any(i.priority in ("user", "urgent") for i in pending)

    # ---- routing -------------------------------------------------------

    @property
    def frontends(self) -> list[Frontend]:
        return list(self._frontends)

    def frontend(self, name: str) -> Frontend:
        for fe in self._frontends:
            if fe.name == name:
                return fe
        msg = f"unknown frontend {name!r}"
        raise ValueError(msg)

    def default_agent(self, kind: Kind) -> str | None:
        for fe in self._frontends:
            if kind in fe.kinds and (agent := fe.agent_for(kind)):
                return agent
        return None

    async def materialize(self, conv: Conversation) -> ConversationBinding | None:
        for fe in self._frontends:
            if conv.kind not in fe.kinds:
                continue
            binding = await fe.materialize(conv)
            if binding is not None:
                return binding
        return None

    # ---- §3.1 api ------------------------------------------------------

    async def spawn(
        self,
        *,
        kind: Kind,
        agent: str | None = None,
        seed: str = "clean",
        parent: Conversation | None = None,
        text: str | None = None,
        title: str | None = None,
        window: int | None = None,
        origin: str = "api",
        binding: tuple[str, str] | None = None,
    ) -> Conversation:
        """Create a conversation and queue its seed turn (§8.2).

        ``binding`` = ``(frontend, external_id)`` puts it into a window that
        already exists (a topic the user created) instead of asking the
        home frontend to ``materialize`` one. ``text`` rides with the seed
        as the first thing the user said, whatever the seed mode; without
        it the seed waits in ``flags`` and opens the first turn, so a fresh
        window stays silent and costs nothing until someone speaks.
        """
        if seed not in SEEDS:
            msg = f"unknown seed {seed!r}"
            raise ValueError(msg)
        if seed == "brief" and not text:
            msg = "seed=brief needs text"
            raise ValueError(msg)
        if agent is None and kind == "branch" and parent is not None:
            agent = parent.agent_name
        if kind == "branch" and parent is not None and parent.kind == "master":
            await self.set_flags(parent, {"streak": 0})
        agent = agent or self.default_agent(kind)
        if agent is None:
            msg = f"no default agent for kind {kind!r}; pass `agent`"
            raise ValueError(msg)
        session_id: str | None = None
        if seed == "copy":
            if parent is None or parent.session_id is None:
                msg = "seed=copy needs a parent with a session"
                raise ValueError(msg)
            session_id = await self._copy_session(
                parent, window=window, strip_tools=False
            )
        conv = await self.create(
            kind=kind,
            agent=agent,
            parent=parent,
            title=title,
            origin=origin,
            session_id=session_id,
        )
        if binding is not None:
            await self.bind(conv, frontend=binding[0], external_id=binding[1])
        else:
            await self.materialize(conv)
        ctx = SeedContext(
            kind=kind, seed=seed, agent=agent, parent=parent, text=text, title=title
        )
        if text is None:
            return await self.set_flags(conv, {"seed": seed, "seed_window": window})
        await self._queue.push(
            conversation_id=cast("int", conv.id),
            priority="user",
            origin=f"сид:{seed}" if seed == "brief" else origin,
            text=await self._seed_text(ctx, window=window),
        )
        self._ensure_worker(cast("int", conv.id))
        return conv

    async def _pending_seed(self, conv: Conversation) -> str | None:
        """A seed nobody has spoken after yet: rendered now, spent once."""
        seed = conv.flags.get("seed")
        if not seed:
            return None
        parent = await self.get_row(conv.parent_id) if conv.parent_id else None
        ctx = SeedContext(
            kind=as_kind(conv.kind),
            seed=str(seed),
            agent=conv.agent_name,
            parent=parent,
            text=None,
            title=conv.title,
        )
        window = conv.flags.get("seed_window")
        text = await self._seed_text(
            ctx, window=window if isinstance(window, int) else None
        )
        await self.set_flags(conv, {"seed": None, "seed_window": None})
        return text

    async def fork(
        self,
        conv: Conversation,
        prompt: str,
        *,
        window: int | None = None,
        strip_tools: bool = False,
        title: str | None = None,
    ) -> ForkResult:
        session_id = await self._copy_session(
            conv, window=window, strip_tools=strip_tools
        )
        child = await self.create(
            kind="fork",
            agent=conv.agent_name,
            parent=conv,
            title=title or f"fork: {conv.title or conv.external_id}",
            origin="system",
            session_id=session_id,
        )
        try:
            text, capture = await self.run_text_turn(
                child, prompt, origin="fork", tools=False
            )
        finally:
            await self._backend(conv.agent_name).close(child.external_id)
        child = await self.set_status(child, "closed")
        return ForkResult(conversation=child, text=text, capture=capture)

    async def read(self, conv: Conversation, *, window: int | None = None) -> str:
        return render_messages(await self.history(conv), window=window)

    async def history(self, conv: Conversation) -> list[dict[str, Any]]:
        if conv.session_id is None:
            async with self._db.session() as session:
                return await load_messages(
                    session, conversation_id=cast("int", conv.id)
                )
        return messages_from_entries(cast("Any", await self.entries(conv)))

    async def first_user_texts(self, ids: Iterable[int]) -> dict[int, str]:
        wanted = list(ids)
        if not wanted:
            return {}
        async with self._db.session() as session:
            rows = (
                await session.exec(
                    select(ConversationMessage).where(
                        col(ConversationMessage.conversation_id).in_(wanted),
                        ConversationMessage.seq == 0,
                        ConversationMessage.role == "user",
                    )
                )
            ).all()
        return {
            r.conversation_id: text_of(json.loads(r.content_json)).strip() for r in rows
        }

    async def implied_title(self, conv: Conversation) -> str | None:
        if conv.title:
            return conv.title
        text = (await self.first_user_texts([cast("int", conv.id)])).get(
            cast("int", conv.id)
        )
        return implied_title(text)

    async def adopt(self, *, kind: Kind, first_user_text: str) -> Conversation | None:
        """The one unbound, session-less conversation whose history starts here.

        Rows from before the SDK cut-over have canonical messages but no
        window and no session; a vault file that begins with the same
        prompt is that conversation continued.
        """
        text = first_user_text.strip()
        if not text:
            return None
        bound = select(ConversationBinding.conversation_id).where(
            col(ConversationBinding.visible).is_(True)
        )
        async with self._db.session() as session:
            rows = (
                await session.exec(
                    select(Conversation).where(
                        Conversation.kind == kind,
                        Conversation.status == "open",
                        col(Conversation.session_id).is_(None),
                        col(Conversation.id).not_in(bound),
                    )
                )
            ).all()
        firsts = await self.first_user_texts(cast("int", r.id) for r in rows)
        hits = [r for r in rows if firsts.get(cast("int", r.id)) == text]
        return hits[0] if len(hits) == 1 else None

    async def entries(self, conv: Conversation, *, subpath: str = "") -> list[Any]:
        if conv.session_id is None:
            return []
        key = {**self._store_key(conv), "subpath": subpath}
        return list(await self._store.load(cast("Any", key)) or [])

    async def subpaths(self, conv: Conversation) -> list[str]:
        if conv.session_id is None:
            return []
        return list(await self._store.list_subkeys(cast("Any", self._store_key(conv))))

    async def inject(
        self,
        conv: Conversation,
        text: str,
        *,
        urgency: Priority = "normal",
        origin: str = "system",
        interrupt: bool = True,
    ) -> InjectQueueItem:
        item = await self._queue.push(
            conversation_id=cast("int", conv.id),
            priority=urgency,
            origin=origin,
            text=text,
        )
        self._bus.publish(
            "inject.queued",
            conversation_id=conv.external_id,
            item=item.id,
            priority=urgency,
            origin=origin,
        )
        if urgency == "urgent" and interrupt:
            backend = self._backend(conv.agent_name)
            if await backend.interrupt(conv.external_id):
                _log.info(
                    "conversation %s: interrupted for urgent inject", conv.external_id
                )
        self._ensure_worker(cast("int", conv.id))
        return item

    async def post(
        self, conv: Conversation, text: str, *, origin: str = "user"
    ) -> InjectQueueItem:
        item = await self._queue.push(
            conversation_id=cast("int", conv.id),
            priority="user",
            origin=origin,
            text=text,
        )
        await self.touch_user(conv)
        self._bus.publish(
            "message.queued",
            conversation_id=conv.external_id,
            item=item.id,
            origin=origin,
        )
        self._ensure_worker(cast("int", conv.id))
        return item

    async def say(self, conv: Conversation, text: str) -> dict[str, Any]:
        runner = self._runners.get(cast("int", conv.id))
        _log.info("say[%s]: %s", conv.external_id, text[:200])
        return self._bus.publish(
            "say",
            conversation_id=conv.external_id,
            text=text,
            turn_id=runner.turn_id if runner is not None else None,
        )

    async def merge(self, conv: Conversation) -> ForkResult:
        if conv.parent_id is None:
            msg = "merge needs a parent conversation"
            raise ValueError(msg)
        parent = await self.get_row(conv.parent_id)
        if parent is None:
            msg = "parent conversation vanished"
            raise LookupError(msg)
        result = await self.fork(
            conv,
            self._texts.merge_prompt,
            title=f"слив: {conv.title or conv.external_id}",
        )
        if result.text.strip():
            await self.inject(parent, result.text, urgency="normal", origin="слив")
        await self.set_status(conv, "merged")
        self._bus.publish(
            "conversation.merged",
            conversation_id=conv.external_id,
            parent=parent.external_id,
            fork=result.conversation.external_id,
        )
        return result

    async def schedule(
        self, conv: Conversation, at: str, text: str, *, dedupe_key: str | None = None
    ) -> tuple[int | None, datetime]:
        if self.scheduler is None:
            msg = "no scheduler; `schedule` is unavailable"
            raise RuntimeError(msg)
        return await self.scheduler.schedule(conv, at, text, dedupe_key=dedupe_key)

    async def schedules(self, conv: Conversation | None = None) -> list[dict[str, Any]]:
        return await self.scheduler.scheduled(conv) if self.scheduler else []

    # ---- §4.5 rotation -------------------------------------------------

    async def handout(self, conv: Conversation, ctx: HandoutContext) -> str:
        """The closing master's last turn: the handout prompt from the config."""
        source = self._texts.handout
        if isinstance(source, str):
            prompt = source.format(day=ctx.day.isoformat(), reason=ctx.reason)
        else:
            produced: Any = source(ctx)
            prompt = await produced if inspect.isawaitable(produced) else produced
        self._bus.publish(
            "handout.start", conversation_id=conv.external_id, day=ctx.day.isoformat()
        )
        try:
            text, _ = await self.run_text_turn(conv, prompt, origin="handout")
        except Exception:  # noqa: BLE001
            _log.exception("handout turn on %s failed", conv.external_id)
            text = ""
        self._bus.publish(
            "handout.end",
            conversation_id=conv.external_id,
            day=ctx.day.isoformat(),
            text=text[:2000],
        )
        return text

    async def close(self, conv: Conversation) -> Conversation:
        row = await self.set_status(conv, "closed")
        with contextlib.suppress(LookupError):
            await self._backend(conv.agent_name).close(conv.external_id)
        return row

    async def reparent(self, conv: Conversation, parent: Conversation) -> Conversation:
        async def apply(row: Conversation) -> None:
            row.parent_id = parent.id

        return await self._update(conv, apply)

    async def mark_closed(self, conv: Conversation) -> bool:
        marked = False
        for fe in self._frontends:
            if conv.kind in fe.kinds:
                try:
                    marked = await fe.mark_closed(conv) or marked
                except Exception:  # noqa: BLE001
                    _log.exception("%s could not mark %s", fe.name, conv.external_id)
        return marked

    async def new_day(self, conv: Conversation, *, moved: int = 0) -> InjectQueueItem:
        day = datetime.now(UTC).astimezone().date().isoformat()
        text = self._texts.new_day.format(day=day)
        if moved:
            text += f" Инжектов переехало из старого мастера: {moved}."
        return await self.inject(
            conv, text, urgency="urgent", origin="ротация", interrupt=False
        )

    # ---- §3.7 questions ------------------------------------------------

    async def ask(self, key: str, payload: dict[str, Any]) -> str | None:
        """``AskUserQuestion`` reached ``can_use_tool``: show it, wait for the answer.

        Returns the answer text, or ``None`` when nobody answered within
        ``question_timeout`` - the caller then tells the model to finish the
        turn, the frontend has already rendered the question as text.
        """
        conv = await self.get(key)
        if conv is None:
            return None
        runner = self._runners.get(cast("int", conv.id))
        question_id = f"q_{uuid.uuid4().hex[:10]}"
        pending = _Question(
            conversation_id=key,
            turn_id=runner.turn_id if runner is not None else None,
            questions=list(payload.get("questions") or []),
            answer=asyncio.get_running_loop().create_future(),
        )
        self._questions[question_id] = pending
        await self._set_pending_question(conv, value=True)
        self._bus.publish(
            "question",
            conversation_id=key,
            turn_id=pending.turn_id,
            question_id=question_id,
            questions=pending.questions,
            timeout=self._question_timeout,
        )
        try:
            async with asyncio.timeout(self._question_timeout):
                answer = await pending.answer
        except TimeoutError:
            self._bus.publish(
                "question.timeout",
                conversation_id=key,
                turn_id=pending.turn_id,
                question_id=question_id,
            )
            return None
        finally:
            self._questions.pop(question_id, None)
            await self._set_pending_question(conv, value=False)
        self._bus.publish(
            "question.answered",
            conversation_id=key,
            turn_id=pending.turn_id,
            question_id=question_id,
            answer=answer,
        )
        return answer

    def answer(self, question_id: str, answer: str) -> bool:
        pending = self._questions.get(question_id)
        if pending is None or pending.answer.done():
            return False
        pending.answer.set_result(answer)
        return True

    def pending_question(self, key: str) -> tuple[str, list[dict[str, Any]]] | None:
        for question_id, pending in self._questions.items():
            if pending.conversation_id == key and not pending.answer.done():
                return question_id, pending.questions
        return None

    def answer_text(self, answer: str | None) -> str:
        if answer is None:
            return self._texts.unanswered.format(
                minutes=round(self._question_timeout / 60)
            )
        return self._texts.answered.format(answer=answer)

    async def _set_pending_question(self, conv: Conversation, *, value: bool) -> None:
        async def apply(row: Conversation) -> None:
            row.pending_question = value

        with contextlib.suppress(LookupError):
            await self._update(conv, apply)

    # ---- turns ---------------------------------------------------------

    async def turn(
        self,
        conv: Conversation,
        *,
        messages: Sequence[Any],
        origin: str,
        capture: TurnCapture | None = None,
        session_id: str | None = None,
        use_session: bool = True,
        tools: bool = True,
        turn_id: str | None = None,
        item_origin: str | None = None,
    ) -> AsyncIterator[MessageStreamEvent]:
        """Run one turn on ``conv`` under its lock; the only path to the backend.

        ``use_session=False`` withholds the stored ``session_id`` so the
        backend reseeds from ``messages`` (markdown file diverged).
        """
        row_id = cast("int", conv.id)
        runner = self._runner(row_id)
        backend = self._backend(conv.agent_name)
        turn_id = turn_id or f"turn_{uuid.uuid4().hex[:12]}"
        capture = capture or TurnCapture()
        resume = session_id if session_id is not None else conv.session_id
        async with runner.lock:
            runner.turn_id = turn_id
            runner.origin = origin
            runner.text = _prompt_preview(messages)
            runner.started_at = datetime.now(UTC)
            runner.tools = {}
            await self._mark_running(conv, turn_id)
            self._bus.publish(
                "turn.start",
                conversation_id=conv.external_id,
                turn_id=turn_id,
                origin=origin,
                item_origin=item_origin,
                text=runner.text,
            )
            stop = "error"
            cut = False
            try:
                events = backend.complete(
                    agent=self._claude_agent(conv.agent_name),
                    messages=messages,
                    conversation_id=conv.external_id,
                    session_id=resume if use_session else None,
                    reseed=not use_session,
                    capture=capture,
                    kind=conv.kind,
                    pinned=conv.kind == "master",
                    tools=tools,
                    observer=self._observer(conv, runner, turn_id, origin),
                    turn_id=turn_id,
                )
                async for event in events:
                    yield event
                stop = "interrupted" if capture.interrupted else "end_turn"
            except asyncio.CancelledError:
                cut = True
                raise
            finally:
                runner.turn_id = None
                await self._mark_done(conv, capture, cut=cut)
                self._bus.publish(
                    "turn.end",
                    conversation_id=conv.external_id,
                    turn_id=turn_id,
                    origin=origin,
                    item_origin=item_origin,
                    stop=stop,
                    usage=_usage_dict(capture),
                )

    async def run_text_turn(
        self,
        conv: Conversation,
        text: str,
        *,
        origin: str,
        tools: bool = True,
        turn_id: str | None = None,
        item_origin: str | None = None,
    ) -> tuple[str, TurnCapture]:
        capture = TurnCapture()
        acc = StreamAccumulator()
        agent = self._claude_agent(conv.agent_name)
        async for event in self.turn(
            conv,
            messages=[{"role": "user", "content": text}],
            origin=origin,
            capture=capture,
            tools=tools,
            turn_id=turn_id,
            item_origin=item_origin,
        ):
            acc.feed(event)
        message = acc.finalize(model=agent.model)
        reply = "\n\n".join(
            getattr(b, "text", "")
            for b in message.content
            if getattr(b, "type", "") == "text"
        ).strip()
        return reply, capture

    # ---- lifecycle -----------------------------------------------------

    async def start(self) -> None:
        await self.recover()
        for row_id in await self._queue.conversations_with_pending():
            self._ensure_worker(row_id)
        self._idle_task = asyncio.create_task(self._idle_loop())

    async def stop(self) -> None:
        tasks = list(self._tasks)
        if self._idle_task is not None:
            tasks.append(self._idle_task)
        for task in tasks:
            task.cancel()
        for task in tasks:
            with contextlib.suppress(BaseException):
                await task
        self._tasks.clear()
        self._idle_task = None

    async def recover(self) -> list[Conversation]:
        """Restart path (§8.1): repair transcripts of turns cut mid-flight."""
        async with self._db.session() as session:
            result = await session.exec(
                select(Conversation).where(col(Conversation.running_turn).is_not(None))
            )
            cut = list(result.all())
        for conv in cut:
            fixed = 0
            if conv.session_id is not None:
                backend = self._backend(conv.agent_name)
                try:
                    fixed = await backend.repair_session(conv.session_id)
                except Exception:  # noqa: BLE001
                    _log.exception("repair of %s failed", conv.session_id)
            turn_id = conv.running_turn

            async def clear(row: Conversation) -> None:
                row.running_turn = None
                row.pending_question = False

            await self._update(conv, clear)
            note = f"тёрн {turn_id} оборван рестартом gateway"
            if fixed:
                note += (
                    f"; {fixed} незакрытых тул-коллов получили tool_result "
                    f"«{self._texts.interrupted}»"
                )
            await self.inject(conv, note, urgency="normal", origin="система")
            _log.warning("conversation %s: %s", conv.external_id, note)
        interrupted = await self._queue.interrupted()
        for item in interrupted:
            _log.warning(
                "queue item #%s (%s) was running at restart; marked interrupted",
                item.id,
                item.priority,
            )
        return cut

    # ---- internals -----------------------------------------------------

    def _backend(self, agent: str) -> ClaudeSdkBackend:
        backend = self._backends.get(agent)
        if backend is None or not hasattr(backend, "repair_session"):
            msg = f"agent {agent!r} has no Claude SDK backend"
            raise LookupError(msg)
        return cast("ClaudeSdkBackend", backend)

    def _claude_agent(self, name: str) -> ClaudeAgent:
        agent = self._agents.get(name)
        if agent is None or not hasattr(agent, "cwd"):
            msg = f"unknown Claude agent {name!r}"
            raise LookupError(msg)
        return cast("ClaudeAgent", agent)

    def _store_key(self, conv: Conversation) -> dict[str, str]:
        agent = self._claude_agent(conv.agent_name)
        return {
            "project_key": project_key_for_directory(str(agent.cwd)),
            "session_id": cast("str", conv.session_id),
        }

    async def _copy_session(
        self, conv: Conversation, *, window: int | None, strip_tools: bool
    ) -> str:
        if conv.session_id is None:
            msg = f"conversation {conv.external_id} has no session to copy"
            raise ValueError(msg)
        live = self._pool.get(conv.external_id)
        if live is not None and live.dirty:
            msg = f"conversation {conv.external_id} has a mirror gap; not forking"
            raise RuntimeError(msg)
        agent = self._claude_agent(conv.agent_name)
        forked = await fork_session_via_store(
            self._store, conv.session_id, directory=str(agent.cwd)
        )
        if window is not None or strip_tools:
            key = {
                "project_key": project_key_for_directory(str(agent.cwd)),
                "session_id": forked.session_id,
            }
            entries = await self._store.load(cast("Any", key)) or []
            trimmed = window_entries(cast("Any", entries), window=window)
            if strip_tools:
                trimmed = strip_tool_entries(trimmed)
            await self._store.delete(cast("Any", key))
            await self._store.append(cast("Any", key), cast("Any", trimmed))
        _log.info(
            "forked session %s -> %s (window=%s, strip_tools=%s)",
            conv.session_id,
            forked.session_id,
            window,
            strip_tools,
        )
        return forked.session_id

    async def _seed_text(self, ctx: SeedContext, *, window: int | None) -> str:
        stamp = datetime.now(UTC).astimezone().strftime("%Y-%m-%d %H:%M")
        title = f" «{ctx.title}»" if ctx.title else ""
        head = f"[сид: {ctx.seed}] {ctx.kind}{title}, {stamp}."
        body: str | None = None
        if self._texts.seed is not None:
            produced: Any = self._texts.seed(ctx)
            if inspect.isawaitable(produced):
                produced = await produced
            body = cast("str | None", produced)
        if body is None:
            if ctx.seed == "brief":
                body = ctx.text
            elif ctx.seed == "copy":
                scope = f"последние {window} тёрнов" if window else "вся история"
                body = f"История родителя скопирована ({scope}); продолжай в ней."
            elif ctx.seed == "morning":
                body = "Хендаут не приехал."
        parts = [head, body, ctx.text if ctx.seed != "brief" else None]
        return "\n\n".join(p for p in parts if p)

    def _runner(self, row_id: int) -> _Runner:
        runner = self._runners.get(row_id)
        if runner is None:
            runner = _Runner()
            self._runners[row_id] = runner
        return runner

    def _ensure_worker(self, row_id: int) -> None:
        runner = self._runner(row_id)
        runner.wake.set()
        if runner.task is None or runner.task.done():
            runner.task = asyncio.create_task(self._worker(row_id))
            self._tasks.add(runner.task)
            runner.task.add_done_callback(self._tasks.discard)

    async def _worker(self, row_id: int) -> None:
        runner = self._runner(row_id)
        while True:
            items = await self._queue.pending(row_id)
            batch, wait = self._pick(items)
            if batch is None:
                runner.wake.clear()
                if wait is None:
                    return
                with contextlib.suppress(TimeoutError):
                    await asyncio.wait_for(runner.wake.wait(), timeout=wait)
                continue
            conv = await self.get_row(row_id)
            if conv is None:
                await self._queue.finish(batch, status="failed")
                return
            await self._run_batch(conv, batch)

    def _pick(
        self, items: list[InjectQueueItem]
    ) -> tuple[list[InjectQueueItem] | None, float | None]:
        if not items:
            return None, None
        normals = [i for i in items if i.priority == "normal"]
        head = items[0]
        if head.priority == "urgent":
            return [head], None
        if head.priority == "user":
            return [head, *normals], None
        age = (datetime.now(UTC) - _aware(head.created_at)).total_seconds()
        if age >= self._normal_window:
            return normals, None
        return None, max(self._normal_window - age, 1.0)

    async def _run_batch(
        self, conv: Conversation, batch: list[InjectQueueItem]
    ) -> None:
        head = batch[0]
        turn_id = f"turn_{uuid.uuid4().hex[:12]}"
        await self._queue.start(batch, turn_id)
        if head.priority == "user":
            origin = "user"
            prompt = head.text
            envelope = await self._envelope_for(conv, injects=len(batch) - 1)
            if envelope:
                prompt += "\n\n" + envelope
            if len(batch) > 1:
                prompt += "\n\n" + _bundle(batch[1:])
        else:
            origin = "inject"
            prompt = "\n\n".join(f"{inject_header(i.origin)}\n{i.text}" for i in batch)
        seed = await self._pending_seed(conv)
        if seed:
            prompt = f"{seed}\n\n{prompt}"
        try:
            text, capture = await self.run_text_turn(
                conv, prompt, origin=origin, turn_id=turn_id, item_origin=head.origin
            )
        except Exception:  # noqa: BLE001
            _log.exception("turn %s on %s failed", turn_id, conv.external_id)
            await self._queue.finish(batch, status="failed")
            return
        await self._queue.finish(
            batch, status="interrupted" if capture.interrupted else "done"
        )
        if origin == "user":
            self._bus.publish(
                "reply",
                conversation_id=conv.external_id,
                turn_id=turn_id,
                item=head.id,
                item_origin=head.origin,
                source="queue",
                prompt=prompt,
                user_text=head.text,
                text=text,
            )

    async def _envelope_for(self, conv: Conversation, *, injects: int) -> str | None:
        if conv.kind != "master":
            return None
        streak = int(conv.flags.get("streak", 0) or 0)
        await self.set_flags(conv, {"streak": streak + 1})
        if self._envelope is None:
            return None
        return self._envelope.build(streak=streak, injects=injects)

    def _observer(
        self, conv: Conversation, runner: _Runner, turn_id: str, origin: str
    ) -> Callable[[Any], None]:
        conversation_id = conv.external_id

        def observe(message: Any) -> None:
            parent = getattr(message, "parent_tool_use_id", None)
            if isinstance(message, RateLimitEvent):
                self._observe_rate_limit(conv, message)
            elif isinstance(message, StreamEvent):
                self._bus.publish(
                    "stream",
                    conversation_id=conversation_id,
                    turn_id=turn_id,
                    origin=origin,
                    parent_tool_use_id=parent,
                    event=message.event,
                )
            elif isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, ToolUseBlock):
                        event = self._bus.publish(
                            "tool",
                            conversation_id=conversation_id,
                            turn_id=turn_id,
                            origin=origin,
                            parent_tool_use_id=parent,
                            tool_use_id=block.id,
                            name=block.name,
                            input=block.input,
                        )
                        runner.tools[block.id] = {
                            "tool_use_id": block.id,
                            "name": block.name,
                            "input": block.input,
                            "parent_tool_use_id": parent,
                            "started_at": event["ts"],
                            "ended_at": None,
                            "is_error": None,
                            "content": None,
                        }
            elif isinstance(message, UserMessage):
                blocks = message.content if isinstance(message.content, list) else ()
                for block in blocks:
                    if isinstance(block, ToolResultBlock):
                        event = self._bus.publish(
                            "tool.result",
                            conversation_id=conversation_id,
                            turn_id=turn_id,
                            origin=origin,
                            parent_tool_use_id=parent,
                            tool_use_id=block.tool_use_id,
                            is_error=bool(block.is_error),
                            content=_result_preview(block.content),
                        )
                        node = runner.tools.get(block.tool_use_id)
                        if node is not None:
                            node["ended_at"] = event["ts"]
                            node["is_error"] = event["is_error"]
                            node["content"] = event["content"]
            elif isinstance(message, ResultMessage) and parent is None:
                self._bus.publish(
                    "result",
                    conversation_id=conversation_id,
                    turn_id=turn_id,
                    origin=origin,
                    subtype=message.subtype,
                    is_error=message.is_error,
                    num_turns=message.num_turns,
                )

        return observe

    def _observe_rate_limit(self, conv: Conversation, message: RateLimitEvent) -> None:
        info = message.rate_limit_info
        row = RateLimit(
            window=info.rate_limit_type or "unknown",
            status=info.status,
            utilization=info.utilization,
            resets_at=_from_unix(info.resets_at),
            overage_status=info.overage_status,
            overage_resets_at=_from_unix(info.overage_resets_at),
            agent_name=conv.agent_name,
            session_id=message.session_id,
            raw=dict(info.raw),
        )
        self._bus.publish(
            "rate_limit",
            conversation_id=conv.external_id,
            window=row.window,
            status=row.status,
            utilization=row.utilization,
            resets_at=_iso(row.resets_at),
            overage_status=row.overage_status,
        )
        task = asyncio.create_task(self._record_rate_limit(row))
        self._tasks.add(task)
        task.add_done_callback(self._tasks.discard)

    async def _record_rate_limit(self, row: RateLimit) -> None:
        try:
            async with self._db.session() as session:
                session.add(row)
                await session.commit()
        except Exception:  # noqa: BLE001
            _log.exception("rate limit write failed")

    async def _mark_running(self, conv: Conversation, turn_id: str) -> None:
        async def apply(row: Conversation) -> None:
            row.running_turn = turn_id
            row.last_activity_at = datetime.now(UTC)

        await self._update(conv, apply)

    async def _mark_done(
        self, conv: Conversation, capture: TurnCapture, *, cut: bool = False
    ) -> None:
        """Close the turn; a cancelled one keeps ``running_turn`` for ``recover``."""

        async def apply(row: Conversation) -> None:
            if not cut:
                row.running_turn = None
            row.last_activity_at = datetime.now(UTC)
            if capture.session_id is not None:
                row.session_id = capture.session_id

        await self._update(conv, apply)

    async def _idle_loop(self) -> None:
        while True:
            try:
                await self._emit_idle()
            except Exception:  # noqa: BLE001
                _log.exception("idle watcher failed")
            await asyncio.sleep(self._idle_interval)

    async def _emit_idle(self) -> None:
        if not self._idle_days:
            return
        now = datetime.now(UTC)
        for conv in await self.find(status="open", limit=10_000):
            last = _aware(conv.last_activity_at or conv.created_at)
            days = int((now - last).total_seconds() // 86400)
            due = [d for d in self._idle_days if days >= d]
            if not due:
                continue
            notified = int(conv.flags.get("idle_notified", 0) or 0)
            if due[-1] <= notified:
                continue
            await self.set_flags(conv, {"idle_notified": due[-1]})
            bindings = await self.bindings(conv)
            self._bus.publish(
                "conversation.idle",
                conversation_id=conv.external_id,
                kind=conv.kind,
                agent=conv.agent_name,
                days=due[-1],
                bindings=[
                    {"frontend": b.frontend, "external_id": b.external_id}
                    for b in bindings
                    if b.visible
                ],
            )


def parse_at(at: str) -> datetime:
    raw = at.strip()
    match = _RELATIVE.match(raw.replace(" ", ""))
    if match:
        amount, unit = match.groups()
        return datetime.now(UTC) + timedelta(seconds=int(amount) * _UNITS[unit])
    parsed = datetime.fromisoformat(raw)
    if parsed.tzinfo is None:
        parsed = parsed.astimezone()
    return parsed.astimezone(UTC)


def _bundle(items: Sequence[InjectQueueItem]) -> str:
    lines = [f"[инжекты, накопившиеся с {_iso(items[0].created_at)}; это не Бобёр]"]  # noqa: RUF001
    lines.extend(f"- [{i.origin}] {i.text}" for i in items)
    return "\n".join(lines)


def _aware(value: datetime) -> datetime:
    return value if value.tzinfo is not None else value.replace(tzinfo=UTC)


def _iso(value: datetime | None) -> str | None:
    return _aware(value).isoformat(timespec="seconds") if value is not None else None


TITLE_MAX = 80


def implied_title(text: str | None) -> str | None:
    if not text:
        return None
    line = text.strip().splitlines()[0].strip()
    return line if len(line) <= TITLE_MAX else line[: TITLE_MAX - 1] + "…"


def _prompt_preview(messages: Sequence[Any], limit: int = 400) -> str | None:
    if not messages:
        return None
    text = text_of(messages[-1].get("content"))
    return text[:limit] if text else None


def _from_unix(value: int | None) -> datetime | None:
    return datetime.fromtimestamp(value, tz=UTC) if value is not None else None


def _result_preview(
    content: str | list[dict[str, Any]] | None, limit: int = 400
) -> str:
    """Short text of a tool result for the panel: the transcript keeps the whole."""
    if content is None:
        return ""
    text = (
        content
        if isinstance(content, str)
        else "\n".join(
            str(part.get("text", ""))
            for part in content
            if isinstance(part, dict) and part.get("type") == "text"
        )
    )
    return text if len(text) <= limit else text[:limit] + "…"


def _usage_dict(capture: TurnCapture) -> dict[str, Any] | None:
    usage = capture.usage
    if usage is None:
        return None
    return {
        "input": usage.input_tokens,
        "output": usage.output_tokens,
        "cache_read": usage.cache_read_tokens,
        "cache_creation": usage.cache_creation_tokens,
        "cost_usd": usage.cost_usd,
        "duration_ms": usage.duration_ms,
    }