simbiot-harness/harness/tg_inbox.py

"""Simbiot Harness — Telegram-инбокс.

Входящие сообщения Алекса → события MessageReceived в стор →
минутный уровень видит их в контексте и может ответить через
`cli.py tg-reply <msg_id> <text>` (событие ReplySent закрывает входящее).

Поллер — отдельный процесс (systemd simbiot-harness-tg), long polling
getUpdates offset=last+1. Идемпотентность: update_id → aggregate_id.
"""
import json
import re
import os
import sys
import time
import uuid
import urllib.request
import urllib.parse
import urllib.error

from . import es

from . import paths as _paths  # t-261004113351-2858c1: .env инстанса

_HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
_ENV_PATH = _paths.env_path()

TG_TOKEN = None
ALLOWED = set()


def _load_env():
    global TG_TOKEN, ALLOWED
    if os.path.exists(_ENV_PATH):
        with open(_ENV_PATH) as f:
            for line in f:
                line = line.strip()
                if line.startswith("SIMBIOT_TG_TOKEN="):
                    TG_TOKEN = line.split("=", 1)[1].strip()
                elif line.startswith("SIMBIOT_TG_ALLOWED="):
                    ALLOWED = {x.strip() for x in line.split("=", 1)[1].split(",") if x.strip()}
    _tok = os.environ.get("SIMBIOT_TG_TOKEN", "").strip()  # systemd EnvironmentFile задаёт переменные, но не пишет .env
    if not TG_TOKEN and _tok:
        TG_TOKEN = _tok
    _allow = os.environ.get("SIMBIOT_TG_ALLOWED", "").strip()
    if TG_TOKEN and not ALLOWED and _allow:
        ALLOWED = {x.strip() for x in _allow.split(",") if x.strip()}


def _api(method: str, params: dict, timeout: int = 35):
    url = f"https://api.telegram.org/bot{TG_TOKEN}/{method}"
    data = urllib.parse.urlencode(params).encode()
    req = urllib.request.Request(url, data=data)
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode())


def send_message(chat_id, text: str, reply_to=None, timeout: int = 35):
    """Отправка сообщения (используется tg-reply и алертами).

    timeout: сек-тик (без-LLM команды) передаёт короткий — рефлекс не
    должен висеть на TG дольше пары секунд.
    """
    _load_env()
    params = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"}
    if reply_to:
        params["reply_to_message_id"] = reply_to
        # сообщение удалили — ответ всё равно уходит (без цитаты), не теряется
        params["allow_sending_without_reply"] = "true"
    try:
        r = _api("sendMessage", params, timeout=timeout)
    except urllib.error.HTTPError as e:
        # 400 от кривого HTML (незакрытый тег, «a < b» в тексте) — шлём как plain text.
        # Иначе ответ/доставка молча теряются: исключение глотается на верхнем уровне.
        if e.code != 400:
            raise
        params.pop("parse_mode", None)
        r = _api("sendMessage", params, timeout=timeout)
    _record_outbound(chat_id, r, text, reply_to)
    return r


def _record_outbound(chat_id, r, text: str, reply_to=None):
    """Журнал исходящих (аудит 2026-10-17): message_id → полный текст. Alex
    ответил «Ок делай» на итог рефлексии, а Telegram отдаёт в reply только
    начало, харнес хранил 300 символов — тик не видел самого предложения.
    Best effort: журнал не мешает отправке."""
    try:
        res = (r or {}).get("result") if isinstance(r, dict) else None
        if not (isinstance(res, dict) and res.get("message_id")):
            return
        es.append_event("outbound", f"{chat_id}:{res['message_id']}", "OutboundMessage", {
            "chat_id": chat_id, "message_id": res["message_id"], "text": (text or "")[:4000],
            "reply_to": reply_to})
    except Exception:
        pass


def _strip_html(s: str) -> str:
    return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", "", s or "")).strip()


def resolve_bot_message(chat_id, message_id, prefix: str = "") -> str:
    """Полный текст НАШЕГО сообщения, на которое ответили: журнал исходящих по
    (chat, message_id); для старых — по совпадению начала с доставками/ответами."""
    import sqlite3
    try:
        db = sqlite3.connect(f"file:{es.DB_PATH}?mode=ro", uri=True, timeout=30)
        try:
            row = db.execute("SELECT data FROM events WHERE aggregate='outbound' AND aggregate_id=? "
                             "LIMIT 1", (f"{chat_id}:{message_id}",)).fetchone()
            if row:
                return json.loads(row[0]).get("text") or ""
            want = _strip_html(prefix)[:150]
            if len(want) < 20:
                return ""
            for (d,) in db.execute(
                    "SELECT data FROM events WHERE event_type IN ('DeliverySent','ReplySent',"
                    "'FollowupSent','OutboundMessage') ORDER BY id DESC LIMIT 3000"):
                t = json.loads(d).get("text") or ""
                if _strip_html(t).startswith(want):
                    return t
        finally:
            db.close()
    except Exception:
        pass
    return ""


def set_reaction(chat_id, message_id: int, emoji: str = "👀", is_big: bool = False):
    """setMessageReaction: видимый маркер «агент взял/смотрит» (e-261002202857).

    Идентичность Telegram: повторная постановка ТОГО ЖЕ эмодзи — no-op API;
    смена эмодзи на другой видна как изменение реакции. Идемпотентный повтор
    безопасен. Реакция допустима на сообщениях не старше 48 часов.
    """
    # Bot API: параметр reaction — МАССИВ ReactionType. Было reaction_type с
    # одиночным объектом: Telegram видел пустую реакцию — в личке молча
    # «снимал реакции» (ok, MessageSeen писался, 👀 не появлялся), в группе
    # отвечал 400 REACTION_EMPTY (аудит 2026-10-17).
    _load_env()
    params = {"chat_id": chat_id, "message_id": message_id,
              "reaction": json.dumps([{"type": "emoji", "emoji": emoji}]),
              "is_big": json.dumps(is_big)}
    return _api("setMessageReaction", params, timeout=10)


