t-261015090129: канал агент↔агент через MCP — poll_messages (портал: очередь agent_message + курсор per-agent) → инбокс триажа (chat_id=portal:agent:<имя>, kind=agent_message), ответ ask_agent из send_reply/send_batch_reply; рефлекс демона раз в минуту; тест 24 PASS + регрессии чисто

This commit is contained in:
Simbiot Harness 2026-10-15 10:11:11 +00:00
parent acedc438f1
commit 9fa188bccb
4 changed files with 403 additions and 1 deletions

184
harness/agent_channel.py Normal file
View file

@ -0,0 +1,184 @@
# -*- 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

View file

@ -968,6 +968,19 @@ def _reflex_tick(level_cfg, seq: int, model: dict):
raise
except Exception as e:
print(f"[portal-channel] рефлекс не смог: {e}", flush=True)
# канал агент↔агент (t-261015090129): очередь ask_agent/poll_messages
# MCP-шлюза. Лучший эффект — на мин-уровне: сообщения других агентов
# попадают в инбокс триажа как обычные входящие.
try:
from . import agent_channel as _ach
res = _ach.sec_reflex()
if not res.get("skipped") and res.get("ingested"):
payload["agent_channel"] = res
_log(f"agent-channel: {res}")
except SystemExit:
raise
except Exception as e:
print(f"[agent-channel] рефлекс не смог: {e}", flush=True)
es.append_event("cycle", f"reflex:{seq}", "ReflexObserved", payload)
return payload, None

View file

@ -346,6 +346,18 @@ def receive_message(update: dict):
target = _resolve_reply_target(chat.get("id"), rt)
if target:
data["reply_to_desc"] = target
# t-261015080211: явный author_* блок + received_at — агентам проще
# читать метаданные автора одним полем, не собирая по отдельным ключам
data["author"] = {
"user_id": data.get("user_id"),
"username": data.get("username"),
"first_name": data.get("first_name"),
"chat_id": data.get("chat_id"),
"chat_type": data.get("chat_type"),
"chat_title": data.get("chat_title"),
"date": data.get("date"),
}
data["received_at"] = data.get("date")
es.append_event("inbox", f"m-{update['update_id']}", "MessageReceived", data)
return f"m-{update['update_id']}"
@ -368,6 +380,20 @@ def send_reply(msg_id: str, text: str, timeout: int = 35):
# HTTP-push'ем на портал — БЕЗ Telegram. У агента может не быть TG-бота
# (simbiot-деплой делает TG опциональным), а портал — не телеграм-юзер:
# разговоры не смешиваются (по chat_id), доставка — отдельный транспорт.
if str(m.get("chat_id") or "").startswith("portal:agent:"):
# канал агент↔агент (t-261015090129): ответ уходит ask_agent'ом
# в очередь того агента, чьё сообщение триажируем. Транспорт — MCP.
from . import agent_channel
to_agent = str(m.get("chat_id")).split(":", 2)[-1]
kind = {"help": "help", "review": "review", "handoff": "handoff"}.get(
m.get("agent_kind"), "help")
ok = agent_channel.push_agent(to_agent, text, kind)
if not ok:
raise RuntimeError(f"ask_agent({to_agent}) не доставил — ответят на следующем тике")
es.append_event("inbox", msg_id, "ReplySent", {
"to_chat": m["chat_id"], "text": text[:2000], "channel": "agent_mcp",
})
return msg_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])
@ -439,6 +465,32 @@ 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-261015090129): батч сообщений других агентов —
# сводный ответ одним ask_agent каждому адресату. Chat_id вида
# portal:agent:<имя>; портальный UI-чат ('portal:<имя>' без agent:) —
# отдельная ветка ниже.
agent_chats = [c for c in by_chat if str(c).startswith("portal:agent:")]
if agent_chats:
if len(by_chat) != len(agent_chats):
raise ValueError("батч смешивает агент-канал и другие каналы — раздели ответы")
from . import agent_channel
for chat_id in agent_chats:
to_agent = str(chat_id).split(":", 2)[-1]
m0 = by_chat[chat_id][0][1]
kind = {"help": "help", "review": "review", "handoff": "handoff"}.get(
m0.get("agent_kind"), "help")
ok = agent_channel.push_agent(to_agent, text, kind)
if not ok:
raise RuntimeError(f"ask_agent({to_agent}) не доставил — попробуем на следующем тике")
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": "agent_mcp",
"batch": {"id": batch_id, "size": len(pending), "head": head_mid},
})
return head_mid
# ── файловый портал-канал (t-261001123121): батч целиком из чата 'portal:*'
# доставляется HTTP-push'ем на портал, без Telegram. Смешивать чаты
# нельзя — один сводный текст на один транспорт.
@ -525,8 +577,18 @@ def poll_loop():
user = (msg.get("from") or {}).get("id")
if ALLOWED and str(user) not in ALLOWED:
# чужой: пишем событие, но не отвечаем (потенциально спам)
# t-261015080211: метаданные автора и в игнор-событии
# (кто писал, чат, время) — не только в MessageReceived
_ig_chat = msg.get("chat") or {}
es.append_event("inbox", f"m-{u['update_id']}", "MessageIgnored", {
"user_id": user, "username": (msg.get("from") or {}).get("username"),
"user_id": user,
"username": (msg.get("from") or {}).get("username"),
"first_name": (msg.get("from") or {}).get("first_name"),
"chat_id": _ig_chat.get("id"),
"chat_type": _ig_chat.get("type"),
"chat_title": _ig_chat.get("title"),
"date": msg.get("date"),
"message_id": msg.get("message_id"),
"reason": "not_allowed", "text": (msg.get("text") or "")[:500],
})
continue

