1bc33d5678
Инцидент: Alex получал презентацию трижды (m-516173239/241). Две причины: 1) гонка push (тик) vs pull (delivery-демон): оба проходили проверку delivery_status != 'sent', пока TG-отправка (до 35с) была в полёте; 2) ad-hoc curl sendDocument из мин-тиков вообще без метки delivered. Фикс — delivery_ledger (INSERT OR IGNORE: факт доставки раз и навсегда) + delivery_locks (атомарный claim ДО отправки, TTL 120с; одно SQLite- выражение = нет гонок между потоками и процессами). deliver_result / deliver_artifact: ledger → claim → отправка → запись; сбой → release, демон досыплет. Requeue сбрасывает ledger (доработка доставляется заново). Новое: cli.py tg-deliver-file (идемпотентная отправка файлов, --force), tg-notify -k <ключ>, tg_inbox.send_document (multipart на чистом urllib), deliver_pending pre-filter по ledger, мандаты уровней: прямые curl/api.telegram отправки запрещены. DeliverySkipped логирует пропуски. Тесты: tests/test_delivery_idempotency.py — 49 проверок (гонка потоков, межпроцессный claim, сбои+ретраи, requeue, CLI). Регрессия: 6/6 тестов ОК.
335 lines
No EOL
18 KiB
Python
335 lines
No EOL
18 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 не нужен")
|
||
# идемпотентная доставка (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
|
||
|
||
|
||
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, 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) -> 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
|
||
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)
|
||
es.append_event("harness", artifact_id, "ArtifactDelivered", {
|
||
"artifact_id": artifact_id, "kind": kind, "tg_ok": True,
|
||
"caption": (caption or "")[:500], "owner": owner,
|
||
})
|
||
return True
|
||
|
||
|
||
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:],
|
||
} |