From 0523135bf1f52c2b6fa5f27ab3a6460aefcebb7e Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Thu, 1 Oct 2026 13:34:26 +0000 Subject: [PATCH] =?UTF-8?q?feat(portal-chat):=20=D0=BF=D1=80=D1=8F=D0=BC?= =?UTF-8?q?=D0=BE=D0=B9=20=D0=BA=D0=B0=D0=BD=D0=B0=D0=BB=20=D0=BF=D0=BE?= =?UTF-8?q?=D1=80=D1=82=D0=B0=D0=BB=E2=86=94=D0=B0=D0=B3=D0=B5=D0=BD=D1=82?= =?UTF-8?q?=20=D0=B1=D0=B5=D0=B7=20Matrix=20(t-261001123122-5ea554)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - portal_channel.py: HTTP-клиент портала — pull входящих чата портала (GET /api/agent/<имя>/channel/poll?since=ms), push ответов (POST outbox, ack входящих по msg_id), сек-рефлекс (≤20с) + cron main(). - tg_inbox: чаты 'portal:*' доставляются push'ем на портал (ReplySent), reply_to = портальный update_id (portal: ack consumed входящего). - scheduler: ReflexObserved хук на portal_channel.sec_reflex — pull работает и без LLM-тика, и на машинах без TG. - es: проекционный бранчPortalDelivered выпилен (маркер — ReplySent от tg_inbox); MessageIgnored branch остался нетронутым. - Диагноз: SSH портал→агенты закрыт (timeout :22/:8765 к #83) — канал на исходящем HTTP, AGENT_TOKEN+PORTAL_URL уже пекутся в /etc/coop-reporter.env, agent-scope env добавлены в cloud-init .env (SIMBIOT_AGENT_TOKEN/NAME/CHAT=1). --- harness/es.py | 2 + harness/portal_channel.py | 271 ++++++++++++++++++++++++++++++++++++++ harness/scheduler.py | 13 ++ harness/tg_inbox.py | 39 ++++++ 4 files changed, 325 insertions(+) create mode 100644 harness/portal_channel.py diff --git a/harness/es.py b/harness/es.py index d076048..6715a21 100644 --- a/harness/es.py +++ b/harness/es.py @@ -456,6 +456,8 @@ def _apply_event(model: dict, ev: dict): "ts": ev["timestamp"], } elif etype == "MessageIgnored": + # канал портала (t-261001123121): доставка реплая помечена ReplySent'ом + # (portal ReplySent — признак push'а) — отдельный маркер не нужен m = model.setdefault("inbox", {}).get(aid) if m: m["ignored"] = True diff --git a/harness/portal_channel.py b/harness/portal_channel.py new file mode 100644 index 0000000..e4942c5 --- /dev/null +++ b/harness/portal_channel.py @@ -0,0 +1,271 @@ +"""Портальный чат-канал (t-261001123121): HTTP-клиент портала для агента. + +Мандат: «чат напрямую через API харнеса, без Matrix». У харнеса нет +принимающего веб-порта (scheduler — цикл, dashboard — read-only), а машина +агента общается с порталом ТОЛЬКО исходящим HTTP (SSH вход из портала закрыт +— проверено 01.10, timeout 22/8765 к агентам). Поэтому канал — pull/push: + + ВХОД (портал → агент): GET /api/agent//channel/poll?since= + X-Agent-Token: → {messages: [{msg_id, text, ts}]} + ingеcт в стор (MessageReceived, chat_id='portal:', id=портальный) + → далее штатный инбокс (батч-секция мин-тика, tg-reply, ReplySent). + ВЫХОД (агент → портал): POST /api/agent//channel/poll + {"outbox": [{msg_id, text, reply_to, ts}]} → портал пишет в + agent_portal_chat (UNIQUE msg_id) — UI видит мгновенно. + ПУШ-ответ из тика: send_reply/send_batch_reply для chat 'portal:*' зовут + push() вместо Telegram — реплай попадает в портал сразу же (≤2с), + не дожидаясь следующего отложенного POST outbox. + +Токены: SIMBIOT_AGENT_TOKEN — тот же reporter-токен (agt_…), что и у +coop-reporter; PORTAL_URL — как в .harness-env. Логин/секреты не храню — +клиент без состояния, дедуп на портале (msg_id UNIQUE). +""" +import os +import json +import time +import datetime +import urllib.parse +import urllib.request + +REPORTER_ENV = "/etc/coop-reporter.env" +STATE_FILE = "/root/simbiot-harness/portal/inbox/.channel-last" +PENDING_DIR = "/root/simbiot-harness/portal/inbox/pending" # legacy-путь (пусто) +OUTBOX_DIR = "/root/simbiot-harness/portal/inbox/outbox" # legacy fallback + +HTTP_TIMEOUT = 8 # портал отвечает быстро; висеть дольше — вред +MIN_REFLEX_GAP = 20 # сек-рефлекс не дёргает канал чаще, чем раз в 20с +MAX_TEXT = 8000 +_last_reflex = {"ts": 0.0} + + +def _now_ms() -> int: + return int(time.time() * 1000) + + +def _cfg() -> dict: + cfg = {"portal_url": "", "agent_token": "", "agent_name": ""} + for k, ek in (("portal_url", "SIMBIOT_PORTAL_URL"), + ("agent_token", "SIMBIOT_AGENT_TOKEN"), + ("agent_name", "SIMBIOT_AGENT_NAME")): + v = os.environ.get(ek) + if v: + cfg[k] = v.strip().rstrip("/") + try: + with open(REPORTER_ENV) as f: + for line in f: + if "=" in line: + k, v = line.split("=", 1) + if k.strip() == "PORTAL_URL" and not cfg["portal_url"]: + cfg["portal_url"] = v.strip().rstrip("/") + elif k.strip() == "AGENT_TOKEN" and not cfg["agent_token"]: + cfg["agent_token"] = v.strip() + except OSError: + pass + if not cfg["agent_name"]: + try: + with open("/etc/hostname") as f: + cfg["agent_name"] = f.read().strip()[:60] or "" + except OSError: + pass + return cfg + + +def _es(): + import sys + if "/root/simbiot-harness" not in sys.path: + sys.path.insert(0, "/root/simbiot-harness") + from harness import es + return es + + +def _agent_chat_id(cfg) -> str: + return f"portal:{cfg['agent_name']}" + + +def store_id(portal_msg_id: str) -> str: + """Ключ входящего в сторе: 'm-' + id (та же схема scheduler для TG).""" + return "m-" + str(portal_msg_id or "").strip()[:80] + + +def _ingest_messages(msgs: list) -> int: + """[{msg_id, text, ts}] → MessageReceived в стор (идемпотентно по msg_id). + + Дедуп по read model: повторно пришедшие (POST-подтверждение не доехало) + пропускаются — LLM их не увидит второй раз, чат не дублируется. + """ + if not msgs: + return 0 + try: + es = _es() + model = es.run_projections() + except Exception: + return 0 # стор занят —pull повторится позже (между тиками) + existing = {str(m.get("update_id")) for m in (model.get("inbox") or {}).values() + if isinstance(m, dict) and m.get("update_id") is not None} + ingested = 0 + cfg = _cfg() + for d in msgs: + mid = str((d or {}).get("msg_id") or "").strip()[:80] + text = ((d or {}).get("text") or "").strip() + if not mid or not text or mid in existing: + continue + data = { + "update_id": mid, + "chat_id": _agent_chat_id(cfg), + "user_id": None, + "username": ((d.get("author") or "portal")[:60] + " · portal")[:80], + "first_name": str(d.get("author_label") or "")[:80] or None, + "text": text[:MAX_TEXT], + "kind": "portal", + "file_id": None, "file_unique_id": None, "duration": None, + "date": (d.get("ts") or _now_ms()) // 1000, + "source": "portal", + } + try: + es.append_event("inbox", store_id(mid), "MessageReceived", data) + except Exception: + continue # стор занят на этом msg — следующий проход + ingested += 1 + return ingested + + +def pull(last_ms: int = 0) -> int: + """ВХОД: GET /channel/poll?since= → ingest. Возвращает число + принятых сообщений (0 = тихо/недоступно — ошибки не наружу).""" + cfg = _cfg() + if not (cfg["portal_url"] and cfg["agent_token"] and cfg["agent_name"]): + return 0 # деплой без репортера: канал выключен, не дёргаем + url = (f"{cfg['portal_url']}/api/agent/{urllib.parse.quote(cfg['agent_name'])}" + f"/channel/poll?since={int(last_ms or 0)}") + req = urllib.request.Request(url, headers={"X-Agent-Token": cfg["agent_token"]}) + try: + with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as r: + data = json.loads(r.read().decode()) + except Exception: + return 0 # сеть/портал недоступны: тихо (cron повторит) + msgs = data.get("messages") or [] + n = _ingest_messages(msgs) + # отметить съеденные: POST'ом их msg_id — портал пометит consumed + # (вторая попытка того же цикла недоступности не ворошит пустое) + if n > 0: + _post_outbox([{"msg_id": m.get("msg_id")} for m in msgs + if m.get("msg_id")]) + return n + + +def push(chat_id: str, text: str, reply_mids=None) -> bool: + """ВЫХОД (немедленный): реплай уровня на сообщение чата портала. + + Вызывается из tg_inbox.send_reply/send_batch_reply для chat 'portal:*'. + True — портал принял (caller пишет ReplySent и закрывает инбокс); + False — не доставлено (caller ReplySent НЕ пишет → повтор на след. тик). + """ + text = (text or "").strip() + if not text: + return False + items = [{"msg_id": f"reply-{int(time.time() * 1000)}-{(m or '?')}", + "text": text[:MAX_TEXT], "reply_to": m, "ts": _now_ms()} + for m in (reply_mids or [None])][:20] + return bool(_post_outbox(items, timeout=15)) + + +def _post_outbox(items: list, timeout: int = HTTP_TIMEOUT) -> bool: + """POST /channel/poll {"outbox": items} — портал пишет в agent_portal_chat.""" + if not items: + return True + cfg = _cfg() + if not (cfg["portal_url"] and cfg["agent_token"] and cfg["agent_name"]): + return False + url = (f"{cfg['portal_url']}/api/agent/{urllib.parse.quote(cfg['agent_name'])}" + f"/channel/poll") + req = urllib.request.Request( + url, data=json.dumps({"outbox": items[:100]}).encode(), + headers={"Content-Type": "application/json", + "X-Agent-Token": cfg["agent_token"]}) + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + return bool(json.loads(r.read().decode()).get("ok")) + except Exception: + return False + + +# ══════════════ фоновые вызовы (без LLM) ══════════════ + +def sec_reflex() -> dict: + """Сек-рефлекс: pull портала + flush outbox-файлов. Дешёво и НЕ чаще + MIN_REFLEX_GAP (портал может быть недоступен — рефлекс не должен висеть).""" + now = time.time() + if now - _last_reflex["ts"] < MIN_REFLEX_GAP: + return {"skipped": "rate-limit"} + if os.environ.get("SIMBIOT_PORTAL_CHAT", "") == "0": + return {"skipped": "disabled"} # мастер-харнес: канал выключен флагом + _last_reflex["ts"] = now + out = {} + out["ingested"] = pull() + out["outbox_files"] = _flush_outbox_files() + # legacy: file-pending (если портал когда-то писал файлами) + try: + import glob + pfiles = glob.glob(os.path.join(PENDING_DIR, "*.json")) + if pfiles: + msgs = [] + for f in pfiles: + try: + with open(f) as fh: + msgs.append(json.load(fh)) + os.remove(f) + except Exception: + pass + out["ingested_files"] = _ingest_messages(msgs) + except Exception: + pass + return out + + +def _flush_outbox_files() -> int: + """Fallback-путь через outbox-файлы: push не прошёл → файл; портал + принимает их теперь тем же POST (файл = item, id тот же).""" + import glob + if not os.path.isdir(OUTBOX_DIR): + return 0 + files = sorted(glob.glob(os.path.join(OUTBOX_DIR, "*.json"))) + if not files: + return 0 + items = [] + flushed = 0 + for f in files: + try: + with open(f) as fh: + d = json.load(fh) + items.append({"msg_id": d.get("id"), "text": d.get("text"), + "reply_to": d.get("reply_to"), "ts": d.get("ts")}) + except Exception: + try: + os.remove(f) # битый файл не должен гнить навсегда + except OSError: + pass + continue + if not items: + return 0 + if _post_outbox(items, timeout=10): + for d in items: + mid = d.get("msg_id") + try: + os.remove(os.path.join(OUTBOX_DIR, + f"{str(mid).replace(':', '_')}.json")) + except OSError: + pass + flushed = len(items) + return flushed + + +def main(): + """cron-тело (каждую минуту): pull портала + flush outbox.""" + n = pull() + f = _flush_outbox_files() + if n or f: + print(f"[portal-channel] pulled={n} flushed={f}", flush=True) + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/harness/scheduler.py b/harness/scheduler.py index 67b92e2..0ff0b51 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -617,6 +617,19 @@ def _reflex_tick(level_cfg, seq: int, model: dict): except Exception as e: # команды — часть рефлекса, но не должны валить телеметрию print(f"[tg-cmd] ответ на команды не удался: {e}", flush=True) + # файловый/HTTP канал чата портала (t-261001123121, без Matrix): pull + # входящих чата портала + немедленный push уровня-репляев. Канал + # независим от TG: работает и на машинах без TG-токена. + try: + from . import portal_channel as _pch + res = _pch.sec_reflex() + if not res.get("skipped") and (res.get("ingested") or res.get("outbox_files")): + payload["portal_channel"] = res + _log(f"portal-channel: {res}") + except SystemExit: + raise + except Exception as e: + print(f"[portal-channel] рефлекс не смог: {e}", flush=True) es.append_event("cycle", f"reflex:{seq}", "ReflexObserved", payload) return payload, None diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index f18e42d..75554a0 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -129,6 +129,21 @@ def send_reply(msg_id: str, text: str, timeout: int = 35): 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 _load_env() # sendChatAction идёт до send_message: в свежем процессе # (демон/сек-рефлекс) токен ещё не загружен — иначе «печатает…» молча теряется try: # индикатор — best effort: недоступность TG-метода не мешает ответу @@ -138,6 +153,7 @@ def send_reply(msg_id: str, text: str, timeout: int = 35): 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 @@ -179,6 +195,29 @@ def send_batch_reply(msg_ids: list, text: str, timeout: int = 35): 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(): try: # индикатор — best effort, как в send_reply _api("sendChatAction", {"chat_id": chat_id, "action": "typing"}, timeout=10)