143
tests/test_agent_channel.py Normal file
View file

@ -0,0 +1,143 @@
# -*- coding: utf-8 -*-
"""t-261015090129-ef5029: канал агент↔агент через MCP (ask_agent/poll_messages).
Запуск: ./venv/bin/python tests/test_agent_channel.py
Сеть НЕ трогается: harness.agent_channel._rpc_tool подменяется фейком —
"портал" отвечает заготовками ask_agent/poll_messages. Проверяет:
- pull(): agent_message → MessageReceived в стор (kind=agent_message,
chat_id=portal:agent:<from>, update_id=id события портала);
- дедуп: повторный pull с теми же id не дублирует инбокс (курсор портала
не сдвинулся — стор второй линией отсекает);
- push_agent(): ask_agent(to, text, kind) — ok:true → True; error → False;
- send_reply() на agent-сообщение: ветка portal:agent:* уходит ask_agent'ом
(fake перехватывает), ReplySent пишется, сообщение закрывается;
- send_batch_reply(): сводный ответ агенту одним ask_agent;
- sec_reflex(): rate-limit (не чаще MIN_REFLEX_GAP), после сброса — pull.
"""
import json
import os
import sys
_DST = "/tmp/simbiot-test-agent-channel.db"
for p in (_DST, _DST + "-wal", _DST + "-shm"):
if os.path.exists(p):
os.unlink(p)
os.environ["SIMBIOT_DB"] = _DST
os.environ["SIMBIOT_PORTAL_MCP_TOKEN"] = "agt_test_agent_channel"
os.environ.pop("SIMBIOT_HOME", None)
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from harness import agent_channel, es, tg_inbox
fails = []
def check(name, cond, extra=""):
print(("PASS" if cond else "FAIL"), name, ("" if cond else ("| " + str(extra))))
if not cond:
fails.append(name)
es.init_db()
# ── фейк MCP-шлюза ──────────────────────────────────────────────────────────
CALLS = []
def _fake_rpc(tool, arguments):
CALLS.append((tool, arguments))
if tool == "poll_messages":
return json.dumps({"messages": FAKE_QUEUE.get("messages", []),
"count": len(FAKE_QUEUE.get("messages", [])),
"note": "ok"})
if tool == "ask_agent":
return json.dumps({"ok": True, "to": arguments.get("to"),
"delivered": "agent_message"})
return json.dumps({"error": "unknown"})
agent_channel._rpc_tool = _fake_rpc
FAKE_QUEUE = {"messages": [
{"id": 9001, "from": "OpenCode", "kind": "help",
"text": "🆘 Запрос помощи от «OpenCode»:\n\nСимбиот, глянь deploy_app для groupsign",
"ts": "2026-10-15 09:00:00"},
{"id": 9002, "from": "Roy", "kind": "handoff",
"text": "🤝 Передача задачи от «Roy»:\n\nЗабери мою задачу по туннелям",
"ts": "2026-10-15 09:05:00"},
]}
# ── 1. pull → инбокс ────────────────────────────────────────────────────────
n = agent_channel.pull()
check("pull инжестит 2 сообщения", n == 2, n)
model = es.run_projections()
inbox = model.get("inbox") or {}
m1 = inbox.get("m-9001")
check("m-9001 в сторе", m1 is not None)
if m1:
check("chat_id агентный", m1.get("chat_id") == "portal:agent:OpenCode", m1.get("chat_id"))
check("kind=agent_message", m1.get("kind") == "agent_message")
check("username=агент", m1.get("username") == "OpenCode")
check("agent_kind=help", m1.get("agent_kind") == "help")
check("m-9002 в сторе", inbox.get("m-9002") is not None)
# ── 2. дедуп: тот же poll снова ─────────────────────────────────────────────
n2 = agent_channel.pull()
check("повторный pull — 0 новых (дедуп стора)", n2 == 0, n2)
# ── 3. push_agent ───────────────────────────────────────────────────────────
ok = agent_channel.push_agent("OpenCode", "deploy ок, лог чист", "help")
check("push_agent ok", ok is True and CALLS[-1][0] == "ask_agent", CALLS[-1:])
args = CALLS[-1][1]
check("ask_agent аргументы", args.get("to") == "OpenCode"
and args.get("text") == "deploy ок, лог чист"
and args.get("kind") == "help", args)
ok = agent_channel.push_agent("", "пустой адресат")
check("push_agent без адресата — False", ok is False)
ok = agent_channel.push_agent("Nobody", "")
check("push_agent без текста — False", ok is False)
# ── 4. send_reply на агентское сообщение ────────────────────────────────────
rid = tg_inbox.send_reply("m-9001", "Симбиот: deploy посмотрел, всё чисто. Детали в тике.")
check("send_reply вернул id", rid == "m-9001", rid)
check("ушёл ask_agent (не TG)", CALLS[-1][0] == "ask_agent", CALLS[-1:])
check("адресат из chat_id", CALLS[-1][1].get("to") == "OpenCode", CALLS[-1][1])
model = es.run_projections()
m1 = (model.get("inbox") or {}).get("m-9001")
check("m-9001 закрыт (replied)", bool(m1 and m1.get("replied")))
check("канал ответа agent_mcp", m1 and m1.get("reply_text") == "Симбиот: deploy посмотрел, всё чисто. Детали в тике.")
# ── 5. батч: сводный ответ агенту ───────────────────────────────────────────
FAKE_QUEUE["messages"] = [{"id": 9003, "from": "Roy", "kind": "review",
"text": "Проверь мой отчёт по туннелям", "ts": "2026-10-15 09:10:00"},
{"id": 9004, "from": "Roy", "kind": "review",
"text": "И заодно по слотам", "ts": "2026-10-15 09:11:00"}]
n = agent_channel.pull()
check("pull инжестит вторую партию 2 сообщения", n == 2, n)
CALLS.clear()
head = tg_inbox.send_batch_reply(["m-9003", "m-9004"], "Принял оба: отчёт и слоты проверю.")
check("батч закрыт головным", head == "m-9003", head)
ask_calls = [c for c in CALLS if c[0] == "ask_agent"]
check("один ask_agent на батч", len(ask_calls) == 1, CALLS)
model = es.run_projections()
m3 = (model.get("inbox") or {}).get("m-9003")
m4 = (model.get("inbox") or {}).get("m-9004")
check("оба закрыты", bool(m3 and m3.get("replied") and m4 and m4.get("replied")))
# ── 6. sec_reflex rate-limit ────────────────────────────────────────────────
agent_channel._last_reflex["ts"] = 0.0
r1 = agent_channel.sec_reflex()
r2 = agent_channel.sec_reflex()
check("первый рефлекс работает", "ingested" in r1, r1)
check("второй — rate-limit", r2.get("skipped") == "rate-limit", r2)
# ── 7. pull без токена — тихо ───────────────────────────────────────────────
_tok = os.environ.pop("SIMBIOT_PORTAL_MCP_TOKEN")
n = agent_channel.pull()
check("без токена — 0, без исключений", n == 0, n)
os.environ["SIMBIOT_PORTAL_MCP_TOKEN"] = _tok
print()
print("FAILS:", len(fails), fails if fails else "")
sys.exit(1 if fails else 0)