From 7f4db9f085bc05d8082e68cd07e046557a084bdf Mon Sep 17 00:00:00 2001 From: Simbiot Harness Date: Wed, 30 Sep 2026 01:31:03 +0000 Subject: [PATCH] =?UTF-8?q?feat(delivery):=20=D0=B0=D0=B2=D1=82=D0=BE?= =?UTF-8?q?=D0=B4=D0=BE=D1=81=D1=82=D0=B0=D0=B2=D0=BA=D0=B0=20=D0=B3=D0=BE?= =?UTF-8?q?=D1=82=D0=BE=D0=B2=D1=8B=D1=85=20=D0=B0=D1=80=D1=82=D0=B5=D1=84?= =?UTF-8?q?=D0=B0=D0=BA=D1=82=D0=BE=D0=B2=20=D0=B2=20TG=20+=20orphan-reque?= =?UTF-8?q?ue=20=D1=81=D0=B8=D1=80=D0=BE=D1=82=20=D1=80=D0=B5=D1=81=D1=82?= =?UTF-8?q?=D0=B0=D1=80=D1=82=D0=BE=D0=B2=20(t-260929205927-8c20ea)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - события 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 --- README.md | 21 +++++-- cli.py | 16 ++++- harness/commands.py | 92 +++++++++++++++++++++++++++++ harness/es.py | 26 ++++++++ harness/levels.py | 4 +- harness/scheduler.py | 138 +++++++++++++++++++++++++++++++++++++++++++ 6 files changed, 291 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 4d7873e..b211b45 100644 --- a/README.md +++ b/README.md @@ -20,9 +20,10 @@ Strands Harness SDK как исполнительное ядро, модели `CycleTicked`, `TickAbsorbed`, `ReflexObserved`, `DigestWritten`, `TaskSubmitted`, `TaskClaimed`, `TaskCompleted`, `TaskFailed`, `TaskPromoted`, -`EscalationRaised`, `EscalationAcknowledged`, -`BudgetSpent`, `CycleError`, `HarnessInit`, `PolicySet` — append-only, -проекции пересобираются replay'ем, стейт = производное от событий. +`TaskAwaitDelivery`, `DeliverySent`, `EscalationRaised`, +`EscalationAcknowledged`, `BudgetSpent`, `CycleError`, `HarnessInit`, +`PolicySet` — append-only, проекции пересобираются replay'ем, стейт = +производное от событий. ## Структура @@ -32,13 +33,15 @@ Strands Harness SDK как исполнительное ядро, модели - `harness/isolated_runner.py` + `_llm_worker.py` — LLM-тик в сабпроцессе (глобальный asyncio-рантайм Strands не переживает 2+ `create_harness`) - `harness/commands.py` — command side -- `cli.py` — init/task/tasks/status/events/requeue +- `cli.py` — init/task/task-await/tasks/status/events/requeue ## CLI ```bash ./venv/bin/python cli.py init --identity "…" ./venv/bin/python cli.py task "…" -m 240 # -m minutes-to-live estimate: S/M/L +./venv/bin/python cli.py task "…" -a # автодоставка результата в TG при готовности +./venv/bin/python cli.py task-await # пометить существующую: доставить в TG ./venv/bin/python cli.py tasks ./venv/bin/python cli.py status ./venv/bin/python cli.py events -n 50 @@ -50,6 +53,16 @@ Strands Harness SDK как исполнительное ядро, модели ./venv/bin/python cli.py tg-reply "текст" # ответ на входящее TG ``` +## Автодоставка артефактов (t-260929205927) + +Задача, заказанная из TG (`cli.py task -a` / `task-await`), помечается +`TaskAwaitDelivery`. При завершении (`TaskCompleted`) результат (answer) +уходит в чат сам — `DeliverySent`, без повторного запроса. Push в тике +(сразу после завершения) + pull-страховка (delivery-демон каждые 15с: +досылает недоставленное, если тик умер между завершением и отправкой). +Идемпотентно: один раз доставлено — повторно не шлётся; после requeue +доработка доставляется снова. Формат: «✅ Готово: » + выжимка ≤1800с. + ## Эскалации Важное с нижних уровней поднимается наверх событием `EscalationRaised` diff --git a/cli.py b/cli.py index 5eecfee..cb95c79 100644 --- a/cli.py +++ b/cli.py @@ -18,6 +18,8 @@ def main(): p_task.add_argument("-d", "--details", default="") p_task.add_argument("-m", "--minutes", type=int, default=5, help="оценка длительности (класс S≤5м M≤60м L≤8ч XL)") + p_task.add_argument("-a", "--await-delivery", action="store_true", + help="результат доставить в TG-чат автоматически при готовности") p_pol = sub.add_parser("policy", help="задать политику") p_pol.add_argument("--identity", default=None) @@ -35,6 +37,10 @@ def main(): p_tgr.add_argument("msg_id", help="id входящего (m-NNN)") p_tgr.add_argument("text") + p_taw = sub.add_parser("task-await", help="пометить задачу: результат доставить в TG при готовности") + p_taw.add_argument("task_id") + p_taw.add_argument("-r", "--reason", default="заказано из TG") + p_tgt = sub.add_parser("tg-transcribe", help="расшифровать голосовое (m-NNN) вручную") p_tgt.add_argument("msg_id", help="id входящего (m-NNN)") p_tgt.add_argument("--force", action="store_true", help="перерасшифровать заново") @@ -78,7 +84,15 @@ def main(): print("OK: стор инициализирован, политика записана") elif args.cmd == "task": tid, size = commands.submit_task(args.title, args.details, args.minutes) - print(f"OK: задача {tid} [{size}] подана") + if args.await_delivery: + commands.mark_awaiting_delivery(tid, reason="подана из TG с автодоставкой") + print(f"OK: задача {tid} [{size}] подана (результат уйдёт в TG при готовности)") + else: + print(f"OK: задача {tid} [{size}] подана") + elif args.cmd == "task-await": + es.init_db() + commands.mark_awaiting_delivery(args.task_id, reason=args.reason) + print(f"OK: {args.task_id} помечена — результат будет доставлен в TG при готовности") elif args.cmd == "policy": pol = commands.set_policy(identity=args.identity, cpu_alert_threshold=args.cpu_threshold) print("OK:", json.dumps(pol, ensure_ascii=False)) diff --git a/harness/commands.py b/harness/commands.py index d05e028..5009e2e 100644 --- a/harness/commands.py +++ b/harness/commands.py @@ -113,6 +113,98 @@ def requeue_task(task_id: str, reason: str = "manual requeue"): 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() diff --git a/harness/es.py b/harness/es.py index f951152..ade2718 100644 --- a/harness/es.py +++ b/harness/es.py @@ -174,6 +174,32 @@ def _apply_event(model: dict, ev: dict): t.update(size=data["to"]) t.update(status="ready", level=None, session_id=None, promoted_to=data.get("to"), promotions=(t.get("promotions") or 0) + 1) + # requeue сбрасывает факт доставки: работа будет сделана заново, + # новый результат должен снова уйти в чат (старый ≠ новый) + if t.get("await_delivery"): + t.update(delivery_status="await") + elif etype == "TaskAwaitDelivery": + # автодоставка (t-260929205927): задачу заказали из TG — результат + # должен уйти в чат сам, без повторного запроса. Флаг на задаче; + # scheduler-демон ловит TaskCompleted и рассылает DeliverySent. + t = model["tasks"].get(aid) + if t: + t.update(await_delivery=True, delivery_status="await", + delivery_reply_to=data.get("reply_to"), + delivery_reason=(data.get("reason") or "")[:300]) + elif etype == "DeliverySent": + # доставлено в чат (результат готовой задачи); best-effort флаг: + # доставка идемпотентна — повторная не отправит (sent раз и навсегда) + t = model["tasks"].get(aid) + if t: + t.update(delivery_status="sent", delivery_sent_at=ev["timestamp"], + delivered_text=(data.get("text") or "")[:2000], + delivery_tg_ok=bool(data.get("tg_ok"))) + else: + model.setdefault("deliveries", {})[aid] = { + "task_id": data.get("task_id"), "ts": ev["timestamp"], + "tg_ok": bool(data.get("tg_ok")), + } elif etype == "DigestWritten": key = f"{data['level']}:{data['period']}" model["digests"][key] = { diff --git a/harness/levels.py b/harness/levels.py index e36337c..85723cb 100644 --- a/harness/levels.py +++ b/harness/levels.py @@ -53,7 +53,9 @@ LEVELS = { "isolated_runner.py) и не разбирай падения тиков — это зона верхних уровней. " "Отвечай из контекста тика, максимум пара шагов, иначе таймаут тика. " - "Если просьба большая — подай её как задачу (cli.py task) и скажи это. " + "Если просьба большая — подай её как задачу (cli.py task -a — флаг " + "автодоставки: результат сам придёт в чат при готовности, не проси " + "Alex спрашивать дважды) и скажи это. " "Важное, что не по зубам уровню (падение тика, повторная ошибка, " "требуется фикс с большим контекстом) — эскалируй наверх: " "cli.py escalate \"причина\" -c \"контекст\" -s critical|warn|info; " diff --git a/harness/scheduler.py b/harness/scheduler.py index 3c193e9..89b13fc 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -229,6 +229,19 @@ def run_level_tick(level_cfg, seq: int, model: dict): 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 @@ -297,6 +310,126 @@ def _reflex_tick(level_cfg, seq: int, model: dict): 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-тик одного уровня больше не блокирует остальные @@ -305,6 +438,11 @@ def main_loop(): 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: