feat(distill,conversations,scheduler,api,ui): close deep chats with a distiller fork, digest and index, idle and weekly state jobs

src/beaver_gateway/cli.py

@@ -181,6 +181,7 @@ async def _async_main() -> None:
             texts=gateway.texts,
             frontends=gateway.frontends,
             envelope=Envelope(watch=gateway.watch, tz=gateway.tz),
+            distiller=gateway.distiller,
         )
         late.conversations = conversations
         scheduler = Scheduler(

src/beaver_gateway/core/conversations.py

@@ -40,6 +40,18 @@ from claude_agent_sdk import (
 from sqlmodel import col, select
 
 from beaver_gateway.core.conversation_store import load_messages
+from beaver_gateway.core.distill import (
+    Digest,
+    DistillContext,
+    Distiller,
+    LineCap,
+    append_index,
+    check_digest,
+    find_digest,
+    index_line,
+    trim_summary,
+    written_paths,
+)
 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 (
@@ -83,6 +95,7 @@ __all__ = [
     "SEEDS",
     "ConversationTexts",
     "Conversations",
+    "DistillResult",
     "ForkResult",
     "SeedContext",
 ]
@@ -96,8 +109,19 @@ _DEFAULT_MERGE_PROMPT = (
     "что не сделано и почему, открытые вопросы. Идентификаторы и ссылки - "
     "дословно. Коротко, прошедшее время."
 )
+_DEFAULT_DISTILL_PROMPT = (
+    "Глубокий чат «{chat}» закрыт ({reason}), сегодня {day}. Напиши выжимку "
+    "файлом и слив текстом ответа: до 5 строк, третье лицо."
+)
+_DEFAULT_DISTILL_PROMPT_NO_MEMORY = (
+    "Глубокий чат «{chat}» закрыт ({reason}), сегодня {day}. Память для него "
+    "выключена: файл не пиши, только слив текстом ответа - до 5 строк, третье лицо."
+)
 _RELATIVE = re.compile(r"^\+(\d+)\s*([smhd])$")
 _UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400}
+_CLOSE_WAIT = 0.25
+_CLOSE_TRIES = 40
+_CAP_TRIES = 3
 
 
 @dataclass(frozen=True, slots=True)
@@ -132,6 +156,18 @@ class ConversationTexts:
         "на утро, не задание - прошедшее время, без повелительного наклонения."
     )
     new_day: str = "Новый день: мастер сменился, хендаут за {day} записан."
+    distill: Callable[[DistillContext], Awaitable[str] | str] | None = None
+    """The distiller fork's first message (§8.4): which chat, what day, where
+    the digest goes; ``None`` uses a path-less default."""
+
+    closed: str = "Закрыт глубокий чат [[{chat}]]{digest}.\n{text}"
+    """What the master hears about a closed deep chat; ``{digest}`` is
+    ``, выжимка [[…]]`` or nothing, ``{text}`` the distiller's merge."""
+
+    too_long: str = (
+        "`{name}`: {lines} строк при потолке {max_lines}. Запись отбита, файл "
+        "возвращён к прежней версии. Сократи и перепиши."
+    )
 
 
 @dataclass(frozen=True, slots=True)
@@ -141,6 +177,18 @@ class ForkResult:
     capture: TurnCapture
 
 
+@dataclass(frozen=True, slots=True)
+class DistillResult:
+    """A closed deep chat: the merge text and, if memory was on, the digest."""
+
+    conversation: Conversation
+    fork: Conversation
+    text: str
+    digest: Digest | None
+    error: str | None
+    trimmed: bool
+
+
 @dataclass
 class _Runner:
     lock: asyncio.Lock = field(default_factory=asyncio.Lock)
@@ -191,8 +239,10 @@ class Conversations:
         idle_interval: float = 3600.0,
         question_timeout: float = 600.0,
         envelope: Envelope | None = None,
+        distiller: Distiller | None = None,
     ) -> None:
         self._db = db
+        self._distiller = distiller
         self._agents = agents
         self._backends = backends
         self._bus = bus
@@ -573,6 +623,7 @@ class Conversations:
         window: int | None = None,
         origin: str = "api",
         binding: tuple[str, str] | None = None,
+        flags: dict[str, Any] | None = None,
     ) -> Conversation:
         """Create a conversation and queue its seed turn (§8.2).
 
@@ -610,6 +661,7 @@ class Conversations:
             title=title,
             origin=origin,
             session_id=session_id,
+            flags=flags,
         )
         if binding is not None:
             await self.bind(conv, frontend=binding[0], external_id=binding[1])
@@ -658,13 +710,21 @@ class Conversations:
         window: int | None = None,
         strip_tools: bool = False,
         title: str | None = None,
+        agent: str | None = None,
     ) -> ForkResult:
+        """Copy the history into a one-off session and run ``prompt`` on it.
+
+        ``agent`` runs the fork under another agent's prompt (the distiller
+        closing a deep chat, §8.4); the copy then lives under that agent's
+        project key. Forks get no MCP (§3.2).
+        """
+        agent = agent or conv.agent_name
         session_id = await self._copy_session(
-            conv, window=window, strip_tools=strip_tools
+            conv, window=window, strip_tools=strip_tools, agent=agent
         )
         child = await self.create(
             kind="fork",
-            agent=conv.agent_name,
+            agent=agent,
             parent=conv,
             title=title or f"fork: {conv.title or conv.external_id}",
             origin="system",
@@ -675,7 +735,7 @@ class Conversations:
                 child, prompt, origin="fork", tools=False
             )
         finally:
-            await self._backend(conv.agent_name).close(child.external_id)
+            await self._backend(agent).close(child.external_id)
         child = await self.set_status(child, "closed")
         return ForkResult(conversation=child, text=text, capture=capture)
 
@@ -882,6 +942,261 @@ class Conversations:
             await self._backend(conv.agent_name).close(conv.external_id)
         return row
 
+    # ---- §8.4 closing a deep chat ---------------------------------------
+
+    async def request_close(self, conv: Conversation) -> Conversation:
+        """``close_chat`` from inside a turn: the chat closes once the turn ends."""
+        if conv.kind != "deep":
+            msg = f"only deep chats close this way, {conv.external_id} is {conv.kind}"
+            raise ValueError(msg)
+        return await self.set_flags(conv, {"close_requested": True})
+
+    async def idle(
+        self,
+        *,
+        kind: str,
+        days: int,
+        since: datetime | None = None,
+        limit: int | None = None,
+    ) -> list[Conversation]:
+        """Open conversations of ``kind`` quiet for ``days``, oldest first.
+
+        Only those with a session (something to fork) and, with ``since``,
+        with activity after it - the pile from before the system went live
+        is left alone (§4.5).
+        """
+        now = datetime.now(UTC)
+        cutoff = now - timedelta(days=days)
+        out: list[tuple[datetime, Conversation]] = []
+        for conv in await self.find(status="open", kind=kind, limit=10_000):
+            if conv.session_id is None:
+                continue
+            last = _aware(conv.last_activity_at or conv.created_at)
+            if last > cutoff or (since is not None and last < since):
+                continue
+            out.append((last, conv))
+        out.sort(key=lambda pair: pair[0])
+        rows = [conv for _, conv in out]
+        return rows[:limit] if limit is not None else rows
+
+    async def distill(
+        self, conv: Conversation, *, reason: str = "api"
+    ) -> DistillResult:
+        """Close a deep chat (§8.4): fork it under the distiller, two channels.
+
+        The digest is a file the fork writes under ``Distiller.dir`` - the
+        gateway checks it appeared with a valid frontmatter and puts a line
+        into the index; the merge is the fork's text, at most five lines,
+        injected into the open master. ``flags.memory=False`` skips the
+        file. The chat's own file is untouched; ``status=closed``.
+        """
+        if self._distiller is None:
+            msg = "no distiller configured (Gateway(distiller=...))"
+            raise RuntimeError(msg)
+        if conv.kind != "deep":
+            msg = f"only deep chats are distilled, {conv.external_id} is {conv.kind}"
+            raise ValueError(msg)
+        row = await self.get_row(cast("int", conv.id)) or conv
+        if row.status != "open":
+            msg = f"conversation {row.external_id} is {row.status}"
+            raise ValueError(msg)
+        if await self.busy(row):
+            msg = f"conversation {row.external_id} is busy"
+            raise RuntimeError(msg)
+        memory = bool(row.flags.get("memory", True))
+        chat_name = await self._chat_name(row)
+        ctx = DistillContext(
+            conversation=row,
+            title=await self.implied_title(row),
+            source=await self._window_of(row),
+            chat_name=chat_name,
+            memory=memory,
+            reason=reason,
+            day=datetime.now(UTC).astimezone().date(),
+        )
+        prompt = await self._distill_prompt(ctx)
+        started = datetime.now(UTC)
+        self._bus.publish(
+            "distill.start",
+            conversation_id=row.external_id,
+            reason=reason,
+            memory=memory,
+        )
+        result = await self.fork(
+            row,
+            prompt,
+            strip_tools=True,
+            agent=self._distiller.agent,
+            title=f"выжимка: {chat_name}",
+        )
+        text, trimmed = trim_summary(result.text)
+        digest: Digest | None = None
+        error: str | None = None
+        if memory:
+            written = written_paths(result.capture.synthesized_messages)
+            path = find_digest(self._distiller, since=started, written=written)
+            if path is None:
+                error = "файл выжимки не появился"
+            else:
+                checked = check_digest(path, self._distiller)
+                if isinstance(checked, str):
+                    error = f"{path.name}: {checked}"
+                else:
+                    digest = checked
+                    append_index(self._distiller, index_line(digest, chat_name))
+        if error is not None:
+            _log.warning("distill of %s: %s", row.external_id, error)
+        master = await self._open_master()
+        if master is not None and text:
+            note = self._texts.closed.format(
+                chat=chat_name,
+                digest=f", выжимка [[{digest.path.stem}]]" if digest else "",
+                text=text,
+            )
+            await self.inject(master, note, urgency="normal", origin="выжимка")
+        await self.close(row)
+        row = await self.set_flags(
+            row,
+            {
+                "close_requested": None,
+                "closed_reason": reason,
+                "digest": str(digest.path) if digest else None,
+                "digest_error": error,
+            },
+        )
+        self._bus.publish(
+            "conversation.distilled",
+            conversation_id=row.external_id,
+            fork=result.conversation.external_id,
+            reason=reason,
+            memory=memory,
+            digest=str(digest.path) if digest else None,
+            error=error,
+            text=text,
+            trimmed=trimmed,
+            master=master.external_id if master is not None else None,
+        )
+        return DistillResult(
+            conversation=row,
+            fork=result.conversation,
+            text=text,
+            digest=digest,
+            error=error,
+            trimmed=trimmed,
+        )
+
+    async def _distill_prompt(self, ctx: DistillContext) -> str:
+        source = self._texts.distill
+        if source is None:
+            template = (
+                _DEFAULT_DISTILL_PROMPT
+                if ctx.memory
+                else _DEFAULT_DISTILL_PROMPT_NO_MEMORY
+            )
+            return template.format(
+                chat=ctx.chat_name, reason=ctx.reason, day=ctx.day.isoformat()
+            )
+        produced: Any = source(ctx)
+        return await produced if inspect.isawaitable(produced) else produced
+
+    async def _open_master(self) -> Conversation | None:
+        masters = await self.find(kind="master", status="open", limit=1)
+        return masters[0] if masters else None
+
+    async def _window_of(self, conv: Conversation) -> str | None:
+        for binding in await self.bindings(conv):
+            if binding.visible:
+                return binding.external_id
+        return None
+
+    async def _chat_name(self, conv: Conversation) -> str:
+        """What ``[[…]]`` to the chat says: the file's stem when it has one."""
+        window = await self._window_of(conv)
+        if window and window.endswith(".md"):
+            return window.rsplit("/", 1)[-1][: -len(".md")]
+        return await self.implied_title(conv) or conv.external_id
+
+    async def _close_after_turn(self, conv: Conversation) -> None:
+        for _ in range(_CLOSE_TRIES):
+            if await self.busy(conv):
+                await asyncio.sleep(_CLOSE_WAIT)
+                continue
+            try:
+                await self.distill(conv, reason="close_chat")
+            except RuntimeError as exc:
+                _log.info("closing %s: %s, retrying", conv.external_id, exc)
+                await asyncio.sleep(_CLOSE_WAIT)
+                continue
+            except Exception:  # noqa: BLE001
+                _log.exception("closing %s after its turn failed", conv.external_id)
+            return
+        _log.warning("closing %s: still busy, giving up", conv.external_id)
+
+    # ---- turn hooks ------------------------------------------------------
+
+    async def _before_turn(self, conv: Conversation) -> str | None:
+        """Snapshot the capped file so a too-long rewrite can be bounced."""
+        cap = LineCap.from_flags(conv.flags.get("line_cap"))
+        if cap is None:
+            return None
+        try:
+            return cap.path.read_text(encoding="utf-8") if cap.path.exists() else ""
+        except OSError:
+            _log.exception("line cap: cannot read %s", cap.path)
+            return None
+
+    async def _after_turn(self, conv: Conversation, before: str | None) -> None:
+        row = await self.get_row(cast("int", conv.id))
+        if row is None:
+            return
+        if row.kind == "deep" and row.flags.get("close_requested"):
+            task = asyncio.create_task(self._close_after_turn(row))
+            self._tasks.add(task)
+            task.add_done_callback(self._tasks.discard)
+        cap = LineCap.from_flags(row.flags.get("line_cap"))
+        if cap is not None and before is not None:
+            await self._enforce_cap(row, cap, before)
+
+    async def _enforce_cap(self, conv: Conversation, cap: LineCap, before: str) -> None:
+        if not cap.path.exists():
+            return
+        after = cap.path.read_text(encoding="utf-8")
+        lines = sum(1 for line in after.splitlines() if line.strip())
+        if lines <= cap.max_lines:
+            return
+        if before:
+            cap.path.write_text(before, encoding="utf-8")
+        else:
+            cap.path.unlink()
+        attempts = int(conv.flags.get("line_cap_attempts", 0) or 0) + 1
+        await self.set_flags(conv, {"line_cap_attempts": attempts})
+        self._bus.publish(
+            "line_cap.bounced",
+            conversation_id=conv.external_id,
+            path=str(cap.path),
+            lines=lines,
+            max_lines=cap.max_lines,
+            attempt=attempts,
+        )
+        _log.warning(
+            "line cap: %s came back with %d lines (cap %d), restored; attempt %d",
+            cap.path,
+            lines,
+            cap.max_lines,
+            attempts,
+        )
+        if attempts > _CAP_TRIES:
+            return
+        await self.inject(
+            conv,
+            self._texts.too_long.format(
+                name=cap.path.name, lines=lines, max_lines=cap.max_lines
+            ),
+            urgency="urgent",
+            origin="потолок",
+            interrupt=False,
+        )
+
     async def reparent(self, conv: Conversation, parent: Conversation) -> Conversation:
         async def apply(row: Conversation) -> None:
             row.parent_id = parent.id
@@ -1020,6 +1335,7 @@ class Conversations:
             runner.started_at = datetime.now(UTC)
             runner.tools = {}
             await self._mark_running(conv, turn_id)
+            before = await self._before_turn(conv)
             self._bus.publish(
                 "turn.start",
                 conversation_id=conv.external_id,
@@ -1053,6 +1369,11 @@ class Conversations:
             finally:
                 runner.turn_id = None
                 await self._mark_done(conv, capture, cut=cut)
+                if stop != "error":
+                    try:
+                        await self._after_turn(conv, before)
+                    except Exception:  # noqa: BLE001
+                        _log.exception("after-turn hook on %s failed", conv.external_id)
                 self._bus.publish(
                     "turn.end",
                     conversation_id=conv.external_id,
@@ -1177,7 +1498,12 @@ class Conversations:
         }
 
     async def _copy_session(
-        self, conv: Conversation, *, window: int | None, strip_tools: bool
+        self,
+        conv: Conversation,
+        *,
+        window: int | None,
+        strip_tools: bool,
+        agent: str | None = None,
     ) -> str:
         if conv.session_id is None:
             msg = f"conversation {conv.external_id} has no session to copy"
@@ -1186,21 +1512,26 @@ class Conversations:
         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)
+        source = self._claude_agent(conv.agent_name)
+        target = self._claude_agent(agent) if agent else source
         forked = await fork_session_via_store(
-            self._store, conv.session_id, directory=str(agent.cwd)
+            self._store, conv.session_id, directory=str(source.cwd)
         )