def mark_seen(msg_id: str, emoji: str = "👀", is_big: bool = False):
    """Реакция на входящее по id + событие MessageSeen.

    Один флаг seen в проекции → реакция ставится ровно один раз на
    сообщение (двумя контурами: поллер при приёме и мин-LLM после
    взятия батча; флаг общий, гонки нет — тот, кто первый).
    Успех события НЕ зависит от доступности TG на момент постановки:
    ставим MessageSeen, только если реакция ушла в TG успешно; иначе
    флаг не подаём и следующий маркер повторит попытку.
    """
    _load_env()
    model = es.run_projections()
    m = (model.get("inbox") or {}).get(msg_id)
    if m is None:
        return False
    if m.get("seen"):
        return False  # уже помечено — идемпотентный SKIP (реакция уже стоит)
    chat_id = m.get("chat_id")
    tg_mid = m.get("message_id") or (m.get("update_id") or 0) % 100000
    if str(chat_id or "").startswith("portal"):
        try:
            from . import portal_channel
            portal_channel.mark_seen(msg_id)
        except Exception:
            pass
        es.append_event("inbox", msg_id, "MessageSeen", {"emoji": "portal"})
        return True
    try:
        set_reaction(chat_id, tg_mid, emoji=emoji, is_big=is_big)
    except Exception:
        return False  # TG недоступен: флаг не подаём — повторит следующий маркер
    es.append_event("inbox", msg_id, "MessageSeen", {"emoji": emoji})
    return True


def typing_keepalive(chat_ids: list, stop, interval: float = 2.0):
    """Фоновый индикатор «работаю» на длинных тиках (e-261002202857).

    sendChatAction typing живёт ~5с и сам не обновляется — обновляем каждые
    interval секунд, пока caller не остановит (threading.Event в finally
    run_level_tick). Живой пульс виден весь тик: «печатает…» == «взял,
    готовлю ответ». Индикатор — best effort: сбои TG не валят тик.
    """
    _load_env()  # токен для _api: в процессе scheduler может быть не загружен
    while not stop.wait(max(0.5, min(interval, 5.0))):
        for chat_id in set(chat_ids):
            try:
                _api("sendChatAction", {"chat_id": chat_id, "action": "typing"},
                     timeout=10)
            except Exception:
                pass


# ── входящие документы/фото → диск (t-261005135853) ─────────────────────────
# Голосовые качает voice-пайплайн (voice/, кэш ASR), а documents/photo раньше
# терялись: file_id есть в событии, файл не скачивался (кейс 4 PDF-выписок
# Alex 05.10 — доставал вручную getFile'ом). Теперь гейтвей сам кладёт файл
# в inbox/files/<file_unique_id>.<ext> и пишет путь в MessageReceived.
_SAFE_EXT_RE = None  # ленивый импорт re


def _file_ext(name: str, mime: str = "") -> str:
    """Расширение из имени/mime, безопасное (без путей и мусора)."""
    global _SAFE_EXT_RE
    if _SAFE_EXT_RE is None:
        import re
        _SAFE_EXT_RE = re.compile(r"[^A-Za-z0-9]")
    ext = ""
    if name and "." in name:
        ext = name.rsplit(".", 1)[-1]
    if not ext and "/" in (mime or ""):
        sub = mime.split("/", 1)[-1]
        ext = {"pdf": "pdf", "jpeg": "jpg", "png": "png", "webp": "webp",
               "octet-stream": "bin"}.get(sub, sub)
    ext = _SAFE_EXT_RE.sub("", ext)[:8]
    return ext or "bin"


def save_document(msg: dict):
    """Скачать document/photo из сообщения → inbox/files/.

    Возвращает абсолютный путь к файлу или None: сбой скачивания не должен
    валить поллер — путь просто не попадёт в событие, file_id в нём
    останется для ручного достёгивания (как было до фикса).
    Идемпотентно: file_unique_id в имени → повторный апдейт не перекачивает
    (файл на диске = готовый ответ).
    """
    doc = msg.get("document")
    photo = msg.get("photo")
    if not doc and not photo:
        return None
    if doc:
        file_id = doc.get("file_id")
        uniq = doc.get("file_unique_id") or file_id
        ext = _file_ext(doc.get("file_name") or "", doc.get("mime_type") or "")
    else:
        biggest = photo[-1]  # самый большой размер
        file_id = biggest.get("file_id")
        uniq = biggest.get("file_unique_id") or file_id
        ext = "jpg"
    if not file_id:
        return None
    files_dir = _paths.inbox_files_dir()
    abs_path = os.path.join(files_dir, f"{uniq}.{ext}")
    if not os.path.exists(abs_path):  # кэш: один файл — одна закачка
        try:
            _load_env()
            if not TG_TOKEN:
                return None
            resp = _api("getFile", {"file_id": file_id})
            if not resp.get("ok"):
                return None
            fp = (resp.get("result") or {}).get("file_path")
            if not fp:
                return None
            durl = f"https://api.telegram.org/file/bot{TG_TOKEN}/{fp}"
            with urllib.request.urlopen(durl, timeout=90) as r:
                blob = r.read()
            if not blob:
                return None
            os.makedirs(files_dir, exist_ok=True)
            tmp = abs_path + ".part"
            with open(tmp, "wb") as f:
                f.write(blob)
            os.replace(tmp, abs_path)  # атомарно: нет наполовину записанных
        except urllib.error.HTTPError as e:
            # 400/410 = file истёк у TG — не ошибка поллера, тихий отказ
            print(f"[tg] getFile HTTP {e.code}: файл {uniq} недоступен", flush=True)
            return None
        except Exception as e:
            print(f"[tg] download {uniq} error: {e}", flush=True)
            return None
    return abs_path


