"""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 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,
    ResultMessage,
    StreamEvent,
    ToolResultBlock,
    ToolUseBlock,
    UserMessage,
    fork_session_via_store,
    project_key_for_directory,
)
from sqlmodel import col, select

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,
    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,
    InjectQueueItem,
    Schedule,
)

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Awaitable, Callable, 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.events import MessageStreamEvent
    from beaver_gateway.core.injects import Priority
    from beaver_gateway.core.registry import AgentRegistry
    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


@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


@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,
    ) -> 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._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)
        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
        return out

    # ---- 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
        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:
        if conv.session_id is None:
            return ""
        entries = await self._store.load(cast("Any", self._store_key(conv)))
        if not entries:
            return ""
        return render_messages(
            messages_from_entries(cast("Any", entries)), window=window
        )

    async def inject(
        self,
        conv: Conversation,
        text: str,
        *,
        urgency: Priority = "normal",
        origin: str = "system",
    ) -> 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":
            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) -> Schedule:
        row = Schedule(
            conversation_id=cast("int", conv.id), execute_at=parse_at(at), text=text
        )
        async with self._db.session() as session:
            session.add(row)
            await session.commit()
            await session.refresh(row)
        self._bus.publish(
            "schedule.created",
            conversation_id=conv.external_id,
            schedule=row.id,
            execute_at=_iso(row.execute_at),
        )
        return row

    # ---- §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)

    async def schedules(self, conv: Conversation | None = None) -> list[Schedule]:
        stmt = select(Schedule).order_by(col(Schedule.execute_at))
        if conv is not None:
            stmt = stmt.where(Schedule.conversation_id == conv.id)
        async with self._db.session() as session:
            return list((await session.exec(stmt)).all())

    # ---- 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
            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,
            )
            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,
                    capture=capture,
                    kind=conv.kind,
                    pinned=conv.kind == "master",
                    tools=tools,
                    observer=self._observer(conv.external_id, 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
            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,
            )

    def _observer(
        self, conversation_id: str, turn_id: str, origin: str
    ) -> Callable[[Any], None]:
        def observe(message: Any) -> None:
            parent = getattr(message, "parent_tool_use_id", None)
            if 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):
                        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,
                        )
            elif isinstance(message, UserMessage):
                blocks = message.content if isinstance(message.content, list) else ()
                for block in blocks:
                    if isinstance(block, ToolResultBlock):
                        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),
                        )
            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

    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


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,
    }