fix(telegram,conversations): attachments by day with sweep in both modes, merge renames the topic, gone topics unbind, failures visible

src/beaver_gateway/agents/claude.py

@@ -3,8 +3,10 @@
 The system prompt is ``system_prompt`` verbatim or, per conversation kind,
 the granules named in ``prompts`` assembled at every session spawn (see
 ``core/prompt.py``). ``skill_sets`` are directories of ``<skill>/SKILL.md``
-folders; each becomes a local SDK plugin. Nothing from disk is loaded
+folders; each becomes a local SDK plugin, either the same tuple for every
-otherwise: the adapter runs with ``setting_sources=[]``.
+kind or a ``SkillSets`` with a tuple per kind (§4.3: the master never sees
+what a branch opens). Nothing from disk is loaded otherwise: the adapter
+runs with ``setting_sources=[]``.
 """
 
 from __future__ import annotations
@@ -19,7 +21,7 @@ from beaver_gateway.core.kinds import KINDS, Kind
 from beaver_gateway.core.policy import PolicyRule  # noqa: TC001 - pydantic runtime
 from beaver_gateway.core.prompt import PromptSource  # noqa: TC001 - pydantic runtime
 
-__all__ = ["ClaudeAgent", "ClaudeOptions", "Prompts"]
+__all__ = ["ClaudeAgent", "ClaudeOptions", "Prompts", "SkillSets"]
 
 
 class ClaudeOptions(BaseModel):
@@ -70,6 +72,21 @@ class Prompts(BaseModel):
         return tuple(k for k in KINDS if self.for_kind(k) is not None)
 
 
+class SkillSets(BaseModel):
+    """Skill-set directories per conversation kind; ``None`` means none."""
+
+    model_config = ConfigDict(frozen=True)
+
+    master: tuple[Path, ...] | None = None
+    branch: tuple[Path, ...] | None = None
+    deep: tuple[Path, ...] | None = None
+    job: tuple[Path, ...] | None = None
+    fork: tuple[Path, ...] | None = None
+
+    def for_kind(self, kind: Kind) -> tuple[Path, ...]:
+        return getattr(self, kind) or ()
+
+
 class ClaudeAgent(BaseAgent):
     cwd: Path
     system_prompt: str = ""
@@ -81,7 +98,11 @@ class ClaudeAgent(BaseAgent):
     rest. Defaults to the kinds ``prompts`` covers, or ``("deep",)`` for a
     verbatim ``system_prompt``."""
 
-    skill_sets: tuple[Path, ...] = ()
+    skill_sets: tuple[Path, ...] | SkillSets = ()
+    """Skill-set directories, each a local plugin: one tuple for every kind,
+    or ``SkillSets`` to give each kind its own (a kind left ``None`` gets
+    no skills)."""
+
     gateway_tools: tuple[str, ...] = ()
     """Gateway tools exposed in-process (``read_conversation``, ``spawn``,
     ``say``, ``schedule``, ``inject``); empty = no gateway MCP server."""
@@ -106,5 +127,10 @@ class ClaudeAgent(BaseAgent):
     def prompt_for(self, kind: Kind) -> tuple[PromptSource, ...] | None:
         return self.prompts.for_kind(kind)
 
+    def skills_for(self, kind: Kind) -> tuple[Path, ...]:
+        if isinstance(self.skill_sets, SkillSets):
+            return self.skill_sets.for_kind(kind)
+        return self.skill_sets
+
     def serves(self, kind: str) -> bool:
         return kind in self.kinds

src/beaver_gateway/backends/claude_sdk.py

@@ -108,6 +108,7 @@ if TYPE_CHECKING:
     from beaver_gateway.agents.base import BaseAgent
     from beaver_gateway.agents.claude import ClaudeAgent
     from beaver_gateway.core.events import MessageStreamEvent
+    from beaver_gateway.core.kinds import Kind
 
 
 _log = logging.getLogger("beaver_gateway.backends.claude_sdk")
@@ -574,8 +575,9 @@ class ClaudeSdkBackend:
         if self._runner.home is not None:
             env["HOME"] = str(self._runner.home)
             env.setdefault("CLAUDE_CONFIG_DIR", str(self._runner.home / ".claude"))
-        plugins = self._plugins()
+        kind = as_kind(spec.kind)
-        sources = agent.prompt_for(as_kind(spec.kind))
+        plugins = self._plugins(kind)
+        sources = agent.prompt_for(kind)
         system_prompt = (
             prompt_assembly.assemble(sources) if sources else agent.system_prompt
         )
@@ -683,10 +685,10 @@ class ClaudeSdkBackend:
 
         return {"PreToolUse": [HookMatcher(hooks=[cast("Any", pre_tool_use)])]}
 
-    def _plugins(self) -> list[dict[str, str]]:
+    def _plugins(self, kind: Kind) -> list[dict[str, str]]:
         plugins: list[dict[str, str]] = []
-        root = self._work_dir / "plugins" / self._agent.name
+        root = self._work_dir / "plugins" / self._agent.name / kind
-        for raw in sorted(self._agent.skill_sets, key=str):
+        for raw in sorted(self._agent.skills_for(kind), key=str):
             source = Path(str(raw))
             name = source.name
             target = root / name

src/beaver_gateway/cli.py

@@ -210,6 +210,7 @@ async def _async_main() -> None:
             rotation=Rotation(
                 conversations, gateway.rotation or RotationPolicy(tz=gateway.tz)
             ),
+            tz=gateway.tz,
         )
         conversations.scheduler = scheduler
 

src/beaver_gateway/core/conversations.py

@@ -23,7 +23,7 @@ import logging
 import re
 import uuid
 from dataclasses import dataclass, field
-from datetime import UTC, date, datetime, timedelta
+from datetime import UTC, date, datetime, timedelta, tzinfo
 from typing import TYPE_CHECKING, Any, cast
 
 from claude_agent_sdk import (
@@ -298,6 +298,10 @@ class Conversations:
         self._tasks: set[asyncio.Task[None]] = set()
         self._idle_task: asyncio.Task[None] | None = None
 
+    @property
+    def db(self) -> Database:
+        return self._db
+
     @property
     def queue(self) -> InjectQueue:
         return self._queue
@@ -964,6 +968,7 @@ class Conversations:
         if result.text.strip():
             await self.inject(parent, result.text, urgency="normal", origin="слив")
         await self.set_status(conv, "merged")
+        await self.mark_closed(conv)
         self._bus.publish(
             "conversation.merged",
             conversation_id=conv.external_id,
@@ -1946,7 +1951,7 @@ class Conversations:
             )
 
 
-def parse_at(at: str) -> datetime:
+def parse_at(at: str, tz: tzinfo = UTC) -> datetime:
     raw = at.strip()
     match = _RELATIVE.match(raw.replace(" ", ""))
     if match:
@@ -1954,7 +1959,7 @@ def parse_at(at: str) -> datetime:
         return datetime.now(UTC) + timedelta(seconds=int(amount) * _UNITS[unit])
     parsed = datetime.fromisoformat(raw)
     if parsed.tzinfo is None:
-        parsed = parsed.astimezone()
+        parsed = parsed.replace(tzinfo=tz)
     return parsed.astimezone(UTC)
 
 

src/beaver_gateway/core/scheduler.py

@@ -14,19 +14,27 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+import functools
 import json
 import logging
+import traceback
 from dataclasses import dataclass
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime, timedelta, tzinfo
 from typing import TYPE_CHECKING, Any, cast
+from zoneinfo import ZoneInfo
 
+from croniter import croniter
 from pgqueuer import PgQueuer, Queries
+from pgqueuer.core.executors import ScheduleExecutor
 from pgqueuer.models import JobId
+from sqlalchemy import func
+from sqlmodel import col, select
 from starlette.applications import Starlette
 from starlette.responses import JSONResponse
 from starlette.routing import Route
 
 from beaver_gateway.core.conversations import parse_at
+from beaver_gateway.storage.models import JobRunRecord
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable, Coroutine, Sequence
@@ -42,12 +50,27 @@ if TYPE_CHECKING:
     from beaver_gateway.core.rotation import Rotation
     from beaver_gateway.storage.models import Conversation
 
-__all__ = ["INJECT", "Budget", "Job", "JobRun", "Scheduler"]
+__all__ = ["INJECT", "Budget", "Job", "JobRun", "LocalCron", "Scheduler", "next_run"]
 
 _log = logging.getLogger("beaver_gateway.core.scheduler")
 
 INJECT = "inject"
 RETRY = timedelta(minutes=15)
+ERROR_MAX = 2000
+
+
+def next_run(expression: str, tz: tzinfo, now: datetime | None = None) -> datetime:
+    """The cron's next fire time, read in ``tz``, returned in UTC."""
+    start = (now or datetime.now(UTC)).astimezone(tz)
+    return croniter(expression, start_time=start).get_next(datetime).astimezone(UTC)
+
+
+@dataclass
+class LocalCron(ScheduleExecutor):
+    tz: tzinfo = UTC
+
+    def get_next(self) -> datetime:
+        return next_run(self.parameters.expression, self.tz)
 
 
 @dataclass(frozen=True, slots=True)
@@ -165,9 +188,11 @@ class Scheduler:
         budget: Budget | None = None,
         rotation: Rotation | None = None,
         heartbeat: timedelta = timedelta(seconds=30),
+        tz: str = "UTC",
     ) -> None:
         self.conversations = conversations
         self.rotation = rotation
+        self.tz = ZoneInfo(tz)
         self.budget = budget or Budget()
         self._jobs = {job.name: job for job in jobs}
         self._driver = driver
@@ -229,7 +254,12 @@ class Scheduler:
 
         if job.cron:
 
-            @pgq.schedule(job.name, job.cron, clean_old=True)
+            @pgq.schedule(
+                job.name,
+                job.cron,
+                clean_old=True,
+                executor_factory=functools.partial(LocalCron, tz=self.tz),
+            )
             async def cron(_schedule: PgSchedule) -> None:
                 await self._dispatch(job, trigger="cron", payload={})
 
@@ -257,19 +287,33 @@ class Scheduler:
             return
         started = datetime.now(UTC)
         bus.publish("job.start", job=job.name, trigger=trigger)
-        status = "done"
+        status, error = "done", None
         try:
             await job.run(JobRun(job, trigger, payload, self))