def _describe_inbox_message(m: dict) -> str:
    """Короткое описание входящего для якоря реплая (t-261013131228).

    Документ/фото без текста описываем тем, что есть: kind, file_name,
    начало caption (text), путь к файлу — этого хватает триажу, чтобы
    понять, НА ЧТО отвечает Alex, не открывая файл.
    """
    kind = m.get("kind") or "text"
    text = (m.get("text") or "").strip()
    if text:
        return f"{kind}: {text[:200]}"
    fname = m.get("file_name") or ""
    fpath = m.get("file_path") or ""
    if fname or fpath:
        return f"{kind} (файл): {fname or (fpath or '').split('/')[-1]}"
    return f"[{kind} без текста]"


def _resolve_reply_target(chat_id, reply_to_message_id: int):
    """Найти в инбоксе сообщение, на которое отвечает Alex (t-261013131228).

    Telegram кладёт в reply_to_message только текстовые реплики; цель без
    текста (документ/фото/стикер) остаётся «чёрным ящиком». Ищем её в
    инбоксе по (chat_id, message_id) и описываем. None — цели в инбоксе
    нет (старее капа 100 или из чата, который бот не видел).
    """
    try:
        model = es.run_projections()
    except Exception:
        return None
    for m in (model.get("inbox") or {}).values():
        if m.get("chat_id") == chat_id and m.get("message_id") == reply_to_message_id:
            return _describe_inbox_message(m)
    return None


_BOT = {}


def bot_identity() -> dict:
    """getMe один раз на процесс: {"id", "username"} (пусто, если TG недоступен)."""
    if not _BOT:
        try:
            _load_env()
            r = (_api("getMe", {}, timeout=15) or {}).get("result") or {}
            _BOT.update(id=r.get("id"), username=(r.get("username") or "").lower())
        except Exception:
            return {}
    return _BOT


def _mentions_bot(msg: dict) -> bool:
    """@username бота в тексте/подписи или text_mention на бота."""
    me = bot_identity()
    text = msg.get("text") or msg.get("caption") or ""
    for ent in (msg.get("entities") or []) + (msg.get("caption_entities") or []):
        if ent.get("type") == "mention" and me.get("username"):
            frag = text[ent.get("offset", 0): ent.get("offset", 0) + ent.get("length", 0)]
            if frag.lstrip("@").lower() == me["username"]:
                return True
        if ent.get("type") == "text_mention" and me.get("id") \
                and (ent.get("user") or {}).get("id") == me["id"]:
            return True
    return bool(me.get("username")) and ("@" + me["username"]) in text.lower()


