simbiot-harness/harness/isolated_runner.py

"""Simbiot Harness — изолированный раннер LLM-тика.

Strands держит глобальный asyncio-рантайм: несколько create_harness()
в одном процессе валит его («cannot schedule new futures after shutdown»).
Решение — каждый LLM-тик в ОТДЕЛЬНОМ сабпроцессе: чистый 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")
_WORKER = os.path.join(_HARNESS_DIR, "harness", "_llm_worker.py")
MARKER = "@@SIMBIOT_RESULT@@"


def run_llm_tick(level_cfg, prompt: str, session_id=None, trace_prefix=None, timeout=None):
    """Запускает LLM-тик в сабпроцессе. Возвращает (text, usage, trace_id).

    События трассы воркер пишет в стор сам (у него есть es).
    Протокол: воркер пишет результат в JSON-ФАЙЛ (путь передан в env);
    stdout/stderr только для лога — библиотечный шум strands не мешает.
    """
    import tempfile
    result_path = tempfile.mktemp(prefix="simbiot_tick_", suffix=".json", dir="/tmp")
    # BudgetSpent пишет сам воркер — ИНКРЕМЕНТАЛЬНО, после каждого модельного
    # вызова (см. _llm_worker._wire_budget_hooks). Раньше событие писалось
    # здесь, ПОСЛЕ возврата воркера: таймаут убивал процесс до записи, и
    # spend молча терял весь тик (task t-260929170544-adebea).
    payload = {
        "model": level_cfg.model,
        "effort": level_cfg.effort,
        "instructions": level_cfg.mandate,
        "prompt": prompt,
        "session_id": None,  # сессии пока выключены: .agent/sessions растёт без очистки
        "trace_prefix": trace_prefix,
        "can_shell": level_cfg.can_shell,
        "ctx_tokens": level_cfg.ctx_tokens,
        "result_path": result_path,
        "level": level_cfg.name,  # для записи BudgetSpent самим воркером
    }
    env = dict(os.environ)
    env["SIMBIOT_DB"] = env.get("SIMBIOT_DB", os.path.join(_HARNESS_DIR, "harness", "harness.db"))
    if timeout is None:
        # таймаут уровня: тик может быть длиннее интервала — занятый уровень
        # просто поглощает такты (TickAbsorbed), это норма архитектуры
        timeout = getattr(level_cfg, "tick_timeout", 300)
        timeout = max(timeout, int(level_cfg.every * 0.8))
    try:
        proc = subprocess.run(
            [_VENV_PY, _WORKER],
            input=json.dumps(payload).encode(),
            capture_output=True,
            timeout=timeout,
            cwd=_HARNESS_DIR,
            env=env,
        )
    except subprocess.TimeoutExpired as e:
        # воркер убит по таймауту. 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):
        try:
            with open(result_path) as f:
                out = json.load(f)
        except ValueError:
            out = None
    try:
        os.unlink(result_path)
    except OSError:
        pass
    if out is None:
        return None, {"error": f"worker result file missing: stdout={(proc.stdout or b'')[:200]!r} stderr={(proc.stderr or b'').decode(errors='replace')[-300:]}"}, None
    if not out.get("ok"):
        return None, {"error": (out.get("error") or "unknown")[:500]}, None
    return out.get("text", ""), out.get("usage", {}), out.get("trace_id")