4e57de63bd
- scheduler: _events_digest() — содержательные события стора за ~90 мин (Message/Reply/Tasks/Escalations/AUM/MCP), шум (CycleTicked/Tool*/) отсечён; секция «СОБЫТИЯ ХАРНЕСА» в build_context по policy.events_context=on - commands/cli: policy --events-context on|off (переключатель, одобрен Alex) - tests/test_events_context.py: 9 hermetic-проверок (on/off/дефолт/пустой стор)
489 lines
No EOL
28 KiB
Python
489 lines
No EOL
28 KiB
Python
"""Simbiot Harness — command side (CQRS): валидация + события.
|
||
|
||
Задачи подаются через CLI/Telegram/файл-инбокс; каждая команда
|
||
валидируется и только потом пишет событие.
|
||
"""
|
||
import os
|
||
import re
|
||
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
|
||
|
||
|
||
# мусорные заголовки: LLM-тик нижнего уровня звал `cli.py task list` / `cli.py task <id>`,
|
||
# а argparse принял субкомандо-подобное слово / id как title → мусорные задачи 'list'
|
||
# (e-261001082241, e-261003094628). Ловим на входе command side.
|
||
_SUSPECT_TITLES = {"list", "tasks", "status", "events", "help", "complete", "promote", "requeue"}
|
||
_TASK_ID_RE = re.compile(r"^t-\d{12}-[0-9a-f]{6}$")
|
||
|
||
|
||
def _validate_task_title(title: str, details: str):
|
||
"""Guard: заголовок-субкоманда CLI или чужой id задачи без деталей — это просмотр, не подача."""
|
||
if not title or not title.strip():
|
||
raise ValueError("пустой заголовок задачи")
|
||
t = title.strip()
|
||
if not details.strip() and (t.lower() in _SUSPECT_TITLES or _TASK_ID_RE.match(t)):
|
||
raise ValueError(
|
||
f"подозрительный заголовок {t!r} без деталей: похоже на субкоманду CLI или id задачи "
|
||
f"(cli.py task list / cli.py task <id> — это просмотр, не создание). "
|
||
"Подача: cli.py task \"заголовок\" -d \"детали\"; доска: cli.py tasks"
|
||
)
|
||
|
||
|
||
def submit_task(title: str, details: str = "", minutes: int = 5, source: str = "cli"):
|
||
"""SubmitTask: класс по оценке длительности. Идемпотентно по id."""
|
||
_validate_task_title(title, details)
|
||
task_id = f"t-{datetime.now(timezone.utc).strftime('%y%m%d%H%M%S')}-{secrets.token_hex(3)}"
|
||
size = size_class(minutes)
|
||
data = {"id": task_id, "title": title, "details": details,
|
||
"size": size, "est_minutes": minutes, "source": source}
|
||
# круг доверия (harness/trust.py): задача из LLM-тика наследует круг тика —
|
||
# просьба правления потом не исполняется часом с правами владельца
|
||
tick_circle = os.environ.get("SIMBIOT_TICK_CIRCLE", "").strip()
|
||
if tick_circle.isdigit():
|
||
data["circle"] = int(tick_circle)
|
||
es.append_event("task", task_id, "TaskSubmitted", data)
|
||
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 close_task(task_id: str, reason: str = "manual close") -> str:
|
||
"""Закрыть failed-задачу без работы: TaskClosed → статус closed.
|
||
|
||
Авто-схлопывание старых фейлов (просьба Alex 03.10, e-261003094432):
|
||
дубли и прах не должны висеть красной группой на доске. closed —
|
||
терминальный, НО обратимый: requeue принимает closed → ready.
|
||
Идемпотентно: уже closed/done — просто возвращаем задачу (no-op).
|
||
"""
|
||
model = es.run_projections()
|
||
t = model["tasks"].get(task_id)
|
||
if t is None:
|
||
raise ValueError(f"задача {task_id} не найдена")
|
||
if t["status"] in ("closed", "done"):
|
||
return task_id # уже закрыта — no-op (идемпотентность)
|
||
if t["status"] != "failed":
|
||
raise ValueError(
|
||
f"задача {task_id} в статусе {t['status']} — close только для failed; "
|
||
f"для живой работы используй requeue, для завершения — complete")
|
||
es.append_event("task", task_id, "TaskClosed", {"by": "manual", "reason": reason[:300]})
|
||
return task_id
|
||
|
||
|
||
def archive_stale_failed(max_age_hours: float = 24.0, now=None) -> list:
|
||
"""Демон-архиватор (e-261003094432): failed старше max_age_hours → closed.
|
||
|
||
Зовёт scheduler-демон каждые DELIVERY_POLL: красная группа фейлов
|
||
не растёт — старше суток уходят в «архив» (статус closed), где их
|
||
видно при желании, но они не забивают доску. requeue возвращает
|
||
закрытую в очередь, если она всё-таки нужна.
|
||
now — для тестов (симуляция «прошло 24ч» без подделки событий).
|
||
"""
|
||
from datetime import datetime, timezone
|
||
model = es.run_projections()
|
||
now = now or datetime.now(timezone.utc)
|
||
closed = []
|
||
for tid, t in (model.get("tasks") or {}).items():
|
||
if t.get("status") != "failed":
|
||
continue
|
||
fin = t.get("finished") or ""
|
||
try:
|
||
age = (now - datetime.fromisoformat(fin)) if fin else None
|
||
except ValueError:
|
||
age = None
|
||
if age is None or age.total_seconds() < max_age_hours * 3600:
|
||
continue
|
||
es.append_event("task", tid, "TaskClosed", {
|
||
"by": "archive", "reason": f"auto: failed > {max_age_hours:g}ч без работы"})
|
||
closed.append(tid)
|
||
return closed
|
||
|
||
|
||
def block_task(task_id: str, reason: str = "ожидает уточнений") -> str:
|
||
"""Заблокировать живую задачу: TaskBlocked → статус blocked (t-261004122019).
|
||
|
||
Задача есть, но не готова к работе — ждёт уточнений от Alex или внешнего
|
||
события. scheduler берёт только ready → blocked не попадает в батчи.
|
||
Обратно: requeue (blocked → ready). Идемпотентно: уже blocked — no-op.
|
||
"""
|
||
model = es.run_projections()
|
||
t = model["tasks"].get(task_id)
|
||
if t is None:
|
||
raise ValueError(f"задача {task_id} не найдена")
|
||
if t["status"] == "blocked":
|
||
return task_id # уже заблокирована — no-op
|
||
if t["status"] in ("done", "closed"):
|
||
raise ValueError(
|
||
f"задача {task_id} в статусе {t['status']} — blocked только для живых; "
|
||
f"оживить можно requeue")
|
||
es.append_event("task", task_id, "TaskBlocked",
|
||
{"by": "manual", "reason": reason[:300]})
|
||
return task_id
|
||
|
||
|
||
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} не найдена")
|
||
# closed тоже возвращаем (авто-схлопывание старых фейлов, e-261003094432):
|
||
# закрытая в архив задача может понадобиться снова — осознанная доработка.
|
||
if t["status"] not in ("failed", "closed", "done", "running", "blocked"):
|
||
raise ValueError(f"задача {task_id} в статусе {t['status']} — requeue не нужен")
|
||
# идемпотентная доставка (t-260930083937): requeue = новая работа над
|
||
# задачей, её результат должен уйти в чат заново → сбрасываем факт
|
||
# доставки и захват. Старая доставка остаётся в истории событий.
|
||
es.forget_delivery(task_id)
|
||
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
|
||
|
||
|
||
import re as _re
|
||
|
||
# Guard (e-261001074527-f0c1b4): ночной тик 20727 пачкой закрыл 5 L-задач
|
||
# шаблонной отпиской воркера «бюджет исчерпан» — это стоп-фраза тика, а не
|
||
# результат. Такие «результаты» закрывать задачу не могут.
|
||
_BUDGET_STUB_RE = _re.compile(r"бюджет\s+(?:шагов\s+)?исчерпан", _re.IGNORECASE)
|
||
_BUDGET_STUB_MARKERS_RE = _re.compile(r"тик\s+завершён|tool-вызов|шагов", _re.IGNORECASE)
|
||
|
||
|
||
def looks_like_budget_stub(answer: str) -> bool:
|
||
"""True, если answer — шаблонная отписка воркера о бюджете шагов.
|
||
|
||
Требование маркера («тик завершён», «tool-вызов», «шагов») отсекает
|
||
ложные срабатывания: честный результат работы, где по делу упомянут
|
||
исчерпанный бюджет («проверил, бюджет месяца исчерпан»), под гвард
|
||
не попадает — а в стоп-фразе воркера маркер всегда есть.
|
||
"""
|
||
s = answer or ""
|
||
return bool(_BUDGET_STUB_RE.search(s) and _BUDGET_STUB_MARKERS_RE.search(s))
|
||
|
||
|
||
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")
|
||
if looks_like_budget_stub(answer):
|
||
requeue_task(task_id, reason="guard: 'бюджет исчерпан' — не результат, задача в ready (инцидент e-261001074527)")
|
||
raise ValueError(
|
||
f"задача {task_id} ПЕРЕВОДЖЕНА в ready: отпиской «бюджет исчерпан» "
|
||
f"задачу закрывать нельзя. Либо доделай и завершай с реальным "
|
||
f"результатом, либо явно requeue/failed.")
|
||
# TaskCompleted — независимо от доставки (событие-истина о завершении)
|
||
es.append_event("task", task_id, "TaskCompleted", {
|
||
"result_ref": result_ref, "answer": answer[:2000],
|
||
})
|
||
delivered = deliver_result(task_id, answer, owner=f"complete:{level}")
|
||
return task_id, delivered
|
||
|
||
|
||
def deliver_result(task_id: str, answer: str, owner: str = "tick") -> bool:
|
||
"""Доставка результата готовой задачи в TG-чат (единственный путь).
|
||
|
||
Идемпотентность (t-260930083937) — ТРИ барьера, гонка уровней закрыта:
|
||
1) delivery_recorded(task_id) — артефакт уже доставлен → SKIP (событие
|
||
DeliverySkipped с причиной, без отправки);
|
||
2) delivery_claim(task_id) — атомарный захват ДО отправки: второй
|
||
отправитель (параллельный тик/демон) получает False и пропускает;
|
||
3) record_delivery(task_id) — ПОСЛЕ успешной отправки; INSERT OR IGNORE
|
||
гарантирует, что даже два параллельных успеха зафиксируются один раз,
|
||
а второй sender увидит recorded и не отправит повторно.
|
||
|
||
Порядок барьеров: (ledger) → (claim) → отправка → (ledger-write) →
|
||
событие DeliverySent. Между claim и record_delivery дубль невозможен;
|
||
сбой отправки → release лока, демон досыплет на след. проходе.
|
||
|
||
Три сценария использования, все через эту функцию:
|
||
1) push в тике сразу после TaskCompleted (scheduler.run_level_tick);
|
||
2) complete_task_with_delivery — то же для исполнителя вручную;
|
||
3) pull-страховка deliver_pending (демон): тик умер между
|
||
завершением и отправкой — досылает по read model.
|
||
Пустой answer доставляется как краткое уведомление о готовности.
|
||
Возвращает True, только если отправка выполнена в ЭТОМ вызове.
|
||
"""
|
||
model = es.run_projections()
|
||
t = model["tasks"].get(task_id)
|
||
if not t or not t.get("await_delivery"):
|
||
return False
|
||
if t.get("status") != "done":
|
||
return False # доставка только готового результата
|
||
# барьер 1: уже доставлено (ledger — истина поверх stale read model:
|
||
# модель могла не догнать события, а ledger атомарен). Пропуск
|
||
# логируем событием — «уже доставлено» видно в истории задачи.
|
||
if es.delivery_recorded(task_id):
|
||
es.append_event("task", task_id, "DeliverySkipped", {
|
||
"task_id": task_id, "reason": "already-delivered",
|
||
})
|
||
return False
|
||
if t.get("delivery_status") == "sent":
|
||
# модель говорит «отправлено», но ledger пуст — доставка из ДО
|
||
# фикса (легаси-задачи): молча пропускаем, без события (демон с
|
||
# pre-filter'ом сюда не попадает, повторов нет)
|
||
return False
|
||
# барьер 2: атомарный захват права отправки ДО отправки —
|
||
# параллельный отправитель (мин/час/демон) здесь не пройдёт
|
||
if not es.delivery_claim(task_id, owner=f"{owner}:{os.getpid()}"):
|
||
# кто-то уже отправляет: не мешаем и не считаем дублем — просто ждём
|
||
return False
|
||
try:
|
||
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:
|
||
# TG недоступен: освобождаем захват — демон досыплет на след.
|
||
# проходе (15с), а не будет ждать истечения TTL лока
|
||
es.delivery_release(task_id)
|
||
return False
|
||
# барьер 3: фиксируем факт доставки (первый раз — True) и событие
|
||
es.record_delivery(task_id, kind="task")
|
||
es.append_event("task", task_id, "DeliverySent", {
|
||
"task_id": task_id, "text": text[:2000], "tg_ok": bool(tg_ok),
|
||
})
|
||
return True
|
||
except Exception:
|
||
# сбой отправки: освобождаем захват — демон досыплет на след. проходе
|
||
es.delivery_release(task_id)
|
||
raise
|
||
|
||
|
||
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 deliver_artifact(artifact_id: str, kind: str, send, owner: str = "manual",
|
||
caption: str = "", force: bool = False, msg: str = None) -> bool:
|
||
"""Идемпотентная доставка ПРОИЗВОЛЬНОГО артефакта в чат (t-260930083937).
|
||
|
||
Рут-кейс дублей: мин-тик шёл файлом «скинь сюда» curl'ом sendDocument —
|
||
без метки delivered каждый следующий тик видел просьбу в истории чата
|
||
и слал заново (Alex получил презентацию трижды). Здесь тот же барьер,
|
||
что и у задач: ledger по artifact_id → атомарный claim → отправка →
|
||
запись. Параллельные тики (мин/час) и повторные попытки не дублируют.
|
||
|
||
artifact_id — стабильный идентификатор вещи (ключ dedup): путь файла,
|
||
«task:<id>», «file:<path>», «page:<url>» и т.п. force=True — повторная
|
||
доставка (игнорирует ledger, но всё равно атомарна: параллель не
|
||
отправит дважды). send — callable без аргументов, возвращает True
|
||
при успехе (уже знает получателя/текст — tg_inbox.* или мок в тестах).
|
||
Возвращает True, только если отправка выполнена в ЭТОМ вызове.
|
||
"""
|
||
if not force and es.delivery_recorded(artifact_id):
|
||
es.append_event("harness", artifact_id, "DeliverySkipped", {
|
||
"artifact_id": artifact_id, "kind": kind,
|
||
"reason": "already-delivered",
|
||
})
|
||
return False
|
||
# Анти-гонка уровней (t-261009160906): артефакт привязан к входящему
|
||
# m-NNN, на которое уже отвечено ДРУГИМ артефактом → SKIP. Два уровня
|
||
# (min/hour) на один запрос не должны слать два разных файла; dedup по
|
||
# artifact_id=file:<путь> их не различает (разные пути = разные ключи).
|
||
if msg and not force:
|
||
model = es.run_projections()
|
||
m = (model.get("inbox") or {}).get(msg)
|
||
if (m is not None and m.get("replied")
|
||
and m.get("reply_artifact") not in (None, "", artifact_id)):
|
||
es.append_event("harness", artifact_id, "DeliverySkipped", {
|
||
"artifact_id": artifact_id, "kind": kind,
|
||
"reason": "msg-already-answered-by-other-artifact",
|
||
"msg": msg, "reply_artifact": m.get("reply_artifact"),
|
||
})
|
||
return False
|
||
if force:
|
||
# ручной повтор — явное решение человека: лок недавней доставки
|
||
# не должен блокировать его на TTL (120с). Снимаем свой же захват
|
||
# и берём заново. Параллельная гонка force vs force всё равно
|
||
# атомарна: один claim заберёт право, второй увидит False.
|
||
es.delivery_release(artifact_id)
|
||
if not es.delivery_claim(artifact_id, owner=f"{owner}:{os.getpid()}"):
|
||
return False # параллельный отправитель уже в полёте
|
||
try:
|
||
ok = bool(send())
|
||
except Exception:
|
||
es.delivery_release(artifact_id)
|
||
raise
|
||
if not ok:
|
||
# отправка не удалась: освобождаем захват под ретрай
|
||
es.delivery_release(artifact_id)
|
||
return False
|
||
# force не перезатирает чужую запись: INSERT OR IGNORE — первая доставка
|
||
# в ledger остаётся фактом; повторный force просто отправит ещё раз,
|
||
# но ledger при этом не «молодеет» (kind/ts не переписываются)
|
||
es.record_delivery(artifact_id, kind=kind)
|
||
ev = {
|
||
"artifact_id": artifact_id, "kind": kind, "tg_ok": True,
|
||
"caption": (caption or "")[:500], "owner": owner,
|
||
}
|
||
if msg:
|
||
ev["msg"] = msg # анти-гонка уровней (t-261009160906)
|
||
es.append_event("harness", artifact_id, "ArtifactDelivered", ev)
|
||
return True
|
||
|
||
|
||
def set_policy(identity: str = None, cpu_alert_threshold: int = None,
|
||
events_context: str = None):
|
||
"""PolicyChanged: идентичность + пороги. Мержим поверх текущей политики.
|
||
|
||
events_context (t-261017110430): 'on' — в контекст тика добавляется секция
|
||
свежих содержательных событий EventStore; 'off' — секции нет. Это и есть
|
||
«переключатель подачи событий EventStore в контекст», одобренный Alex.
|
||
"""
|
||
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
|
||
if events_context is not None:
|
||
patch["events_context"] = events_context
|
||
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:],
|
||
} |