def receive_message(update: dict):
    """MessageReceived в стор. Идемпотентно по update_id."""
    msg = update.get("message") or update.get("edited_message") or {}
    chat = msg.get("chat") or {}
    user = msg.get("from") or {}
    # нетекстовые сообщения: помечаем, что это (стикер/фото/голос/…),
    # иначе пустой text выглядит как «пустое сообщение» и триаж не понимает, что случилось
    kind = "text"
    if msg.get("sticker"):
        kind = "sticker"
    elif msg.get("photo"):
        kind = "photo"
    elif msg.get("voice"):
        kind = "voice"
    elif msg.get("video_note") or msg.get("video"):
        kind = "video"
    elif msg.get("audio"):
        kind = "audio"
    elif msg.get("document"):
        kind = "document"
    elif msg.get("animation"):
        kind = "animation"
    # file_id нужен, чтобы потом скачать файл (голосовое → транскрипция, фото → OCR)
    file = (msg.get("voice") or msg.get("audio") or msg.get("document")
            or msg.get("video") or msg.get("video_note") or {})
    file_id = file.get("file_id")
    duration = file.get("duration") or file.get("duration_seconds")
    photo = msg.get("photo")
    if photo and not file_id:
        file_id = photo[-1].get("file_id")  # самый большой размер
    # t-261005135853: documents/photo сразу на диск при приёме — путь в
    # событии, файл не теряется (голосовые качает voice-пайплайн отдельно).
    # Сбой скачивания не валит приём: file_path=None, file_id остаётся.
    doc_rel = None
    if kind in ("document", "photo"):
        try:
            doc_rel = save_document(msg)
        except Exception as e:
            print(f"[tg] save_document error: {e}", flush=True)
    data = {
        "update_id": update["update_id"],
        "chat_id": chat.get("id"),
        # t-261013113629: тип чата (private/group/supergroup/channel) и название —
        # маршрутизация ответов в чат-источник (группа vs личка) и дефолтный
        # чат уведомлений (личка Alex, а не последняя активная группа)
        "chat_type": chat.get("type"),
        "chat_title": chat.get("title"),
        # tg message_id нужен для setMessageReaction (реакция «seen», e-261002202857):
        # update_id ≠ message_id (глобальный стрим vs чат), фолбэк — старый hash
        "message_id": msg.get("message_id"),
        "user_id": user.get("id"),
        "username": user.get("username"),
        "first_name": user.get("first_name"),
        "text": (msg.get("text") or msg.get("caption") or "").strip()[:10000],
        "kind": kind,
        "file_id": file_id,
        # file_unique_id: ключ аудио-кэша и идемпотентности расшифровки
        # (file_id ротируется, unique — постоянный для одного файла)
        "file_unique_id": file.get("file_unique_id") or (photo[-1].get("file_unique_id") if photo else None),
        # t-261005135853: путь к скачанному файлу (абсолютный, открывается
        # напрямую с любого уровня) + исходное имя/размер/mime — триаж видит
        # что за файл, без getFile. None = скачать не удалось (file_id остаётся).
        "file_path": doc_rel,
        "file_name": (msg.get("document") or {}).get("file_name"),
        "file_size": (msg.get("document") or {}).get("file_size")
                     or (photo[-1].get("file_size") if photo else None),
        "mime_type": (msg.get("document") or {}).get("mime_type"),
        "duration": duration,
        "date": msg.get("date"),
        # reply_to_message_id (t-261002085926): когда Alex отвечает (reply) на
        # конкретное сообщение, харнес должен знать, на какое именно — иначе
        # ссылка «я же сделал ответ на твоём сообщение» теряется и триаж
        # видит только голый текст. message_thread_id (форумы/топики) — тоже.
        "reply_to_message_id": (msg.get("reply_to_message") or {}).get("message_id"),
        "reply_to_text": ((msg.get("reply_to_message") or {}).get("text") or "")[:300],
        "message_thread_id": msg.get("message_thread_id"),
    }
    # t-261013131228: Alex реплаит на сообщение без текста (документ/фото/
    # стикер) — Telegram не отдаёт его содержимое в reply_to_message, и
    # триаж видит голый «[reply на сообщение без текста]» без контекста
    # (кейс 13.10: реплай на концепт ИИШНИЦЫ → мин-уровень анонсировал
    # не то событие). Резолвим цель по инбоксу того же чата: (chat_id,
    # reply_to_message_id) → описание цели (текст/caption/файл).
    rt = data.get("reply_to_message_id")
    if rt is not None and not (data.get("reply_to_text") or "").strip():
        target = _resolve_reply_target(chat.get("id"), rt)
        if target:
            data["reply_to_desc"] = target
    # t-261015080211: явный author_* блок + received_at — агентам проще
    # читать метаданные автора одним полем, не собирая по отдельным ключам
    data["author"] = {
        "user_id": data.get("user_id"),
        "username": data.get("username"),
        "first_name": data.get("first_name"),
        "chat_id": data.get("chat_id"),
        "chat_type": data.get("chat_type"),
        "chat_title": data.get("chat_title"),
        "date": data.get("date"),
    }
    data["received_at"] = data.get("date")
    # адресация в группах (аудит 2026-10-17, «Коди встревает в разговор»):
    # кому reply (боту или человеку) и упомянут ли бот — сырьё для
    # harness/addressing.py, решающего, будить ли LLM вообще
    rfrom = (msg.get("reply_to_message") or {}).get("from") or {}
    if rfrom:
        data["reply_to_user_id"] = rfrom.get("id")
        data["reply_to_is_bot"] = bool(rfrom.get("is_bot"))
        data["reply_to_first_name"] = rfrom.get("first_name")
    data["mentions_bot"] = _mentions_bot(msg)
    if data.get("reply_to_is_bot") and data.get("reply_to_message_id"):
        full = resolve_bot_message(chat.get("id"), data["reply_to_message_id"],
                                   data.get("reply_to_text") or "")
        if full:
            data["reply_to_full"] = full[:4000]
    es.append_event("inbox", f"m-{update['update_id']}", "MessageReceived", data)
    return f"m-{update['update_id']}"


def _group_reply_to(items) -> "int | None":
    """В группе отвечаем reply'ем на последнее сообщение батча — видно, кому
    и на что ответ (аудит 2026-10-17: ответ отдельным сообщением через
    20–30 с выглядел как «бот не реагирует»). Личка — без цитаты, как было."""
    from .addressing import is_group
    items = [m for m in items if m and m.get("message_id")]
    if not items or not is_group(items[0]):
        return None
    return max(items, key=lambda m: m.get("ts") or "").get("message_id")


