Initial Codex Telegram admin bot

bot/__init__.py

@@ -0,0 +1 @@
+"""Codex Server Admin Telegram Bot."""

bot/codex_runner.py

@@ -0,0 +1,286 @@
+"""Runs tasks through the OpenAI Codex Python SDK and yields progress events."""
+
+import asyncio
+from dataclasses import dataclass
+from typing import Any, AsyncIterator
+
+from openai_codex import ApprovalMode, AsyncCodex, CodexConfig, Sandbox
+from openai_codex.types import ReasoningEffort, ThreadItem, ThreadTokenUsage, Turn
+
+from . import config
+
+TOOL_INPUT_PREVIEW_LEN = 100
+
+
+@dataclass
+class ToolEvent:
+    name: str
+    brief: str
+
+
+@dataclass
+class TextEvent:
+    text: str
+
+
+@dataclass
+class DoneEvent:
+    thread_id: str | None
+    result_text: str | None
+    is_error: bool
+    status: str
+    duration_ms: int | None
+    usage: ThreadTokenUsage | None
+
+
+def _shorten(text: str, limit: int = TOOL_INPUT_PREVIEW_LEN) -> str:
+    text = " ".join(text.split())
+    if len(text) > limit:
+        return text[: limit - 1] + "..."
+    return text
+
+
+def _enum_value(value: Any) -> str:
+    return getattr(value, "value", str(value))
+
+
+def _item_root(item: Any) -> Any:
+    return item.root if hasattr(item, "root") else item
+
+
+def _thread_item_type(item: Any) -> str:
+    return str(getattr(_item_root(item), "type", "item"))
+
+
+def _tool_event_from_item(item: Any, completed: bool = False) -> ToolEvent | None:
+    root = _item_root(item)
+    item_type = getattr(root, "type", None)
+
+    if item_type == "commandExecution":
+        status = _enum_value(getattr(root, "status", ""))
+        exit_code = getattr(root, "exit_code", None)
+        suffix = f" -> exit {exit_code}" if completed and exit_code is not None else ""
+        return ToolEvent("command", f"{status}{suffix}: {_shorten(root.command)}")
+
+    if item_type == "fileChange":
+        status = _enum_value(getattr(root, "status", ""))
+        changes = getattr(root, "changes", []) or []
+        paths = ", ".join(_shorten(getattr(change, "path", ""), 40) for change in changes[:3])
+        more = f" +{len(changes) - 3}" if len(changes) > 3 else ""
+        return ToolEvent("file", f"{status}: {paths}{more}" if paths else status)
+
+    if item_type == "mcpToolCall":
+        status = _enum_value(getattr(root, "status", ""))
+        server = getattr(root, "server", "")
+        tool = getattr(root, "tool", "")
+        return ToolEvent("mcp", f"{status}: {server}.{tool}")
+
+    if item_type == "dynamicToolCall":
+        status = _enum_value(getattr(root, "status", ""))
+        namespace = getattr(root, "namespace", None)
+        tool = getattr(root, "tool", "")
+        name = f"{namespace}.{tool}" if namespace else tool
+        return ToolEvent("tool", f"{status}: {name}")
+
+    if item_type == "webSearch":
+        return ToolEvent("web", _shorten(getattr(root, "query", "")))
+
+    if item_type == "collabAgentToolCall":
+        status = _enum_value(getattr(root, "status", ""))
+        tool = _enum_value(getattr(root, "tool", "agent"))
+        prompt = getattr(root, "prompt", None)
+        return ToolEvent("agent", f"{status}: {tool} {_shorten(prompt or '')}".strip())
+
+    return None
+
+
+def _text_event_from_item(item: Any) -> TextEvent | None:
+    root = _item_root(item)
+    item_type = getattr(root, "type", None)
+    if item_type == "plan":
+        text = getattr(root, "text", "").strip()
+        return TextEvent(text) if text else None
+    if item_type == "agentMessage":
+        phase = getattr(root, "phase", None)
+        if phase is not None and getattr(phase, "name", "") == "final_answer":
+            return None
+        text = getattr(root, "text", "").strip()
+        return TextEvent(text) if text else None
+    return None
+
+
+def _final_response_from_items(items: list[ThreadItem]) -> str | None:
+    fallback: str | None = None
+    for item in reversed(items):
+        root = _item_root(item)
+        if getattr(root, "type", None) != "agentMessage":
+            continue
+        text = getattr(root, "text", None)
+        if not text:
+            continue
+        phase = getattr(root, "phase", None)
+        if getattr(phase, "name", "") == "final_answer" or getattr(phase, "value", "") == "final_answer":
+            return text
+        if phase is None and fallback is None:
+            fallback = text
+    return fallback
+
+
+def _sandbox() -> Sandbox:
+    raw = config.CODEX_SANDBOX.strip().lower().replace("_", "-")
+    aliases = {
+        "read-only": Sandbox.read_only,
+        "readonly": Sandbox.read_only,
+        "workspace-write": Sandbox.workspace_write,
+        "workspace": Sandbox.workspace_write,
+        "full-access": Sandbox.full_access,
+        "full": Sandbox.full_access,
+        "danger-full-access": Sandbox.full_access,
+    }
+    if raw not in aliases:
+        valid = ", ".join(sorted(aliases))
+        raise RuntimeError(f"Unsupported CODEX_SANDBOX={config.CODEX_SANDBOX!r}; use one of: {valid}")
+    return aliases[raw]
+
+
+def _approval_mode() -> ApprovalMode:
+    try:
+        return ApprovalMode(config.CODEX_APPROVAL_MODE)
+    except ValueError as exc:
+        valid = ", ".join(mode.value for mode in ApprovalMode)
+        raise RuntimeError(
+            f"Unsupported CODEX_APPROVAL_MODE={config.CODEX_APPROVAL_MODE!r}; use one of: {valid}"
+        ) from exc
+
+
+def _effort(value: str | None) -> ReasoningEffort | None:
+    if value is None:
+        return None
+    try:
+        return ReasoningEffort(value)
+    except ValueError as exc:
+        valid = ", ".join(effort.value for effort in ReasoningEffort)
+        raise RuntimeError(f"Unsupported reasoning effort={value!r}; use one of: {valid}") from exc
+
+
+def _codex_config() -> CodexConfig:
+    return CodexConfig(codex_bin=config.CODEX_BIN)
+
+
+async def _get_thread(
+    codex: AsyncCodex,
+    thread_id: str | None,
+    model: str | None,
+    effort: str | None,
+) -> tuple[Any, bool]:
+    selected_model = model or config.CODEX_DEFAULT_MODEL
+    thread_config = {"model_reasoning_effort": effort} if effort else None
+    kwargs = {
+        "approval_mode": _approval_mode(),
+        "config": thread_config,
+        "cwd": config.WORK_DIR,
+        "developer_instructions": config.SYSTEM_PROMPT,
+        "model": selected_model,
+        "sandbox": _sandbox(),
+    }
+    if thread_id:
+        try:
+            return await codex.thread_resume(thread_id, **kwargs), True
+        except Exception:
+            return await codex.thread_start(**kwargs), False
+    return await codex.thread_start(**kwargs), False
+
+
+def _done_from_turn(
+    turn: Turn | None,
+    thread_id: str | None,
+    items: list[ThreadItem],
+    usage: ThreadTokenUsage | None,
+) -> DoneEvent:
+    if turn is None:
+        raise RuntimeError("turn completed event not received")
+    status = _enum_value(turn.status)
+    is_error = status in {"failed", "interrupted"}
+    result_text = _final_response_from_items(items)
+    if is_error and not result_text:
+        error = getattr(turn, "error", None)
+        result_text = getattr(error, "message", None) or f"Codex turn finished with status: {status}"
+    return DoneEvent(
+        thread_id=thread_id,
+        result_text=result_text,
+        is_error=is_error,
+        status=status,
+        duration_ms=getattr(turn, "duration_ms", None),
+        usage=usage,
+    )
+
+
+async def run_task(
+    prompt: str,
+    resume_thread_id: str | None,
+    model: str | None = None,
+    effort: str | None = None,
+) -> AsyncIterator[ToolEvent | TextEvent | DoneEvent]:
+    """Execute one task, yielding progress events and finally a DoneEvent."""
+    async with AsyncCodex(config=_codex_config()) as codex:
+        if config.OPENAI_API_KEY:
+            await codex.login_api_key(config.OPENAI_API_KEY)
+
+        thread, resumed = await _get_thread(codex, resume_thread_id, model, effort)
+        if resume_thread_id and not resumed:
+            yield ToolEvent("thread", "previous thread was unavailable; started a new one")
+
+        selected_model = model or config.CODEX_DEFAULT_MODEL
+        turn = await thread.turn(
+            prompt,
+            approval_mode=_approval_mode(),
+            cwd=config.WORK_DIR,
+            effort=_effort(effort),
+            model=selected_model,
+            sandbox=_sandbox(),
+        )
+
+        items: list[ThreadItem] = []
+        usage: ThreadTokenUsage | None = None
+        completed_turn: Turn | None = None
+
+        try:
+            async for event in turn.stream():
+                payload = event.payload
+                if event.method == "item/started":
+                    tool_event = _tool_event_from_item(payload.item)
+                    if tool_event is not None:
+                        yield tool_event
+                    continue
+
+                if event.method == "item/completed":
+                    items.append(payload.item)
+                    text_event = _text_event_from_item(payload.item)
+                    if text_event is not None:
+                        yield text_event
+                    if _thread_item_type(payload.item) in {
+                        "commandExecution",
+                        "fileChange",
+                        "mcpToolCall",
+                        "dynamicToolCall",
+                        "collabAgentToolCall",
+                    }:
+                        tool_event = _tool_event_from_item(payload.item, completed=True)
+                        if tool_event is not None:
+                            yield tool_event
+                    continue
+
+                if event.method == "thread/tokenUsage/updated":
+                    usage = payload.token_usage
+                    continue
+
+                if event.method == "turn/completed":
+                    completed_turn = payload.turn
+                    continue
+        except asyncio.CancelledError:
+            try:
+                await turn.interrupt()
+            finally:
+                raise
+
+        yield _done_from_turn(completed_turn, thread.id, items, usage)

