simbiot-harness/harness/scheduler.py
Simbiot Harness 7f4db9f085 feat(delivery): автодоставка готовых артефактов в TG + orphan-requeue сирот рестартов (t-260929205927-8c20ea)
- события TaskAwaitDelivery / DeliverySent: флаг 'результат доставить в чат'
  на задаче, доставка идемпотентна (один раз sent — повторов нет)
- commands.mark_awaiting_delivery / deliver_result (единственный путь отправки) /
  complete_task_with_delivery: завершение и отправка идут парой —
  «сделал и промолчал» больше не случается
- scheduler: push после TaskCompleted + delivery-демон (pull каждые 15с —
  досылает, если тик умер между завершением и отправкой); requeue сбрасывает
  доставку — доработка снова уйдёт в чат
- orphan_requeue: running-задачи с мёртвым воркером (рестарт демона убивает
  воркеров, TaskCompleted не пишется) возвращаются в очередь автоматически;
  корень эскалации e-260929234911: 'повторный таймаут часового воркера'
  был убийством живого воркера рестартом в 23:12/00:03, не таймаутом
  (трейс и BudgetSpent шли вплоть до секунды рестарта)
- cli: task -a (подать с доставкой) + task-await (пометить существующую);
  мин-уровень помечает задачи-просьбы из TG
- 18 тестов на копии БД: submit→await→complete→deliver→idempotency→
  requeue→re-deliver→pull→orphan→rebuild
2026-09-30 01:31:03 +00:00

474 lines
No EOL
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""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)
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-сообщения, ждущие ответа (в приоритете).
# Голосовое с пустой расшифровкой (тишина/отказ) тоже показываем —
# минутный уровень должен знать, что голосовое пришло, но текста нет,
# и ответить «не расслышал», а не делать вид, что сообщения не было.
inbox = [m for m in (model.get("inbox") or {}).values()
if not m.get("replied") and not m.get("ignored")
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" # тик этого такта уже был (идемпотентность)
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."""
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}"
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()