def send_reply(msg_id: str, text: str, timeout: int = 35):
    """Ответ на входящее: Telegram-отправка + событие ReplySent (закрывает входящее).

    Перед отправкой шлём sendChatAction typing — в TG видно «печатает…».
    sendChatAction живёт ~5с и НЕ обновляется сам — этого достаточно для
    минутного пульса (типичный лаг ответа < 5с после начала отправки).
    timeout — как в send_message (короткий для сек-рефлекса).
    """
    model = es.run_projections()
    m = (model.get("inbox") or {}).get(msg_id)
    if m is None:
        raise ValueError(f"входящее {msg_id} не найдено")
    if m.get("replied"):
        raise ValueError(f"на {msg_id} уже отвечено")
    # ── файловый портал-канал (t-261001123121): чаты 'portal:*' отвечаются
    # HTTP-push'ем на портал — БЕЗ Telegram. У агента может не быть TG-бота
    # (simbiot-деплой делает TG опциональным), а портал — не телеграм-юзер:
    # разговоры не смешиваются (по chat_id), доставка — отдельный транспорт.
    if str(m.get("chat_id") or "").startswith("portal:agent:"):
        # канал агент↔агент (t-261015090129): ответ уходит ask_agent'ом
        # в очередь того агента, чьё сообщение триажируем. Транспорт — MCP.
        from . import agent_channel
        to_agent = str(m.get("chat_id")).split(":", 2)[-1]
        kind = {"help": "help", "review": "review", "handoff": "handoff"}.get(
            m.get("agent_kind"), "help")
        ok = agent_channel.push_agent(to_agent, text, kind)
        if not ok:
            raise RuntimeError(f"ask_agent({to_agent}) не доставил — ответят на следующем тике")
        es.append_event("inbox", msg_id, "ReplySent", {
            "to_chat": m["chat_id"], "text": text[:2000], "channel": "agent_mcp",
        })
        return msg_id
    if str(m.get("chat_id") or "").startswith("portal"):
        from . import portal_channel
        ok = portal_channel.push(m["chat_id"], text, [m.get("update_id") or msg_id])
        if not ok:
            # доставка не состоялась: ReplySent НЕ пишем — сообщение остаётся
            # без ответа, следующий рефлекс/тик повторит (гейт replied).
            raise RuntimeError("портал недоступен (push не доставил) — ответят на следующем тике")
        es.append_event("inbox", msg_id, "ReplySent", {
            "to_chat": m["chat_id"], "text": text[:2000], "channel": "portal",
        })
        return msg_id
    # ── TUI-канал (t-261004132716): чат 'tui:*' — локальный терминал агента.
    # Доставка = запись в этот же стор: TUI-процесс (tools/tui_chat.py) читает
    # reply_text проекции и печатает. Внешнего транспорта нет — ReplySent
    # пишем сразу, ответ не может «потеряться в сети».
    if str(m.get("chat_id") or "").startswith("tui:"):
        es.append_event("inbox", msg_id, "ReplySent", {
            "to_chat": m["chat_id"], "text": text[:2000], "channel": "tui",
        })
        return msg_id
    _load_env()  # sendChatAction идёт до send_message: в свежем процессе
    # (демон/сек-рефлекс) токен ещё не загружен — иначе «печатает…» молча теряется
    try:  # индикатор — best effort: недоступность TG-метода не мешает ответу
        _api("sendChatAction", {"chat_id": m["chat_id"], "action": "typing"}, timeout=10)
    except Exception:
        pass
    r = send_message(m["chat_id"], text, reply_to=_group_reply_to([m]), timeout=timeout)
    es.append_event("inbox", msg_id, "ReplySent", {
        "to_chat": m["chat_id"], "text": text[:2000], "tg_ok": bool(r.get("ok")),
        "channel": "tg",
    })
    return msg_id


def send_followup(msg_id: str, text: str, timeout: int = 35) -> bool:
    """Досылка результата в ЧАТ входящего, на которое уже ответили (аудит 2026-10-16).

    Правило мин-уровня «сначала короткий ответ, потом работа»: первый
    tg-reply закрывает входящее (ReplySent), итог работы уходит сюда — в тот
    же чат (группа → группа, личка → личка, портал/агент — своим каналом),
    а не в личку по умолчанию, как tg-notify. Идемпотентность — снаружи
    (commands.deliver_artifact по ключу followup:<id>:<хеш текста>).
    """
    model = es.run_projections()
    m = (model.get("inbox") or {}).get(msg_id)
    if m is None:
        raise ValueError(f"входящее {msg_id} не найдено")
    chat = str(m.get("chat_id") or "")
    if chat.startswith("portal:agent:"):
        from . import agent_channel
        ok = agent_channel.push_agent(chat.split(":", 2)[-1], text, "help")
    elif chat.startswith("portal"):
        from . import portal_channel
        ok = portal_channel.push(m["chat_id"], text, [m.get("update_id") or msg_id])
    elif chat.startswith("tui:"):
        ok = True  # TUI читает стор: события FollowupSent достаточно
    else:
        r = send_message(m["chat_id"], text, reply_to=_group_reply_to([m]), timeout=timeout)
        ok = bool((r or {}).get("ok"))
    if ok:
        es.append_event("inbox", msg_id, "FollowupSent", {
            "to_chat": m.get("chat_id"), "text": text[:2000]})
    return ok