bot/config.py

@@ -0,0 +1,64 @@
+"""Configuration loaded from environment variables (.env supported via python-dotenv)."""
+
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+PROJECT_DIR = Path(__file__).resolve().parent.parent
+
+TELEGRAM_API_ID = int(os.environ["TELEGRAM_API_ID"])
+TELEGRAM_API_HASH = os.environ["TELEGRAM_API_HASH"]
+TELEGRAM_BOT_TOKEN = os.environ["TELEGRAM_BOT_TOKEN"]
+
+ADMIN_IDS = [
+    int(part)
+    for part in os.environ.get("ADMIN_IDS", "").replace(" ", "").split(",")
+    if part
+]
+if not ADMIN_IDS:
+    raise RuntimeError("ADMIN_IDS is empty - set at least one Telegram user ID")
+
+WORK_DIR = os.environ.get("WORK_DIR", "/")
+
+SESSIONS_FILE = Path(os.environ.get("SESSIONS_FILE", PROJECT_DIR / "sessions.json"))
+
+OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") or None
+CODEX_DEFAULT_MODEL = os.environ.get("CODEX_DEFAULT_MODEL") or None
+CODEX_SANDBOX = os.environ.get("CODEX_SANDBOX", "full-access")
+CODEX_APPROVAL_MODE = os.environ.get("CODEX_APPROVAL_MODE", "auto_review")
+CODEX_BIN = os.environ.get("CODEX_BIN") or None
+
+SYSTEM_PROMPT = os.environ.get(
+    "SYSTEM_PROMPT",
+    "You are the administrator of this Linux server, operating via a Telegram bot. "
+    "Execute the tasks you are given carefully. Answer concisely: the reply is shown "
+    "in a Telegram chat. Reply in the same language the user writes in.",
+)
+
+
+def _parse_model_choices(raw: str) -> list[tuple[str, str]]:
+    choices: list[tuple[str, str]] = []
+    for item in raw.split(","):
+        item = item.strip()
+        if not item:
+            continue
+        if "=" in item:
+            label, value = item.split("=", 1)
+        else:
+            label = value = item
+        label = label.strip()
+        value = value.strip()
+        if label and value:
+            choices.append((label, value))
+    return choices
+
+
+MODEL_CHOICES = _parse_model_choices(
+    os.environ.get(
+        "CODEX_MODEL_CHOICES",
+        "GPT-5.5=gpt-5.5,GPT-5.4=gpt-5.4,GPT-5.4 mini=gpt-5.4-mini",
+    )
+)