-        if window is not None or strip_tools:
-            key = {
-                "project_key": project_key_for_directory(str(agent.cwd)),
+        source_key = {
+            "project_key": project_key_for_directory(str(source.cwd)),
             "session_id": forked.session_id,
         }
-            entries = await self._store.load(cast("Any", key)) or []
+        target_key = {
+            "project_key": project_key_for_directory(str(target.cwd)),
+            "session_id": forked.session_id,
+        }
+        if window is not None or strip_tools or target_key != source_key:
+            entries = await self._store.load(cast("Any", source_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))
+            await self._store.delete(cast("Any", source_key))
+            await self._store.append(cast("Any", target_key), cast("Any", trimmed))
         _log.info(
             "forked session %s -> %s (window=%s, strip_tools=%s)",
             conv.session_id,

src/beaver_gateway/core/distill.py

@@ -0,0 +1,197 @@
+"""Closing a deep chat: the digest, the index, the file line cap (§6.4, §8.4).
+
+The gateway knows no path by itself (§0.8): ``Distiller`` from ``config.py``
+names the agent, says where digests land and where the index lives, and
+the distiller writes the file on its own. What the gateway does is check that a file
+with a valid frontmatter appeared under ``Distiller.dir`` during the fork
+turn, put one line into the index, and cap the merge text at
+``SUMMARY_LINES``. ``LineCap`` is the same idea for a file a job rewrites
+(``состояние.md``): a result longer than the cap is bounced - the file
+goes back to what it was and the job is told to shorten.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date, datetime
+from pathlib import Path
+from typing import TYPE_CHECKING, Any
+
+import frontmatter
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable, Sequence
+
+    from beaver_gateway.storage.models import Conversation
+
+__all__ = [
+    "SUMMARY_LINES",
+    "Digest",
+    "DistillContext",
+    "Distiller",
+    "LineCap",
+    "append_index",
+    "check_digest",
+    "find_digest",
+    "index_line",
+    "trim_summary",
+    "written_paths",
+]
+
+SUMMARY_LINES = 5
+"""A merge into the master is at most this many lines (§6.4)."""
+
+
+@dataclass(frozen=True, slots=True)
+class Distiller:
+    """Who closes deep chats, where digests land, where the index lives."""
+
+    agent: str
+    dir: Path
+    index: Path
+    type: str = "выжимка"
+    """Value the ``type`` frontmatter key must carry."""
+
+    index_header: str = "# индекс\n\nстрока на выжимку: чат → его выжимка.\n"  # noqa: RUF001
+
+
+@dataclass(frozen=True, slots=True)
+class DistillContext:
+    """What the setup's distill prompt is built from."""
+
+    conversation: Conversation
+    title: str | None
+    source: str | None
+    """The window the chat lives in (a vault-relative path for markdown)."""
+
+    chat_name: str
+    """What a ``[[wikilink]]`` to the chat is called."""
+
+    memory: bool
+    reason: str
+    day: date
+
+
+@dataclass(frozen=True, slots=True)
+class Digest:
+    path: Path
+    source: str
+    date: date
+
+
+@dataclass(frozen=True, slots=True)
+class LineCap:
+    """A file a job rewrites may not exceed ``max_lines`` after its turn."""
+
+    path: Path
+    max_lines: int
+
+    def as_flags(self) -> dict[str, Any]:
+        return {"path": str(self.path), "max_lines": self.max_lines}
+
+    @classmethod
+    def from_flags(cls, value: Any) -> LineCap | None:
+        if not isinstance(value, dict):
+            return None
+        path, max_lines = value.get("path"), value.get("max_lines")
+        if not isinstance(path, str) or not isinstance(max_lines, int):
+            return None
+        return cls(path=Path(path), max_lines=max_lines)
+
+
+def written_paths(messages: Iterable[dict[str, Any]]) -> list[Path]:
+    """Paths the turn's ``Write``/``Edit`` calls targeted, in order."""
+    out: list[Path] = []
+    for message in messages:
+        content = message.get("content")
+        if not isinstance(content, list):
+            continue
+        for block in content:
+            if not isinstance(block, dict) or block.get("type") != "tool_use":
+                continue
+            if block.get("name") not in ("Write", "Edit", "MultiEdit"):
+                continue
+            target = (block.get("input") or {}).get("file_path")
+            if isinstance(target, str) and target:
+                out.append(Path(target))
+    return out
+
+
+def find_digest(
+    digests: Distiller, *, since: datetime, written: Sequence[Path] = ()
+) -> Path | None:
+    """The digest the fork turn produced.
+
+    A written path under ``dir`` first, otherwise the newest file there
+    modified since ``since``.
+    """
+    root = digests.dir.resolve()
+    for path in reversed(list(written)):
+        candidate = path if path.is_absolute() else digests.dir / path
+        try:
+            inside = candidate.resolve().relative_to(root)
+        except (OSError, ValueError):
+            continue
+        if (digests.dir / inside).is_file():
+            return digests.dir / inside
+    if not digests.dir.is_dir():
+        return None
+    stamp = since.timestamp() - 1
+    fresh = [
+        p
+        for p in digests.dir.glob("*.md")
+        if p.is_file() and p.stat().st_mtime >= stamp
+    ]
+    return max(fresh, key=lambda p: p.stat().st_mtime) if fresh else None
+
+
+def check_digest(path: Path, digests: Distiller) -> Digest | str:
+    """The digest's frontmatter parsed, or why it is not a digest."""
+    try:
+        post = frontmatter.load(str(path))
+    except (OSError, ValueError) as exc:
+        return f"не читается: {exc}"
+    meta = post.metadata
+    if meta.get("type") != digests.type:
+        return f"`type` должен быть `{digests.type}`, не {meta.get('type')!r}"
+    source = meta.get("source")
+    if not isinstance(source, str) or not source.strip():
+        return "`source` пустой"
+    when = meta.get("date")
+    if isinstance(when, datetime):
+        when = when.date()
+    elif isinstance(when, str):
+        try:
+            when = date.fromisoformat(when.strip())
+        except ValueError:
+            return f"`date` не дата: {when!r}"
+    if not isinstance(when, date):
+        return "`date` отсутствует"
+    if not post.content.strip():
+        return "тело пустое"
+    return Digest(path=path, source=source.strip(), date=when)
+
+
+def index_line(digest: Digest, chat_name: str) -> str:
+    return f"- {digest.date.isoformat()} [[{chat_name}]] → [[{digest.path.stem}]]"
+
+
+def append_index(digests: Distiller, line: str) -> None:
+    index = digests.index
+    text = index.read_text(encoding="utf-8") if index.exists() else ""
+    if line in text.splitlines():
+        return
+    if not text:
+        text = digests.index_header
+    if not text.endswith("\n"):
+        text += "\n"
+    index.parent.mkdir(parents=True, exist_ok=True)
+    index.write_text(text + line + "\n", encoding="utf-8")
+
+
+def trim_summary(text: str, limit: int = SUMMARY_LINES) -> tuple[str, bool]:
+    """The merge text cut to ``limit`` non-empty lines; ``True`` if it was."""
+    lines = [line.rstrip() for line in text.strip().splitlines() if line.strip()]
+    if len(lines) <= limit:
+        return "\n".join(lines), False
+    return "\n".join(lines[:limit]), True

src/beaver_gateway/core/gateway_tools.py

@@ -26,7 +26,7 @@ __all__ = ["SERVER_NAME", "TOOL_NAMES", "build_tool_server"]
 _log = logging.getLogger("beaver_gateway.core.gateway_tools")
 
 SERVER_NAME = "gateway"
-TOOL_NAMES = ("read_conversation", "spawn", "say", "schedule", "inject")
+TOOL_NAMES = ("read_conversation", "spawn", "say", "schedule", "inject", "close_chat")
 
 
 def build_tool_server(
@@ -171,7 +171,23 @@ def _tools(conversations: Conversations, key: str) -> list[SdkMcpTool[Any]]:
         )
         return _text(f"queued #{item.id}")
 
-    return [read_conversation, spawn, say, schedule, inject]
+    @tool(
+        "close_chat",
+        "Close this deep chat once the current reply is finished: the "
+        "distiller forks it, writes the digest and hands a short merge to the "
+        "master. Call it when the human says the discussion is over; the chat "
+        "file stays as it is.",
+        {"type": "object", "properties": {}},
+    )
+    async def close_chat(_args: dict[str, Any]) -> dict[str, Any]:
+        conv = await current()
+        try:
+            await conversations.request_close(conv)
+        except ValueError as exc:
+            return _error(str(exc))
+        return _text("ok: the chat closes after this reply")
+
+    return [read_conversation, spawn, say, schedule, inject, close_chat]
 
 
 def _text(text: str) -> dict[str, Any]:

src/beaver_gateway/core/registry.py

@@ -17,6 +17,7 @@ if TYPE_CHECKING:
 
     from beaver_gateway.agents.base import BaseAgent
     from beaver_gateway.core.conversations import ConversationTexts
+    from beaver_gateway.core.distill import Distiller
     from beaver_gateway.core.rotation import RotationPolicy
     from beaver_gateway.core.scheduler import Budget, Job
     from beaver_gateway.core.watch import VaultWatch
@@ -95,6 +96,8 @@ class Gateway:
     """Vault watcher feeding the envelope (§3.5, §4.6); ``None`` = no vault block."""
     budget: Budget | None = None
     """Subscription window past which non-critical jobs wait (§4.5)."""
+    distiller: Distiller | None = None
+    """Who closes deep chats and where the digests and the index live (§8.4)."""
     tz: str = "UTC"
     """Local zone for the envelope clock and the rotation hour."""
     host: str = "0.0.0.0"  # noqa: S104

src/beaver_gateway/core/scheduler.py

@@ -36,7 +36,8 @@ if TYPE_CHECKING:
     from pgqueuer.ports.driver import Driver
     from starlette.requests import Request
 
-    from beaver_gateway.core.conversations import Conversations
+    from beaver_gateway.core.conversations import Conversations, DistillResult
+    from beaver_gateway.core.distill import LineCap
     from beaver_gateway.core.injects import Priority
     from beaver_gateway.core.rotation import Rotation
     from beaver_gateway.storage.models import Conversation
@@ -98,12 +99,45 @@ class JobRun:
         return True
 
     async def spawn_job(
-        self, *, agent: str, text: str, title: str | None = None
+        self,
+        *,
+        agent: str,
+        text: str,
+        title: str | None = None,
+        line_cap: LineCap | None = None,
     ) -> Conversation:
+        """A headless job turn; ``line_cap`` bounces a rewrite past the cap."""
         return await self.conversations.spawn(
-            kind="job", agent=agent, seed="brief", text=text, title=title, origin="job"
+            kind="job",
+            agent=agent,
+            seed="brief",
+            text=text,
+            title=title,
+            origin="job",
+            flags={"line_cap": line_cap.as_flags()} if line_cap else None,
         )
 
+    async def close_idle(
+        self,
+        *,
+        kind: str = "deep",
+        days: int = 2,
+        limit: int = 3,
+        since: datetime | None = None,
+    ) -> list[DistillResult]:
+        """§4.5: close chats quiet for ``days``, at most ``limit`` per run."""
+        out: list[DistillResult] = []
+        for conv in await self.conversations.idle(
+            kind=kind, days=days, since=since, limit=limit
+        ):
+            try:
+                out.append(
+                    await self.conversations.distill(conv, reason=f"idle {days}d")
+                )
+            except Exception:  # noqa: BLE001
+                _log.exception("closing idle %s failed", conv.external_id)
+        return out
+
     async def retry_in(self, delay: timedelta) -> None:
         await self.scheduler.trigger(
             self.job, self.payload, delay=delay, trigger=self.trigger

src/beaver_gateway/frontends/api/frontend.py

@@ -441,6 +441,36 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa
             "text": result.text,
         }
 
+    @app.post("/conversations/{public_id}/close")
+    async def post_close(public_id: str, request: Request) -> dict[str, Any]:
+        token = await require_token(request, runtime, scope=SCOPE)
+        conv = await conv_of(public_id)
+        if conv.kind != "deep":
+            closed = await conversations.close(conv)
+            return {"id": closed.external_id, "status": closed.status}
+        try:
+            result = await conversations.distill(conv, reason="api")
+        except (ValueError, LookupError) as exc:
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
+        except RuntimeError as exc:
+            raise HTTPException(status.HTTP_409_CONFLICT, str(exc)) from exc
+        await audit.log(
+            runtime,
+            actor=f"token:{token}",
+            kind="api_close",
+            agent_name=conv.agent_name,
+            conversation=conv.external_id,
+            digest=str(result.digest.path) if result.digest else None,
+        )
+        return {
+            "id": conv.external_id,
+            "status": "closed",
+            "fork": result.fork.external_id,
+            "text": result.text,
+            "digest": str(result.digest.path) if result.digest else None,
+            "error": result.error,
+        }
+
     @app.post("/conversations/{public_id}/fork")
     async def post_fork(public_id: str, request: Request) -> dict[str, Any]:
         await require_token(request, runtime, scope=SCOPE)

tests/test_api.py

@@ -420,3 +420,31 @@ async def test_pre_sdk_rows_show_canonical_history_and_a_title(world: World) ->
     assert shown["title"] == "так смотри, план на май" and shown["session_id"] is None
     history = await api.get(f"/conversations/{conv.external_id}/history")
     assert [m["role"] for m in history["messages"]] == ["user", "assistant"]
+
+
+async def test_close_distills_a_deep_chat(world: World) -> None:
+    from test_distill import DistillerClient, deep_chat, distiller
+
+    config = distiller(world, DistillerClient)
+    api = Api(world)
+    master = await world.conversations.create(kind="master", agent="a", origin="test")
+    chat = await deep_chat(world)
+    res = await api.http.post(
+        f"/conversations/{chat.external_id}/close", headers=HEADERS
+    )
+    assert res.status_code == 200, res.text
+    body = res.json()
+    assert body["status"] == "closed" and body["error"] is None
+    assert body["digest"] == str(config.dir / "2026-08-29 - тема.md")
+    assert body["text"].count("\n") == 2
+    assert (await world.conversations.get(chat.external_id)).status == "closed"
+    assert (await world.conversations.queue.recent(master.id))[0].origin == "выжимка"
+    again = await api.http.post(
+        f"/conversations/{chat.external_id}/close", headers=HEADERS
+    )
+    assert again.status_code == 400
+    branch = await world.conversations.create(kind="branch", agent="a", origin="test")
+    plain = await api.http.post(
+        f"/conversations/{branch.external_id}/close", headers=HEADERS
+    )
+    assert plain.status_code == 200 and plain.json()["status"] == "closed"

tests/test_distill.py

@@ -0,0 +1,419 @@
+import asyncio
+import uuid
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+from claude_agent_sdk import (
+    AssistantMessage,
+    ResultMessage,
+    TextBlock,
+    ToolResultBlock,
+    ToolUseBlock,
+    UserMessage,
+    project_key_for_directory,
+)
+from test_conversations import ScriptedClient, World, world
+
+from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions
+from beaver_gateway.backends.claude_sdk import ClaudeSdkBackend
+from beaver_gateway.core.conversations import ConversationTexts
+from beaver_gateway.core.distill import (
+    Distiller,
+    DistillContext,
+    LineCap,
+    check_digest,
+    trim_summary,
+)
+from beaver_gateway.core.gateway_tools import _tools, build_tool_server
+from beaver_gateway.core.registry import AgentRegistry
+from beaver_gateway.core.scheduler import Job, JobRun, Scheduler
+from beaver_gateway.core.transcript import build_entries
+from beaver_gateway.storage.models import Conversation
+
+__all__ = ["world"]
+
+DIGEST = """---
+type: выжимка
+source: "[[{chat}]]"
+date: 2026-08-29
+---
+# тема
+
+## контекст
+зачем открывали
+
+## решили
+- одно
+"""
+
+
+class DistillerClient(ScriptedClient):
+    """The distiller: writes a digest file when told to, answers in N lines."""
+
+    digest_dir: Path | None = None
+    lines = 3
+    write = True
+    frontmatter = DIGEST
+
+    async def receive_response(self):
+        prompt = self.prompts[-1]
+        written: list[ToolUseBlock] = []
+        if self.write and self.digest_dir is not None and "файл не пиши" not in prompt:
+            chat = prompt.split("«", 1)[1].split("»", 1)[0] if "«" in prompt else "чат"
+            path = self.digest_dir / "2026-08-29 - тема.md"
+            path.write_text(self.frontmatter.format(chat=chat), encoding="utf-8")
+            written.append(
+                ToolUseBlock(id="w1", name="Write", input={"file_path": str(path)})
+            )
+        for block in written:
+            yield AssistantMessage(content=[block], model="m")
+            yield UserMessage(content=[ToolResultBlock(tool_use_id=block.id)])
+        text = "\n".join(f"строка {i} про {prompt[:20]!r}" for i in range(self.lines))
+        yield AssistantMessage(content=[TextBlock(text=text)], model="m")
+        yield ResultMessage(
+            subtype="success",
+            duration_ms=1,
+            duration_api_ms=1,
+            is_error=False,
+            num_turns=1,
+            session_id=self.session_id,
+            stop_reason="end_turn",
+            total_cost_usd=0.0,
+            usage={"input_tokens": 1, "output_tokens": 1},
+        )
+
+
+class ClosingClient(ScriptedClient):
+    """A deep chat that calls ``close_chat`` inside its reply."""
+
+    conversations: Any = None
+    key = ""
+
+    async def receive_response(self):
+        prompt = self.prompts[-1]
+        if "обсудили" in prompt:
+            assert "gateway" in self.options.mcp_servers
+            tools = _tools(ClosingClient.conversations, ClosingClient.key)
+            close = next(t for t in tools if t.name == "close_chat")
+            result = await close.handler({})
+            assert "closes after this reply" in result["content"][0]["text"]
+            yield AssistantMessage(
+                content=[
+                    ToolUseBlock(id="c1", name="mcp__gateway__close_chat", input={})
+                ],
+                model="m",
+            )
+        yield AssistantMessage(content=[TextBlock(text=f"ok:{prompt}")], model="m")
+        yield ResultMessage(
+            subtype="success",
+            duration_ms=1,
+            duration_api_ms=1,
+            is_error=False,
+            num_turns=1,
+            session_id=self.session_id,
+            stop_reason="end_turn",
+            total_cost_usd=0.0,
+            usage={"input_tokens": 1, "output_tokens": 1},
+        )
+
+
+class CapClient(ScriptedClient):
+    """A job that rewrites the capped file with ``lines`` lines."""
+
+    target: Path | None = None
+    lines = 70
+
+    async def receive_response(self):
+        assert self.target is not None
+        self.target.write_text(
+            "\n".join(f"- строка {i}" for i in range(self.lines)), encoding="utf-8"
+        )
+        CapClient.lines = 10
+        async for event in super().receive_response():
+            yield event
+
+
+def distiller(world: World, client: type[ScriptedClient]) -> Distiller:
+    vault = world.root / "vault"
+    digests = vault / "мета" / "бобер" / "выжимки"
+    digests.mkdir(parents=True)
+    agent = ClaudeAgent(
+        name="x",
+        model="m",
+        system_prompt="distill",
+        cwd=vault,
+        kinds=("fork", "job"),
+        options=ClaudeOptions(
+            effort="medium", tools=("Read", "Write"), include_partial_messages=False
+        ),
+    )
+    backend = ClaudeSdkBackend(
+        agent=agent,
+        mcp_internal_urls={},
+        session_store=world.store,
+        client_factory=client,
+        work_dir=world.root / "work",
+        pool=world.pool,
+    )
+    world.conversations._agents = AgentRegistry(  # noqa: SLF001
+        [world.agent, world.deep_agent, agent]
+    )
+    world.conversations._backends["x"] = backend  # noqa: SLF001
+    config = Distiller(agent="x", dir=digests, index=digests.parent / "индекс.md")
+    world.conversations._distiller = config  # noqa: SLF001
+    DistillerClient.digest_dir = digests
+    DistillerClient.lines = 3
+    DistillerClient.write = True
+    DistillerClient.frontmatter = DIGEST
+    return config
+
+
+async def deep_chat(world: World, name: str = "2026-08-20 - тема чата") -> Conversation:
+    sid = str(uuid.uuid4())
+    await world.store.append(
+        world.key(sid),
+        build_entries(
+            [
+                {"role": "user", "content": "q"},
+                {
+                    "role": "assistant",
+                    "content": [
+                        {"type": "tool_use", "id": "t1", "name": "Read", "input": {}}
+                    ],
+                },
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "tool_result", "tool_use_id": "t1", "content": "x"}
+                    ],
+                },
+                {"role": "assistant", "content": "a"},
+            ],
+            session_id=sid,
+            cwd=str(world.root),
+            model="m",
+        ),
+    )
+    conv = await world.conversations.create(
+        kind="deep", agent="d", origin="markdown", session_id=sid
+    )
+    await world.conversations.bind(
+        conv, frontend="markdown", external_id=f"2026-08/{name}.md"
+    )
+    return conv
+
+
+async def test_distill_writes_the_digest_indexes_it_and_merges_short(
+    world: World,
+) -> None:
+    config = distiller(world, DistillerClient)
+    prompts: list[DistillContext] = []
+
+    def distill_prompt(ctx: DistillContext) -> str:
+        prompts.append(ctx)
+        return f"Чат «{ctx.chat_name}» закрыт ({ctx.reason})."
+
+    world.conversations._texts = ConversationTexts(distill=distill_prompt)  # noqa: SLF001
+    master = await world.conversations.create(kind="master", agent="a", origin="test")
+    chat = await deep_chat(world)
+
+    result = await world.conversations.distill(chat, reason="api")
+
+    assert prompts[0].chat_name == "2026-08-20 - тема чата"
+    assert prompts[0].memory is True
+    assert result.digest is not None and result.error is None
+    assert result.digest.path == config.dir / "2026-08-29 - тема.md"
+    assert result.digest.source == "[[2026-08-20 - тема чата]]"
+    index = config.index.read_text(encoding="utf-8")
+    assert index.startswith("# индекс")
+    assert "- 2026-08-29 [[2026-08-20 - тема чата]] → [[2026-08-29 - тема]]" in index
+    assert result.text.count("\n") == 2 and not result.trimmed
+    closed = await world.conversations.get(chat.external_id)
+    assert closed.status == "closed"
+    assert closed.flags["digest"] == str(result.digest.path)
+    assert closed.flags["closed_reason"] == "api"
+    fork = await world.conversations.get(result.fork.external_id)
+    assert fork.kind == "fork" and fork.agent_name == "x" and fork.status == "closed"
+    items = await world.conversations.queue.recent(master.id)
+    assert items[0].origin == "выжимка"
+    assert items[0].text.startswith(
+        "Закрыт глубокий чат [[2026-08-20 - тема чата]], выжимка [[2026-08-29 - тема]]."
+    )
+    assert items[0].text.endswith(result.text)
+    forked = ScriptedClient.instances[-1]
+    assert forked.options.resume is not None
+    vault_key = {
+        "project_key": project_key_for_directory(str(config.dir.parents[2])),
+        "session_id": forked.options.resume,
+    }
+    entries = await world.store.load(vault_key)
+    assert entries
+    assert all(
+        not isinstance(e.get("message", {}).get("content"), list)
+        or all(b.get("type") == "text" for b in e["message"]["content"])
+        for e in entries
+    )
+
+
+async def test_memory_off_merges_without_a_file(world: World) -> None:
+    config = distiller(world, DistillerClient)
+    master = await world.conversations.create(kind="master", agent="a", origin="test")
+    chat = await deep_chat(world)
+    await world.conversations.set_flags(chat, {"memory": False})
+
+    result = await world.conversations.distill(chat, reason="api")
+
+    assert result.digest is None and result.error is None
+    assert list(config.dir.iterdir()) == []
+    assert not config.index.exists()
+    assert (await world.conversations.get(chat.external_id)).status == "closed"
+    items = await world.conversations.queue.recent(master.id)
+    assert len(items) == 1 and items[0].text.endswith(result.text)
+    assert ", выжимка" not in items[0].text
+    assert "файл не пиши" in ScriptedClient.instances[-1].prompts[0]
+
+
+async def test_bad_frontmatter_and_long_merge_are_reported(world: World) -> None:
+    distiller(world, DistillerClient)
+    DistillerClient.frontmatter = "---\ntype: заметка\n---\n# тема\n"
+    DistillerClient.lines = 9
+    await world.conversations.create(kind="master", agent="a", origin="test")
+    chat = await deep_chat(world)
+
+    result = await world.conversations.distill(chat, reason="idle 2d")
+
+    assert result.digest is None
+    assert result.error is not None and "`type`" in result.error
+    assert result.trimmed and result.text.count("\n") == 4
+    closed = await world.conversations.get(chat.external_id)
+    assert closed.status == "closed"
+    assert closed.flags["digest_error"] == result.error
+
+
+def test_check_digest_rejects_what_is_not_a_digest(tmp_path: Path) -> None:
+    config = Distiller(agent="x", dir=tmp_path, index=tmp_path / "i.md")
+    path = tmp_path / "d.md"
+    path.write_text("---\ntype: выжимка\nsource: ''\ndate: 2026-08-29\n---\nx\n")
+    assert check_digest(path, config) == "`source` пустой"
+    path.write_text("---\ntype: выжимка\nsource: '[[a]]'\ndate: вчера\n---\nx\n")
+    assert "`date`" in check_digest(path, config)
+    path.write_text("---\ntype: выжимка\nsource: '[[a]]'\ndate: 2026-08-29\n---\n\n")
+    assert check_digest(path, config) == "тело пустое"
+    path.write_text("---\ntype: выжимка\nsource: '[[a]]'\ndate: 2026-08-29\n---\nx\n")
+    digest = check_digest(path, config)
+    assert not isinstance(digest, str) and digest.date.isoformat() == "2026-08-29"
+    assert trim_summary("a\n\nb\nc\nd\ne\nf") == ("a\nb\nc\nd\ne", True)
+
+
+async def test_close_chat_tool_closes_after_the_reply(world: World) -> None:
+    distiller(world, DistillerClient)
+    world.deep_agent = ClaudeAgent(
+        name="d",
+        model="m",
+        system_prompt="deep",
+        cwd=world.root,
+        gateway_tools=("close_chat",),
+    )
+    world.conversations._agents = AgentRegistry(  # noqa: SLF001
+        [world.agent, world.deep_agent, world.conversations._agents.get("x")]  # noqa: SLF001
+    )
+    world.conversations._backends["d"] = ClaudeSdkBackend(  # noqa: SLF001
+        agent=world.deep_agent,
+        mcp_internal_urls={},
+        session_store=world.store,
+        client_factory=ClosingClient,
+        work_dir=world.root / "work",
+        pool=world.pool,
+        tool_server=lambda key, _kind: build_tool_server(
+            world.conversations, conversation_key=key, names=("close_chat",)
+        ),
+    )
+    master = await world.conversations.create(kind="master", agent="a", origin="test")
+    chat = await deep_chat(world)
+    ClosingClient.conversations = world.conversations
+    ClosingClient.key = chat.external_id
+
+    await world.conversations.post(chat, "ок, обсудили")
+    await world.settle(chat, 1)
+    for _ in range(100):
+        row = await world.conversations.get(chat.external_id)
+        if row.status == "closed":
+            break
+        await asyncio.sleep(0.05)
+    else:
+        raise AssertionError("chat did not close")
+
+    assert row.flags["closed_reason"] == "close_chat"
+    assert row.flags["digest"] is not None
+    items = await world.conversations.queue.recent(master.id)
+    assert items[0].origin == "выжимка"
+
+
+async def test_idle_picks_quiet_chats_after_launch_at_most_limit(world: World) -> None:
+    distiller(world, DistillerClient)
+    now = datetime.now(UTC)
+    launch = now - timedelta(days=10)
+
+    async def aged(name: str, days: float) -> Conversation:
+        conv = await deep_chat(world, name)
+
+        async def apply(row: Conversation) -> None:
+            row.last_activity_at = now - timedelta(days=days)
+
+        return await world.conversations._update(conv, apply)  # noqa: SLF001
+
+    old = await aged("old", 30)
+    a = await aged("a", 5)
+    b = await aged("b", 4)
+    c = await aged("c", 3)
+    d = await aged("d", 2.5)
+    fresh = await aged("fresh", 1)
+    no_session = await world.conversations.create(kind="deep", agent="d", origin="t")
+
+    idle = await world.conversations.idle(kind="deep", days=2, since=launch)
+    assert [x.external_id for x in idle] == [
+        a.external_id,
+        b.external_id,
+        c.external_id,
+        d.external_id,
+    ]
+    assert old.external_id not in {x.external_id for x in idle}
+    assert fresh.external_id not in {x.external_id for x in idle}
+    assert no_session.external_id not in {x.external_id for x in idle}
+
+    scheduler = Scheduler(conversations=world.conversations)
+    run = JobRun(Job("закрытие", lambda _run: asyncio.sleep(0)), "cron", {}, scheduler)
+    closed = await run.close_idle(kind="deep", days=2, limit=3, since=launch)
+    assert [r.conversation.external_id for r in closed] == [
+        a.external_id,
+        b.external_id,
+        c.external_id,
+    ]
+    assert (await world.conversations.get(d.external_id)).status == "open"
+    assert len(await world.conversations.idle(kind="deep", days=2, since=launch)) == 1
+
+
+async def test_line_cap_bounces_a_long_rewrite_and_asks_to_shorten(
+    world: World,
+) -> None:
+    distiller(world, CapClient)
+    state = world.root / "vault" / "состояние.md"
+    state.write_text("# состояние\n- было так\n", encoding="utf-8")
+    CapClient.target = state
+    CapClient.lines = 70
+    scheduler = Scheduler(conversations=world.conversations)
+    run = JobRun(Job("память", lambda _run: asyncio.sleep(0)), "cron", {}, scheduler)
+
+    job = await run.spawn_job(
+        agent="x", text="перепиши", line_cap=LineCap(state, max_lines=60)
+    )
+    await world.settle(job, 2)
+
+    client = ScriptedClient.instances[-1]
+    assert len(client.prompts) == 2
+    assert "70 строк при потолке 60" in client.prompts[1]
+    assert "[инжект: потолок" in client.prompts[1]
+    assert state.read_text(encoding="utf-8").count("\n") == 10 - 1
+    row = await world.conversations.get(job.external_id)
+    assert row.flags["line_cap_attempts"] == 1

