simbiot-harness/harness/portal_channel.py

"""Портальный чат-канал (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()