98 lines
No EOL
5.1 KiB
Python
98 lines
No EOL
5.1 KiB
Python
"""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))
|
||
# e-261005110300: раньше subprocess.run + /proc/{e.pid}/status → у
|
||
# TimeoutExpired НЕТ pid, обработчик валило AttributeError'ом и маскировал
|
||
# нормальный SchedulerTimeout (тик «упал» вместо «убит по таймауту»).
|
||
# Теперь: Popen в СВОЕЙ сессии (pgid = pid воркера, дети в той же группе)
|
||
# + communicate(timeout) → по таймауту убиваем ГРУППУ целиком и ждём
|
||
# фактической смерти (сироты-LLM-подпроцессы, race со стором — в прошлом).
|
||
proc = subprocess.Popen(
|
||
[_VENV_PY, _WORKER],
|
||
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||
cwd=_HARNESS_DIR, env=env, start_new_session=True,
|
||
)
|
||
try:
|
||
out_bytes, err_bytes = proc.communicate(
|
||
input=json.dumps(payload).encode(), timeout=timeout)
|
||
except subprocess.TimeoutExpired:
|
||
# Добиваем всю процесс-группу и ЖДЁМ фактической смерти.
|
||
try:
|
||
os.killpg(proc.pid, signal.SIGKILL)
|
||
except (ProcessLookupError, PermissionError, OSError):
|
||
pass
|
||
try:
|
||
proc.communicate(timeout=10) # reap: собираем зомби, чистим пайпы
|
||
except subprocess.TimeoutExpired:
|
||
pass
|
||
raise
|
||
# у_Popen нет полей stdout/stderr: для ошибок ниже используем собранное
|
||
if out_bytes is None:
|
||
out_bytes = b""
|
||
if err_bytes is None:
|
||
err_bytes = b""
|
||
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={out_bytes[:200]!r} stderr={err_bytes.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") |