def send_batch_reply(msg_ids: list, text: str, timeout: int = 35):
    """Сводный ответ на БАТЧ входящих (t-260930064924): одна Telegram-отправка
    закрывает несколько сообщений одного разговора.

    Минутный тик собирает ВСЕ неотвеченные сообщения разом (секция инбокса)
    и отвечает одним сводным текстом — вместо по одному за тик
    (m-516173222..228 обрабатывались 7 тиками подряд). Событие ReplySent
    пишем на КАЖДОЕ закрываемое входящее (проекция ставит replied=True) —
    сводный ответ не должен оставлять «хвостов», на которые гейт важности
    потом будит минуту снова. Полный текст храним у головного (первого по
    ts), у остальных — ссылку на него, чтобы не дублировать текст в сторе.

    Сообщения разных чатов не смешиваются: батч делится по chat_id, каждому
    чату — своя отправка (сводный текст один, белый список — не только личка).

    Идемпотентность: перед отправкой перечитываем read model; уже отвеченные
    (гонка с сек-рефлексом/ручным tg-reply) выпадают из батча. Если выпали
    все — ValueError (тихий пропуск, не ошибка тика).
    """
    model = es.run_projections()
    inbox = model.get("inbox") or {}
    pending = []
    for mid in msg_ids:
        m = inbox.get(mid)
        if m is None:
            raise ValueError(f"входящее {mid} не найдено")
        if m.get("replied") or m.get("ignored"):
            continue  # гонка: уже закрыто/чужое — не трогаем
        pending.append((mid, m))
    if not pending:
        raise ValueError("на все сообщения батча уже отвечено")
    _load_env()
    head_mid = pending[0][0]
    batch_id = f"b-{pending[0][1].get('update_id')}-{len(pending)}"
    by_chat = {}
    for mid, m in pending:
        by_chat.setdefault(m["chat_id"], []).append((mid, m))
    # ── канал агент↔агент (t-261015090129): батч сообщений других агентов —
    # сводный ответ одним ask_agent каждому адресату. Chat_id вида
    # portal:agent:<имя>; портальный UI-чат ('portal:<имя>' без agent:) —
    # отдельная ветка ниже.
    agent_chats = [c for c in by_chat if str(c).startswith("portal:agent:")]
    if agent_chats:
        if len(by_chat) != len(agent_chats):
            raise ValueError("батч смешивает агент-канал и другие каналы — раздели ответы")
        from . import agent_channel
        for chat_id in agent_chats:
            to_agent = str(chat_id).split(":", 2)[-1]
            m0 = by_chat[chat_id][0][1]
            kind = {"help": "help", "review": "review", "handoff": "handoff"}.get(
                m0.get("agent_kind"), "help")
            ok = agent_channel.push_agent(to_agent, text, kind)
            if not ok:
                raise RuntimeError(f"ask_agent({to_agent}) не доставил — попробуем на следующем тике")
            for mid, m in by_chat[chat_id]:
                es.append_event("inbox", mid, "ReplySent", {
                    "to_chat": chat_id,
                    "text": (text if mid == head_mid
                             else f"(сводный ответ батча {batch_id} — см. {head_mid})")[:2000],
                    "channel": "agent_mcp",
                    "batch": {"id": batch_id, "size": len(pending), "head": head_mid},
                })
        return head_mid
    # ── файловый портал-канал (t-261001123121): батч целиком из чата 'portal:*'
    # доставляется HTTP-push'ем на портал, без Telegram. Смешивать чаты
    # нельзя — один сводный текст на один транспорт.
    portal_chats = [c for c in by_chat if str(c).startswith("portal")]
    if portal_chats:
        if len(by_chat) != 1:
            raise ValueError("батч смешивает портал и Telegram — доставка неоднозначна")
        from . import portal_channel
        chat_id = portal_chats[0]
        ids = [by_chat[chat_id][0][1].get("update_id") or mid
               for mid, _ in by_chat[chat_id]]
        ok = portal_channel.push(chat_id, text, ids)
        if not ok:
            raise RuntimeError("портал недоступен (push не доставил) — попробуем на следующем тике")
        for mid, m in by_chat[chat_id]:
            es.append_event("inbox", mid, "ReplySent", {
                "to_chat": chat_id,
                "text": (text if mid == head_mid
                         else f"(сводный ответ батча {batch_id} — см. {head_mid})")[:2000],
                "channel": "portal",
                "batch": {"id": batch_id, "size": len(pending), "head": head_mid},
            })
        return head_mid
    for chat_id, items in by_chat.items():
        if str(chat_id).startswith("tui:"):
            # TUI-канал (t-261004132716): локальная доставка событием в стор,
            # без Telegram — TUI-процесс читает reply_text проекции.
            for mid, m in items:
                es.append_event("inbox", mid, "ReplySent", {
                    "to_chat": chat_id,
                    "text": (text if mid == head_mid
                             else f"(сводный ответ батча {batch_id} — см. {head_mid})")[:2000],
                    "channel": "tui",
                    "batch": {"id": batch_id, "size": len(pending), "head": head_mid},
                })
            continue
        try:  # индикатор — best effort, как в send_reply
            _api("sendChatAction", {"chat_id": chat_id, "action": "typing"}, timeout=10)
        except Exception:
            pass
        r = send_message(chat_id, text, reply_to=_group_reply_to([m for _, m in items]),
                         timeout=timeout)
        for mid, m in items:
            es.append_event("inbox", mid, "ReplySent", {
                "to_chat": chat_id,
                "text": (text if mid == head_mid
                         else f"(сводный ответ батча {batch_id} — см. {head_mid})")[:2000],
                "tg_ok": bool(r.get("ok")),
                "batch": {"id": batch_id, "size": len(pending), "head": head_mid},
            })
    return head_mid


def _already_received(msg: dict) -> bool:
    """Есть ли в инбоксе входящее с тем же (chat_id, message_id)."""
    cid, mid = (msg.get("chat") or {}).get("id"), msg.get("message_id")
    inbox = es.run_projections().get("inbox") or {}
    return any(m.get("chat_id") == cid and m.get("message_id") == mid for m in inbox.values())


