"""Simbiot Harness — Telegram-инбокс. Входящие сообщения Алекса → события MessageReceived в стор → минутный уровень видит их в контексте и может ответить через `cli.py tg-reply ` (событие ReplySent закрывает входящее). Поллер — отдельный процесс (systemd simbiot-harness-tg), long polling getUpdates offset=last+1. Идемпотентность: update_id → aggregate_id. """ import json 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 try: return _api("sendMessage", params, timeout=timeout) except urllib.error.HTTPError as e: # 400 от кривого HTML (незакрытый тег, «a < b» в тексте) — шлём как plain text. # Иначе ответ/доставка молча теряются: исключение глотается на верхнем уровне. if e.code == 400: params.pop("parse_mode", None) return _api("sendMessage", params, timeout=timeout) raise def set_reaction(chat_id, message_id: int, emoji: str = "👀", is_big: bool = False): """setMessageReaction: видимый маркер «агент взял/смотрит» (e-261002202857). Идентичность Telegram: повторная постановка ТОГО ЖЕ эмодзи — no-op API; смена эмодзи на другой видна как изменение реакции. Идемпотентный повтор безопасен. Реакция допустима на сообщениях не старше 48 часов. """ params = {"chat_id": chat_id, "message_id": message_id, "reaction_type": 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 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") # самый большой размер data = { "update_id": update["update_id"], "chat_id": chat.get("id"), # 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), "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"), } es.append_event("inbox", f"m-{update['update_id']}", "MessageReceived", data) return f"m-{update['update_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"): 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=None, 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_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-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=None, 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 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}) # 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) msg = u.get("message") or u.get("edited_message") if not msg: continue user = (msg.get("from") or {}).get("id") if ALLOWED and str(user) not in ALLOWED: # чужой: пишем событие, но не отвечаем (потенциально спам) es.append_event("inbox", f"m-{u['update_id']}", "MessageIgnored", { "user_id": user, "username": (msg.get("from") or {}).get("username"), "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 не мешает приёму/обработке. 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 # некому: чат ещё не известен return sorted(inbox, 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 # отправка в чат харнеса") sys.exit(0) poll_loop()