bot/main.py

@@ -0,0 +1,271 @@
+"""Entry point: kurigram bot that runs admin tasks via Codex."""
+
+import asyncio
+import logging
+
+from pyrogram import Client, filters
+from pyrogram.enums import ChatAction, ParseMode
+from pyrogram.errors import RPCError
+from pyrogram.types import (
+    CallbackQuery,
+    InlineKeyboardButton,
+    InlineKeyboardMarkup,
+    Message,
+)
+
+from . import codex_runner, config
+from .codex_runner import DoneEvent, TextEvent, ToolEvent
+from .sessions import ChatStore
+from .telegram_ui import ProgressReporter, send_long_message
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+)
+log = logging.getLogger("codex-tg-bot")
+
+app = Client(
+    "codex_admin_bot",
+    api_id=config.TELEGRAM_API_ID,
+    api_hash=config.TELEGRAM_API_HASH,
+    bot_token=config.TELEGRAM_BOT_TOKEN,
+    workdir=str(config.PROJECT_DIR),
+)
+
+chats = ChatStore(config.SESSIONS_FILE)
+running_tasks: dict[int, asyncio.Task] = {}
+
+admin = filters.user(config.ADMIN_IDS)
+
+MODELS: list[tuple[str, str | None]] = [
+    *config.MODEL_CHOICES,
+    ("По умолчанию", None),
+]
+
+EFFORTS: list[tuple[str, str | None]] = [
+    ("minimal", "minimal"),
+    ("low", "low"),
+    ("medium", "medium"),
+    ("high", "high"),
+    ("xhigh", "xhigh"),
+    ("По умолчанию", None),
+]
+
+HELP_TEXT = (
+    "**Codex Server Admin Bot**\n\n"
+    "Любое текстовое сообщение - задача для Codex на сервере.\n"
+    "Контекст беседы сохраняется между сообщениями в этом чате.\n\n"
+    "Команды:\n"
+    "/new - начать новую беседу (сбросить контекст)\n"
+    "/model - выбрать модель и уровень reasoning effort\n"
+    "/status - статус текущей задачи и thread\n"
+    "/stop - отменить выполняющуюся задачу\n"
+    "/help - эта справка\n\n"
+    f"Рабочая директория: `{config.WORK_DIR}`\n"
+    f"Sandbox: `{config.CODEX_SANDBOX}`"
+)
+
+
+def _model_keyboard(chat_id: int) -> InlineKeyboardMarkup:
+    current_model = chats.get(chat_id, "model")
+    current_effort = chats.get(chat_id, "effort")
+
+    def label(name: str, value: str | None, current: str | None) -> str:
+        return f"* {name}" if value == current else name
+
+    rows = [[InlineKeyboardButton("- Модель -", callback_data="noop")]]
+    for i in range(0, len(MODELS), 2):
+        rows.append(
+            [
+                InlineKeyboardButton(
+                    label(name, value, current_model),
+                    callback_data=f"model:{value or ''}",
+                )
+                for name, value in MODELS[i : i + 2]
+            ]
+        )
+    rows.append([InlineKeyboardButton("- Reasoning effort -", callback_data="noop")])
+    for i in range(0, len(EFFORTS), 3):
+        rows.append(
+            [
+                InlineKeyboardButton(
+                    label(name, value, current_effort),
+                    callback_data=f"effort:{value or ''}",
+                )
+                for name, value in EFFORTS[i : i + 3]
+            ]
+        )
+    return InlineKeyboardMarkup(rows)
+
+
+MODEL_TEXT = (
+    "**Настройки Codex**\n\n"
+    "Выберите модель и reasoning effort для задач в этом чате.\n"
+    "\"По умолчанию\" использует настройки Codex runtime или CODEX_DEFAULT_MODEL."
+)
+
+
+@app.on_message(filters.command(["start", "help"]) & admin)
+async def cmd_help(_, message: Message):
+    await message.reply_text(HELP_TEXT, parse_mode=ParseMode.MARKDOWN)
+
+
+@app.on_message(filters.command("new") & admin)
+async def cmd_new(_, message: Message):
+    if chats.clear_thread(message.chat.id):
+        await message.reply_text("Контекст сброшен, начинаем новую беседу.")
+    else:
+        await message.reply_text("Контекста и не было - беседа новая.")
+
+
+@app.on_message(filters.command("model") & admin)
+async def cmd_model(_, message: Message):
+    await message.reply_text(
+        MODEL_TEXT,
+        parse_mode=ParseMode.MARKDOWN,
+        reply_markup=_model_keyboard(message.chat.id),
+    )
+
+
+@app.on_callback_query(admin)
+async def on_settings_callback(_, query: CallbackQuery):
+    data = query.data or ""
+    if data == "noop":
+        await query.answer()
+        return
+    key, _, raw = data.partition(":")
+    if key not in ("model", "effort"):
+        await query.answer()
+        return
+    value = raw or None
+    chat_id = query.message.chat.id
+    chats.set(chat_id, key, value)
+    shown = value or "по умолчанию"
+    await query.answer(f"{'Модель' if key == 'model' else 'Reasoning'}: {shown}")
+    try:
+        await query.message.edit_reply_markup(_model_keyboard(chat_id))
+    except RPCError:
+        pass
+
+
+@app.on_message(filters.command("status") & admin)
+async def cmd_status(_, message: Message):
+    chat_id = message.chat.id
+    task = running_tasks.get(chat_id)
+    state = "задача выполняется" if task and not task.done() else "бот свободен"
+    thread_id = chats.get(chat_id, "thread_id")
+    thread_line = f"`{thread_id}`" if thread_id else "нет (новая беседа)"
+    model = chats.get(chat_id, "model") or config.CODEX_DEFAULT_MODEL or "по умолчанию"
+    effort = chats.get(chat_id, "effort") or "по умолчанию"
+    await message.reply_text(
+        f"{state}\nThread: {thread_line}\n"
+        f"Модель: `{model}`\nReasoning: `{effort}`\n"
+        f"Рабочая директория: `{config.WORK_DIR}`\nSandbox: `{config.CODEX_SANDBOX}`",
+        parse_mode=ParseMode.MARKDOWN,
+    )
+
+
+@app.on_message(filters.command("stop") & admin)
+async def cmd_stop(_, message: Message):
+    task = running_tasks.get(message.chat.id)
+    if task and not task.done():
+        task.cancel()
+        await message.reply_text("Останавливаю задачу...")
+    else:
+        await message.reply_text("Сейчас ничего не выполняется.")
+
+
+@app.on_message(
+    filters.text
+    & ~filters.command(["start", "help", "new", "model", "status", "stop"])
+    & admin
+)
+async def on_task(client: Client, message: Message):
+    chat_id = message.chat.id
+    existing = running_tasks.get(chat_id)
+    if existing and not existing.done():
+        await message.reply_text(
+            "Задача уже выполняется. Дождитесь завершения или отмените: /stop"
+        )
+        return
+    running_tasks[chat_id] = asyncio.create_task(_execute(client, message))
+
+
+@app.on_message()
+async def on_other(_, message: Message):
+    user = message.from_user
+    if user and user.id in config.ADMIN_IDS:
+        await message.reply_text("Я понимаю только текстовые сообщения.")
+        return
+    log.warning(
+        "Ignoring message from unauthorized user id=%s username=%s",
+        user.id if user else "?",
+        user.username if user else "?",
+    )
+
+
+async def _execute(client: Client, message: Message):
+    chat_id = message.chat.id
+    await client.send_chat_action(chat_id, ChatAction.TYPING)
+    progress = ProgressReporter(await message.reply_text("Работаю..."))
+    done: DoneEvent | None = None
+    try:
+        async for event in codex_runner.run_task(
+            message.text,
+            chats.get(chat_id, "thread_id"),
+            model=chats.get(chat_id, "model"),
+            effort=chats.get(chat_id, "effort"),
+        ):
+            if isinstance(event, ToolEvent):
+                progress.add_line(f"{event.name}: {event.brief}")
+                await progress.flush()
+            elif isinstance(event, TextEvent):
+                first_line = event.text.strip().splitlines()[0]
+                progress.add_line(f"note: {first_line[:100]}")
+                await progress.flush()
+            elif isinstance(event, DoneEvent):
+                done = event
+    except asyncio.CancelledError:
+        await progress.delete()
+        await message.reply_text("Задача отменена.")
+        return
+    except Exception:
+        log.exception("Task failed in chat %s", chat_id)
+        await progress.delete()
+        await message.reply_text(
+            "Ошибка при выполнении задачи, подробности в логах бота."
+        )
+        return
+    finally:
+        running_tasks.pop(chat_id, None)
+
+    await progress.delete()
+    if done is None:
+        await message.reply_text("Codex завершился без результата.")
+        return
+    if done.thread_id:
+        chats.set(chat_id, "thread_id", done.thread_id)
+
+    status = "ERROR" if done.is_error else "OK"
+    duration = f" · {done.duration_ms / 1000:.1f}s" if done.duration_ms else ""
+    tokens = _usage_summary(done)
+    thread_short = f" · thread {done.thread_id[:8]}..." if done.thread_id else ""
+    footer = f"\n\n{status} · {done.status}{duration}{tokens}{thread_short}"
+    body = done.result_text or "(пустой ответ)"
+    await send_long_message(message, body + footer)
+
+
+def _usage_summary(done: DoneEvent) -> str:
+    if done.usage is None:
+        return ""
+    last = done.usage.last
+    return f" · {last.total_tokens} tokens"
+
+
+def main():
+    log.info("Starting bot, admins: %s, work dir: %s", config.ADMIN_IDS, config.WORK_DIR)
+    app.run()
+
+
+if __name__ == "__main__":
+    main()

