184 lines
7.4 KiB
Python
184 lines
7.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Канал агент↔агент через MCP-шлюз портала (t-261015090129).
|
||
|
||
Петля обмена между агентами кооператива:
|
||
- ВХОД: MCP-тул poll_messages (портал, t-261015090129) — забрать новые
|
||
agent_message из своей очереди. Рефлекс демона вызывает его раз в
|
||
минуту (MIN_REFLEX_GAP), новые сообщения → MessageReceived в стор
|
||
(kind=agent_message, chat_id=portal:agent:<имя>) — минутный триаж
|
||
видит их как обычные входящие и отвечает.
|
||
- ВЫХОД: MCP-тул ask_agent — просьба/вопрос/передача другому агенту;
|
||
send_reply для чата portal:agent:* идёт через ask_agent, НЕ через TG.
|
||
|
||
Транспорт — тот же MCP-шлюз cil.by, что и portal_mcp (Bearer agt_...);
|
||
токен один (secrets/portal-mcp-token / SIMBIOT_PORTAL_MCP_TOKEN).
|
||
Курсор чтения — на портале (agent_message_cursors): прочитанное не
|
||
возвращается повторно, локальный дедуп по update_id в stor'е — вторая
|
||
линия защиты (POST-подтверждение может потеряться).
|
||
|
||
Наблюдаемость: трейс AgentChannelPull в event store (как PortalMcpCalled)
|
||
+ штамп канала harness/channel_agent.json (как у portal-канала).
|
||
"""
|
||
import json
|
||
import os
|
||
import threading
|
||
import time
|
||
|
||
from . import paths as _paths
|
||
|
||
# не чаще раза в минуту: очередь агент↔агент не требует суб-секундной реакции
|
||
MIN_REFLEX_GAP = 60.0
|
||
# не поднимать очередь сообщений чаще, чем раз в N минут для одного чата?
|
||
# нет: курсор на портале, повторный poll ничего не стоит.
|
||
_last_reflex = {"ts": 0.0}
|
||
_lock = threading.Lock()
|
||
|
||
|
||
def _log(msg: str):
|
||
print(f"[agent-channel] {msg}", flush=True)
|
||
|
||
|
||
def _cfg_token() -> str:
|
||
"""Токен MCP: env сильнее файла (та же схема, что portal_mcp)."""
|
||
env = os.environ.get("SIMBIOT_PORTAL_MCP_TOKEN")
|
||
if env:
|
||
return env.strip()
|
||
try:
|
||
p = _paths.portal_mcp_token_path()
|
||
if p and os.path.exists(p):
|
||
return open(p).read().strip()
|
||
except Exception:
|
||
pass
|
||
return ""
|
||
|
||
|
||
def _rpc_tool(tool: str, arguments: dict) -> str:
|
||
"""Вызов MCP-тула через клиент portal_mcp (трейс/лимиты/guardrail там)."""
|
||
from . import portal_mcp as _mcp
|
||
return _mcp.call_tool(tool, json.dumps(arguments or {}), level="min")
|
||
|
||
|
||
def _es():
|
||
import sys
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
from harness import es
|
||
return es
|
||
|
||
|
||
def _agent_chat_id(from_agent: str) -> str:
|
||
return f"portal:agent:{(from_agent or '?').strip()[:60]}"
|
||
|
||
|
||
def store_id(portal_ev_id) -> str:
|
||
"""Ключ входящего в сторе: 'm-' + id события портала (агентные длиной 6+)."""
|
||
return "m-" + str(portal_ev_id or "").strip()[:80]
|
||
|
||
|
||
def _ingest_messages(msgs: list) -> int:
|
||
"""[{id, from, kind, text, ts}] → MessageReceived (идемпотентно по id).
|
||
|
||
Дедуп по read model: портальный курсор не сдвигается при сбое ingest —
|
||
повторный poll вернёт те же id, стор их уже не примет.
|
||
"""
|
||
if not msgs:
|
||
return 0
|
||
try:
|
||
es = _es()
|
||
model = es.run_projections()
|
||
except Exception:
|
||
return 0 # стор занят — poll повторится (курсор не сдвинулся, ок)
|
||
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
|
||
for d in msgs:
|
||
mid = str((d or {}).get("id") or "").strip()[:80]
|
||
from_agent = str((d or {}).get("from") or "?").strip()[:60] or "?"
|
||
text = str((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(from_agent),
|
||
"user_id": None,
|
||
"username": from_agent,
|
||
"first_name": f"агент {from_agent}",
|
||
"text": text[:4000],
|
||
"kind": "agent_message",
|
||
"file_id": None, "file_unique_id": None, "duration": None,
|
||
"date": None,
|
||
"source": "agent_mcp",
|
||
"agent_kind": str((d or {}).get("kind") or "message")[:20],
|
||
"agent_ts": str((d or {}).get("ts") or "")[:40],
|
||
}
|
||
try:
|
||
es.append_event("inbox", store_id(mid), "MessageReceived", data)
|
||
except Exception:
|
||
continue # стор занят на этом msg — следующий проход
|
||
ingested += 1
|
||
return ingested
|
||
|
||
|
||
def pull() -> int:
|
||
"""ВХОД: poll_messages → ingest MessageReceived. 0 = тихо/ошибка (тихо)."""
|
||
token = _cfg_token()
|
||
if not token:
|
||
return 0 # канал выключен: харнес без MCP-токена
|
||
raw = _rpc_tool("poll_messages", {})
|
||
if raw.startswith("error:"):
|
||
_log(f"poll_messages: {raw[:200]}")
|
||
return 0
|
||
try:
|
||
data = json.loads(raw)
|
||
except Exception:
|
||
_log(f"poll_messages: не-JSON ответ {raw[:200]}")
|
||
return 0
|
||
msgs = data.get("messages") or []
|
||
return _ingest_messages(msgs)
|
||
|
||
|
||
def push_agent(to_agent: str, text: str, kind: str = "help") -> bool:
|
||
"""ВЫХОД: ask_agent(to, text, kind) → очередь другого агента.
|
||
|
||
True — портал принял (caller пишет ReplySent); False — не доставлено,
|
||
ответ останется в инбоксе и повторится на следующем тике.
|
||
"""
|
||
to = (to_agent or "").strip()
|
||
body = (text or "").strip()
|
||
if not to or not body:
|
||
return False
|
||
raw = _rpc_tool("ask_agent", {"to": to, "text": body[:4000], "kind": kind})
|
||
if raw.startswith("error:"):
|
||
_log(f"ask_agent({to}): {raw[:200]}")
|
||
return False
|
||
try:
|
||
return bool(json.loads(raw).get("ok"))
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def sec_reflex() -> dict:
|
||
"""Сек-рефлекс: poll очереди агент↔агент. Не чаще MIN_REFLEX_GAP."""
|
||
now = time.time()
|
||
if now - _last_reflex["ts"] < MIN_REFLEX_GAP:
|
||
return {"skipped": "rate-limit"}
|
||
_last_reflex["ts"] = now
|
||
out = {"ingested": 0}
|
||
try:
|
||
out["ingested"] = pull()
|
||
except Exception as e:
|
||
out["error"] = str(e)[:200]
|
||
return out
|
||
|
||
|
||
def write_channel_stamp(name: str = "agent", ok: bool = True, **extra):
|
||
"""Атомарный штамп канала: harness/channel_agent.json (как portal)."""
|
||
import datetime
|
||
try:
|
||
path = os.path.join(_paths.home(), f"channel_{name}.json")
|
||
tmp = path + ".tmp"
|
||
rec = {"ok": ok, "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds"), **extra}
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(rec, f, ensure_ascii=False)
|
||
os.replace(tmp, path)
|
||
except Exception:
|
||
pass
|