-        except Exception:  # noqa: BLE001
+        except Exception as exc:  # noqa: BLE001
-            status = "failed"
+            status, error = "failed", _error_text(exc)
             _log.exception("job %s (%s) failed", job.name, trigger)
-        self._runs[job.name] = {
+        record = JobRunRecord(
-            "trigger": trigger,
+            job=job.name,
-            "started_at": started.isoformat(timespec="seconds"),
+            trigger=trigger,
-            "status": status,
+            started_at=started,
-        }
+            finished_at=datetime.now(UTC),
+            status=status,
+            error=error,
+            payload=payload,
+        )
+        await self._record(record)
+        self._runs[job.name] = self._run_public(record)
         bus.publish("job.end", job=job.name, trigger=trigger, status=status)
 
+    async def _record(self, record: JobRunRecord) -> None:
+        try:
+            async with self.conversations.db.session() as session:
+                session.add(record)
+                await session.commit()
+        except Exception:  # noqa: BLE001
+            _log.exception("job %s: run not recorded", record.job)
+
     async def trigger(
         self,
         job: Job,
@@ -325,14 +369,14 @@ class Scheduler:
         if self._queries is None:
             msg = "scheduler needs postgres; `schedule` is unavailable"
             raise RuntimeError(msg)
-        when = parse_at(at)
+        when = parse_at(at, self.tz).astimezone(self.tz)
         delay = max(when - datetime.now(UTC), timedelta(0))
         payload = _encode(
             {
                 "conversation": conv.external_id,
                 "text": text,
                 "urgency": urgency,
-                "at": when.isoformat(timespec="seconds"),
+                "at": when.isoformat(timespec="minutes"),
             }
         )
         ids = await self._queries.enqueue(
@@ -347,7 +391,7 @@ class Scheduler:
             "schedule.created",
             conversation_id=conv.external_id,
             job=job_id,
-            execute_at=when.isoformat(timespec="seconds"),
+            execute_at=when.isoformat(timespec="minutes"),
         )
         return job_id, when
 
@@ -411,16 +455,22 @@ class Scheduler:
             for row in await self._queries.peek_schedule():
                 crons[str(row.entrypoint)] = row
             queue = [
-                _job_public(row) for row in await self._queries.browse_queue(limit=200)
+                _job_public(row, self.tz)
+                for row in await self._queries.browse_queue(limit=200)
             ]
         utilization = await self.utilization()
+        last = await self._last_runs()
         return {
             "enabled": self.enabled,
+            "tz": str(self.tz),
             "utilization": utilization,
             "throttled": utilization is not None
             and utilization > self.budget.threshold,
             "threshold": self.budget.threshold,
-            "jobs": [self._job_public(job, crons.get(job.name)) for job in self.jobs],
+            "jobs": [
+                self._job_public(job, crons.get(job.name), last.get(job.name))
+                for job in self.jobs
+            ],
             "queue": queue,
         }
 
@@ -428,24 +478,60 @@ class Scheduler:
         if self._queries is None:
             return []
         rows = await self._queries.browse_queue(limit=500, entrypoints=[INJECT])
-        out = [_job_public(row) for row in rows]
+        out = [_job_public(row, self.tz) for row in rows]
         if conv is not None:
             out = [
                 j for j in out if j["payload"].get("conversation") == conv.external_id
             ]
         return out
 
-    def _job_public(self, job: Job, cron: PgSchedule | None) -> dict[str, Any]:
+    async def runs(self, name: str, limit: int = 50) -> list[dict[str, Any]]:
+        stmt = (
+            select(JobRunRecord)
+            .where(JobRunRecord.job == name)
+            .order_by(col(JobRunRecord.id).desc())
+            .limit(limit)
+        )
+        async with self.conversations.db.session() as session:
+            rows = (await session.exec(stmt)).all()
+        return [self._run_public(row) for row in rows]
+
+    async def _last_runs(self) -> dict[str, dict[str, Any]]:
+        newest = select(func.max(JobRunRecord.id)).group_by(JobRunRecord.job)
+        stmt = select(JobRunRecord).where(col(JobRunRecord.id).in_(newest))
+        try:
+            async with self.conversations.db.session() as session:
+                rows = (await session.exec(stmt)).all()
+        except Exception:  # noqa: BLE001
+            _log.exception("job runs unreadable, using memory")
+            return dict(self._runs)
+        return {row.job: self._run_public(row) for row in rows}
+
+    def _job_public(
+        self, job: Job, cron: PgSchedule | None, run: dict[str, Any] | None
+    ) -> dict[str, Any]:
         return {
             "name": job.name,
             "cron": job.cron,
             "webhook": job.webhook,
             "events": list(job.events),
             "critical": job.critical,
-            "next_run": _iso(cron.next_run) if cron is not None else None,
+            "next_run": _iso(cron.next_run, self.tz) if cron is not None else None,
-            "last_run": _iso(cron.last_run) if cron is not None else None,
+            "last_run": _iso(cron.last_run, self.tz) if cron is not None else None,
             "status": str(cron.status) if cron is not None else None,
-            "run": self._runs.get(job.name),
+            "run": run or self._runs.get(job.name),
+        }
+
+    def _run_public(self, row: JobRunRecord) -> dict[str, Any]:
+        return {
+            "id": row.id,
+            "job": row.job,
+            "trigger": row.trigger,
+            "started_at": _iso(row.started_at, self.tz),
+            "finished_at": _iso(row.finished_at, self.tz),
+            "status": row.status,
+            "error": row.error,
+            "payload": row.payload,
         }
 
     # ---- http ----------------------------------------------------------
@@ -506,21 +592,28 @@ def _decode(payload: bytes | None) -> dict[str, Any]:
     return data if isinstance(data, dict) else {"raw": data}
 
 
-def _job_public(row: PgJob) -> dict[str, Any]:
+def _job_public(row: PgJob, tz: tzinfo) -> dict[str, Any]:
     return {
         "id": int(row.id),
         "entrypoint": row.entrypoint,
         "status": str(row.status),
-        "execute_after": _iso(row.execute_after),
+        "execute_after": _iso(row.execute_after, tz),
-        "created": _iso(row.created),
+        "created": _iso(row.created, tz),
         "attempts": row.attempts,
         "payload": _decode(row.payload),
     }
 
 
+def _error_text(exc: BaseException) -> str:
+    head = f"{type(exc).__name__}: {exc}"
+    return f"{head}\n{''.join(traceback.format_exception(exc))}"[:ERROR_MAX]
+
+
 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:
+def _iso(value: datetime | None, tz: tzinfo = UTC) -> str | None:
-    return _aware(value).isoformat(timespec="seconds") if value is not None else None
+    if value is None:
+        return None
+    return _aware(value).astimezone(tz).isoformat(timespec="seconds")

src/beaver_gateway/core/transcript.py

@@ -22,7 +22,9 @@ from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
 
 try:
-    from claude_agent_sdk._cli_version import __cli_version__ as _cli_version
+    from claude_agent_sdk._cli_version import __cli_version__
+
+    _cli_version = str(__cli_version__)
 except ImportError:  # pragma: no cover
     _cli_version = "2.1.248"
 

src/beaver_gateway/frontends/api/frontend.py

@@ -62,7 +62,7 @@ if TYPE_CHECKING:
     from pathlib import Path
 
     from beaver_gateway.core.conversations import Conversations
-    from beaver_gateway.core.scheduler import Scheduler
+    from beaver_gateway.core.scheduler import Job, Scheduler
     from beaver_gateway.frontends.base import GatewayRuntime
 
 _log = logging.getLogger("beaver_gateway.frontends.api")
@@ -580,15 +580,27 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa
         await require_token(request, runtime, scope=SCOPE)
         return await scheduler_of().snapshot()
 
+    def job_of(scheduler: Scheduler, name: str) -> Job:
+        job = scheduler.job(name)
+        if job is None:
+            raise HTTPException(status.HTTP_404_NOT_FOUND, f"no job {name!r}")
+        return job
+
     @app.post("/jobs/{name}/run", status_code=status.HTTP_202_ACCEPTED)
     async def run_job(name: str, request: Request) -> dict[str, Any]:
         await require_token(request, runtime, scope=SCOPE)
         scheduler = scheduler_of()
-        job = scheduler.job(name)
+        job = job_of(scheduler, name)
-        if job is None:
-            raise HTTPException(status.HTTP_404_NOT_FOUND, f"no job {name!r}")
         return {"job": await scheduler.trigger(job, await body_of(request))}
 
+    @app.get("/jobs/{name}/runs")
+    async def job_runs(name: str, request: Request) -> dict[str, Any]:
+        await require_token(request, runtime, scope=SCOPE)
+        scheduler = scheduler_of()
+        job = job_of(scheduler, name)
+        limit = min(int(request.query_params.get("limit", 50)), 500)
+        return {"job": job.name, "runs": await scheduler.runs(job.name, limit)}
+
     @app.delete("/jobs/queue/{job_id}")
     async def cancel_job(job_id: int, request: Request) -> dict[str, Any]:
         await require_token(request, runtime, scope=SCOPE)

src/beaver_gateway/frontends/telegram/drafts.py

@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING
 
 from aiogram.exceptions import TelegramAPIError, TelegramBadRequest
 
-from beaver_gateway.frontends.telegram.render import to_html
+from beaver_gateway.frontends.telegram.render import to_html, to_html_tail
 
 if TYPE_CHECKING:
     from aiogram import Bot
@@ -112,12 +112,12 @@ class Draft:
     async def _push(self) -> None:
         self._dirty = False
         self._last_sent = time.monotonic()
-        text = self._render()
+        rendered, plain = self._render()
         try:
             try:
-                await self._send(to_html(text), "HTML")
+                await self._send(rendered, "HTML")
             except TelegramBadRequest:
-                await self._send(text, None)
+                await self._send(plain, None)
         except TelegramAPIError as exc:
             self._broken = True
             _log.warning(
@@ -133,6 +133,9 @@ class Draft:
             parse_mode=parse_mode,
         )
 
-    def _render(self) -> str:
+    def _render(self) -> tuple[str, str]:
         tail = self.text[-_TAIL:]
-        return f"{self.status}\n\n{tail}" if tail.strip() else self.status
+        if not tail.strip():
+            return to_html(self.status), self.status
+        rendered = f"{to_html(self.status)}\n\n{to_html_tail(self.text, _TAIL)}"
+        return rendered, f"{self.status}\n\n{tail}"

src/beaver_gateway/frontends/telegram/frontend.py

@@ -19,8 +19,10 @@ import logging
 import tempfile
 import time
 from dataclasses import dataclass, field
+from datetime import datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, cast
+from zoneinfo import ZoneInfo
 
 from aiogram import Bot
 from aiogram.client.default import DefaultBotProperties
@@ -71,14 +73,21 @@ _HELP = "General - мастер, любой другой топик - ветка
 class Attachments:
     """Where files from Telegram go.
 
-    ``ephemeral`` - under the gateway's data dir, swept after ``keep_days``;
+    Files land in ``<root>/YYYY-MM-DD/<unixts>-<name>`` (the day in ``tz``)
-    ``vault`` - into a directory the agent owns. Pulling what is worth
+    and whatever is older than ``keep_days`` is swept; ``None`` never
-    keeping into the vault is the agent's job either way.
+    sweeps. ``ephemeral`` - under the gateway's data dir; ``vault`` -
+    ``dir`` is an inbox inside the agent's zone: the agent moves keepers
+    next to the note, the rest is swept after ``keep_days``.
     """
 
     mode: Literal["ephemeral", "vault"] = "ephemeral"
     dir: Path | None = None
-    keep_days: int = 7
+    keep_days: int | None = 7
+    tz: str = "UTC"
+
+    def day(self, now: float | None = None) -> str:
+        stamp = datetime.fromtimestamp(now or time.time(), tz=ZoneInfo(self.tz))
+        return stamp.date().isoformat()
 
     @property
     def root(self) -> Path:
@@ -174,7 +183,6 @@ class TelegramFrontend(Frontend):
             self.chat_id,
             getattr(me, "has_topics_enabled", None),
         )