bot/sessions.py

@@ -0,0 +1,55 @@
+"""Per-chat state persistence: thread_id, model and effort in a JSON file."""
+
+import json
+from pathlib import Path
+from typing import Any
+
+
+class ChatStore:
+    def __init__(self, path: Path):
+        self._path = path
+        self._chats: dict[str, dict[str, Any]] = {}
+        if path.exists():
+            try:
+                data = json.loads(path.read_text(encoding="utf-8"))
+                self._chats = {
+                    k: (v if isinstance(v, dict) else {"thread_id": v})
+                    for k, v in data.items()
+                }
+                self._migrate_session_ids()
+            except (json.JSONDecodeError, OSError):
+                self._chats = {}
+
+    def get(self, chat_id: int, key: str) -> str | None:
+        value = self._chats.get(str(chat_id), {}).get(key)
+        return value if isinstance(value, str) else None
+
+    def set(self, chat_id: int, key: str, value: str | None) -> None:
+        chat = self._chats.setdefault(str(chat_id), {})
+        if value is None:
+            chat.pop(key, None)
+        else:
+            chat[key] = value
+        self._save()
+
+    def clear_thread(self, chat_id: int) -> bool:
+        chat = self._chats.get(str(chat_id), {})
+        removed = False
+        for key in ("thread_id", "session_id"):
+            removed = chat.pop(key, None) is not None or removed
+        if removed:
+            self._save()
+        return removed
+
+    def _migrate_session_ids(self) -> None:
+        changed = False
+        for chat in self._chats.values():
+            if "thread_id" not in chat and "session_id" in chat:
+                chat["thread_id"] = chat["session_id"]
+                changed = True
+        if changed:
+            self._save()
+
+    def _save(self) -> None:
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        self._path.write_text(json.dumps(self._chats, indent=2), encoding="utf-8")