def poll_loop():
    """Long polling getUpdates → MessageReceived. Отдельный процесс."""
    _load_env()
    if not TG_TOKEN:
        raise SystemExit("SIMBIOT_TG_TOKEN не задан в .env")
    es.init_db()
    # (t-260929210755) список команд бота — виден в чате как меню "/":
    # регистрируем при старте поллера (идемпотентно) и после рестартов.
    try:
        from . import tg_commands
        mode = tg_commands.register_bot_commands()
        print(f"[tg] команды бота зарегистрированы ({mode})", flush=True)
    except Exception as e:
        print(f"[tg] регистрация команд не удалась: {e}", flush=True)
    offset = 0
    print("[tg] инбокс-поллер запущен", flush=True)
    while True:
        try:
            updates = _api("getUpdates", {
                "offset": offset, "timeout": 30, "limit": 16,
                # явно: Telegram помнит прошлый allowed_updates; callback_query —
                # кнопки ядра Коди («это было мне?»)
                "allowed_updates": json.dumps(["message", "edited_message", "callback_query"])})
            # getUpdates ответил — канал жив: штамп для дашборда
            # (t-261004103718-d8112b). getUpdates сам НЕ пробуем в дашборде:
            # конфликт long polling с этим процессом (сдвиг offset).
            from . import portal_channel as _pch
            _pch.write_channel_stamp(
                "tg", ok=True,
                updates=len((updates or {}).get("result") or []))
            for u in updates.get("result", []):
                offset = max(offset, u["update_id"] + 1)
                if u.get("callback_query"):
                    # кнопки ядра Коди («это было мне?») → метка человека
                    try:
                        from . import kodi_core, trust
                        if not kodi_core.on_callback(u["callback_query"], ALLOWED):
                            trust.on_callback(u["callback_query"])  # одобрение владельца
                    except Exception as e:
                        print(f"[tg] callback не обработан: {e}", flush=True)
                    continue
                msg = u.get("message") or u.get("edited_message")
                if not msg:
                    continue
                if u.get("edited_message") and _already_received(msg):
                    # правка уже принятого сообщения — не новое входящее (дубль
                    # m-516174914/915: исправил опечатку → два одинаковых в батче)
                    print(f"[tg] правка сообщения {msg.get('message_id')} — пропуск", flush=True)
                    continue
                user = (msg.get("from") or {}).get("id")
                if ALLOWED and str(user) not in ALLOWED:
                    # чужой: пишем событие, но не отвечаем (потенциально спам)
                    # t-261015080211: метаданные автора и в игнор-событии
                    # (кто писал, чат, время) — не только в MessageReceived
                    _ig_chat = msg.get("chat") or {}
                    es.append_event("inbox", f"m-{u['update_id']}", "MessageIgnored", {
                        "user_id": user,
                        "username": (msg.get("from") or {}).get("username"),
                        "first_name": (msg.get("from") or {}).get("first_name"),
                        "chat_id": _ig_chat.get("id"),
                        "chat_type": _ig_chat.get("type"),
                        "chat_title": _ig_chat.get("title"),
                        "date": msg.get("date"),
                        "message_id": msg.get("message_id"),
                        "reason": "not_allowed", "text": (msg.get("text") or "")[:500],
                    })
                    continue
                mid = receive_message(u)
                print(f"[tg] {mid} от {(msg.get('from') or {}).get('username') or user}: "
                      f"{(msg.get('text') or '')[:60]}", flush=True)
                # Реакция 👀 — Alex сразу ВИДИТ, что сообщение принято и взято в
                # работу, без «тишины на весь тик» (e-261002202857).
                # best effort: недоступность TG не мешает приёму/обработке.
                # В группах 👀 ставит триаж адресации (harness/addressing.py) —
                # и только на адресованное боту: не метим чужие разговоры.
                if (msg.get("chat") or {}).get("type") not in ("group", "supergroup"):
                    try:
                        mark_seen(mid)
                    except Exception as e:
                        print(f"[tg] {mid} реакция 👀 не удалась: {e}", flush=True)
                # голосовые/аудио/видео-заметки: расшифровка сразу при приёме,
                # чтобы минутный уровень уже видел текст (не качает файлы сам —
                # таймаут тика мал). Сбой транскрипции не валит поллер.
                fobj = (msg.get("voice") or msg.get("audio")
                        or msg.get("video_note") or msg.get("video"))
                if fobj:
                    try:
                        from . import voice
                        kind = ("voice" if msg.get("voice") else
                                "audio" if msg.get("audio") else
                                "video_note" if msg.get("video_note") else
                                "video")
                        text, engine = voice.transcribe_message(
                            mid,
                            file_id=fobj.get("file_id") or "",
                            file_unique_id=fobj.get("file_unique_id") or "",
                            duration=fobj.get("duration"),
                            kind=kind,
                        )
                        print(f"[tg] {mid} транскрипция ({engine}): {(text or '')[:80]}", flush=True)
                    except Exception as e:
                        print(f"[tg] {mid} ошибка транскрипции: {e}", flush=True)
        except Exception as e:
            print(f"[tg] ошибка поллинга: {e}", flush=True)
            time.sleep(5)


def notify_chat(text: str):
    """Отправить сообщение в чат харнеса (без входящего-родителя — уведомления).

    Чат берём из последнего входящего MessageReceived (это личка Алекса);
    если входящих ещё не было — тихо пропускаем (некому показывать).
    """
    chat_id = _default_chat_id()
    if chat_id is None:
        return None
    try:
        return send_message(chat_id, text)
    except Exception:
        return None  # уведомление — best effort, не валить тик


def _default_chat_id():
    """chat_id последнего живого чата харнеса (личка Alex) или None."""
    _load_env()
    if not TG_TOKEN:
        return None
    model = es.run_projections()
    # portal:* — портал-канал (t-261001123121), tui:* — терминал-чат
    # (t-261004132716): оба НЕ Telegram, уходят своими транспортами.
    # Без фильтра E2E/чат портала или TUI «перебивает» личку Alex
    # дефолтным чатом → send_message/send_document падают 'chat not found'
    # (портал — найдено 02.10; tui — 04.10, SKIP уведомлений при живом чате).
    inbox = [m for m in (model.get("inbox") or {}).values()
             if m.get("chat_id")
             and not str(m.get("chat_id")).startswith(("portal", "tui"))
             and not m.get("ignored")]
    if not inbox:
        return None  # некому: чат ещё не известен
    # t-261013113629: уведомления/файлы — в ЛИЧКУ Alex, а не в последний
    # активный чат (иначе ответ в группе делает её дефолтом, и файлы/дайджесты
    # уходят всем участникам группы). Личный чат = private по chat_type;
    # старые записи без chat_type — chat_id > 0 (у групп он отрицательный).
    def _is_private(m):
        if m.get("chat_type"):
            return m.get("chat_type") == "private"
        try:  # старые записи без chat_type: у групп chat_id < 0
            return int(m.get("chat_id")) > 0
        except (TypeError, ValueError):
            return False
    privates = [m for m in inbox if _is_private(m)]
    pool = privates or inbox
    return sorted(pool, key=lambda m: m.get("ts") or "")[-1]["chat_id"]


def _post_multipart(method: str, fields: dict, file_path: str,
                    file_field: str, timeout: int = 90):
    """multipart/form-data POST без requests (sendDocument) — чистый urllib.

    Файл читается целиком: артефакты харнеса (HTML-презентации, отчёты,
    картинки) — килобайты/мегабайты, стримить нечего. Таймаут больше
    текстового: на файле TG бывает медленнее.
    """
    _load_env()
    boundary = "----simbiot" + uuid.uuid4().hex
    parts = []
    for k, v in fields.items():
        parts.append(
            f"--{boundary}\r\nContent-Disposition: form-data; name=\"{k}\"\r\n\r\n{v}\r\n"
            .encode())
    with open(file_path, "rb") as f:
        content = f.read()
    fname = os.path.basename(file_path)
    parts.append(
        (f"--{boundary}\r\nContent-Disposition: form-data; "
         f"name=\"{file_field}\"; filename=\"{fname}\"\r\n"
         f"Content-Type: application/octet-stream\r\n\r\n").encode()
        + content + b"\r\n")
    parts.append(f"--{boundary}--\r\n".encode())
    body = b"".join(parts)
    url = f"https://api.telegram.org/bot{TG_TOKEN}/{method}"
    req = urllib.request.Request(
        url, data=body,
        headers={"Content-Type": f"multipart/form-data; boundary={boundary}"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode())


def send_document(chat_id, file_path: str, caption: str = "", timeout: int = 90):
    """Отправить ФАЙЛ в чат (sendDocument). Исключение наружу — пусть
    вызывающий (deliver_artifact) решает, ретраить или валить: у нас
    идемпотентная обёртка со своим retry-циклом демона."""
    return _post_multipart(
        "sendDocument",
        {"chat_id": str(chat_id), "caption": caption[:1000]},
        file_path, "document", timeout=timeout)


def send_document_to_chat(file_path: str, caption: str = "") -> bool:
    """Файл в дефолтный чат харнеса. False — некому отправить (чат не известен).

    Исключения НЕ глотаем (в отличие от notify_chat): файл — ценный
    артефакт, сбой должен быть виден вызывающему, а не превращаться в
    молчаливый SKIP.
    """
    chat_id = _default_chat_id()
    if chat_id is None:
        return False
    return bool(send_document(chat_id, file_path, caption=caption))


def ladder_snapshot() -> str:
    """Срез лестницы для Telegram: уровни, занятость, задачи, эскалации.

    Занятость: busy в read model всегда False (истинное состояние живёт
    в памяти планировщика, событий для него нет), поэтому выводим её из
    running-задач: TaskClaimed закрепляет уровень за задачей в проекции.
    Уровень без задачи (чистый дайджест-тик) здесь выглядит как «ожидает» —
    приемлемо: дайджест-тик короткий, а взятие задачи — главный кейс.
    """
    model = es.run_projections()
    cs = model.get("cycle_state") or {}
    tasks_all = list((model.get("tasks") or {}).values())
    running_levels = {t.get("level") for t in tasks_all if t.get("status") == "running"}
    order = ["sec", "min", "hour", "day"]
    lines = []
    for name in order:
        st = cs.get(name) or {}
        if name in running_levels:
            busy = "🔧 занят (running-задача)"
        else:
            busy = "ожидает"
        lines.append(f"{'🔹' if name != 'sec' else '⚡'} {name}: {busy}, тиков {st.get('ticks', 0)}"
                     + (f", ошибок {st['errors']}" if st.get("errors") else ""))
    tasks = tasks_all
    tasks.sort(key=lambda t: t.get("created") or "", reverse=True)
    board = [t for t in tasks if t.get("status") in ("ready", "running")]
    lines.append(f"📋 готово/в работе: {sum(1 for t in tasks if t['status'] == 'ready')}/"
                 f"{sum(1 for t in tasks if t['status'] == 'running')}")
    for t in board[:8]:
        lines.append(f"  [{t['size']}] {t['status']}: {t['title'][:60]}")
    open_esc = [e for e in (model.get("escalations") or {}).values()
                if isinstance(e, dict) and e.get("status") == "open"]
    lines.append(f"🚨 открытых эскалаций: {len(open_esc)}")
    for e in open_esc[:5]:
        lines.append(f"  [{e.get('severity', 'info')}] {e.get('reason', '')[:80]}")
    return "\n".join(lines)


if __name__ == "__main__":
    # Единственный легитимный поллер — systemd-сервис simbiot-harness-tg.
    # `python -m harness.tg_inbox` с любыми аргументами (--help, list, …)
    # раньше поднимал второго поллера → 409 Conflict в легитимном.
    if len(sys.argv) > 1:
        print("usage: python -m harness.tg_inbox            # long-poll поллер (systemd)")
        print("       cli.py ladder                         # срез лестницы (текст)")
        print("       cli.py tg-notify <text>               # отправка в чат харнеса")
        sys.exit(0)
    poll_loop()