diff --git a/harness/isolated_runner.py b/harness/isolated_runner.py index aec571f..870ebc7 100644 --- a/harness/isolated_runner.py +++ b/harness/isolated_runner.py @@ -7,8 +7,10 @@ Strands держит глобальный asyncio-рантайм: несколь """ import json import os +import signal import subprocess import sys +import time _HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) _VENV_PY = os.path.join(_HARNESS_DIR, "venv", "bin", "python") @@ -58,10 +60,37 @@ def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, tim env=env, ) except subprocess.TimeoutExpired as e: - # воркер убит по таймауту. Он пишет BudgetSpent сам после каждого - # модельного вызова, так что spend этого тика уже в сторе — - # теряется максимум хвост вызова, шедшего на момент убийства. - # Снаружи дописывать нечего: история воркера недоступна после SIGKILL. + # воркер убит по таймауту. subprocess.run убивает только основной + # процесс (Popen.kill через os.kill без группы) — сабпроцесс _llm_worker + # успевал сам заспавнить LLM-подпроцессы, которые оставались ЖИВЫМИ и + # продолжали писать в стор после SchedulerTimeout (BudgetSpent/ToolCalled + # с завершённым trace — race, burn 9router, t-261002090242-cf7cf9). + # Добиваем всю процесс-группу и ЖДЁМ фактической смерти. + pgid = None + try: + with open(f"/proc/{e.pid}/status") as f: + for line in f: + if line.startswith("Tgid:"): + pgid = line.split()[1] + break + except OSError: + pass + try: + if e.pid > 1: + os.killpg(pgid or e.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + pass + for _ in range(10): # до 5с на смерть группы, иначе — SIGKILL ещё раз + try: + os.killpg(pgid or e.pid, 0) + time.sleep(0.5) + except (ProcessLookupError, PermissionError) as ex: + break + else: + try: + os.killpg(pgid or e.pid, signal.SIGKILL) + except OSError: + pass raise out = None if os.path.exists(result_path): diff --git a/harness/levels.py b/harness/levels.py index 00ddb2f..1d27874 100644 --- a/harness/levels.py +++ b/harness/levels.py @@ -79,6 +79,9 @@ LEVELS = { "cli.py escalate \"причина\" -c \"контекст\" -s critical|warn|info; " "эскалацию увидит часовой/дневной уровень в своём контексте."), can_shell=True, + # t-261002090242-cf7cf9: 300с мин-тику тесно (занятый мин-уровень + # штатно поглощает такты) — таймаут не должен обрывать рабочий тик. + tick_timeout=480, ), "hour": CycleLevel( name="hour", label="Час · поток", every=HOUR, uses_llm=True, diff --git a/harness/portal_channel.py b/harness/portal_channel.py index e4942c5..368a112 100644 --- a/harness/portal_channel.py +++ b/harness/portal_channel.py @@ -31,6 +31,7 @@ REPORTER_ENV = "/etc/coop-reporter.env" STATE_FILE = "/root/simbiot-harness/portal/inbox/.channel-last" PENDING_DIR = "/root/simbiot-harness/portal/inbox/pending" # legacy-путь (пусто) OUTBOX_DIR = "/root/simbiot-harness/portal/inbox/outbox" # legacy fallback +_HARNESS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) HTTP_TIMEOUT = 8 # портал отвечает быстро; висеть дольше — вред MIN_REFLEX_GAP = 20 # сек-рефлекс не дёргает канал чаще, чем раз в 20с @@ -144,6 +145,18 @@ def pull(last_ms: int = 0) -> int: 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 # (вторая попытка того же цикла недоступности не ворошит пустое) @@ -189,7 +202,140 @@ def _post_outbox(items: list, timeout: int = HTTP_TIMEOUT) -> bool: return False -# ══════════════ фоновые вызовы (без LLM) ══════════════ +# ══════════════ /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 [\nallowed=]`. + + 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(os.path.join(_HARNESS_DIR, ".env")) 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 + try: + req = urllib.request.Request(f"https://api.telegram.org/bot{token}/getMe") + with urllib.request.urlopen(req, timeout=10) 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]) + uname = me["result"].get("username") or "bot" + except Exception as e: + _report(f"❌ TG-токен не принят Telegram (getMe): {e}") + print(f"[set-tg] токен не валиден: {e}", flush=True) + return + # 2) аллоулист: из команды (allowed=…), иначе сохранённый в .env + allowed = payload.get("allowed") or payload.get("owner_tg") or "" + if not allowed: + try: + with open(os.path.join(_HARNESS_DIR, ".env")) 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 = os.path.join(_HARNESS_DIR, ".env") + 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-файлов. Дешёво и НЕ чаще