simbiot-harness/harness/commands.py

477 lines
No EOL
27 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 — 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)
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 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):
"""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:],
}