451 lines
No EOL
20 KiB
Python
451 lines
No EOL
20 KiB
Python
"""Портальный чат-канал (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
|
||
|
||
from . import paths as _paths # t-261004113351-2858c1: SIMBIOT_HOME
|
||
|
||
REPORTER_ENV = "/etc/coop-reporter.env"
|
||
STATE_FILE = _paths.channel_state_file() # t-261004113351-2858c1: SIMBIOT_HOME
|
||
PENDING_DIR = _paths.inbox_pending_dir()
|
||
OUTBOX_DIR = _paths.inbox_outbox_dir()
|
||
_HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
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)
|
||
|
||
|
||
# ── живой статус каналов для дашборда (t-261004103718-d8112b) ──────────────
|
||
# Каждый реальный пользователь канала (сек-рефлекс портала, TG-поллер) после
|
||
# УСПЕШНОГО обмена пишет штамп {"ok":bool,"ts":ms,...} в harness/channel_<n>.json.
|
||
# Дашборд читает штампы (не дёргает портал сам) — возраст штампа = возраст данных.
|
||
STAMP_DIR = _paths.stamp_dir() # legacy: корень репо; SIMBIOT_HOME → <home>/run
|
||
|
||
|
||
def write_channel_stamp(name: str, ok: bool, **extra) -> None:
|
||
"""Атомарный штамп успеха канала: harness/channel_<portal|tg>.json."""
|
||
try:
|
||
path = os.path.join(STAMP_DIR, "harness", f"channel_{name}.json")
|
||
d = {"ok": bool(ok), "ts": _now_ms()}
|
||
d.update(extra)
|
||
tmp = path + ".tmp"
|
||
with open(tmp, "w") as f:
|
||
json.dump(d, f)
|
||
os.replace(tmp, path)
|
||
except OSError:
|
||
pass # штамп — диагностика, не должен ломать канал
|
||
|
||
|
||
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 []
|
||
# ── /set-tg-bot (кнопка TG в UI портала): перехват ДО инбокса — токен
|
||
# секрет и не должен попасть в стор/LLM. Ack сразу (consumed=1),
|
||
# применение в фоновом потоке: getMe+рестарт не задерживают рефлекс.
|
||
_settgs = [(str((m or {}).get("msg_id") or ""), str((m or {}).get("text") or ""))
|
||
for m in msgs if "cmd-settg-" in str((m or {}).get("msg_id") or "")]
|
||
if _settgs:
|
||
import threading as _thr
|
||
for _mid, _txt in _settgs:
|
||
_ack_cmd(_mid)
|
||
_thr.Thread(target=_apply_tg_command, args=(_txt,), daemon=True).start()
|
||
_ids = {mid for mid, _ in _settgs}
|
||
msgs = [m for m in msgs if str((m or {}).get("msg_id") or "") not in _ids]
|
||
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
|
||
|
||
|
||
# ══════════════ /set-tg-bot: подключение TG из UI портала ══════════════
|
||
# UI дашборда портала (Детали → Каналы) не может зайти на машину агента по
|
||
# SSH (закрыт, e2e 01.10) → команда идёт через pull/push канал
|
||
# (agent_portal_chat, X-Agent-Token). pull() перехватывает её ДО инбокса:
|
||
# токен — секрет, в стор/LLM-контекст не попадает. Апплаер идемпотентен
|
||
# (getMe-самотест до записи, атомарный .env, рестарт только поллера).
|
||
|
||
|
||
def _ack_cmd(msg_id: str) -> None:
|
||
"""Ack команды порталу (portal consumed=1) — best effort."""
|
||
try:
|
||
_post_outbox([{"msg_id": msg_id}])
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _write_outbox_file(item: dict) -> bool:
|
||
"""Fallback: push не прошёл → outbox-файл, дос flush вытолкнет позже.
|
||
Формат совместим с _flush_outbox_files (ключи id/text/reply_to/ts)."""
|
||
try:
|
||
os.makedirs(OUTBOX_DIR, exist_ok=True)
|
||
mid = str(item.get("msg_id") or "settg").replace(":", "_")
|
||
with open(os.path.join(OUTBOX_DIR, f"{mid}.json"), "w") as f:
|
||
json.dump({"id": item.get("msg_id"), "text": item.get("text"),
|
||
"reply_to": item.get("reply_to"), "ts": item.get("ts")}, f)
|
||
return True
|
||
except OSError:
|
||
return False
|
||
|
||
|
||
def _report(text: str) -> None:
|
||
"""Отчёт апплаера в чат портала: push, при недоступности — outbox-файл."""
|
||
item = {"msg_id": f"settg-{int(time.time() * 1000)}",
|
||
"text": text[:MAX_TEXT], "reply_to": None, "ts": _now_ms()}
|
||
if not _post_outbox([item], timeout=15):
|
||
_write_outbox_file(item)
|
||
|
||
|
||
def _apply_tg_command(text: str) -> None:
|
||
"""Применить `/set-tg-bot <token>[\nallowed=<tg_id>]`.
|
||
|
||
getMe-самотест ДО записи (битый токен не должен ронять poller) →
|
||
атомарный .env (SIMBIOT_TG_TOKEN + SIMBIOT_TG_ALLOWED, старый аллоулист
|
||
не теряем при отсутствии нового) → рестарт ТОЛЬКО simbiot-harness-tg:
|
||
живым процессам харнеса токен на старте не нужен — send_message
|
||
перечитывает .env с диска (_load_env). Отчёт — в чат портала.
|
||
"""
|
||
body = (text or "").strip()
|
||
lines_ = body.splitlines()
|
||
if not lines_ or not lines_[0].startswith("/set-tg-bot"):
|
||
return
|
||
toks = lines_[0].split()
|
||
if len(toks) < 2 or not toks[1]:
|
||
_report("❌ /set-tg-bot: токен не передан")
|
||
return
|
||
token = toks[1]
|
||
payload = {}
|
||
for ln in lines_[1:]:
|
||
if "=" in ln:
|
||
k, v = ln.split("=", 1)
|
||
payload[k.strip()] = v.strip()
|
||
# 0) идемпотентность: тот же токен уже в .env — не применяем повторно
|
||
# (не дёргаем getMe/рестарт, не спамим отчётом при ретраях канала)
|
||
try:
|
||
with open(_paths.env_path()) as f:
|
||
for line in f:
|
||
if line.startswith("SIMBIOT_TG_TOKEN="):
|
||
if line.split("=", 1)[1].strip() == token:
|
||
_report("ℹ️ Этот бот уже подключён — изменение не требуется")
|
||
return
|
||
except OSError:
|
||
pass
|
||
# 1) самотест токена ДО записи: getMe от Telegram
|
||
# (t-261016141XXX: 3 попытки с паузой — таймаут сети не должен ронять привязку)
|
||
me = None
|
||
_last_err = None
|
||
for _attempt in range(3):
|
||
try:
|
||
req = urllib.request.Request(f"https://api.telegram.org/bot{token}/getMe")
|
||
with urllib.request.urlopen(req, timeout=15) as r:
|
||
me = json.loads(r.read().decode())
|
||
if not (me.get("ok") and (me.get("result") or {}).get("username")):
|
||
raise RuntimeError(str(me)[:150])
|
||
break
|
||
except Exception as e:
|
||
_last_err = e
|
||
me = None
|
||
if _attempt < 2:
|
||
time.sleep(3 * (_attempt + 1))
|
||
if me is None:
|
||
_report(f"❌ TG-токен не принят Telegram (getMe, 3 попытки): {_last_err}")
|
||
print(f"[set-tg] токен не валиден: {_last_err}", flush=True)
|
||
return
|
||
uname = me["result"].get("username") or "bot"
|
||
# 2) аллоулист: из команды (allowed=…), иначе сохранённый в .env
|
||
allowed = payload.get("allowed") or payload.get("owner_tg") or ""
|
||
if not allowed:
|
||
try:
|
||
with open(_paths.env_path()) as f:
|
||
for line in f:
|
||
if line.startswith("SIMBIOT_TG_ALLOWED="):
|
||
allowed = line.split("=", 1)[1].strip()
|
||
except OSError:
|
||
pass
|
||
# 3) атомарная запись .env (в .gitignore — в git не попадает)
|
||
env_path = _paths.env_path()
|
||
tmp_path = env_path + ".tmp"
|
||
try:
|
||
with open(env_path) as f:
|
||
keep = [ln for ln in f
|
||
if not ln.startswith("SIMBIOT_TG_TOKEN=")
|
||
and not ln.startswith("SIMBIOT_TG_ALLOWED=")]
|
||
keep.append(f"SIMBIOT_TG_TOKEN={token}\n")
|
||
if allowed:
|
||
keep.append(f"SIMBIOT_TG_ALLOWED={allowed}\n")
|
||
with open(tmp_path, "w") as f:
|
||
f.writelines(keep)
|
||
os.chmod(tmp_path, 0o600)
|
||
os.replace(tmp_path, env_path)
|
||
except Exception as e:
|
||
_report(f"❌ не смог записать {env_path}: {e}")
|
||
print(f"[set-tg] запись .env не удалась: {e}", flush=True)
|
||
return
|
||
# 4) рестарт ТОЛЬКО поллера инбокса: scheduler/dashboard не роняем
|
||
# (самоубийство процесса-тика не нужно — токен подхватится при
|
||
# следующей отправке через _load_env с диска).
|
||
import subprocess
|
||
if os.environ.get("SIMBIOT_SELF_HOSTED", "") != "1":
|
||
try:
|
||
subprocess.run(["systemctl", "restart", "simbiot-harness-tg"],
|
||
capture_output=True, timeout=30)
|
||
except Exception as e:
|
||
print(f"[set-tg] рестарт poller не удался: {e}", flush=True)
|
||
# 5) событие в стор (без секрета: только имя бота) + отчёт в портал
|
||
try:
|
||
_es().append_event("harness", f"set-tg-{int(time.time() * 1000)}",
|
||
"TgChannelConfigured", {"bot": f"@{uname}"})
|
||
except Exception:
|
||
pass
|
||
_report(f"✅ Telegram подключён: @{uname} — самотест прошёл")
|
||
print(f"[set-tg] применён токен @{uname}", flush=True)
|
||
|
||
|
||
|
||
|
||
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() # доехали сюда = портал ответил HTTP (успех обмена)
|
||
write_channel_stamp("portal", ok=True, ingested=out["ingested"])
|
||
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() |