871a3b15a3
- гейт (e-260930062242/2510/2607/2741): collect_signals ранжирует события (unreplied_inbox/ready_task/critical-эскалация = critical; warn-эскалации, high_load = warn), gate_allows_llm решает по порогу уровня (SIMBIOT_GATE_<LEVEL>, дефолты min=critical/hour=warn/day=warn); тихий такт — GateSkipped + дайджест-заглушка, без LLM и без расхода роутера - model в BudgetSpent/InvocationStarted (e-260930062120): видно, какая модель и уровень тратили кредиты (strands не выносит фактическую модель в usage — пишем запрошенный model_id) - per-level лимиты тул-колов уже в дереве: resolve_tool_budget (min=30/hour=300, env-оверрайд) (e-260930060717/060904) - tests/test_gate.py: 28 hermetic-проверок, ALL PASS - simbiot-presentation.html: презентация Simbiot (t-260930053538), 10 слайдов
601 lines
No EOL
31 KiB
Python
601 lines
No EOL
31 KiB
Python
"""Simbiot Harness — scheduler: тактовая лестница + cycle runners.
|
||
|
||
Правила:
|
||
- тик уровня идемпотентен (UNIQUE level:seq);
|
||
- один исполнитель на уровень → сериализация;
|
||
- занятый уровень поглощает тик (TickAbsorbed) и пишет
|
||
детерминированную дайджест-заглушку — без LLM;
|
||
- запущенная задача не прерывается.
|
||
"""
|
||
import json
|
||
import os
|
||
import time
|
||
import traceback
|
||
from datetime import datetime, timezone, timedelta
|
||
|
||
from . import es
|
||
from .levels import LEVELS
|
||
from . import runtime
|
||
from . import commands
|
||
from .commands import _ORDER
|
||
|
||
# интервал демона
|
||
POLL = 1.0
|
||
# уровни, включённые в v0.1 (порядок для лога)
|
||
ACTIVE = ["sec", "min", "hour", "day"]
|
||
|
||
# состояние исполнителей уровня: level → {"busy": bool, "task": id|None}
|
||
_runners = {}
|
||
# когда уровень последний раз слал TG-уведомление о занятости (анти-спам)
|
||
_NOTIFY_TS = {}
|
||
|
||
|
||
def _now():
|
||
return datetime.now(timezone.utc)
|
||
|
||
|
||
def _log(msg: str):
|
||
print(f"[{datetime.now().strftime('%H:%M:%S')}] {msg}", flush=True)
|
||
|
||
|
||
def tick_number(level_cfg, ts: int):
|
||
return ts // level_cfg.every
|
||
|
||
|
||
def emit_tick(level_cfg, seq: int):
|
||
"""CycleTicked идемпотентно. Возвращает True если это новый тик."""
|
||
r = es.append_tick(level_cfg.name, seq)
|
||
return r is not None
|
||
|
||
|
||
def deterministic_digest_stub(level_cfg, seq: int, reason: str):
|
||
"""Заглушка дайджеста для поглощённого тика — детерминированно, без LLM."""
|
||
period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc)
|
||
es.append_event("cycle", f"stub:{level_cfg.name}:{seq}", "DigestWritten", {
|
||
"level": level_cfg.name,
|
||
"period": period.isoformat(),
|
||
"summary_ref": None,
|
||
"text": f"[absorbed:{reason}] уровень {level_cfg.name} был занят/пропущен; такт {seq}",
|
||
"stub": True,
|
||
})
|
||
|
||
|
||
def find_task(level_cfg, model: dict):
|
||
"""Ищет ready-задачу доступного классу уровня. Возвращает task_id или None."""
|
||
classes = level_cfg.task_classes
|
||
for tid, t in model["tasks"].items():
|
||
if t["status"] == "ready" and t.get("size") in classes:
|
||
return tid
|
||
return None
|
||
|
||
|
||
def _lvl_rank(name: str) -> int:
|
||
"""Ранг уровня в лестнице; неизвестный → ниже всех (эскалации ему не адресованы)."""
|
||
return _ORDER.index(name) if name in _ORDER else len(_ORDER)
|
||
|
||
|
||
# ────────────────────────── ГЕЙТ ВАЖНОСТИ (e-260930062242/2510/2607/2741) ──────────────────────────
|
||
# Alex: LLM-тик уровня запускается ТОЛЬКО если есть событие ≥ порога важности
|
||
# этого уровня; иначе тик отрабатывает чисто детерминированной механикой (как sec).
|
||
# Ранжирование: critical — входящее TG без ответа, готовая задача уровня,
|
||
# открытая critical-эскалация; warn — warn-эскалации, высокая нагрузка;
|
||
# всё остальное — тишина, LLM не нужен (не тратим кредиты роутера).
|
||
|
||
# порог по умолчанию на уровень (env-оверрайд: SIMBIOT_GATE_<LEVEL>=critical|warn|off)
|
||
_GATE_DEFAULTS = {"min": "critical", "hour": "warn", "day": "warn"}
|
||
|
||
|
||
def _gate_threshold(level_cfg) -> str:
|
||
"""Порог важности уровня: off — гейт выключен, LLM всегда."""
|
||
env = os.environ.get(f"SIMBIOT_GATE_{level_cfg.name.upper()}")
|
||
if env:
|
||
return env.strip().lower()
|
||
return _GATE_DEFAULTS.get(level_cfg.name, "off")
|
||
|
||
|
||
def _pending_inbox(model: dict):
|
||
"""Входящие TG без ответа (не команды) — материал для минутного LLM."""
|
||
from . import tg_commands
|
||
return [m for m in (model.get("inbox") or {}).values()
|
||
if not m.get("replied") and not m.get("ignored")
|
||
and not tg_commands.parse(m.get("text") or "")
|
||
and (m.get("text") or m.get("kind") in ("voice", "audio", "video_note"))]
|
||
|
||
|
||
def collect_signals(level_cfg, model: dict, load: float = None):
|
||
"""События, важные именно для этого уровня, с рангом важности.
|
||
|
||
Возвращает [(severity, what, detail)] — чистая функция от read model,
|
||
без побочных эффектов; используется гейтом LLM и тестом (hermetic).
|
||
"""
|
||
sigs = []
|
||
# 1) входящее TG без ответа — критично (должен ответить этот или нижний уровень)
|
||
inbox = _pending_inbox(model)
|
||
if inbox:
|
||
ids = ", ".join(("m-" + str(m.get("update_id"))) for m in inbox[:5])
|
||
sigs.append(("critical", "unreplied_inbox",
|
||
f"{len(inbox)} сообщений без ответа ({ids})"))
|
||
# 2) готовая задача доступного уровня классу — критично (тик должен её взять)
|
||
ready = [t for t in (model.get("tasks") or {}).values()
|
||
if t.get("status") == "ready" and t.get("size") in level_cfg.task_classes]
|
||
if ready:
|
||
ids = ", ".join(f"{x['id']}[{x.get('size')}]" for x in
|
||
sorted(ready, key=lambda x: x.get("created", ""))[:5])
|
||
sigs.append(("critical", "ready_task", f"готовые задачи: {ids}"))
|
||
# 3) открытые эскалации, адресованные этому уровню и выше — ранг = severity
|
||
for e in (model.get("escalations") or {}).values():
|
||
if not isinstance(e, dict) or e.get("status") != "open" or e.get("stub"):
|
||
continue
|
||
to_level = e.get("to_level", "day")
|
||
if to_level not in _ORDER or _lvl_rank(level_cfg.name) < _lvl_rank(to_level):
|
||
continue
|
||
sev = e.get("severity", "info")
|
||
if sev in ("critical", "warn", "info"):
|
||
sigs.append((sev, f"escalation:{e.get('id')}",
|
||
(e.get("reason") or "")[:120]))
|
||
# 4) высокая нагрузка — вторично (алерт-порог как у сек-рефлекса)
|
||
if load is not None:
|
||
thr = (model.get("policy") or {}).get("cpu_alert_threshold", 8)
|
||
if load >= thr:
|
||
sigs.append(("warn", "high_load", f"load1 {load:.2f} >= {thr}"))
|
||
return sigs
|
||
|
||
|
||
def gate_allows_llm(level_cfg, signals) -> bool:
|
||
"""Решение гейта: запускать ли LLM-тик по порогу важности уровня."""
|
||
thr = _gate_threshold(level_cfg)
|
||
if thr == "off":
|
||
return True
|
||
order = {"critical": 0, "warn": 1, "info": 2}
|
||
if thr not in order:
|
||
return True
|
||
rank = order[thr]
|
||
return any(order.get(s[0], 3) <= rank for s in signals)
|
||
|
||
|
||
def build_context(level_cfg, model: dict, task=None):
|
||
"""Сборка контекста тика: мандат + память уровня + выжимки нижних + задача."""
|
||
parts = [level_cfg.mandate]
|
||
|
||
# идентичность/цели v0.1 — статично из policy (если есть)
|
||
policy = model.get("policy")
|
||
if policy and policy.get("identity"):
|
||
parts.append("ИДЕНТИЧНОСТЬ: " + policy["identity"])
|
||
|
||
# выжимки нижних уровней: последние дайджесты
|
||
order = ["min", "hour", "day"]
|
||
idx = order.index(level_cfg.name) if level_cfg.name in order else 0
|
||
lower = order[:idx]
|
||
digests = []
|
||
for lvl in lower:
|
||
for key, d in model["digests"].items():
|
||
if d["level"] == lvl and not d.get("stub"):
|
||
digests.append(d)
|
||
digests.sort(key=lambda d: d["ts"], reverse=True)
|
||
if digests:
|
||
parts.append("ВЫЖИМКИ НИЖНИХ УРОВНЕЙ (последние, сжатые):")
|
||
for d in digests[:6]:
|
||
parts.append(f"- [{d['level']}] {d['text'][:400]}")
|
||
|
||
# эскалации: открытые, адресованные этому уровню и выше — в приоритете
|
||
escs = []
|
||
for e in (model.get("escalations") or {}).values():
|
||
if not isinstance(e, dict) or e.get("status") != "open" or e.get("stub"):
|
||
continue
|
||
to_level = e.get("to_level", "day")
|
||
if to_level not in _ORDER or _lvl_rank(level_cfg.name) < _lvl_rank(to_level):
|
||
continue
|
||
escs.append(e)
|
||
if escs:
|
||
escs.sort(key=lambda e: ({'critical': 0, 'warn': 1, 'info': 2}.get(e.get("severity"), 3),
|
||
e.get("ts", "")), reverse=False)
|
||
lines = []
|
||
for e in escs[:10]:
|
||
who = f"от {e.get('from_level')}" if e.get("from_level") else ""
|
||
lines.append(f"- [{e['id']}] ({e.get('severity','info')}) {who} → "
|
||
f"{e.get('to_level')}: {e.get('reason','')[:300]}"
|
||
+ (f"\n контекст: {e.get('context','')[:400]}" if e.get("context") else ""))
|
||
parts.append("ЭСКАЛАЦИИ (важное с нижних уровней; отработай и закрой: "
|
||
"./venv/bin/python cli.py escalation-ack <id>):\n" + "\n".join(lines))
|
||
|
||
# инбокс: входящие TG-сообщения, ждущие ответа (в приоритете).
|
||
# Голосовое с пустой расшифровкой (тишина/отказ) тоже показываем —
|
||
# минутный уровень должен знать, что голосовое пришло, но текста нет,
|
||
# и ответить «не расслышал», а не делать вид, что сообщения не было.
|
||
# Команды бота (/лестница и др.) — НЕ здесь: их мгновенно закрывает
|
||
# сек-рефлекс (t-260929210755), мин-LLM на них не тратится. Фильтруем
|
||
# даже без ответа — иначе гонка на границе тика (команда пришла,
|
||
# сек ещё не отвечал) или лежащий TG заставляли минуту отвечать LLM.
|
||
from . import tg_commands
|
||
inbox = [m for m in (model.get("inbox") or {}).values()
|
||
if not m.get("replied") and not m.get("ignored")
|
||
and not tg_commands.parse(m.get("text") or "")
|
||
and (m.get("text") or m.get("kind") in ("voice", "audio", "video_note"))]
|
||
if inbox:
|
||
lines = []
|
||
for m in sorted(inbox, key=lambda x: x.get("ts") or "")[-10:]:
|
||
m_id = "m-" + str(m.get("update_id"))
|
||
who = m.get("username") or m.get("first_name") or m.get("user_id")
|
||
tag = ""
|
||
if m.get("kind") == "voice":
|
||
if m.get("transcribed") and m.get("text"):
|
||
tag = " [голосовое, расшифровка]"
|
||
else:
|
||
tag = (" [голосовое, расшифровка не удалась — попроси прислать текстом]"
|
||
if m.get("transcribed") else
|
||
" [голосовое, ещё не расшифровано]")
|
||
lines.append(f"- [{m_id}] от {who}{tag}: {m.get('text','')[:300]}")
|
||
parts.append("ВХОДЯЩИЕ СООБЩЕНИЯ (ответь через: ./venv/bin/python cli.py tg-reply <id> \"текст\"):\n" + "\n".join(lines))
|
||
|
||
# доска задач
|
||
tasks = [t for t in model["tasks"].values() if t["status"] in ("ready", "running")]
|
||
if tasks:
|
||
lines = [f"- {t['id']}: {t['title'][:80]} [{t['size']}] {t['status']}" for t in tasks[:15]]
|
||
parts.append("ДОСКА ЗАДАЧ:\n" + "\n".join(lines))
|
||
|
||
if task:
|
||
parts.append(f"ТЕКУЩАЯ ЗАДАЧА: {task['id']}: {task['title']}\n"
|
||
+ (f"ПОСТАНОВКА: {task['details']}\n" if task.get("details") else "")
|
||
+ "Выполни её и выдай результат.")
|
||
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
def _notify_busy(level_cfg, task):
|
||
"""TG-visibility: уровни сами сообщают о занятости.
|
||
|
||
Анти-спам: по одному уведомлению на уровень не чаще раза в 15 минут —
|
||
повторные взятия задач внутри часа не заспамят чат.
|
||
"""
|
||
global _NOTIFY_TS
|
||
now = time.time()
|
||
if now - _NOTIFY_TS.get(level_cfg.name, 0) < 900: # 15 минут
|
||
return
|
||
_NOTIFY_TS[level_cfg.name] = now
|
||
from . import tg_inbox
|
||
tg_inbox.notify_chat(
|
||
f"🔧 Уровень {level_cfg.name} взял задачу [{task.get('size', '?')}] "
|
||
f"«{task.get('title', '')[:70]}»\n"
|
||
f"Срез лестницы по запросу: /лестница"
|
||
)
|
||
|
||
|
||
def run_level_tick(level_cfg, seq: int, model: dict):
|
||
"""Один тик уровня с LLM: дайджест + (опц.) задача. Возвращает текст.
|
||
|
||
LLM-тик гоняется в изолированном сабпроцессе (isolated_runner):
|
||
у Strands глобальный asyncio-рантайм, второй create_harness()
|
||
в том же процессе валит его («cannot schedule new futures»).
|
||
"""
|
||
from . import isolated_runner
|
||
|
||
task_id = find_task(level_cfg, model)
|
||
task = model["tasks"].get(task_id) if task_id else None
|
||
if task:
|
||
es.append_event("task", task_id, "TaskClaimed", {
|
||
"level": level_cfg.name, "session_id": f"s-{level_cfg.name}-{seq}",
|
||
})
|
||
# TG-visibility: взяли задачу — сообщаем в чат (анти-спам: не чаще 15м на уровень)
|
||
try:
|
||
from . import tg_inbox
|
||
_notify_busy(level_cfg, task)
|
||
except Exception:
|
||
pass # уведомление — best effort, не валить тик
|
||
prompt = build_context(level_cfg, model, task)
|
||
text, usage, trace_id = isolated_runner.run_llm_tick(
|
||
level_cfg, prompt,
|
||
session_id=f"s-{level_cfg.name}-{seq}",
|
||
trace_prefix=f"{level_cfg.name}-{seq}",
|
||
)
|
||
if text is None:
|
||
# воркер упал — задача не выполнена, откатываем claim
|
||
if task:
|
||
es.append_event("task", task_id, "TaskFailed", {
|
||
"error": str(usage.get("error", "llm worker failed"))[:500],
|
||
})
|
||
raise RuntimeError(str(usage.get("error", "llm worker failed")))
|
||
# дайджест тика
|
||
period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc)
|
||
es.append_event("cycle", f"digest:{level_cfg.name}:{seq}", "DigestWritten", {
|
||
"level": level_cfg.name, "period": period.isoformat(),
|
||
"summary_ref": trace_id, "text": (text or "")[:2000],
|
||
})
|
||
# BudgetSpent воркер уже записал сам — инкрементально, после каждого
|
||
# модельного вызова (устраняет потерю spend при таймауте воркера).
|
||
# Если usage в результате пуст (воркер не смог разобрать историю —
|
||
# старый воркер / нетекстовая структура), дублируем итог, чтобы тик
|
||
# не терялся в spend вовсе: дедуп по trace_id в read model.
|
||
if usage and not usage.get("model_calls"):
|
||
es.append_event("harness", "budget", "BudgetSpent", {
|
||
"level": level_cfg.name, "trace_id": trace_id, "source": "runner",
|
||
**{k: v for k, v in usage.items() if k != "error"},
|
||
})
|
||
if task:
|
||
es.append_event("task", task_id, "TaskCompleted", {
|
||
"result_ref": trace_id, "answer": (text or "")[:2000],
|
||
})
|
||
# автодоставка (t-260929205927): задачу заказали из TG (await_delivery)
|
||
# — результат уходит в чат сразу после завершения, без повторного запроса.
|
||
# Единственный путь доставки — commands.deliver_result (идемпотентен:
|
||
# без флага / уже sent — молча пропустит). Не отправилось здесь —
|
||
# дослает delivery-демон (deliver_pending) на следующем проходе.
|
||
try:
|
||
from . import commands as _cmd
|
||
if _cmd.deliver_result(task_id, text or ""):
|
||
_log(f"delivery: результат {task_id} доставлен в TG")
|
||
except Exception as e:
|
||
# доставка — best effort: не валить тик; DeliverySent допишет
|
||
# delivery-демон (см. deliver_pending) — задача не потеряется
|
||
_log(f"delivery: отправка {task_id} не удалась ({e}) — доставит delivery-демон")
|
||
return text, task_id
|
||
|
||
|
||
def process_level(level_cfg, model: dict):
|
||
"""Тик одного уровня: занят? → absorb; иначе работа."""
|
||
st = _runners.setdefault(level_cfg.name, {"busy": False, "task": None})
|
||
if st["busy"]:
|
||
seq = tick_number(level_cfg, int(time.time()))
|
||
deterministic_digest_stub(level_cfg, seq, "busy")
|
||
es.append_event("cycle", f"absorbed:{level_cfg.name}:{seq}", "TickAbsorbed",
|
||
{"level": level_cfg.name, "reason": "busy"})
|
||
return "absorbed"
|
||
|
||
seq = tick_number(level_cfg, int(time.time()))
|
||
if not emit_tick(level_cfg, seq):
|
||
return "already" # тик этого такта уже был (идемпотентность)
|
||
|
||
# ГЕЙТ ВАЖНОСТИ (e-260930062242/2510/2607/2741): LLM-тик запускается
|
||
# только если есть событие ≥ порога уровня; тихий такт отрабатывает
|
||
# детерминированной механикой (как sec) — без расхода кредитов роутера.
|
||
if level_cfg.uses_llm:
|
||
try:
|
||
load = os.getloadavg()[0]
|
||
except Exception:
|
||
load = None
|
||
signals = collect_signals(level_cfg, model, load)
|
||
if not gate_allows_llm(level_cfg, signals):
|
||
period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc)
|
||
es.append_event("cycle", f"digest:{level_cfg.name}:{seq}", "DigestWritten", {
|
||
"level": level_cfg.name, "period": period.isoformat(),
|
||
"summary_ref": None,
|
||
"text": f"[gate:quiet] {level_cfg.name}: событий важнее порога нет "
|
||
f"({len(signals)} ниже порога) — LLM не запускался",
|
||
"stub": True,
|
||
})
|
||
es.append_event("cycle", f"gate:{level_cfg.name}:{seq}", "GateSkipped", {
|
||
"level": level_cfg.name,
|
||
"threshold": _gate_threshold(level_cfg),
|
||
"signals": [{"severity": s[0], "what": s[1], "detail": s[2]}
|
||
for s in signals][:10],
|
||
})
|
||
return f"gate:quiet ({len(signals)} signals below threshold)"
|
||
|
||
st["busy"] = True
|
||
try:
|
||
if not level_cfg.uses_llm:
|
||
# секундный уровень: детерминированный рефлекс без LLM
|
||
text, task_id = _reflex_tick(level_cfg, seq, model)
|
||
return f"reflex ok {task_id or ''}".strip()
|
||
text, task_id = run_level_tick(level_cfg, seq, model)
|
||
return f"ran (task={task_id or '-'})"
|
||
except Exception as e:
|
||
es.append_event("harness", "errors", "CycleError", {
|
||
"level": level_cfg.name, "error": str(e)[:1000],
|
||
"trace": traceback.format_exc()[-2000:],
|
||
})
|
||
# воркер LLM-тика умер — наверх, чтобы верхний уровень узнал сразу:
|
||
# нижний сам себя не чинит (контекст мал), фикс идёт сверху.
|
||
try:
|
||
import secrets as _secrets
|
||
esc_id = f"e-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}-{_secrets.token_hex(3)}"
|
||
es.append_event("harness", esc_id, "EscalationRaised", {
|
||
"id": esc_id, "from_level": level_cfg.name,
|
||
"to_level": commands.level_above(level_cfg.name) or "day",
|
||
"reason": f"тик {level_cfg.name} упал: {str(e)[:200]}",
|
||
"context": traceback.format_exc()[-1500:],
|
||
"severity": "warn", "task_id": None, "source": "scheduler",
|
||
})
|
||
except Exception:
|
||
pass # эскалация — best effort: не маскировать исходную ошибку
|
||
return "error: " + str(e)[:120]
|
||
finally:
|
||
st["busy"] = False
|
||
|
||
|
||
def _reflex_tick(level_cfg, seq: int, model: dict):
|
||
"""Секундный рефлекс: телеметрия + алерты + без-LLM команды, без LLM.
|
||
|
||
Команды бота (/лестница, /статус, /задачи, /помощь) — мгновенные
|
||
ответы из read model (t-260929210755): мин-уровень на них не
|
||
тратится, ReplySent закрывает входящее сразу после отправки.
|
||
"""
|
||
import os
|
||
load = None
|
||
try:
|
||
load = os.getloadavg()[0]
|
||
except Exception:
|
||
pass
|
||
period = datetime.fromtimestamp(seq * level_cfg.every, tz=timezone.utc)
|
||
payload = {"level": "sec", "seq": seq, "load1": load,
|
||
"ts": period.isoformat()}
|
||
# алерт-порог из policy (по умолчанию 8 — как simbiotd)
|
||
thr = (model.get("policy") or {}).get("cpu_alert_threshold", 8)
|
||
if load is not None and load >= thr:
|
||
payload["alert"] = f"load {load:.2f} >= {thr}"
|
||
# без-LLM команды: отвечаем на входящие-команды без ответа
|
||
try:
|
||
from . import tg_commands
|
||
handled = tg_commands.answer_pending(model)
|
||
if handled:
|
||
for mid, cmd in handled:
|
||
_log(f"tg-cmd: {mid} /{cmd} → отвечено (сек-рефлекс)")
|
||
payload["commands"] = [c for _, c in handled]
|
||
except Exception as e:
|
||
# команды — часть рефлекса, но не должны валить телеметрию
|
||
print(f"[tg-cmd] ответ на команды не удался: {e}", flush=True)
|
||
es.append_event("cycle", f"reflex:{seq}", "ReflexObserved", payload)
|
||
return payload, None
|
||
|
||
|
||
# ────────────── АВТОДОСТАВКА (t-260929205927) + СИРОТЫ-РЕСТАРТЫ ──────────────
|
||
|
||
DELIVERY_POLL = 15 # сек: проход delivery-демона (push уже сделал большую часть)
|
||
|
||
|
||
def orphan_requeue():
|
||
"""Сироты рестартов: задачи в running без живого воркера → очередь.
|
||
|
||
Раньше: тик-владелец задачи умирал (systemctl restart демона при деплое
|
||
убивает и воркеров), TaskFailed/TaskCompleted не пишется — и задача
|
||
висела в running навсегда (эскалация e-260929234911: cad7af завис дважды
|
||
за два часа; «повторный таймаут часового воркера» был на самом деле
|
||
убийством живого воркера рестартом). Мин-уровень вычищал руками.
|
||
|
||
Теперь демон при старте и каждые DELIVERY_POLL секунд:
|
||
- running-задача с мёртвым воркером (нет свежего следа сессии) →
|
||
TaskPromoted(ready) с reason 'orphan-requeue'; флаг await_delivery
|
||
сохраняется, доработка после requeue тоже уйдёт в чат.
|
||
"""
|
||
model = es.run_projections()
|
||
from datetime import timedelta
|
||
now = _now()
|
||
requeued = []
|
||
for tid, t in (model.get("tasks") or {}).items():
|
||
if t.get("status") != "running":
|
||
continue
|
||
sess = t.get("session_id") or ""
|
||
level = t.get("level") or "hour"
|
||
cfg = LEVELS.get(level)
|
||
# Порог молчания = таймаут убийства воркера (isolated_runner:
|
||
# max(tick_timeout, every*0.8)) + запас. Пока воркер жив, он пишет
|
||
# след (BudgetSpent/Tool*) каждые секунды-минуты; тишина дольше
|
||
# его же таймаута означает: воркер уже убит раннером (TimeoutExpired
|
||
# НЕ пишет TaskFailed — задача и висела в running) или умер с демоном.
|
||
timeout = (cfg.tick_timeout if cfg else 300) or 300
|
||
if cfg and cfg.uses_llm:
|
||
timeout = max(timeout, int(cfg.every * 0.8))
|
||
timeout += 60
|
||
# свежесть следа: трейс сессии (aggregate_id 'hour-497423:96')
|
||
# или BudgetSpent с её trace_id. session 's-hour-497423' →
|
||
# префикс 'hour-497423' безопасен: след. символ ':' или '"'.
|
||
prefix = sess[2:] if sess.startswith("s-") else sess
|
||
db = es._conn()
|
||
try:
|
||
row = db.execute(
|
||
"SELECT MAX(timestamp) AS ts FROM events "
|
||
"WHERE (aggregate='trace' AND aggregate_id LIKE ?) "
|
||
" OR (aggregate='harness' AND event_type='BudgetSpent' AND data LIKE ?)",
|
||
(prefix + "%", "%" + prefix + "%"),
|
||
).fetchone()
|
||
last = row["ts"] if row else None
|
||
finally:
|
||
db.close()
|
||
if last:
|
||
try:
|
||
last_dt = datetime.fromisoformat(last)
|
||
if (now - last_dt) < timedelta(seconds=timeout + 60):
|
||
continue # воркер жив/недавно работал — не трогаем
|
||
except ValueError:
|
||
pass
|
||
es.append_event("task", tid, "TaskPromoted",
|
||
{"to": t.get("size") or "M", "from": t.get("size") or "M",
|
||
"reason": "orphan-requeue: воркер тика-владельца мёртв "
|
||
"(рестарт/таймаут), задача возвращена в очередь"})
|
||
requeued.append(tid)
|
||
_log(f"orphan-requeue: {tid} (сессия {sess}) → ready")
|
||
return requeued
|
||
|
||
|
||
def deliver_pending():
|
||
"""Доставка готовых задач с await_delivery, не получивших push.
|
||
|
||
Push (в run_level_tick) закрывает 99% кейсов немедленно; этот pull —
|
||
страховка: тик умер после TaskCompleted, но до DeliverySent; флаг
|
||
поставлен позже завершения (старый тик) — досылаем по факту.
|
||
Идемпотентно: delivery_status=='sent' не отправляет повторно.
|
||
"""
|
||
from . import commands as _cmd
|
||
model = es.run_projections()
|
||
out = []
|
||
for tid, t in (model.get("tasks") or {}).items():
|
||
if not (t.get("await_delivery") and t.get("status") == "done"
|
||
and t.get("delivery_status") != "sent"):
|
||
continue
|
||
# ответ задачи: TaskCompleted.answer (проекция не хранит — берём из стора)
|
||
db = es._conn()
|
||
try:
|
||
row = db.execute(
|
||
"SELECT data FROM events WHERE aggregate='task' AND aggregate_id=? "
|
||
"AND event_type='TaskCompleted' ORDER BY id DESC LIMIT 1",
|
||
(tid,),
|
||
).fetchone()
|
||
finally:
|
||
db.close()
|
||
if not row:
|
||
continue
|
||
answer = (json.loads(row["data"]).get("answer") or "").strip()
|
||
try:
|
||
if _cmd.deliver_result(tid, answer):
|
||
out.append(tid)
|
||
_log(f"delivery-pull: {tid} доставлен")
|
||
except Exception as e:
|
||
_log(f"delivery-pull: {tid} ошибка {e} — повторим на след. проходе")
|
||
return out
|
||
|
||
|
||
def _delivery_and_orphans_loop():
|
||
"""Фоновый поток демона: доставка + сироты, не мешает тикам уровней."""
|
||
import threading
|
||
_log("delivery-демон запущен (orphan-requeue + автодоставка)")
|
||
# первый проход сразу: чистим сирот рестарта, доим недоставленное
|
||
while True:
|
||
try:
|
||
orphan_requeue()
|
||
deliver_pending()
|
||
except Exception:
|
||
pass # демон не должен умирать: следующий проход попробует снова
|
||
time.sleep(DELIVERY_POLL)
|
||
|
||
|
||
def main_loop():
|
||
"""Демон: уровни тикают независимо, каждый в своём потоке.
|
||
LLM-тик одного уровня больше не блокирует остальные
|
||
(раньше минутный тик замораживал и сек-рефлекс, и час)."""
|
||
import threading
|
||
|
||
es.init_db()
|
||
_log("Simbiot Harness запущен; уровни: " + ", ".join(ACTIVE))
|
||
# delivery-демон: сироты рестартов + автодоставка TG (t-260929205927).
|
||
# Первый проход сразу — чистка сирот до первого тика уровней.
|
||
dt = threading.Thread(target=_delivery_and_orphans_loop, daemon=True,
|
||
name="delivery")
|
||
dt.start()
|
||
last_seq = {}
|
||
threads = {}
|
||
while True:
|
||
for name in ACTIVE:
|
||
cfg = LEVELS[name]
|
||
seq = tick_number(cfg, int(time.time()))
|
||
if last_seq.get(name) == seq:
|
||
continue # этого такта ещё не было — ждём границу
|
||
last_seq[name] = seq
|
||
if cfg.uses_llm and _runners.get(name, {}).get("busy"):
|
||
# уровень ещё работает в предыдущем такте — тик поглощается,
|
||
# но остальные уровни продолжают жить
|
||
continue
|
||
t = threads.get(name)
|
||
if t is not None and t.is_alive():
|
||
continue
|
||
def _run(cfg=cfg, seq=seq):
|
||
model = es.run_projections()
|
||
state = process_level(cfg, model)
|
||
if state not in ("already",):
|
||
_log(f"{cfg.name} #{seq}: {state}")
|
||
t = threading.Thread(target=_run, daemon=True, name=f"tick-{name}")
|
||
threads[name] = t
|
||
t.start()
|
||
time.sleep(POLL)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main_loop() |