feat(portal-chat): прямой канал портал↔агент без Matrix (t-261001123122-5ea554)

- 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

@@ -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

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/<AGENT>/channel/poll?since=<ms>
+        X-Agent-Token:<AGENT_TOKEN из .env>  → {messages: [{msg_id, text, ts}]}
+        ingеcт в стор (MessageReceived, chat_id='portal:<agent>', id=портальный)
+        → далее штатный инбокс (батч-секция мин-тика, tg-reply, ReplySent).
+  ВЫХОД (агент → портал): POST /api/agent/<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=<last_ms> → 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()

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
 

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)