diff --git a/.gitignore b/.gitignore index b843b40..b79f7f8 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ venv/ .agent/ __pycache__/ /tmp/ +.env diff --git a/README.md b/README.md index 831e973..3b3a308 100644 --- a/README.md +++ b/README.md @@ -43,6 +43,7 @@ Strands Harness SDK как исполнительное ядро, модели ./venv/bin/python cli.py events -n 50 ./venv/bin/python cli.py requeue -r "причина" ./venv/bin/python cli.py promote -r "вне класса уровня" # S→M→L +./venv/bin/python cli.py tg-reply "текст" # ответ на входящее TG ``` ## Запуск @@ -52,6 +53,13 @@ systemctl enable --now simbiot-harness journalctl -u simbiot-harness -f ``` +## Telegram-инбокс + +Бот @juno42bot → long polling (`simbiot-harness-tg.service`) → +`MessageReceived` в стор → минутный уровень видит входящие в контексте → +отвечает `cli.py tg-reply` (Telegram-отправка + `ReplySent` закрывает входящее). +Белый список чатов в `.env` (`SIMBIOT_TG_ALLOWED`), чужие → `MessageIgnored`. + ## Модели 9router (router.cil.by), отдельный ключ `sk-simbiot` в `.env`, комбо `auto`: diff --git a/__pycache__/cli.cpython-312.pyc b/__pycache__/cli.cpython-312.pyc index 4c1d888..0c60966 100644 Binary files a/__pycache__/cli.cpython-312.pyc and b/__pycache__/cli.cpython-312.pyc differ diff --git a/cli.py b/cli.py index 2cfc645..95666de 100644 --- a/cli.py +++ b/cli.py @@ -31,6 +31,10 @@ def main(): p_pr.add_argument("task_id") p_pr.add_argument("-r", "--reason", default="manual promote") + p_tgr = sub.add_parser("tg-reply", help="ответить на входящее TG-сообщение") + p_tgr.add_argument("msg_id", help="id входящего (m-NNN)") + p_tgr.add_argument("text") + sub.add_parser("status", help="состояние харнеса") sub.add_parser("tasks", help="доска задач") @@ -61,6 +65,11 @@ def main(): es.init_db() tid, new_size = commands.promote_task(args.task_id, args.reason) print(f"OK: задача {tid} повышена до [{new_size}]") + elif args.cmd == "tg-reply": + es.init_db() + from harness import tg_inbox + mid = tg_inbox.send_reply(args.msg_id, args.text) + print(f"OK: ответ на {mid} отправлен") elif args.cmd == "status": es.init_db() print(json.dumps(commands.status(), ensure_ascii=False, indent=1)) diff --git a/harness/__pycache__/es.cpython-312.pyc b/harness/__pycache__/es.cpython-312.pyc index a39efd1..804bae8 100644 Binary files a/harness/__pycache__/es.cpython-312.pyc and b/harness/__pycache__/es.cpython-312.pyc differ diff --git a/harness/__pycache__/levels.cpython-312.pyc b/harness/__pycache__/levels.cpython-312.pyc index dba2ddd..b367113 100644 Binary files a/harness/__pycache__/levels.cpython-312.pyc and b/harness/__pycache__/levels.cpython-312.pyc differ diff --git a/harness/__pycache__/scheduler.cpython-312.pyc b/harness/__pycache__/scheduler.cpython-312.pyc index 69bca44..e4508ee 100644 Binary files a/harness/__pycache__/scheduler.cpython-312.pyc and b/harness/__pycache__/scheduler.cpython-312.pyc differ diff --git a/harness/es.py b/harness/es.py index 00f6dec..d363571 100644 --- a/harness/es.py +++ b/harness/es.py @@ -194,6 +194,26 @@ def _apply_event(model: dict, ev: dict): spend.append(data) if len(spend) > 500: # кап: не растить модель линейно model["spend"] = spend[-500:] + elif etype == "MessageReceived": + model.setdefault("inbox", {})[aid] = { + **data, "replied": False, "ignored": False, + "ts": ev["timestamp"], + } + inbox = model["inbox"] + if len(inbox) > 100: # кап инбокса: последние 100 входящих + keep = sorted(inbox.items(), key=lambda kv: kv[1].get("ts", ""))[-100:] + model["inbox"] = dict(keep) + elif etype == "ReplySent": + m = model.setdefault("inbox", {}).get(aid) + if m: + m["replied"] = True + m["reply_text"] = data.get("text", "") + elif etype == "MessageIgnored": + m = model.setdefault("inbox", {}).get(aid) + if m: + m["ignored"] = True + else: + model["inbox"][aid] = {"ignored": True, "reason": data.get("reason"), "ts": ev["timestamp"]} # trace-события (InvocationStarted/ToolCalled/ToolResult) не материализуем — они для дебага diff --git a/harness/levels.py b/harness/levels.py index 67aed1c..c1089a8 100644 --- a/harness/levels.py +++ b/harness/levels.py @@ -43,9 +43,12 @@ LEVELS = { "min": CycleLevel( name="min", label="Минута · пульс", every=MIN, uses_llm=True, model="auto", effort="low", ctx_tokens=4_000, task_classes=("S",), - mandate=("Ты — минутный пульс харнеса Simbiot. Триаж входящих, быстрые ответы, " - "мелкие фиксы (S-задачи до 5 минут). Отвечай кратко, не углубляйся. " - "Если задача больше твоего класса — откажись и предложи промоушен."), + mandate=("Ты — минутный пульс харнеса Simbiot. Триаж входящих TG-сообщений: " + "если во ВХОДЯЩИХ есть сообщение без ответа — ответь на него командой " + "cli.py tg-reply \"текст\" (тон: кратко, по-русски, тёпло; можно " + "выполнить просьбу инструментами и ответить результатом). Быстрые " + "мелкие фиксы (S-задачи до 5 минут) тоже твои. Не растекайся. " + "Если просьба большая — подай её как задачу (cli.py task) и скажи это."), can_shell=True, ), "hour": CycleLevel( diff --git a/harness/scheduler.py b/harness/scheduler.py index b0f82f4..dd73f2d 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -89,6 +89,17 @@ def build_context(level_cfg, model: dict, task=None): for d in digests[:6]: parts.append(f"- [{d['level']}] {d['text'][:400]}") + # инбокс: входящие TG-сообщения, ждущие ответа (в приоритете) + inbox = [m for m in (model.get("inbox") or {}).values() + if not m.get("replied") and not m.get("ignored") and m.get("text")] + if inbox: + lines = [] + for m in inbox[:10]: + m_id = "m-" + str(m.get("update_id")) + who = m.get("username") or m.get("first_name") or m.get("user_id") + lines.append(f"- [{m_id}] от {who}: {m['text'][:300]}") + parts.append("ВХОДЯЩИЕ СООБЩЕНИЯ (ответь через: ./venv/bin/python cli.py tg-reply \"текст\"):\n" + "\n".join(lines)) + # доска задач tasks = [t for t in model["tasks"].values() if t["status"] in ("ready", "running")] if tasks: diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py new file mode 100644 index 0000000..4f6c806 --- /dev/null +++ b/harness/tg_inbox.py @@ -0,0 +1,120 @@ +"""Simbiot Harness — Telegram-инбокс. + +Входящие сообщения Алекса → события MessageReceived в стор → +минутный уровень видит их в контексте и может ответить через +`cli.py tg-reply ` (событие ReplySent закрывает входящее). + +Поллер — отдельный процесс (systemd simbiot-harness-tg), long polling +getUpdates offset=last+1. Идемпотентность: update_id → aggregate_id. +""" +import json +import os +import time +import urllib.request +import urllib.parse + +from . import es + +_HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +_ENV_PATH = os.path.join(_HARNESS_DIR, ".env") + +TG_TOKEN = None +ALLOWED = set() + + +def _load_env(): + global TG_TOKEN, ALLOWED + if os.path.exists(_ENV_PATH): + with open(_ENV_PATH) as f: + for line in f: + line = line.strip() + if line.startswith("SIMBIOT_TG_TOKEN="): + TG_TOKEN = line.split("=", 1)[1].strip() + elif line.startswith("SIMBIOT_TG_ALLOWED="): + ALLOWED = {x.strip() for x in line.split("=", 1)[1].split(",") if x.strip()} + + +def _api(method: str, params: dict, timeout: int = 35): + url = f"https://api.telegram.org/bot{TG_TOKEN}/{method}" + data = urllib.parse.urlencode(params).encode() + req = urllib.request.Request(url, data=data) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.loads(r.read().decode()) + + +def send_message(chat_id, text: str, reply_to=None): + """Отправка сообщения (используется tg-reply и алертами).""" + _load_env() + params = {"chat_id": chat_id, "text": text, "parse_mode": "HTML"} + if reply_to: + params["reply_to_message_id"] = reply_to + return _api("sendMessage", params) + + +def receive_message(update: dict): + """MessageReceived в стор. Идемпотентно по update_id.""" + msg = update.get("message") or update.get("edited_message") or {} + chat = msg.get("chat") or {} + user = msg.get("from") or {} + data = { + "update_id": update["update_id"], + "chat_id": chat.get("id"), + "user_id": user.get("id"), + "username": user.get("username"), + "first_name": user.get("first_name"), + "text": (msg.get("text") or "")[:4000], + "date": msg.get("date"), + } + es.append_event("inbox", f"m-{update['update_id']}", "MessageReceived", data) + return f"m-{update['update_id']}" + + +def send_reply(msg_id: str, text: str): + """Ответ на входящее: Telegram-отправка + событие ReplySent (закрывает входящее).""" + model = es.run_projections() + m = (model.get("inbox") or {}).get(msg_id) + if m is None: + raise ValueError(f"входящее {msg_id} не найдено") + if m.get("replied"): + raise ValueError(f"на {msg_id} уже отвечено") + r = send_message(m["chat_id"], text, reply_to=None) + es.append_event("inbox", msg_id, "ReplySent", { + "to_chat": m["chat_id"], "text": text[:2000], "tg_ok": bool(r.get("ok")), + }) + return msg_id + + +def poll_loop(): + """Long polling getUpdates → MessageReceived. Отдельный процесс.""" + _load_env() + if not TG_TOKEN: + raise SystemExit("SIMBIOT_TG_TOKEN не задан в .env") + es.init_db() + offset = 0 + print("[tg] инбокс-поллер запущен", flush=True) + while True: + try: + updates = _api("getUpdates", {"offset": offset, "timeout": 30, "limit": 16}) + for u in updates.get("result", []): + offset = max(offset, u["update_id"] + 1) + msg = u.get("message") or u.get("edited_message") + if not msg: + continue + user = (msg.get("from") or {}).get("id") + if ALLOWED and str(user) not in ALLOWED: + # чужой: пишем событие, но не отвечаем (потенциально спам) + es.append_event("inbox", f"m-{u['update_id']}", "MessageIgnored", { + "user_id": user, "username": (msg.get("from") or {}).get("username"), + "reason": "not_allowed", "text": (msg.get("text") or "")[:500], + }) + continue + mid = receive_message(u) + print(f"[tg] {mid} от {(msg.get('from') or {}).get('username') or user}: " + f"{(msg.get('text') or '')[:60]}", flush=True) + except Exception as e: + print(f"[tg] ошибка поллинга: {e}", flush=True) + time.sleep(5) + + +if __name__ == "__main__": + poll_loop() \ No newline at end of file