7f4db9f085
- события 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
241 lines
No EOL
12 KiB
Python
241 lines
No EOL
12 KiB
Python
"""Simbiot Harness — command side (CQRS): валидация + события.
|
||
|
||
Задачи подаются через CLI/Telegram/файл-инбокс; каждая команда
|
||
валидируется и только потом пишет событие.
|
||
"""
|
||
import os
|
||
import secrets
|
||
from datetime import datetime, timezone
|
||
|
||
from . import es
|
||
from .levels import LEVELS, size_class
|
||
|
||
# лестница сверху вниз: min → hour → day (sec эскалации не шлёт — рефлекс без LLM)
|
||
_ORDER = ["sec", "min", "hour", "day"]
|
||
|
||
|
||
def level_above(level: str):
|
||
"""Уровень над данным или None (day эскалирует некуда — это уже вершина)."""
|
||
if level not in _ORDER:
|
||
return None
|
||
i = _ORDER.index(level)
|
||
return _ORDER[i + 1] if i + 1 < len(_ORDER) else None
|
||
|
||
|
||
def raise_escalation(from_level: str, reason: str, context: str = "", severity: str = "info",
|
||
task_id: str = None, source: str = "cli"):
|
||
"""EscalationRaised: важное с нижнего уровня поднимается наверх.
|
||
|
||
Проекция кладёт эскалацию в model['escalations'], scheduler inject'ит
|
||
открытые в контекст верхних уровней (build_context). Идемпотентно по id.
|
||
"""
|
||
if from_level not in _ORDER:
|
||
raise ValueError(f"неизвестный уровень {from_level}")
|
||
to_level = level_above(from_level)
|
||
if to_level is None:
|
||
raise ValueError(f"у {from_level} нет уровня выше — эскалация некуда")
|
||
if severity not in ("info", "warn", "critical"):
|
||
raise ValueError(f"неизвестный severity {severity} (info|warn|critical)")
|
||
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": from_level, "to_level": to_level,
|
||
"reason": reason, "context": context[:4000], "severity": severity,
|
||
"task_id": task_id, "source": source,
|
||
})
|
||
return esc_id, to_level
|
||
|
||
|
||
def acknowledge_escalation(esc_id: str, level: str = "hour", note: str = ""):
|
||
"""Закрыть эскалацию (верхний уровень принял к сведению/отработал)."""
|
||
model = es.run_projections()
|
||
e = (model.get("escalations") or {}).get(esc_id)
|
||
if e is None:
|
||
raise ValueError(f"эскалация {esc_id} не найдена")
|
||
if e.get("status") == "acked":
|
||
raise ValueError(f"эскалация {esc_id} уже закрыта")
|
||
es.append_event("harness", esc_id, "EscalationAcknowledged", {
|
||
"id": esc_id, "level": level, "note": note[:1000],
|
||
})
|
||
return esc_id
|
||
|
||
|
||
def list_escalations(only_open: bool = False):
|
||
model = es.run_projections()
|
||
out = [e for e in (model.get("escalations") or {}).values()
|
||
if not only_open or e.get("status") == "open"]
|
||
out.sort(key=lambda e: e.get("ts", ""), reverse=True)
|
||
return out
|
||
|
||
|
||
def submit_task(title: str, details: str = "", minutes: int = 5, source: str = "cli"):
|
||
"""SubmitTask: класс по оценке длительности. Идемпотентно по id."""
|
||
task_id = f"t-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}-{secrets.token_hex(3)}"
|
||
size = size_class(minutes)
|
||
es.append_event("task", task_id, "TaskSubmitted", {
|
||
"id": task_id, "title": title, "details": details,
|
||
"size": size, "est_minutes": minutes, "source": source,
|
||
})
|
||
return task_id, size
|
||
|
||
|
||
def promote_task(task_id: str, reason: str = "manual promote"):
|
||
"""Промоушен ready-задачи на уровень выше (TaskPromoted → ready, size выше)."""
|
||
order = ["S", "M", "L"]
|
||
model = es.run_projections()
|
||
t = model["tasks"].get(task_id)
|
||
if t is None:
|
||
raise ValueError(f"задача {task_id} не найдена")
|
||
if t["status"] != "ready":
|
||
raise ValueError(f"задача {task_id} в статусе {t['status']} — promote только из ready")
|
||
i = order.index(t["size"]) if t["size"] in order else 0
|
||
if i >= len(order) - 1:
|
||
raise ValueError(f"задача {task_id} уже размера {t['size']} — выше некуда")
|
||
new_size = order[i + 1]
|
||
es.append_event("task", task_id, "TaskPromoted",
|
||
{"to": new_size, "from": t["size"], "reason": reason})
|
||
return task_id, new_size
|
||
|
||
|
||
def requeue_task(task_id: str, reason: str = "manual requeue"):
|
||
"""Вернуть failed/running/прерванную задачу в очередь (TaskPromoted → ready).
|
||
|
||
running тоже возвращаем: если тик-владелец умер (рестарт демона, таймаут),
|
||
TaskFailed/TaskCompleted не будет — иначе задача зависает в running навсегда.
|
||
"""
|
||
model = es.run_projections()
|
||
t = model["tasks"].get(task_id)
|
||
if t is None:
|
||
raise ValueError(f"задача {task_id} не найдена")
|
||
if t["status"] not in ("failed", "done", "running"):
|
||
raise ValueError(f"задача {task_id} в статусе {t['status']} — requeue не нужен")
|
||
es.append_event("task", task_id, "TaskPromoted",
|
||
{"to": t["size"], "from": t["size"], "reason": reason})
|
||
return task_id
|
||
|
||
|
||
def mark_awaiting_delivery(task_id: str, reason: str = "", reply_to: str = None):
|
||
"""TaskAwaitDelivery: задачу заказали из TG — результат уйдёт в чат сам.
|
||
|
||
Автодоставка артефактов (t-260929205927): когда задача, заказанная из
|
||
TG, завершается (TaskCompleted), результат (answer/результат_ref)
|
||
автоматически отправляется в чат — Alex не должен просить дважды.
|
||
Доставка идемпотентна: после DeliverySent повторная не отправляется.
|
||
Флаг живёт через requeue — доработка после неудачи доставляется тоже.
|
||
"""
|
||
model = es.run_projections()
|
||
t = model["tasks"].get(task_id)
|
||
if t is None:
|
||
raise ValueError(f"задача {task_id} не найдена")
|
||
if t.get("await_delivery"):
|
||
return task_id # уже помечена — идемпотентно
|
||
es.append_event("task", task_id, "TaskAwaitDelivery", {
|
||
"reply_to": reply_to, "reason": (reason or "заказано из TG")[:300],
|
||
})
|
||
return task_id
|
||
|
||
|
||
def complete_task_with_delivery(task_id: str, answer: str, level: str,
|
||
result_ref: str = None):
|
||
"""TaskCompleted + немедленная доставка результата в TG-чат.
|
||
|
||
Используется исполнителем (часовой/дневной тик) как атомарная замена
|
||
голому TaskCompleted для задач с await_delivery: завершение и отправка
|
||
идут парой, «сделал и промолчал» больше не случается. Для задач без
|
||
флага доставки — только TaskCompleted (поведение прежнее).
|
||
|
||
Возвращает (task_id, delivered: bool).
|
||
"""
|
||
model = es.run_projections()
|
||
t = model["tasks"].get(task_id)
|
||
if t is None:
|
||
raise ValueError(f"задача {task_id} не найдена")
|
||
if t["status"] != "running":
|
||
raise ValueError(f"задача {task_id} в статусе {t['status']} — завершать можно только running")
|
||
if (answer or "").strip() == "":
|
||
raise ValueError("пустой результат — доставлять нечего, заверши как failed/requeue")
|
||
# TaskCompleted — независимо от доставки (событие-истина о завершении)
|
||
es.append_event("task", task_id, "TaskCompleted", {
|
||
"result_ref": result_ref, "answer": answer[:2000],
|
||
})
|
||
delivered = deliver_result(task_id, answer)
|
||
return task_id, delivered
|
||
|
||
|
||
def deliver_result(task_id: str, answer: str) -> bool:
|
||
"""Доставка результата готовой задачи в TG-чат (единственный путь).
|
||
|
||
Идемпотентно: без await_delivery / уже sent — молча пропускает.
|
||
Три сценария использования, все через эту функцию:
|
||
1) push в тике сразу после TaskCompleted (scheduler.run_level_tick);
|
||
2) complete_task_with_delivery — то же для исполнителя вручную;
|
||
3) pull-страховка deliver_pending (демон): тик умер между
|
||
завершением и отправкой — досылает по read model.
|
||
Пустой answer доставляется как краткое уведомление о готовности
|
||
(задача завершена, а не отправка «ничего»).
|
||
Возвращает True, если отправка выполнена (событие DeliverySent).
|
||
"""
|
||
model = es.run_projections()
|
||
t = model["tasks"].get(task_id)
|
||
if not t or not t.get("await_delivery") or t.get("delivery_status") == "sent":
|
||
return False
|
||
if t.get("status") != "done":
|
||
return False # доставка только готового результата
|
||
from . import tg_inbox
|
||
text = _delivery_text(t, answer)
|
||
try:
|
||
tg_ok = tg_inbox.notify_chat(text)
|
||
except Exception:
|
||
tg_ok = None # TG недоступен — событие не пишем, pull повторит
|
||
if tg_ok is None:
|
||
return False
|
||
es.append_event("task", task_id, "DeliverySent", {
|
||
"task_id": task_id, "text": text[:2000], "tg_ok": bool(tg_ok),
|
||
})
|
||
return True # доставка зафиксирована событием (ok или not ok — повторов нет)
|
||
|
||
|
||
def _delivery_text(task: dict, answer: str) -> str:
|
||
"""Формат уведомления о готовности: что готово + выжимка результата."""
|
||
title = (task.get("title") or "задача").strip()
|
||
a = (answer or "").strip()
|
||
if not a:
|
||
return f"✅ Готово: «{title[:80]}» (завершено; детали — на дашборде)"
|
||
if len(a) > 1800:
|
||
a = a[:1800] + "…"
|
||
return f"✅ Готово: «{title[:80]}»\n\n{a}"
|
||
|
||
|
||
def set_policy(identity: str = None, cpu_alert_threshold: int = None):
|
||
"""PolicyChanged: идентичность + пороги. Мержим поверх текущей политики."""
|
||
model = es.run_projections()
|
||
current = dict(model.get("policy") or {})
|
||
patch = {}
|
||
if identity is not None:
|
||
patch["identity"] = identity
|
||
if cpu_alert_threshold is not None:
|
||
patch["cpu_alert_threshold"] = cpu_alert_threshold
|
||
current.update(patch)
|
||
es.append_event("harness", "policy", "PolicyChanged", current)
|
||
return current
|
||
|
||
|
||
def list_tasks():
|
||
model = es.run_projections()
|
||
out = []
|
||
for t in model["tasks"].values():
|
||
out.append(t)
|
||
return out
|
||
|
||
|
||
def status():
|
||
model = es.run_projections()
|
||
open_escalations = sum(1 for e in (model.get("escalations") or {}).values()
|
||
if e.get("status") == "open")
|
||
return {
|
||
"cycle_state": model["cycle_state"],
|
||
"tasks_ready": sum(1 for t in model["tasks"].values() if t["status"] == "ready"),
|
||
"tasks_done": sum(1 for t in model["tasks"].values() if t["status"] == "done"),
|
||
"digests": len(model["digests"]),
|
||
"escalations_open": open_escalations,
|
||
"spend": model.get("spend", [])[-8:],
|
||
} |