ui/src/lib/api/client.ts

@@ -234,6 +234,17 @@ export class ApiClient {
     return this.post(`/api/conversations/${id}/merge`);
   }
 
+  close(id: string): Promise<{
+    id: string;
+    status: string;
+    fork?: string;
+    text?: string;
+    digest?: string | null;
+    error?: string | null;
+  }> {
+    return this.post(`/api/conversations/${id}/close`);
+  }
+
   bind(
     id: string,
     frontend: string,

ui/src/lib/panel/conversation-header.svelte

@@ -87,6 +87,15 @@
     }
   }
 
+  async function distill() {
+    const result = await client.close(info.id);
+    if (result.error) {
+      toast.warning(`Digest: ${result.error}`);
+    } else if (result.digest) {
+      toast.message(`Digest: ${result.digest}`);
+    }
+  }
+
   function copyId() {
     navigator.clipboard
       .writeText(info.id)
@@ -153,6 +162,11 @@
               Merge into parent
             </DropdownMenu.Item>
           {/if}
+          {#if info.kind === "deep" && info.status === "open"}
+            <DropdownMenu.Item onclick={() => run(distill, "Chat closed")}>
+              Close with digest
+            </DropdownMenu.Item>
+          {/if}
           {#if info.status === "open"}
             <DropdownMenu.Item
               onclick={() =>