-        self._sweep_attachments()
         await self.bot.set_my_commands(
             [BotCommand(command=c, description=d) for c, d in _COMMAND_HELP]
         )
@@ -183,6 +191,7 @@ class TelegramFrontend(Frontend):
                 tg.create_task(self.inbox.run())
                 tg.create_task(self.outbox.run())
                 tg.create_task(self._events())
+                tg.create_task(self._sweep_loop())
         finally:
             for draft in list(self._drafts.values()):
                 await draft.stop()
@@ -373,7 +382,11 @@ class TelegramFrontend(Frontend):
             return
         if message.chat.id != self.chat_id:
             return
-        in_topic = message.is_topic_message or message.forum_topic_created is not None
+        in_topic = (
+            message.is_topic_message
+            or message.forum_topic_created is not None
+            or message.forum_topic_edited is not None
+        )
         thread_id = message.message_thread_id if in_topic else None
         is_master = (
             thread_id is None or self._ext(thread_id) == await self._master_ext()
@@ -382,10 +395,12 @@ class TelegramFrontend(Frontend):
             return
         if message.forum_topic_edited is not None and thread_id is not None:
             if message.forum_topic_edited.name:
-                self._topic_names[thread_id] = message.forum_topic_edited.name
+                await self._on_topic_renamed(
+                    thread_id, message.forum_topic_edited.name, is_master=is_master
+                )
             return
         text = (message.text or message.caption or "").strip()
-        attachment = await self._save_attachment(message, thread_id)
+        attachment = await self._save_attachment(message)
         if attachment:
             text = f"{text}\n\n{attachment}".strip()
         if not text:
@@ -414,6 +429,14 @@ class TelegramFrontend(Frontend):
         if conv.running_turn or conv.pending_question:
             await self._react(message, item.id)
 
+    async def _on_topic_renamed(
+        self, thread_id: int, name: str, *, is_master: bool
+    ) -> None:
+        self._topic_names[thread_id] = name
+        conv = None if is_master else await self._live(thread_id)
+        if conv is not None:
+            await self.conversations.set_title(conv, name)
+
     async def _live(self, thread_id: int) -> Conversation | None:
         conv = await self.conversations.find_bound(
             frontend=FRONTEND, external_id=self._ext(thread_id)
@@ -446,13 +469,15 @@ class TelegramFrontend(Frontend):
         with contextlib.suppress(TelegramAPIError):
             await self.bot.set_message_reaction(target[0], target[1], reaction=[])
 
-    async def _save_attachment(
+    async def _save_attachment(self, message: Message) -> str | None:
-        self, message: Message, thread_id: int | None
-    ) -> str | None:
         file_id: str | None = None
         name: str | None = None
         kind = ""
         size: int | None = None
+        if message.sticker:
+            return "[вложение: стикер - не поддерживается]"
+        if message.animation:
+            return "[вложение: анимация - не поддерживается]"
         if message.photo:
             photo = message.photo[-1]
             file_id, kind, size = photo.file_id, "фото", photo.file_size
@@ -496,12 +521,13 @@ class TelegramFrontend(Frontend):
                 f"[вложение: {kind} {name}, {size // 1024 // 1024} МБ - "
                 "больше 20 МБ, Telegram не отдаёт ботам]"
             )
-        folder = self.attachments.root / (self._ext(thread_id).replace("/", "_"))
+        now = time.time()
+        folder = self.attachments.root / self.attachments.day(now)
         folder.mkdir(parents=True, exist_ok=True)
         with contextlib.suppress(OSError):
             self.attachments.root.chmod(0o755)
             folder.chmod(0o755)
-        path = folder / f"{int(time.time())}-{Path(name).name}"
+        path = folder / f"{int(now)}-{Path(name).name}"
         try:
             await self.bot.download(file_id, destination=path)
             path.chmod(0o644)
@@ -509,19 +535,33 @@ class TelegramFrontend(Frontend):
             _log.warning("attachment download failed: %s", exc)
             return f"[вложение: {kind} {name} - не скачалось: {exc}]"
         shown = f", {size // 1024} КБ" if size else ""
+        keep = self.attachments.keep_days
+        if keep is None:
             return f"[вложение: {kind} {path}{shown}]"
+        return (
+            f"[вложение: {kind} {path}{shown}; хранится {keep} дн., "
+            "перенеси в vault, если нужно]"
+        )
+
+    async def _sweep_loop(self, interval: float = 6 * 3600) -> None:
+        while True:
+            self._sweep_attachments()
+            await asyncio.sleep(interval)
 
     def _sweep_attachments(self) -> None:
-        if self.attachments.mode != "ephemeral":
+        keep = self.attachments.keep_days
-            return
         root = self.attachments.root
-        if not root.exists():
+        if keep is None or not root.exists():
             return
-        cutoff = time.time() - self.attachments.keep_days * 86400
+        cutoff = time.time() - keep * 86400
         for path in root.rglob("*"):
             with contextlib.suppress(OSError):
                 if path.is_file() and path.stat().st_mtime < cutoff:
                     path.unlink()
+        for folder in root.iterdir():
+            with contextlib.suppress(OSError):
+                if folder.is_dir() and not any(folder.iterdir()):
+                    folder.rmdir()
 
     # ---- commands --------------------------------------------------------
 
@@ -634,7 +674,10 @@ class TelegramFrontend(Frontend):
         if kind == "conversation.bound":
             self._targets.pop(str(event.get("conversation_id")), None)
             return
-        if kind in ("delivery.sent", "delivery.failed", "conversation.created"):
+        if kind == "delivery.failed":
+            await self._on_delivery_failed(event)
+            return
+        if kind in ("delivery.sent", "conversation.created"):
             return
         key = event.get("conversation_id")
         if not isinstance(key, str):
@@ -661,10 +704,13 @@ class TelegramFrontend(Frontend):
                 draft = self._drafts.get(key)
                 if draft is not None:
                     await draft.stop()
-                if event.get("stop") == "error" and event.get("origin") == "user":
+                if event.get("stop") == "error":
+                    origin = event.get("origin")
                     await self._deliver(
                         conv,
-                        "⚠️ тёрн упал, смотри логи gateway",
+                        "⚠️ тёрн упал, смотри логи gateway"
+                        if origin == "user"
+                        else f"⚠️ фоновый тёрн ({origin}) упал, смотри логи gateway",
                         turn_id=event.get("turn_id"),
                         key=f"{event.get('turn_id')}:error",
                     )
@@ -687,6 +733,38 @@ class TelegramFrontend(Frontend):
             case "conversation.merged":
                 await self._deliver(conv, "✅ слито в мастер", key=f"{key}:merged")
 
+    async def _on_delivery_failed(self, event: Event) -> None:
+        row = event.get("conversation_row")
+        _log.error(
+            "delivery to %s/%s (conversation row %s) failed: %s",
+            event.get("chat_id"),
+            event.get("thread_id"),
+            row,
+            event.get("error"),
+        )
+        if not event.get("gone") or not isinstance(row, int):
+            return
+        conv = await self.conversations.get_row(row)
+        if conv is None:
+            return
+        bound = next(
+            (
+                b
+                for b in await self.conversations.bindings(conv)
+                if b.frontend == FRONTEND and b.visible
+            ),
+            None,
+        )
+        if bound is None:
+            return
+        _log.warning(
+            "window %s of %s is gone; unbinding", bound.external_id, conv.external_id
+        )
+        await self.conversations.bind(
+            conv, frontend=FRONTEND, external_id=bound.external_id, visible=False
+        )
+        self._targets.pop(conv.external_id, None)
+
     async def _on_reply(self, conv: Conversation, event: Event) -> None:
         turn_id = str(event.get("turn_id") or "")
         origin = str(event.get("item_origin") or "")
@@ -789,6 +867,7 @@ class TelegramFrontend(Frontend):
         ask = self._asks.get(question_id)
         if ask is None or not qi_raw.isdigit():
             await self._callback_reply(query, "вопрос уже закрыт")
+            await self._strip_keyboard(query)
             return
         qi = int(qi_raw)
         question = ask.questions[qi]
@@ -831,6 +910,17 @@ class TelegramFrontend(Frontend):
             if not self.conversations.answer(question_id, answer):
                 await self._edit_asks(ask, "⌛ время вышло - ответь текстом")
 
+    async def _strip_keyboard(self, query: CallbackQuery) -> None:
+        message = query.message if isinstance(query.message, Message) else None
+        if message is None:
+            return
+        with contextlib.suppress(TelegramAPIError):
+            await self.bot.edit_message_reply_markup(
+                chat_id=message.chat.id,
+                message_id=message.message_id,
+                reply_markup=None,
+            )
+
     async def _callback_reply(self, query: CallbackQuery, text: str | None) -> None:
         with contextlib.suppress(TelegramAPIError):
             await self.bot.answer_callback_query(query.id, text=text)

src/beaver_gateway/frontends/telegram/outbox.py

@@ -143,7 +143,7 @@ class Outbox:
             if "parse" in text and not row.plain:
                 await self._retry(row, str(exc), delay=0.0, plain=True)
             elif any(marker in text for marker in _GONE):
-                await self._fail(row, str(exc))
+                await self._fail(row, str(exc), gone=True)
             else:
                 await self._fail(row, str(exc))
         except (TelegramNotFound, TelegramForbiddenError) as exc:
@@ -180,7 +180,7 @@ class Outbox:
         if delay < 5.0:
             self._wake.set()
 
-    async def _fail(self, row: Delivery, error: str) -> None:
+    async def _fail(self, row: Delivery, error: str, *, gone: bool = False) -> None:
         _log.error(
             "delivery #%s to %s/%s given up: %s",
             row.id,
@@ -196,6 +196,7 @@ class Outbox:
             chat_id=row.chat_id,
             thread_id=row.thread_id,
             error=error,
+            gone=gone,
         )
 
     async def _mark(

src/beaver_gateway/frontends/telegram/render.py

@@ -4,20 +4,38 @@ from __future__ import annotations
 
 import html
 import re
-from typing import Any
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any, Literal
 
-__all__ = ["LIMIT", "chunks", "status_label", "to_html"]
+if TYPE_CHECKING:
+    from collections.abc import Callable
+
+__all__ = ["LIMIT", "chunks", "status_label", "to_html", "to_html_tail"]
 
 LIMIT = 4000
-_FENCE = re.compile(r"```[^\n]*\n(.*?)(?:```|$)", re.DOTALL)
+_SNAP = 200
-_INLINE_CODE = re.compile(r"(`[^`\n]+`)")
+
-_HEADING = re.compile(r"^#{1,6}\s+(.+)$", re.MULTILINE)
+_FENCE_OPEN = re.compile(r"^\s*```(.*)$")
-_BOLD = re.compile(r"\*\*(.+?)\*\*|__(.+?)__", re.DOTALL)
+_FENCE_CLOSE = re.compile(r"^\s*```\s*$")
+_LANG = re.compile(r"[\w+#.-]+")
+_QUOTE = re.compile(r"^\s*(?:>\s?)+")
+_TABLE_SEP = re.compile(r"^\s*\|?\s*:?-+:?\s*(?:\|\s*:?-+:?\s*)*\|?\s*$")
+_CELL_SPLIT = re.compile(r"(?<!\\)\|")
+_HEADING = re.compile(r"^\s{0,3}#{1,6}\s+(.+?)\s*#*\s*$")
+_RULE = re.compile(r"^\s*([-*_])(?:\s*\1){2,}\s*$")
+_TASK = re.compile(r"^(\s*)[-*+]\s+\[([ xX])\]\s*(.*)$")
+_BULLET = re.compile(r"^(\s*)[-*+]\s+(.*)$")
+_NUMBERED = re.compile(r"^(\s*)(\d+)[.)]\s+(.*)$")
+_TOKEN = re.compile(
+    r"`[^`\n]+`|\[\[[^\]\n]+\]\]|!?\[[^\]]*\]\((?:https?|tg)://[^)\s]+\)"
+)
+_WIKILINK = re.compile(r"\[\[([^\]|#]+)(?:#([^\]|]+))?(?:\|([^\]]+))?\]\]")
+_LINK = re.compile(r"(!?)\[([^\]]*)\]\(([^)\s]+)\)")
+_BOLD = re.compile(r"\*\*(.+?)\*\*|__(.+?)__")
 _ITALIC = re.compile(r"(?<![\w*])\*(?!\s)(.+?)(?<!\s)\*(?![\w*])")
 _ITALIC_U = re.compile(r"(?<![\w_])_(?!\s)(.+?)(?<!\s)_(?![\w_])")
-_LINK = re.compile(r"\[([^\]]+)\]\((https?://[^)\s]+)\)")
-_BULLET = re.compile(r"^(\s*)[-*]\s+", re.MULTILINE)
 _STRIKE = re.compile(r"~~(.+?)~~")
+_SPOILER = re.compile(r"\|\|(.+?)\|\|")
 
 _LABELS: dict[str, str] = {
     "Read": "читаю vault",
@@ -41,57 +59,47 @@ _LABELS: dict[str, str] = {
     "mcp__gateway__inject": "передаю в другой разговор",
 }
 
+_Kind = Literal["fence", "quote", "table", "text"]
+
+
+@dataclass(slots=True)
+class _Block:
+    kind: _Kind
+    lines: list[str]
+
 
 def to_html(text: str) -> str:
-    out: list[str] = []
+    return "\n".join(_render(block) for block in _blocks(text)).strip()
-    pos = 0
-    for match in _FENCE.finditer(text):
-        out.append(_inline(text[pos : match.start()]))
-        out.append(f"<pre>{html.escape(match.group(1).rstrip())}</pre>")
-        pos = match.end()
-    out.append(_inline(text[pos:]))
-    return "".join(out).strip()
 
 
-def _inline(text: str) -> str:
+def to_html_tail(text: str, tail_chars: int) -> str:
-    parts = _INLINE_CODE.split(text)
+    start = max(len(text) - tail_chars, 0)
-    for i, part in enumerate(parts):
+    if start:
-        if i % 2:
+        newline = text.find("\n", start, start + _SNAP)
-            parts[i] = f"<code>{html.escape(part[1:-1])}</code>"
+        if newline >= 0:
-            continue
+            start = newline + 1
-        s = html.escape(part, quote=False)
+    opening = _open_fence(text[:start])
-        s = _HEADING.sub(r"<b>\1</b>", s)
+    tail = text[start:]
-        s = _BOLD.sub(lambda m: f"<b>{m.group(1) or m.group(2)}</b>", s)
+    return to_html(f"{opening}\n{tail}" if opening else tail)
-        s = _ITALIC.sub(r"<i>\1</i>", s)
-        s = _ITALIC_U.sub(r"<i>\1</i>", s)
-        s = _STRIKE.sub(r"<s>\1</s>", s)
-        s = _LINK.sub(r'<a href="\2">\1</a>', s)
-        parts[i] = _BULLET.sub(r"\1• ", s)
-    return "".join(parts)
 
 
 def chunks(text: str, limit: int = LIMIT) -> list[str]:
     text = text.strip()
-    if len(text) <= limit:
+    if not text:
-        return [text] if text else []
+        return []
+    if _fits(text, limit):
+        return [text]
     out: list[str] = []
-    while len(text) > limit:
+    cur: list[str] = []
-        cut = _cut_point(text, limit)
+    for block in _blocks(text):
-        out.append(text[:cut].rstrip())
+        for piece in _pieces(block, limit):
-        text = text[cut:].lstrip()
+            if cur and not _fits("\n".join([*cur, piece]), limit):
-    if text:
+                _flush(out, cur)
-        out.append(text)
+            cur.append(piece)
+    _flush(out, cur)
     return out
 
 
-def _cut_point(text: str, limit: int) -> int:
-    for sep in ("\n\n", "\n", ". ", " "):
-        cut = text.rfind(sep, limit // 2, limit)
-        if cut > 0:
-            return cut + len(sep)
-    return limit
-
-
 def status_label(name: str, tool_input: dict[str, Any] | None = None) -> str:
     label = _LABELS.get(name)
     if label is not None:
@@ -109,3 +117,245 @@ def status_label(name: str, tool_input: dict[str, Any] | None = None) -> str:
                 hint = value.strip().splitlines()[0][:60]
                 break
     return f"{name} {hint}" if hint else name
+
+
+def _blocks(text: str) -> list[_Block]:
+    lines = text.split("\n")
+    blocks: list[_Block] = []
+    i = 0
+    while i < len(lines):
+        line = lines[i]
+        if _FENCE_OPEN.match(line):
+            j = i + 1
+            while j < len(lines) and not _FENCE_CLOSE.match(lines[j]):
+                j += 1
+            end = min(j + 1, len(lines))
+            blocks.append(_Block("fence", lines[i:end]))
+            i = end
+        elif _QUOTE.match(line):
+            j = i + 1
+            while j < len(lines) and _QUOTE.match(lines[j]):
+                j += 1
+            blocks.append(_Block("quote", lines[i:j]))
+            i = j
+        elif _is_table_start(lines, i):
+            j = i + 2
+            while j < len(lines) and "|" in lines[j] and lines[j].strip():
+                j += 1
+            blocks.append(_Block("table", lines[i:j]))
+            i = j
+        elif not line.strip():
+            blocks.append(_Block("text", [line]))
+            i += 1
+        else:
+            j = i + 1
+            while j < len(lines) and _is_plain(lines, j):
+                j += 1
+            blocks.append(_Block("text", lines[i:j]))
+            i = j
+    return blocks
+
+
+def _is_table_start(lines: list[str], i: int) -> bool:
+    if "|" not in lines[i] or i + 1 >= len(lines):
+        return False
+    sep = lines[i + 1]
+    return "|" in sep and _TABLE_SEP.match(sep) is not None
+
+
+def _is_plain(lines: list[str], i: int) -> bool:
+    line = lines[i]
+    return bool(
+        line.strip()
+        and not _FENCE_OPEN.match(line)
+        and not _QUOTE.match(line)
+        and not _is_table_start(lines, i)
+    )
+
+
+def _open_fence(text: str) -> str | None:
+    opening: str | None = None
+    for line in text.split("\n"):
+        if opening is None:
+            if _FENCE_OPEN.match(line):
+                opening = line
+        elif _FENCE_CLOSE.match(line):
+            opening = None
+    return opening
+
+
+def _render(block: _Block) -> str:
+    if block.kind == "fence":
+        return _render_fence(block.lines)
+    if block.kind == "quote":
+        body = "\n".join(_line(_QUOTE.sub("", line, 1)) for line in block.lines)
+        return f"<blockquote>{body}</blockquote>" if body.strip() else ""
+    if block.kind == "table":
+        return _render_table(block.lines)
+    return "\n".join(_line(line) for line in block.lines)
+
+
+def _fence_parts(lines: list[str]) -> tuple[str, list[str]]:
+    body = lines[1:]
+    if body and _FENCE_CLOSE.match(body[-1]):
+        body = body[:-1]
+    return lines[0], body
+
+
+def _render_fence(lines: list[str]) -> str:
+    head, body = _fence_parts(lines)
+    code = html.escape("\n".join(body).rstrip())
+    if not code:
+        return ""
+    lang = _language(head)
+    if lang:
+        return f'<pre><code class="language-{lang}">{code}</code></pre>'
+    return f"<pre>{code}</pre>"
+
+
+def _language(head: str) -> str:
+    match = _FENCE_OPEN.match(head)
+    info = match.group(1).strip() if match else ""
+    token = info.split()[0] if info else ""
+    return token if _LANG.fullmatch(token) else ""
+
+
+def _render_table(lines: list[str]) -> str:
+    rows = [_cells(line) for i, line in enumerate(lines) if i != 1]
+    width = max(len(row) for row in rows)
+    rows = [[*row, *[""] * (width - len(row))] for row in rows]
+    widths = [max(len(row[c]) for row in rows) for c in range(width)]
+    text = "\n".join(_table_row(row, widths) for row in rows)
+    return f"<pre>{html.escape(text)}</pre>"
+
+
+def _table_row(row: list[str], widths: list[int]) -> str:
+    cells = (cell.ljust(w) for cell, w in zip(row, widths, strict=True))
+    return " | ".join(cells).rstrip(" |")
+
+
+def _cells(line: str) -> list[str]:
+    inner = line.strip()
+    inner = inner.removeprefix("|").removesuffix("|")
+    return [cell.strip().replace("\\|", "|") for cell in _CELL_SPLIT.split(inner)]
+
+
+def _line(line: str) -> str:
+    if _RULE.match(line):
+        return "———"
+    if match := _HEADING.match(line):
+        return f"<b>{_inline(match.group(1))}</b>"
+    if match := _TASK.match(line):
+        box = "☐" if match.group(2) == " " else "☑"
+        return f"{match.group(1)}{box} {_inline(match.group(3))}"
+    if match := _BULLET.match(line):
+        return f"{match.group(1)}• {_inline(match.group(2))}"
+    if match := _NUMBERED.match(line):
+        return f"{match.group(1)}{match.group(2)}. {_inline(match.group(3))}"
+    return _inline(line)
+
+
+def _inline(text: str) -> str:
+    out: list[str] = []
+    pos = 0
+    for match in _TOKEN.finditer(text):
+        out.append(_format(text[pos : match.start()]))
+        out.append(_token(match.group()))
+        pos = match.end()
+    out.append(_format(text[pos:]))
+    return "".join(out)
+
+
+def _token(token: str) -> str:
+    if token.startswith("`"):
+        return f"<code>{html.escape(token[1:-1])}</code>"
+    if token.startswith("[["):
+        return _wikilink(token)
+    match = _LINK.fullmatch(token)
+    if match is None:
+        return html.escape(token, quote=False)
+    href = html.escape(match.group(3), quote=True)
+    label = _format(match.group(2)) or href
+    return f'<a href="{href}">{label}</a>'
+
+
+def _wikilink(token: str) -> str:
+    match = _WIKILINK.fullmatch(token)
+    if match is None:
+        return html.escape(token, quote=False)
+    note, heading, alias = (g.strip() if g else "" for g in match.groups())
+    label = alias or (f"{note} › {heading}" if heading else note)  # noqa: RUF001
+    return f"<u>{html.escape(label, quote=False)}</u>"
+
+
+def _format(text: str) -> str:
+    s = html.escape(text, quote=False)
+    s = _BOLD.sub(lambda m: f"<b>{m.group(1) or m.group(2)}</b>", s)
+    s = _ITALIC.sub(r"<i>\1</i>", s)
+    s = _ITALIC_U.sub(r"<i>\1</i>", s)
+    s = _STRIKE.sub(r"<s>\1</s>", s)
+    return _SPOILER.sub(r"<tg-spoiler>\1</tg-spoiler>", s)
+
+
+def _fits(text: str, limit: int) -> bool:
+    return len(to_html(text)) <= limit
+
+
+def _flush(out: list[str], cur: list[str]) -> None:
+    chunk = "\n".join(cur).strip()
+    if chunk:
+        out.append(chunk)
+    cur.clear()
+
+
+def _pieces(block: _Block, limit: int) -> list[str]:
+    whole = "\n".join(block.lines)
+    if _fits(whole, limit):
+        return [whole]
+    if block.kind == "fence":
+        head, body = _fence_parts(block.lines)
+        return _pack(body, lambda ls: "\n".join([head, *ls, "```"]), limit)
+    if block.kind == "quote":
+        inner = [_QUOTE.sub("", line, 1) for line in block.lines]
+        return _pack(inner, lambda ls: "\n".join(f"> {s}" for s in ls), limit)
+    if block.kind == "table":
+        header = block.lines[:2]
+        return _pack(block.lines[2:], lambda ls: "\n".join(header + ls), limit)
+    return _pack(block.lines, "\n".join, limit)
+
+
+def _pack(lines: list[str], wrap: Callable[[list[str]], str], limit: int) -> list[str]:
+    pieces: list[str] = []
+    cur: list[str] = []
+    for line in lines:
+        if cur and not _fits(wrap([*cur, line]), limit):
+            pieces.append(wrap(cur))
+            cur = []
+        if _fits(wrap([line]), limit):
+            cur.append(line)
+        else:
+            pieces.extend(wrap([part]) for part in _shatter(line, wrap, limit))
+    if cur:
+        pieces.append(wrap(cur))
+    return pieces
+
+
+def _shatter(line: str, wrap: Callable[[list[str]], str], limit: int) -> list[str]:
+    parts: list[str] = []
+    while line:
+        cut = len(line)
+        while cut > 1 and not _fits(wrap([line[:cut].rstrip()]), limit):
+            rendered = len(to_html(wrap([line[:cut]])))
+            cut = min(cut - 1, _cut_point(line, max(1, cut * limit // rendered)))
+        if part := line[:cut].rstrip():
+            parts.append(part)
+        line = line[cut:].lstrip()
+    return parts
+
+
+def _cut_point(text: str, limit: int) -> int:
+    for sep in ("\n\n", "\n", ". ", " "):
+        cut = text.rfind(sep, limit // 2, limit)
+        if cut > 0:
+            return cut + len(sep)
+    return limit

src/beaver_gateway/storage/models.py

@@ -360,6 +360,32 @@ class RateLimit(SQLModel, table=True):
     )
 
 
+class JobRunRecord(SQLModel, table=True):
+    """One finished run of a scheduler job (§3.6): what fired it, how it ended.
+
+    ``error`` keeps the head of the traceback of a failed run; ``payload``
+    is the trigger's data (``{}`` for cron) and should stay small.
+    """
+
+    __tablename__ = "job_runs"
+
+    id: int | None = Field(default=None, primary_key=True)
+    job: str = Field(index=True)
+    trigger: str
+    started_at: datetime
+    finished_at: datetime
+    status: str
+    error: str | None = Field(default=None)
+    payload: dict[str, Any] = Field(
+        default_factory=dict,
+        sa_column=Column(
+            JSON().with_variant(JSONB(), "postgresql"),
+            nullable=False,
+            server_default=text("'{}'"),
+        ),
+    )
+
+
 __all__ = [
     "AuditLog",
     "Conversation",
@@ -367,6 +393,7 @@ __all__ = [
     "ConversationMessage",
     "Delivery",
     "InjectQueueItem",
+    "JobRunRecord",
     "RateLimit",
     "TelegramUpdate",
     "Token",

tests/test_api.py

@@ -21,6 +21,7 @@ from test_conversations import ScriptedClient, World
 from beaver_gateway.core.conversation_store import rewrite_messages
 from beaver_gateway.core.auth import TokenStore
 from beaver_gateway.core.registry import McpRegistry
+from beaver_gateway.core.scheduler import Job, JobRun, Scheduler
 from beaver_gateway.core.transcript import build_entries
 from beaver_gateway.frontends.admin import AdminFrontend
 from beaver_gateway.frontends.admin.frontend import build_app as build_admin
@@ -96,7 +97,12 @@ class ToolClient(ScriptedClient):
 
 
 class Api:
-    def __init__(self, world: World, memory_root: Path | None = None) -> None:
+    def __init__(
+        self,
+        world: World,
+        memory_root: Path | None = None,
+        scheduler: Scheduler | None = None,
+    ) -> None:
         self.world = world
         self.store = TokenStore(world.db, bootstrap={"t": TOKEN})
         self.runtime = GatewayRuntime(
@@ -112,6 +118,7 @@ class Api:
             conversations=world.conversations,
             bus=world.bus,
             pool=world.pool,
+            scheduler=scheduler,
         )
         self.app = build_api(self.runtime, memory_root=memory_root)
         self.http = AsyncClient(
@@ -146,6 +153,34 @@ async def seed_usage(world: World, rows: list[dict[str, Any]]) -> None:
         await session.commit()
 
 
+async def test_job_runs_come_from_the_db(world: World) -> None:
+    async def ping(_run: JobRun) -> None:
+        pass
+
+    scheduler = Scheduler(
+        conversations=world.conversations, jobs=[Job("ping", ping)], tz="Europe/Warsaw"
+    )
+    api = Api(world, scheduler=scheduler)
+    await scheduler.start()
+    res = await api.http.post("/jobs/ping/run", headers=HEADERS)
+    assert res.status_code == 202, res.text
+    for _ in range(100):
+        runs = (await api.get("/jobs/ping/runs", {"limit": 10}))["runs"]
+        if runs:
+            break
+        await asyncio.sleep(0.02)
+    assert [(r["trigger"], r["status"], r["payload"]) for r in runs] == [
+        ("manual", "done", {})
+    ]
+    jobs = await api.get("/jobs")
+    assert jobs["tz"] == "Europe/Warsaw"
+    assert jobs["jobs"][0]["run"]["id"] == runs[0]["id"]
+    assert jobs["jobs"][0]["run"]["finished_at"].endswith("+02:00")
+    missing = await api.http.get("/jobs/nope/runs", headers=HEADERS)
+    assert missing.status_code == 404
+    await scheduler.stop()
+
+
 async def test_usage_groups_by_agent_day_and_model(world: World) -> None:
     api = Api(world)
     now = datetime.now(UTC).replace(tzinfo=None)

tests/test_claude_sdk_backend.py

@@ -25,7 +25,7 @@ from claude_agent_sdk import (
 )
 
 from beaver_gateway.agents.base import ExposedMcp
-from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions, Prompts
+from beaver_gateway.agents.claude import ClaudeAgent, ClaudeOptions, Prompts, SkillSets
 from beaver_gateway.backends.claude_sdk import (
     ClaudeSdkBackend,
     RunnerConfig,
@@ -355,29 +355,58 @@ async def test_mcp_deny_and_usage_sink(cwd: Path) -> None:
     assert seen[0].session_id == "fresh-session"
 
 
-async def test_skill_sets_become_sorted_plugins(cwd: Path) -> None:
+def _skill_sets(cwd: Path, *names: str) -> tuple[Path, ...]:
     sets = cwd / "skills"
-    for name in ("zeta", "общие"):
+    for name in names:
         (sets / name / "demo").mkdir(parents=True)
         (sets / name / "demo" / "SKILL.md").write_text("---\nname: demo\n---\n")
-    backend = _backend(
+    return tuple(sets / name for name in names)
-        cwd, InMemorySessionStore(), skill_sets=(sets / "zeta", sets / "общие")
+
-    )
+
+async def _plugins_for(backend: ClaudeSdkBackend, kind: str) -> list[Path]:
     await _drain(
         backend.complete(
             agent=backend.agent,
             messages=[{"role": "user", "content": "x"}],
-            conversation_id="c",
+            conversation_id=f"c-{kind}",
+            kind=kind,
         )
     )
-    plugins = FakeClient.instances[0].options.plugins
+    plugins = FakeClient.instances[-1].options.plugins
-    assert [p["type"] for p in plugins] == ["local", "local"]
+    assert all(p["type"] == "local" for p in plugins)
     paths = [Path(p["path"]) for p in plugins]
-    assert [p.name for p in paths] == ["zeta", "общие"]
     for path in paths:
+        assert path.parent.name == kind
+        assert path.parent.parent.name == backend.agent.name
         assert (path / ".claude-plugin" / "plugin.json").exists()
         assert (path / "skills" / "demo" / "SKILL.md").exists()
+    return paths
+
+
+async def test_skill_sets_become_sorted_plugins(cwd: Path) -> None:
+    zeta, common = _skill_sets(cwd, "zeta", "общие")
+    backend = _backend(cwd, InMemorySessionStore(), skill_sets=(zeta, common))
+    paths = await _plugins_for(backend, "deep")
+    assert [p.name for p in paths] == ["zeta", "общие"]
     assert FakeClient.instances[0].options.skills == "all"
+    assert [p.name for p in await _plugins_for(backend, "job")] == ["zeta", "общие"]
+
+
+async def test_skill_sets_per_kind(cwd: Path) -> None:
+    a, b = _skill_sets(cwd, "a", "b")
+    backend = _backend(
+        cwd,
+        InMemorySessionStore(),
+        kinds=("master", "branch", "fork"),
+        skill_sets=SkillSets(master=(a,), branch=(a, b)),
+    )
+    master = await _plugins_for(backend, "master")
+    branch = await _plugins_for(backend, "branch")
+    assert [p.name for p in master] == ["a"]
+    assert [p.name for p in branch] == ["a", "b"]
+    assert master[0] != branch[0]
+    assert await _plugins_for(backend, "fork") == []
+    assert FakeClient.instances[-1].options.skills is None
 
 
 async def test_prompt_sources_are_assembled(cwd: Path) -> None:

tests/test_conversations.py

@@ -4,6 +4,7 @@ import uuid
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any
+from zoneinfo import ZoneInfo
 
 import pytest
 from claude_agent_sdk import (
@@ -629,6 +630,9 @@ async def test_schedule_without_scheduler_and_parse_at(world: World) -> None:
     assert parse_at("2026-09-01T10:00:00+02:00") == datetime(
         2026, 9, 1, 8, 0, tzinfo=UTC
     )
+    assert parse_at("2026-09-01T10:00", ZoneInfo("Europe/Warsaw")) == datetime(
+        2026, 9, 1, 8, 0, tzinfo=UTC
+    )
     delta = (parse_at("+15m") - datetime.now(UTC)).total_seconds()
     assert 14 * 60 < delta <= 15 * 60
     with pytest.raises(ValueError, match="Invalid isoformat"):

tests/test_scheduler.py

@@ -1,9 +1,11 @@
 import asyncio
+import inspect
 import os
 import tempfile
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
+from zoneinfo import ZoneInfo
 
 import psycopg
 import pytest
@@ -11,14 +13,12 @@ from httpx import ASGITransport, AsyncClient
 from pgqueuer import PsycopgDriver, Queries
 from test_conversations import World
 
-from beaver_gateway.core.scheduler import Budget, Job, JobRun, Scheduler
+from beaver_gateway.core.conversations import parse_at
+from beaver_gateway.core.scheduler import Budget, Job, JobRun, Scheduler, next_run
 from beaver_gateway.storage.models import RateLimit
 
 DATABASE_URL = os.environ.get("TEST_DATABASE_URL")
-
+WARSAW = ZoneInfo("Europe/Warsaw")
-pytestmark = pytest.mark.skipif(
-    not DATABASE_URL, reason="TEST_DATABASE_URL (postgres) is not set"
-)
 
 
 class Pg:
@@ -37,6 +37,8 @@ class Pg:
 
 @pytest.fixture
 async def pg() -> Pg:
+    if not DATABASE_URL:
+        pytest.skip("TEST_DATABASE_URL (postgres) is not set")
     handle = Pg()
     queries = Queries(await handle.driver())
     if await queries.has_table("pgqueuer"):
@@ -61,6 +63,8 @@ async def until(pred, timeout: float = 5.0) -> Any:
     deadline = asyncio.get_running_loop().time() + timeout
     while asyncio.get_running_loop().time() < deadline:
         value = pred()
+        if inspect.isawaitable(value):
+            value = await value
         if value:
             return value
         await asyncio.sleep(0.05)
@@ -68,16 +72,91 @@ async def until(pred, timeout: float = 5.0) -> Any:
     raise AssertionError(msg)
 
 
+def test_cron_is_read_in_the_gateway_tz() -> None:
+    summer = datetime(2026, 7, 1, 12, tzinfo=UTC)
+    assert next_run("0 4 * * *", WARSAW, summer) == datetime(2026, 7, 2, 2, tzinfo=UTC)
+    assert next_run("0 4 * * *", UTC, summer) == datetime(2026, 7, 2, 4, tzinfo=UTC)
+    before_fallback = datetime(2026, 10, 24, 12, tzinfo=UTC)
+    assert next_run("0 4 * * *", WARSAW, before_fallback) == datetime(
+        2026, 10, 25, 3, tzinfo=UTC
+    )
+    during_fallback = datetime(2026, 10, 25, 1, 30, tzinfo=UTC)
+    assert next_run("0 4 * * *", WARSAW, during_fallback) == datetime(
+        2026, 10, 25, 3, tzinfo=UTC
+    )
+    assert next_run("*/15 * * * *", WARSAW, during_fallback) == datetime(
+        2026, 10, 25, 1, 45, tzinfo=UTC
+    )
+    soon = next_run("* * * * *", WARSAW)
+    assert soon.tzinfo is UTC
+    assert timedelta(0) < soon - datetime.now(UTC) <= timedelta(minutes=1)
+
+
+def test_parse_at_reads_naive_iso_in_tz() -> None:
+    assert parse_at("2026-09-01T10:00", WARSAW) == datetime(
+        2026, 9, 1, 8, 0, tzinfo=UTC
+    )
+    assert parse_at("2026-12-01T10:00", WARSAW) == datetime(
+        2026, 12, 1, 9, 0, tzinfo=UTC
+    )
+    assert parse_at("2026-09-01T10:00:00+02:00", WARSAW) == datetime(
+        2026, 9, 1, 8, 0, tzinfo=UTC
+    )
+    assert parse_at("2026-09-01T10:00") == datetime(2026, 9, 1, 10, 0, tzinfo=UTC)
+
+
+async def test_runs_are_recorded_without_postgres(world: World) -> None:
+    async def ok(run: JobRun) -> None:
+        run.payload["seen"] = True
+
+    async def boom(_run: JobRun) -> None:
+        msg = "no network"
+        raise ConnectionError(msg)
+
+    scheduler = Scheduler(
+        conversations=world.conversations,
+        jobs=[Job("ok", ok, webhook=True), Job("boom", boom)],
+        tz="Europe/Warsaw",
+    )
+    await scheduler.start()
+    await scheduler.hook("ok", {"stack": "x"})
+    await scheduler.trigger(scheduler.job("boom"))
+    await until(lambda: len(scheduler._runs) == 2)  # noqa: SLF001
+    runs = await scheduler.runs("boom")
+    assert [r["status"] for r in runs] == ["failed"]
+    assert runs[0]["error"].startswith("ConnectionError: no network\n")
+    assert runs[0]["trigger"] == "manual"
+    assert runs[0]["started_at"] <= runs[0]["finished_at"]
+    assert runs[0]["started_at"].endswith("+02:00")
+    ok_runs = await scheduler.runs("ok")
+    assert ok_runs[0]["payload"] == {"stack": "x", "seen": True}
+    assert ok_runs[0]["trigger"] == "webhook"
+    await scheduler.hook("ok", {"stack": "y"})
+    await until(lambda: scheduler._runs["ok"]["payload"].get("stack") == "y")  # noqa: SLF001
+    assert len(await scheduler.runs("ok")) == 2
+    assert len(await scheduler.runs("ok", limit=1)) == 1
+    snapshot = await scheduler.snapshot()
+    assert snapshot["tz"] == "Europe/Warsaw"
+    runs_by_name = {j["name"]: j["run"] for j in snapshot["jobs"]}
+    assert runs_by_name["ok"]["payload"]["stack"] == "y"
+    assert runs_by_name["boom"]["status"] == "failed"
+    await scheduler.stop()
+
+
 async def test_schedule_survives_a_restart(world: World, pg: Pg) -> None:
     conv = await world.conversations.create(kind="master", agent="a", origin="test")
-    first = Scheduler(conversations=world.conversations, driver=await pg.driver())
+    first = Scheduler(
+        conversations=world.conversations, driver=await pg.driver(), tz="Europe/Warsaw"
+    )
     world.conversations.scheduler = first
     await first.start()
     job_id, when = await world.conversations.schedule(conv, "+1s", "push X")
     assert job_id is not None
     assert 0 < (when - datetime.now(UTC)).total_seconds() <= 1
+    assert when.utcoffset() == datetime.now(WARSAW).utcoffset()
     queued = await world.conversations.schedules(conv)
     assert [q["payload"]["text"] for q in queued] == ["push X"]
+    assert queued[0]["payload"]["at"] == when.isoformat(timespec="minutes")
     await first.stop()
 
     second = Scheduler(conversations=world.conversations, driver=await pg.driver())
@@ -138,9 +217,14 @@ async def test_webhook_runs_the_job_with_its_payload(world: World, pg: Pg) -> No
         assert missing.status_code == 404
     await until(lambda: seen)
     assert seen == [{"trigger": "webhook", "stack": "x"}]
+    await until(lambda: scheduler._runs.get("ping"))  # noqa: SLF001
     snapshot = await scheduler.snapshot()
     assert [j["name"] for j in snapshot["jobs"]] == ["ping", "quiet"]
     assert snapshot["enabled"] and snapshot["jobs"][0]["webhook"]
+    assert snapshot["jobs"][0]["run"]["status"] == "done"
+    assert snapshot["jobs"][0]["run"]["trigger"] == "webhook"
+    runs = await scheduler.runs("ping")
+    assert [r["payload"] for r in runs] == [{"stack": "x"}]
     await scheduler.stop()
 
 
@@ -170,6 +254,7 @@ async def test_non_critical_jobs_wait_while_the_window_is_hot(
         ],
         driver=await pg.driver(),
         budget=Budget(threshold=0.7),
+        tz="Europe/Warsaw",
     )
     await scheduler.start()
     events: list[dict[str, Any]] = []
@@ -193,5 +278,13 @@ async def test_non_critical_jobs_wait_while_the_window_is_hot(
     assert snapshot["throttled"] and snapshot["utilization"] == 0.9
     crons = {j["name"]: j["cron"] for j in snapshot["jobs"]}
     assert crons == {"vibegram": "*/10 * * * *", "rotation": "0 * * * *"}
-    await until(lambda: all(j["next_run"] for j in snapshot["jobs"]) or True)
+
+    async def crons_ready() -> dict[str, Any] | None:
+        snap = await scheduler.snapshot()
+        return snap if all(j["next_run"] for j in snap["jobs"]) else None
+
+    snapshot = await until(crons_ready)
+    expected = datetime.now(WARSAW).utcoffset()
+    for job_info in snapshot["jobs"]:
+        assert datetime.fromisoformat(job_info["next_run"]).utcoffset() == expected
     await scheduler.stop()

tests/test_telegram.py

@@ -13,8 +13,14 @@ from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny
 from beaver_gateway.core.registry import McpRegistry
 from beaver_gateway.core.transcript import build_entries
 from beaver_gateway.frontends.base import GatewayRuntime
-from beaver_gateway.frontends.telegram import TelegramFrontend
+from beaver_gateway.frontends.telegram import Attachments, TelegramFrontend
-from beaver_gateway.frontends.telegram.render import chunks, status_label, to_html
+from beaver_gateway.frontends.telegram.render import (
+    LIMIT,
+    chunks,
+    status_label,
+    to_html,
+    to_html_tail,
+)
 from beaver_gateway.storage.models import ConversationBinding, Delivery, TelegramUpdate
 from sqlmodel import select
 from test_conversations import ScriptedClient, StubFrontend, World
@@ -33,6 +39,7 @@ class FakeBot:
         self.reactions: list[tuple[int, list[Any]]] = []
         self.fail_sends = 0
         self.reject_html = False
+        self.gone_threads: set[int] = set()
         self.order: list[tuple[str, int | None]] = []
         self._message_id = 100
         self.session = SimpleNamespace(close=self._close)
@@ -64,6 +71,11 @@ class FakeBot:
                 method=SendMessage(chat_id=0, text="x"),
                 message="Bad Request: can't parse entities",
             )
+        if message_thread_id in self.gone_threads:
+            raise TelegramBadRequest(
+                method=SendMessage(chat_id=0, text="x"),
+                message="Bad Request: message thread not found",
+            )
         self._message_id += 1
         self.order.append(("message", message_thread_id))
         self.sent.append(
@@ -86,7 +98,8 @@ class FakeBot:
 
     async def create_forum_topic(self, chat_id, name, **_: Any) -> Any:
         self.topics.append(name)
-        return SimpleNamespace(message_thread_id=900 + len(self.topics), name=name)
+        created = sum(1 for t in self.topics if not t.startswith("edit:"))
+        return SimpleNamespace(message_thread_id=900 + created, name=name)
 
     async def edit_forum_topic(
         self, chat_id, message_thread_id, name=None, **_: Any
@@ -148,6 +161,33 @@ class FakeBot:
             }
         )
 
+    def topic_edited(self, name: str, thread: int) -> None:
+        self.push(
+            {
+                "message": {
+                    "message_id": 10 + len(self.updates),
+                    "date": 1700000000,
+                    "chat": {"id": USER, "type": "private"},
+                    "from": {"id": USER, "is_bot": False, "first_name": "h"},
+                    "message_thread_id": thread,
+                    "is_topic_message": True,
+                    "forum_topic_edited": {"name": name},
+                }
+            }
+        )
+
+    def media(self, body: dict[str, Any], *, thread: int | None = None) -> None:
+        message: dict[str, Any] = {
+            "message_id": 10 + len(self.updates),
+            "date": 1700000000,
+            "chat": {"id": USER, "type": "private"},
+            "from": {"id": USER, "is_bot": False, "first_name": "h"},
+            **body,
+        }
+        if thread is not None:
+            message |= {"message_thread_id": thread, "is_topic_message": True}
+        self.push({"message": message})
+
     def callback(self, data: str, message_id: int) -> None:
         self.push(
             {
@@ -455,9 +495,10 @@ async def test_commands_status_merge_and_new(stack: Stack) -> None:
         frontend="telegram", external_id=f"{USER}/11"
     )
     assert branch.status == "merged"
+    assert stack.bot.topics == ["🦫 General", "edit:11:✅ work"]
     stack.bot.message("/new отчёт")
     await stack.until(lambda: stack.sent_with("в новом топике"), what="new topic")
-    assert stack.bot.topics == ["🦫 General", "отчёт"]
+    assert stack.bot.topics == ["🦫 General", "edit:11:✅ work", "отчёт"]
     child = await stack.world.conversations.find_bound(
         frontend="telegram", external_id=f"{USER}/902"
     )
@@ -583,18 +624,112 @@ def test_render_helpers() -> None:
         == "<b>жирно</b> и <code>code</code> &lt;b&gt;"
     )
     assert to_html("# Заголовок\n- пункт") == "<b>Заголовок</b>\n• пункт"
-    assert to_html("```py\nx = 1\n```") == "<pre>x = 1</pre>"
+    assert to_html("```\nx < 1 & y\n```") == "<pre>x &lt; 1 &amp; y</pre>"
+    assert (
+        to_html("```py\nunterminated\n")
+        == '<pre><code class="language-py">unterminated</code></pre>'
+    )
     assert (
         to_html("[док](https://a.b/c?x=1&y=2)")
         == '<a href="https://a.b/c?x=1&amp;y=2">док</a>'
     )
+    assert (
+        to_html("[tg](tg://user?id=1) https://x.y/a_b_c")
+        == '<a href="tg://user?id=1">tg</a> https://x.y/a_b_c'
+    )
+    assert to_html("![alt](https://x.y/i.png) ![](https://x.y/j.png)") == (
+        '<a href="https://x.y/i.png">alt</a> <a href="https://x.y/j.png">https://x.y/j.png</a>'
+    )
+    assert to_html("~~нет~~ _к_ *к* __ж__") == "<s>нет</s> <i>к</i> <i>к</i> <b>ж</b>"
+    assert to_html("---\n***\n___") == "———\n———\n———"
+    assert to_html("`**raw** [[x]]`") == "<code>**raw** [[x]]</code>"
     parts = chunks("абв\n\n" + "г" * 30 + "\n\n" + "д" * 30, limit=40)
     assert parts == ["абв\n\n" + "г" * 30, "д" * 30]
+    assert chunks("   \n ") == []
     assert status_label("Read", {"file_path": "/x"}) == "читаю vault"
     assert status_label("mcp__firefly__list_accounts", None) == "firefly: list_accounts"
     assert status_label("Foo", {"command": "ls -la"}) == "Foo ls -la"
 
 
+def test_render_wikilinks() -> None:
+    assert to_html("[[Note]]") == "<u>Note</u>"
+    assert to_html("[[Note|alias]]") == "<u>alias</u>"
+    assert to_html("[[Note#heading]]") == "<u>Note › heading</u>"
+    assert to_html("[[A & B|**x**]]") == "<u>**x**</u>"
+
+
+def test_render_blocks() -> None:
+    assert (
+        to_html("> цитата **жирно**\n> вторая\n\nтекст")
+        == "<blockquote>цитата <b>жирно</b>\nвторая</blockquote>\n\nтекст"
+    )
+    assert (
+        to_html("1. один\n2) два\n  - [ ] нет\n- [x] да")
+        == "1. один\n2. два\n  ☐ нет\n☑ да"
+    )
+    assert (
+        to_html("| a | bb |\n|---|:--:|\n| ccc | d |\n| e |")
+        == "<pre>a   | bb\nccc | d\ne</pre>"
+    )
+    assert (
+        to_html("```python\nprint(1)\n```")
+        == '<pre><code class="language-python">print(1)</code></pre>'
+    )
+    assert (
+        to_html("||секрет|| и `||нет||`")
+        == "<tg-spoiler>секрет</tg-spoiler> и <code>||нет||</code>"
+    )
+
+
+def test_chunks_split_fence_into_valid_fences() -> None:
+    body = "\n".join(f"x = {i} & {i}" for i in range(400))
+    text = f"до\n```python\n{body}\n```\nпосле"
+    parts = chunks(text, limit=1500)
+    assert len(parts) > 2
+    assert parts[0] == "до"
+    assert all(p.startswith("```python\n") and p.endswith("```") for p in parts[1:-1])
+    assert parts[-1].endswith("```\nпосле")
+    fences = [p.removesuffix("\nпосле") for p in parts if p.startswith("```")]
+    inner = [p.removeprefix("```python\n").removesuffix("\n```") for p in fences]
+    assert "\n".join(inner) == body
+    for p in parts:
+        rendered = to_html(p)
+        assert 0 < len(rendered) <= 1500
+    for p in fences:
+        assert to_html(p).startswith('<pre><code class="language-python">')
+        assert to_html(p).endswith("</code></pre>")
+
+
+def test_chunks_split_blockquote_and_stay_under_limit() -> None:
+    text = "\n".join(f"> строка {i} **ж** <" for i in range(300))
+    parts = chunks(text, limit=1000)
+    assert len(parts) > 1
+    for p in parts:
+        rendered = to_html(p)
+        assert len(rendered) <= 1000
+        assert rendered.startswith("<blockquote>") and rendered.endswith(
+            "</blockquote>"
+        )
+    assert "".join(parts).count("строка") == 300
+    long_line = "слово " * 2000
+    parts = chunks(long_line, limit=LIMIT)
+    assert all(0 < len(to_html(p)) <= LIMIT for p in parts)
+    assert " ".join(parts).split() == long_line.split()
+
+
+def test_to_html_tail_inside_fence() -> None:
+    full = (
+        "intro\n```py\n" + "\n".join(f"code {i} <" for i in range(200)) + "\n```\nafter"
+    )
+    rendered = to_html_tail(full, 100)
+    assert rendered.startswith('<pre><code class="language-py">code ')
+    assert rendered.endswith("</code></pre>\nafter")
+    assert "```" not in rendered
+    assert to_html_tail("a\n> b\n> c", 3) == "<blockquote>c</blockquote>"
+    assert to_html_tail("**x** y", 100) == "<b>x</b> y"
+    assert to_html_tail("x" * 50, 10) == "x" * 10
+
+
 async def test_message_with_a_link_preview_is_stored_and_answered(stack: Stack) -> None:
     stack.bot.push(
         {
@@ -614,3 +749,171 @@ async def test_message_with_a_link_preview_is_stored_and_answered(stack: Stack)
     preview = rows[0].payload["message"]["link_preview_options"]
     assert preview["url"] == "https://x.y"
     assert preview.get("is_disabled") is None
+
+
+def _photo(caption: str) -> dict[str, Any]:
+    return {
+        "caption": caption,
+        "photo": [
+            {
+                "file_id": "f1",
+                "file_unique_id": "u1",
+                "width": 1,
+                "height": 1,
+                "file_size": 2048,
+            }
+        ],
+    }
+
+
+async def test_attachments_land_in_day_folder_with_keep_note(stack: Stack) -> None:
+    root = Path(tempfile.mkdtemp(prefix="beaver-att-"))
+    stack.tg.attachments = Attachments(dir=root, keep_days=3, tz="Asia/Bangkok")
+    stack.bot.media(_photo("смотри"))
+    await stack.until(lambda: stack.sent_with("смотри"), what="reply")
+    prompt = next(
+        p for c in ScriptedClient.instances for p in c.prompts if "смотри" in p
+    )
+    (day,) = list(root.iterdir())
+    assert day.name == stack.tg.attachments.day()
+    (saved,) = list(day.iterdir())
+    assert saved.read_bytes() == b"data" and saved.name.endswith("-u1.jpg")
+    assert prompt.endswith(
+        f"смотри\n\n[вложение: фото {saved}, 2 КБ; хранится 3 дн., "
+        "перенеси в vault, если нужно]"
+    )
+    stack.tg.attachments = Attachments(dir=root, keep_days=None)
+    stack.bot.media(_photo("ещё"))
+    await stack.until(lambda: stack.sent_with("ещё"), what="second reply")
+    prompt = next(p for c in ScriptedClient.instances for p in c.prompts if "ещё" in p)
+    assert prompt.endswith(" КБ]") and "хранится" not in prompt
+
+
+async def test_sticker_alone_becomes_an_unsupported_note(stack: Stack) -> None:
+    stack.bot.media(
+        {
+            "sticker": {
+                "file_id": "s1",
+                "file_unique_id": "su1",
+                "type": "regular",
+                "width": 1,
+                "height": 1,
+                "is_animated": False,
+                "is_video": False,
+            }
+        }
+    )
+    await stack.until(lambda: stack.sent_with("стикер"), what="reply")
+    prompt = next(
+        p for c in ScriptedClient.instances for p in c.prompts if "стикер" in p
+    )
+    assert prompt.endswith("[вложение: стикер - не поддерживается]")
+
+
+def test_sweep_drops_old_files_and_empty_day_folders() -> None:
+    root = Path(tempfile.mkdtemp(prefix="beaver-sweep-"))
+    old = root / "2020-01-01" / "1-old.jpg"
+    old.parent.mkdir()
+    old.write_bytes(b"x")
+    import os
+
+    os.utime(old, (1_577_836_800, 1_577_836_800))
+    fresh = root / "2099-01-01" / "2-new.jpg"
+    fresh.parent.mkdir()
+    fresh.write_bytes(b"y")
+    tg = TelegramFrontend(
+        token="t", user_id=USER, attachments=Attachments(dir=root, keep_days=None)
+    )
+    tg._sweep_attachments()  # noqa: SLF001
+    assert old.exists()
+    tg.attachments = Attachments(dir=root, keep_days=7)
+    tg._sweep_attachments()  # noqa: SLF001
+    assert not old.parent.exists() and fresh.exists()
+
+
+async def test_topic_rename_updates_the_title(stack: Stack) -> None:
+    stack.bot.topic_created("план", 7)
+    stack.bot.message("hello", thread=7)
+    await stack.until(lambda: stack.sent_with("hello"), what="reply")
+    stack.bot.topic_edited("новое имя", 7)
+
+    async def titled() -> Any:
+        conv = await stack.world.conversations.find_bound(
+            frontend="telegram", external_id=f"{USER}/7"
+        )
+        return conv if conv is not None and conv.title == "новое имя" else None
+
+    deadline = asyncio.get_running_loop().time() + 5
+    branch = None
+    while branch is None and asyncio.get_running_loop().time() < deadline:
+        branch = await titled()
+        await asyncio.sleep(0.02)
+    assert branch is not None
+    stack.tg._topic_names.clear()  # noqa: SLF001
+    assert await stack.tg.mark_topic(branch)
+    assert stack.bot.topics[-1] == "edit:7:✅ новое имя"
+
+
+async def test_unknown_callback_strips_the_keyboard(stack: Stack) -> None:
+    stack.bot.callback("q:nope:0:1", 555)
+    await stack.until(
+        lambda: {"message_id": 555, "markup": None} in stack.bot.edits,
+        what="keyboard removed",
+    )
+
+
+async def test_failed_background_turn_is_reported_once(stack: Stack) -> None:
+    stack.bot.message("hi")
+    await stack.until(lambda: stack.sent_with("hi"), what="reply")
+    master = await stack.world.conversations.find_bound(
+        frontend="telegram", external_id=GENERAL
+    )
+    event = {
+        "type": "turn.end",
+        "conversation_id": master.external_id,
+        "turn_id": "t-err",
+        "origin": "крон",
+        "stop": "error",
+    }
+    await stack.tg._on_event(event)  # noqa: SLF001
+    await stack.tg._on_event(event)  # noqa: SLF001
+    await stack.until(
+        lambda: stack.sent_with("фоновый тёрн (крон) упал"), what="notice"
+    )
+    await asyncio.sleep(0.2)
+    assert len([m for m in stack.bot.sent if "упал" in m["text"]]) == 1
+    await stack.tg._on_event({**event, "turn_id": "t-user", "origin": "user"})  # noqa: SLF001
+    notice = await stack.until(lambda: stack.sent_with("⚠️ тёрн упал"), what="user")
+    assert "фоновый" not in notice["text"]
+
+
+async def test_gone_topic_unbinds_and_the_next_message_starts_afresh(
+    stack: Stack,
+) -> None:
+    stack.bot.message("hello", thread=7)
+    await stack.until(lambda: stack.sent_with("hello"), what="reply")
+    branch = await stack.world.conversations.find_bound(
+        frontend="telegram", external_id=f"{USER}/7"
+    )
+    stack.bot.gone_threads.add(7)
+    await stack.world.conversations.say(branch, "psst")
+
+    async def unbound() -> bool:
+        rows = await stack.world.conversations.bindings(branch)
+        return all(not b.visible for b in rows)
+
+    deadline = asyncio.get_running_loop().time() + 5
+    while not await unbound() and asyncio.get_running_loop().time() < deadline:
+        await asyncio.sleep(0.02)
+    assert await unbound()
+    assert branch.external_id not in stack.tg._targets  # noqa: SLF001
+    async with stack.world.db.session() as session:
+        rows = list((await session.exec(select(Delivery))).all())
+    assert [r.status for r in rows if r.text == "psst"] == ["failed"]
+    stack.bot.gone_threads.clear()
+    stack.bot.message("again", thread=7)
+    await stack.until(lambda: stack.sent_with("again"), what="fresh reply")
+    fresh = await stack.world.conversations.find_bound(
+        frontend="telegram", external_id=f"{USER}/7"
+    )
+    assert fresh is not None and fresh.id != branch.id and fresh.status == "open"

ui/src/lib/api/types.ts

@@ -270,6 +270,17 @@ export interface QueuedJob {
   status: string;
 }
 
+export interface JobRunInfo {
+  error: string | null;
+  finished_at: string | null;
+  id: number | null;
+  job: string;
+  payload: Record<string, unknown>;
+  started_at: string;
+  status: string;
+  trigger: string;
+}
+
 export interface JobInfo {
   critical: boolean;
   cron: string | null;
@@ -277,7 +288,7 @@ export interface JobInfo {
   last_run: string | null;
   name: string;
   next_run: string | null;
-  run: { trigger: string; started_at: string; status: string } | null;
+  run: JobRunInfo | null;
   status: string | null;
   webhook: boolean;
 }