bot/telegram_ui.py

@@ -0,0 +1,83 @@
+"""Telegram output helpers: throttled progress edits, long-message splitting."""
+
+import asyncio
+import time
+
+from pyrogram.enums import ParseMode
+from pyrogram.errors import FloodWait, MessageNotModified, RPCError
+from pyrogram.types import Message
+
+TELEGRAM_MESSAGE_LIMIT = 4096
+EDIT_INTERVAL_SECONDS = 2.0
+PROGRESS_MAX_LINES = 15
+
+
+class ProgressReporter:
+    """Edits a single working message with throttling as the task progresses."""
+
+    def __init__(self, message: Message):
+        self._message = message
+        self._lines: list[str] = []
+        self._last_edit = 0.0
+        self._last_text = message.text or ""
+        self._lock = asyncio.Lock()
+
+    def add_line(self, line: str) -> None:
+        self._lines.append(line)
+
+    def _render(self) -> str:
+        shown = self._lines[-PROGRESS_MAX_LINES:]
+        hidden = len(self._lines) - len(shown)
+        header = "Работаю..."
+        if hidden > 0:
+            header += f" (скрыто {hidden} строк)"
+        text = header + "\n" + "\n".join(shown)
+        return text[:TELEGRAM_MESSAGE_LIMIT]
+
+    async def flush(self, force: bool = False) -> None:
+        async with self._lock:
+            now = time.monotonic()
+            if not force and now - self._last_edit < EDIT_INTERVAL_SECONDS:
+                return
+            text = self._render()
+            if text == self._last_text:
+                return
+            try:
+                await self._message.edit_text(text, parse_mode=ParseMode.DISABLED)
+                self._last_text = text
+                self._last_edit = time.monotonic()
+            except MessageNotModified:
+                self._last_text = text
+            except FloodWait as e:
+                self._last_edit = time.monotonic() + e.value
+            except RPCError:
+                pass
+
+    async def delete(self) -> None:
+        try:
+            await self._message.delete()
+        except RPCError:
+            pass
+
+
+def split_text(text: str, limit: int = TELEGRAM_MESSAGE_LIMIT) -> list[str]:
+    """Split text into chunks under the Telegram limit, preferring line breaks."""
+    chunks = []
+    while len(text) > limit:
+        cut = text.rfind("\n", 0, limit)
+        if cut < limit // 2:
+            cut = limit
+        chunks.append(text[:cut])
+        text = text[cut:].lstrip("\n")
+    if text:
+        chunks.append(text)
+    return chunks
+
+
+async def send_long_message(message: Message, text: str) -> None:
+    """Reply with text of any length, falling back to plain text on Markdown errors."""
+    for chunk in split_text(text):
+        try:
+            await message.reply_text(chunk, parse_mode=ParseMode.MARKDOWN)
+        except RPCError:
+            await message.reply_text(chunk, parse_mode=ParseMode.DISABLED)