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).
This commit is contained in:
Simbiot Harness
parent
fe1c569888
commit
0523135bf1
4 changed files with 325 additions and 0 deletions
|
|
@ -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
|
||||
|
|
|
|||
271
harness/portal_channel.py
Normal file
271
harness/portal_channel.py
Normal file
|
|
@ -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()
|
||||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in a new issue