diff --git a/harness/agent_channel.py b/harness/agent_channel.py new file mode 100644 index 0000000..c0b2798 --- /dev/null +++ b/harness/agent_channel.py @@ -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 diff --git a/harness/scheduler.py b/harness/scheduler.py index 81da892..16f0484 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -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 diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index b2e8dcf..41ba771 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -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 diff --git a/tests/test_agent_channel.py b/tests/test_agent_channel.py new file mode 100644 index 0000000..7823982 --- /dev/null +++ b/tests/test_agent_channel.py @@ -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:, 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)