diff --git a/README.md b/README.md index 26e9e6f..0d0cb79 100644 --- a/README.md +++ b/README.md @@ -54,6 +54,9 @@ Strands Harness SDK как исполнительное ядро, модели ./venv/bin/python cli.py tg-reply "текст" # ответ на входящее TG # батч-режим (t-260930064924): один сводный ответ закрывает несколько входящих ./venv/bin/python cli.py tg-reply m-1,m-2,m-3 -c "текст" # ReplySent на каждое, одна отправка +# идемпотентные уведомления/файлы (t-260930083937): повтор — SKIP, дублей в чат нет +./venv/bin/python cli.py tg-notify "текст" -k <ключ> +./venv/bin/python cli.py tg-deliver-file <путь> -c "подпись" [--force] ``` ## Автодоставка артефактов (t-260929205927) @@ -66,6 +69,29 @@ Strands Harness SDK как исполнительное ядро, модели Идемпотентно: один раз доставлено — повторно не шлётся; после requeue доработка доставляется снова. Формат: «✅ Готово: » + выжимка ≤1800с. +## Идемпотентность доставки (t-260930083937) + +Инцидент: Alex получал презентацию трижды (m-516173239/241) — гонка +push (тик) vs pull (демон): оба проходили проверку `delivery_status != 'sent'`, +пока TG-отправка (до 35с) была в полёте; плюс ad-hoc curl sendDocument без +метки delivered. Фикс — три атомарных барьера в `deliver_result` / +`deliver_artifact`: + +1. **ledger** (`delivery_ledger`, INSERT OR IGNORE): факт «доставлено раз и + навсегда» по артефакт-идентификатору (задача/файл/ключ сообщения). + Повтор → SKIP + событие `DeliverySkipped` (пропуск логируется); +2. **claim** (`delivery_locks`, атомарный INSERT…WHERE NOT EXISTS): + пометка «доставляю» ДО отправки, TTL 120с > таймаута TG. Параллельный + уровень/демон получает False и пропускает — гонка закрыта на уровне + SQLite (межпроцессно, WAL). Сбой отправки → release, демон досыплет; +3. **запись в ledger** ПОСЛЕ успеха + `DeliverySent`. + +Requeue сбрасывает ledger — доработка доставляется заново. Файлы в чат — +только через `cli.py tg-deliver-file` (тот же барьер); прямые +curl/api.telegram отправки из тиков запрещены (мандаты уровней). +Тесты: `tests/test_delivery_idempotency.py` (49 проверок: гонки, сбои, +requeue, межпроцессный claim, CLI). + ## Эскалации Важное с нижних уровней поднимается наверх событием `EscalationRaised` diff --git a/cli.py b/cli.py index f744b89..9d08c50 100644 --- a/cli.py +++ b/cli.py @@ -57,6 +57,16 @@ def main(): p_lad = sub.add_parser("ladder", help="срез лестницы: уровни, занятость, задачи, эскалации") p_not = sub.add_parser("tg-notify", help="отправить сообщение в чат харнеса (уведомление)") p_not.add_argument("text") + p_not.add_argument("-k", "--key", dest="key", default=None, + help="ключ идемпотентности: сообщение с таким ключом доставляется " + "только один раз (дубль/повтор — SKIP; t-260930083937)") + p_dfl = sub.add_parser("tg-deliver-file", help="отправить файл в чат (идемпотентно, t-260930083937)") + p_dfl.add_argument("file", help="путь к файлу") + p_dfl.add_argument("-c", "--caption", default="", help="подпись к файлу") + p_dfl.add_argument("-k", "--key", dest="key", default=None, + help="ключ идемпотентности (по умолчанию file:<абсолютный путь>)") + p_dfl.add_argument("--force", action="store_true", + help="отправить даже если уже доставлен (повторная доставка вручную)") p_pro = sub.add_parser("proactive", help="инициативная реплика: ритм/тема/отправка (t-260930073419)") p_pro.add_argument("--force", action="store_true", help="отправить сейчас, игнорируя окно/интервал/тишину") @@ -203,8 +213,33 @@ def main(): elif args.cmd == "tg-notify": es.init_db() from harness import tg_inbox - r = tg_inbox.notify_chat(args.text) - print("OK: уведомление отправлено" if r else "SKIP: некому отправить (чат не известен)") + if args.key: + # идемпотентное уведомление (t-260930083937): ключ = артефакт-ид + aid = f"msg:{args.key}" + def _send(): + return bool(tg_inbox.notify_chat(args.text)) + ok = commands.deliver_artifact(aid, kind="message", send=_send, + owner="cli:tg-notify", + caption=args.text) + print("OK: уведомление отправлено" if ok + else "SKIP: уже доставлено (или некому отправить / чужой захват)") + else: + r = tg_inbox.notify_chat(args.text) + print("OK: уведомление отправлено" if r else "SKIP: некому отправить (чат не известен)") + elif args.cmd == "tg-deliver-file": + es.init_db() + from harness import tg_inbox + if not os.path.exists(args.file): + print(f"FAIL: файл не найден: {args.file}") + sys.exit(1) + aid = args.key or f"file:{os.path.abspath(args.file)}" + def _send(): + return tg_inbox.send_document_to_chat(args.file, caption=args.caption) + ok = commands.deliver_artifact(aid, kind="file", send=_send, + owner="cli:tg-deliver-file", + caption=args.caption, force=args.force) + print(f"OK: файл доставлен ({aid})" if ok + else f"SKIP: уже доставлен ({aid}) — используй --force для повтора") elif args.cmd == "proactive": es.init_db() from harness import proactive diff --git a/harness/commands.py b/harness/commands.py index 5009e2e..26cc3e1 100644 --- a/harness/commands.py +++ b/harness/commands.py @@ -108,6 +108,10 @@ def requeue_task(task_id: str, reason: str = "manual requeue"): 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 @@ -157,41 +161,80 @@ def complete_task_with_delivery(task_id: str, answer: str, level: str, es.append_event("task", task_id, "TaskCompleted", { "result_ref": result_ref, "answer": answer[:2000], }) - delivered = deliver_result(task_id, answer) + delivered = deliver_result(task_id, answer, owner=f"complete:{level}") return task_id, delivered -def deliver_result(task_id: str, answer: str) -> bool: +def deliver_result(task_id: str, answer: str, owner: str = "tick") -> bool: """Доставка результата готовой задачи в TG-чат (единственный путь). - Идемпотентно: без await_delivery / уже sent — молча пропускает. + Идемпотентность (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, если отправка выполнена (событие DeliverySent). + Пустой answer доставляется как краткое уведомление о готовности. + Возвращает True, только если отправка выполнена в ЭТОМ вызове. """ 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": + if not t or not t.get("await_delivery"): 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: + # барьер 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 - es.append_event("task", task_id, "DeliverySent", { - "task_id": task_id, "text": text[:2000], "tg_ok": bool(tg_ok), - }) - return True # доставка зафиксирована событием (ok или not ok — повторов нет) + 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: @@ -205,6 +248,57 @@ def _delivery_text(task: dict, answer: str) -> str: 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() diff --git a/harness/es.py b/harness/es.py index a241ed9..94be524 100644 --- a/harness/es.py +++ b/harness/es.py @@ -54,6 +54,9 @@ def init_db(): ); """ ) + # идемпотентная доставка (t-260930083937): ledger «доставлено раз и + # навсегда» + локи захвата отправки; дубли в чат больше невозможны + db.executescript(_DELIVERY_DDL) # идемпотентность тиков: один уровень — один seq db.executescript( """ @@ -67,6 +70,143 @@ def init_db(): db.close() +# ────────────────── ИДЕМПОТЕНТНАЯ ДОСТАВКА (t-260930083937) ────────────────── +# Дубли в чат приходили из гонки push (тик) vs pull (delivery-демон): оба +# проверяли delivery_status из stale-модели, пока TG-отправка (до 35с) была +# в полёте → по два DeliverySent на задачу (разница 23–34с в сторе). +# Фикс — два атомарных примитива поверх SQLite: +# ledger — факт «артефакт доставлен» (INSERT OR IGNORE = только первая); +# claim — захват права отправки на TTL (одно атомарное выражение — без +# BEGIN IMMEDIATE гонок между процессами/потоками не бывает). + +_DELIVERY_DDL = """ +CREATE TABLE IF NOT EXISTS delivery_ledger ( + artifact_id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + ts TEXT NOT NULL +); +CREATE TABLE IF NOT EXISTS delivery_locks ( + artifact_id TEXT PRIMARY KEY, + owner TEXT, + claimed_at REAL NOT NULL, + expires_at REAL NOT NULL +); +""" + +_delivery_tables_ready = False + + +def _ensure_delivery_tables(db): + """Ленивое создание таблиц доставки: живой демон обновляется без + ручной миграции (init_db вызывается не всеми путями).""" + global _delivery_tables_ready + if _delivery_tables_ready: + return + db.executescript(_DELIVERY_DDL) + db.commit() + _delivery_tables_ready = True + + +def delivery_recorded(artifact_id: str) -> bool: + """Артефакт уже доставлен? (метка delivered по артефакт-идентификатору).""" + db = _conn() + try: + _ensure_delivery_tables(db) + row = db.execute("SELECT 1 FROM delivery_ledger WHERE artifact_id=?", + (artifact_id,)).fetchone() + return row is not None + finally: + db.close() + + +def record_delivery(artifact_id: str, kind: str = "task") -> bool: + """Зафиксировать доставку. True — только для ПЕРВОЙ (INSERT OR IGNORE): + повторная пометка не проходит, дубль доставки невозможен даже при + параллельном срабатывании нескольких уровней (мин/час/демон).""" + with _LOCK: + db = _conn() + try: + _ensure_delivery_tables(db) + cur = db.execute( + "INSERT OR IGNORE INTO delivery_ledger (artifact_id, kind, ts) VALUES (?,?,?)", + (artifact_id, kind, datetime.now(timezone.utc).isoformat())) + db.commit() + return cur.rowcount > 0 + finally: + db.close() + + +def forget_delivery(artifact_id: str) -> bool: + """Сброс факта доставки: requeue доработки — новый результат должен + уйти в чат заново (старый ≠ новый).""" + with _LOCK: + db = _conn() + try: + _ensure_delivery_tables(db) + cur = db.execute("DELETE FROM delivery_ledger WHERE artifact_id=?", + (artifact_id,)) + db.execute("DELETE FROM delivery_locks WHERE artifact_id=?", + (artifact_id,)) + db.commit() + return cur.rowcount > 0 + finally: + db.close() + + +def delivery_claim(artifact_id: str, owner: str = "", ttl_sec: int = 120) -> bool: + """Атомарный захват права отправки (пометка «доставляю» ДО отправки). + + Одно выражение INSERT…SELECT…WHERE NOT EXISTS + ON CONFLICT — атомарно + на уровне SQLite, гонки между потоками/процессами исключены: + - лока нет / истёк → захват, True; + - лок жив (кто-то уже отправляет) → False, отправитель пропускает. + TTL > таймаута TG-отправки (35с текст / 90с файл) с запасом: живой + sender не истечёт раньше конца отправки. Успех НЕ освобождает лок — + он истекает сам: ранний release после успешной отправки открыл бы + окно для параллельного дубля (второй уже начал читать модель). + """ + with _LOCK: + db = _conn() + try: + _ensure_delivery_tables(db) + now = time.time() + # WHERE NOT EXISTS отфильтровывает insert при живом локе → + # rowcount=0 (чужой захват); при истёкшем/отсутствующем локе + # INSERT срабатывает, ON CONFLICT обновляет существующую строку. + cur = db.execute( + """ + INSERT INTO delivery_locks (artifact_id, owner, claimed_at, expires_at) + SELECT ?, ?, ?, ? + WHERE NOT EXISTS ( + SELECT 1 FROM delivery_locks + WHERE artifact_id = ? AND expires_at > ? + ) + ON CONFLICT(artifact_id) DO UPDATE SET + owner = excluded.owner, + claimed_at = excluded.claimed_at, + expires_at = excluded.expires_at + """, + (artifact_id, owner, now, now + ttl_sec, artifact_id, now)) + db.commit() + return cur.rowcount > 0 + finally: + db.close() + + +def delivery_release(artifact_id: str): + """Освободить захват при сбое отправки: демон досыплет на след. проходе. + Вызывается ТОЛЬКО при неудаче — после успешной отправки лок истекает сам + (ранний release = окно для дубля).""" + with _LOCK: + db = _conn() + try: + _ensure_delivery_tables(db) + db.execute("DELETE FROM delivery_locks WHERE artifact_id=?", (artifact_id,)) + db.commit() + finally: + db.close() + + def append_event(aggregate: str, aggregate_id: str, event_type: str, data: dict): """Append-only запись события. Возвращает (event_id, version) или None при дубликате.""" with _LOCK: @@ -174,6 +314,11 @@ 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 доработки (from==to): старая доставка не считается — + # новый результат должен уйти в чат заново (t-260930083937). + # Без сброса флаг висел бы 'sent' и резал досылку навсегда. + t.pop("delivery_status", None) + t.pop("delivery_sent_at", None) # requeue сбрасывает факт доставки: работа будет сделана заново, # новый результат должен снова уйти в чат (старый ≠ новый) if t.get("await_delivery"): @@ -200,6 +345,15 @@ def _apply_event(model: dict, ev: dict): "task_id": data.get("task_id"), "ts": ev["timestamp"], "tg_ok": bool(data.get("tg_ok")), } + elif etype == "DeliverySkipped": + # идемпотентная доставка (t-260930083937): артефакт уже доставлен / + # захвачен другим отправителем — повтор НЕ отправлен, пропуск залогирован. + # Сырьё (по одному событию на каждую проверку) наверх не идёт — + # часовой уровень видит сводку skips на задаче. + t = model["tasks"].get(aid) + if t: + t["delivery_skips"] = (t.get("delivery_skips") or 0) + 1 + t["last_skip_reason"] = (data.get("reason") or "")[:200] elif etype == "DigestWritten": key = f"{data['level']}:{data['period']}" model["digests"][key] = { diff --git a/harness/levels.py b/harness/levels.py index dcfcfc1..5416599 100644 --- a/harness/levels.py +++ b/harness/levels.py @@ -65,6 +65,11 @@ LEVELS = { "Если просьба большая — подай её как задачу (cli.py task -a — флаг " "автодоставки: результат сам придёт в чат при готовности, не проси " "Alex спрашивать дважды) и скажи это. " + "Файлы в чат отправляй ТОЛЬКО идемпотентно: cli.py tg-deliver-file " + "<путь> -c \"подпись\" (повторный вызов с тем же файлом — SKIP, дублей " + "в чат нет). НИКОГДА не отправляй в TG напрямую curl'ом/api.telegram " + "(sendDocument/sendMessage из shell) — у сырой отправки нет защиты от " + "повтора, Alex получит файл по три раза. " "Важное, что не по зубам уровню (падение тика, повторная ошибка, " "требуется фикс с большим контекстом) — эскалируй наверх: " "cli.py escalate \"причина\" -c \"контекст\" -s critical|warn|info; " @@ -80,7 +85,11 @@ LEVELS = { "Эскалации от нижних уровней (секция ЭСКАЛАЦИИ в контексте) — твоя " "зона: разберись и закрой cli.py escalation-ack <id> -n \"что сделал\"; " "если фикс больше твоего контекста — переэскалируй " - "cli.py escalate \"причина\" -f hour -s critical."), + "cli.py escalate \"причина\" -f hour -s critical. " + "Файлы/уведомления в чат — только идемпотентно: cli.py tg-deliver-file " + "<путь> -c \"подпись\", cli.py tg-notify \"текст\" -k <ключ> (повтор — SKIP). " + "Прямые curl/api.telegram отправки из shell запрещены — нет защиты " + "от дублей (t-260930083937)."), can_shell=True, ), "day": CycleLevel( @@ -89,7 +98,9 @@ LEVELS = { mandate=("Ты — дневной цикл харнеса Simbiot. Манифест дня: план, выбор задач " "L-класса, дневной отчёт. Думай глубоко, планируй на день вперёд. " "Открытые эскалации (секция ЭСКАЛАЦИИ) — приоритет: это важное, " - "поднятое с нижних уровней; закрой cli.py escalation-ack <id>."), + "поднятое с нижних уровней; закрой cli.py escalation-ack <id>. " + "Файлы в чат — только cli.py tg-deliver-file (идемпотентно, " + "повтор — SKIP); прямые curl/api.telegram отправки запрещены."), can_shell=True, ), } diff --git a/harness/scheduler.py b/harness/scheduler.py index d899e30..d5048c4 100644 --- a/harness/scheduler.py +++ b/harness/scheduler.py @@ -336,7 +336,7 @@ def run_level_tick(level_cfg, seq: int, model: dict): # дослает delivery-демон (deliver_pending) на следующем проходе. try: from . import commands as _cmd - if _cmd.deliver_result(task_id, text or ""): + if _cmd.deliver_result(task_id, text or "", owner=f"tick:{level_cfg.name}"): _log(f"delivery: результат {task_id} доставлен в TG") except Exception as e: # доставка — best effort: не валить тик; DeliverySent допишет @@ -536,6 +536,12 @@ def deliver_pending(): if not (t.get("await_delivery") and t.get("status") == "done" and t.get("delivery_status") != "sent"): continue + # идемпотентная доставка (t-260930083937): ledger — истина поверх + # stale read model (модель могла не догнать события после рестарта). + # Pre-filter без события: DeliverySkipped пишет только deliver_result + # (push-гонка), демон молча пропускает — иначе по событию каждые 15с. + if es.delivery_recorded(tid): + continue # ответ задачи: TaskCompleted.answer (проекция не хранит — берём из стора) db = es._conn() try: @@ -550,7 +556,7 @@ def deliver_pending(): continue answer = (json.loads(row["data"]).get("answer") or "").strip() try: - if _cmd.deliver_result(tid, answer): + if _cmd.deliver_result(tid, answer, owner="daemon"): out.append(tid) _log(f"delivery-pull: {tid} доставлен") except Exception as e: diff --git a/harness/tg_inbox.py b/harness/tg_inbox.py index 157e672..18ecd04 100644 --- a/harness/tg_inbox.py +++ b/harness/tg_inbox.py @@ -11,6 +11,7 @@ import json import os import sys import time +import uuid import urllib.request import urllib.parse import urllib.error @@ -260,6 +261,17 @@ def notify_chat(text: str): Чат берём из последнего входящего MessageReceived (это личка Алекса); если входящих ещё не было — тихо пропускаем (некому показывать). """ + chat_id = _default_chat_id() + if chat_id is None: + return None + try: + return send_message(chat_id, text) + except Exception: + return None # уведомление — best effort, не валить тик + + +def _default_chat_id(): + """chat_id последнего живого чата харнеса (личка Alex) или None.""" _load_env() if not TG_TOKEN: return None @@ -268,11 +280,63 @@ def notify_chat(text: str): if m.get("chat_id") and not m.get("ignored")] if not inbox: return None # некому: чат ещё не известен - chat_id = sorted(inbox, key=lambda m: m.get("ts") or "")[-1]["chat_id"] - try: - return send_message(chat_id, text) - except Exception: - return None # уведомление — best effort, не валить тик + return sorted(inbox, key=lambda m: m.get("ts") or "")[-1]["chat_id"] + + +def _post_multipart(method: str, fields: dict, file_path: str, + file_field: str, timeout: int = 90): + """multipart/form-data POST без requests (sendDocument) — чистый urllib. + + Файл читается целиком: артефакты харнеса (HTML-презентации, отчёты, + картинки) — килобайты/мегабайты, стримить нечего. Таймаут больше + текстового: на файле TG бывает медленнее. + """ + _load_env() + boundary = "----simbiot" + uuid.uuid4().hex + parts = [] + for k, v in fields.items(): + parts.append( + f"--{boundary}\r\nContent-Disposition: form-data; name=\"{k}\"\r\n\r\n{v}\r\n" + .encode()) + with open(file_path, "rb") as f: + content = f.read() + fname = os.path.basename(file_path) + parts.append( + (f"--{boundary}\r\nContent-Disposition: form-data; " + f"name=\"{file_field}\"; filename=\"{fname}\"\r\n" + f"Content-Type: application/octet-stream\r\n\r\n").encode() + + content + b"\r\n") + parts.append(f"--{boundary}--\r\n".encode()) + body = b"".join(parts) + url = f"https://api.telegram.org/bot{TG_TOKEN}/{method}" + req = urllib.request.Request( + url, data=body, + headers={"Content-Type": f"multipart/form-data; boundary={boundary}"}) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.loads(r.read().decode()) + + +def send_document(chat_id, file_path: str, caption: str = "", timeout: int = 90): + """Отправить ФАЙЛ в чат (sendDocument). Исключение наружу — пусть + вызывающий (deliver_artifact) решает, ретраить или валить: у нас + идемпотентная обёртка со своим retry-циклом демона.""" + return _post_multipart( + "sendDocument", + {"chat_id": str(chat_id), "caption": caption[:1000]}, + file_path, "document", timeout=timeout) + + +def send_document_to_chat(file_path: str, caption: str = "") -> bool: + """Файл в дефолтный чат харнеса. False — некому отправить (чат не известен). + + Исключения НЕ глотаем (в отличие от notify_chat): файл — ценный + артефакт, сбой должен быть виден вызывающему, а не превращаться в + молчаливый SKIP. + """ + chat_id = _default_chat_id() + if chat_id is None: + return False + return bool(send_document(chat_id, file_path, caption=caption)) def ladder_snapshot() -> str: diff --git a/tests/test_delivery_idempotency.py b/tests/test_delivery_idempotency.py new file mode 100644 index 0000000..c028635 --- /dev/null +++ b/tests/test_delivery_idempotency.py @@ -0,0 +1,301 @@ +# -*- coding: utf-8 -*- +"""t-260930083937: идемпотентная автодоставка артефактов в TG (hermetic, без LLM/TG). + +Запуск: ./venv/bin/python tests/test_delivery_idempotency.py + +Дубли в чат (m-516173239/241 — «Ты мне её уже три раза отправил») +приходили из гонки уровней: push в тике и delivery-демон оба проходили +проверку delivery_status != 'sent', пока TG-отправка была в полёте +(23–34с между дублями DeliverySent в сторе). Плюс ad-hoc curl sendDocument +вообще без метки delivered — каждый тик слал файл заново. + +Проверяет: ledger (первая запись только одна), атомарный claim +(живой/истёкший лок, release, межпроцессная атомарность), детерминированную +гонку deliver_result (push в полёте + параллельный демон → одна отправка), +SKIP с событием, сбой TG → release → досылка демоном, requeue со сбросом +ledger и флагов модели, pre-filter deliver_pending, deliver_artifact +(dedup/force/сбой+ретрай), CLI tg-notify -k / tg-deliver-file (--force). +""" +import contextlib +import io +import json +import os +import subprocess +import sys +import threading + +# ── изоляция: своя БД в /tmp (prod не трогаем) ── +_DST = "/tmp/simbiot-test-delivery-idem.db" +for p in (_DST, _DST + "-wal", _DST + "-shm"): + if os.path.exists(p): + os.unlink(p) +os.environ["SIMBIOT_DB"] = _DST +sys.path.insert(0, "/root/simbiot-harness") + +from harness import es, tg_inbox +from harness import commands +from harness.scheduler import deliver_pending + +fails = [] + + +def check(name, cond, extra=""): + print(("PASS" if cond else "FAIL"), name, ("| " + str(extra)) if not cond else "") + if not cond: + fails.append(name) + + +def tg_sends(): + """Все события фактической отправки (DeliverySent/ArtifactDelivered) в сторе.""" + db = es._conn() + try: + rows = db.execute( + "SELECT aggregate_id, event_type FROM events " + "WHERE event_type IN ('DeliverySent','ArtifactDelivered')").fetchall() + return [(r["aggregate_id"], r["event_type"]) for r in rows] + finally: + db.close() + + +def count_skips(aggregate_id): + db = es._conn() + try: + return db.execute( + "SELECT COUNT(*) AS n FROM events WHERE aggregate='task' " + "AND aggregate_id=? AND event_type='DeliverySkipped'", + (aggregate_id,)).fetchone()["n"] + finally: + db.close() + + +def run_cli(*argv): + """CLI-команда in-process (stdout перехвачен) — моки tg_inbox действуют.""" + import cli as cli_mod + old = sys.argv + sys.argv = ["cli.py"] + list(argv) + buf = io.StringIO() + try: + with contextlib.redirect_stdout(buf): + cli_mod.main() + finally: + sys.argv = old + return buf.getvalue() + + +# ═══════════════ 1. ledger: доставлено ровно один раз ═══════════════ +es.init_db() +aid = "file:/tmp/pres.html" +check("ledger: первая запись", es.record_delivery(aid, "file") is True) +check("ledger: дубль записи отклонён", es.record_delivery(aid, "file") is False) +check("ledger: recorded видит запись", es.delivery_recorded(aid) is True) +check("ledger: отсутствующий не recorded", + es.delivery_recorded("file:/tmp/none.html") is False) + +# ═══════════════ 2. claim: атомарный захват ДО отправки ═══════════════ +check("claim: первый захват", es.delivery_claim("lock-1", owner="A", ttl_sec=60) is True) +check("claim: живой лок блокирует", es.delivery_claim("lock-1", owner="B", ttl_sec=60) is False) +es.delivery_claim("lock-exp", owner="A", ttl_sec=-1) # уже истёк +check("claim: истёкший лок перехватывается", + es.delivery_claim("lock-exp", owner="B", ttl_sec=60) is True) +es.delivery_release("lock-1") +check("claim: release открывает повторный захват", + es.delivery_claim("lock-1", owner="B", ttl_sec=60) is True) + +# межпроцессная атомарность: два дочерних процесса тянут один claim +code = ("import os,sys;os.environ['SIMBIOT_DB']=%r;" + "sys.path.insert(0,'/root/simbiot-harness');" + "from harness import es;" + "print('1' if es.delivery_claim('lock-xproc', owner='child', ttl_sec=60) else '0')" + % _DST) +r1 = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True) +r2 = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True) +wins = (r1.stdout.strip() + r2.stdout.strip()).replace("\n", "") +check("claim: межпроцессно ровно один победитель", sorted(wins) == ["0", "1"], wins) + +# ═══════════════ 3. deliver_artifact: dedup / force / сбой+ретрай ═══════════════ +calls = [] +ok1 = commands.deliver_artifact("art-1", "file", lambda: (calls.append(1) or True), + owner="test", caption="презентация") +ok2 = commands.deliver_artifact("art-1", "file", lambda: (calls.append(1) or True), + owner="test") +check("artifact: первая доставка True", ok1 is True) +check("artifact: повтор False", ok2 is False) +check("artifact: send вызван один раз", len(calls) == 1, f"calls={len(calls)}") +check("artifact: ledger записан", es.delivery_recorded("art-1") is True) +check("artifact: событие ArtifactDelivered есть", + ("art-1", "ArtifactDelivered") in tg_sends()) + +okf = commands.deliver_artifact("art-1", "file", lambda: (calls.append(1) or True), + owner="test", force=True) +check("artifact: force пробивает живой лок", okf is True) +check("artifact: force — второй send", len(calls) == 2, f"calls={len(calls)}") +check("artifact: force не «молодит» ledger (IGNORE)", + es.record_delivery("art-1", "file") is False) + +flaky = {"n": 0} +def _flaky(): + flaky["n"] += 1 + if flaky["n"] == 1: + raise RuntimeError("TG упал") + return True +try: + commands.deliver_artifact("art-flaky", "file", _flaky, owner="test") + check("artifact: сбой прокинут", False) +except RuntimeError: + check("artifact: сбой прокинут", True) +check("artifact: сбой не записан в ledger", + es.delivery_recorded("art-flaky") is False) +check("artifact: ретрай после сбоя успешен", + commands.deliver_artifact("art-flaky", "file", _flaky, owner="test") is True) + +# ═══════════════ 4. deliver_result: гонка уровней = одна отправка ═══════════════ +# инцидент: push в тике и delivery-демон параллельно → 2× DeliverySent. +# Детерминированная гонка: поток A (тик) берёт claim и «висит в отправке» +# (как TG до 35с), B (демон) идёт в это же время → claim отклоняет. +tid, _ = commands.submit_task("Задача с автодоставкой", "детали", 5) +commands.mark_awaiting_delivery(tid, reason="тест") +es.append_event("task", tid, "TaskClaimed", {"level": "hour", "session_id": "s-test"}) +es.append_event("task", tid, "TaskCompleted", {"answer": "ответ задачи"}) + +_orig_notify = tg_inbox.notify_chat +in_flight = threading.Event() # держим «отправку» потока A +started = threading.Event() # A дошёл до notify + +def _slow_notify(text): + started.set() + in_flight.wait(timeout=10) # имитация полёта TG-отправки + return True + +res_a = {} +def _deliver_a(): + res_a["r"] = commands.deliver_result(tid, "ответ задачи", owner="tick:hour") + +tg_inbox.notify_chat = _slow_notify +try: + ta = threading.Thread(target=_deliver_a) + ta.start() + check("гонка: A дошёл до отправки (claim взят)", started.wait(timeout=5)) + # параллельный уровень/демон в то же время: claim жив → пропуск + res_b = commands.deliver_result(tid, "ответ задачи", owner="daemon") + in_flight.set() # отпускаем A: отправка завершается + ta.join(timeout=10) +finally: + tg_inbox.notify_chat = _orig_notify + +sent = [a for a, t in tg_sends() if a == tid and t == "DeliverySent"] +check("гонка: ровно один DeliverySent", len(sent) == 1, f"sent={len(sent)}") +check("гонка: B (демон) пропущен claim'ом", res_b is False, f"res_b={res_b}") +check("гонка: A доставил (True)", res_a.get("r") is True, f"res_a={res_a}") +check("гонка: ledger записан", es.delivery_recorded(tid) is True) + +# повтор при записанном ledger (stale read model): SKIP с событием +before = count_skips(tid) +check("stale: повтор SKIP", commands.deliver_result(tid, "ответ") is False) +check("stale: SKIP залогирован DeliverySkipped", + count_skips(tid) == before + 1, f"skips={count_skips(tid)}") +model = es.run_projections() +check("stale: проекция delivery_skips видна", + model["tasks"][tid].get("delivery_skips") == 1, + model["tasks"][tid].get("delivery_skips")) + +# ═══════════════ 5. сбой TG: release → демон досылает ═══════════════ +tid2, _ = commands.submit_task("Задача с падением TG", "детали", 5) +commands.mark_awaiting_delivery(tid2, reason="тест") +es.append_event("task", tid2, "TaskClaimed", {"level": "hour", "session_id": "s-test2"}) +es.append_event("task", tid2, "TaskCompleted", {"answer": "ответ 2"}) + +def _dead_notify(text): + raise RuntimeError("TG недоступен") +tg_inbox.notify_chat = _dead_notify +try: + check("сбой TG: deliver_result False", + commands.deliver_result(tid2, "ответ 2", owner="daemon") is False) +finally: + tg_inbox.notify_chat = _orig_notify +check("сбой TG: ledger пуст", es.delivery_recorded(tid2) is False) +db = es._conn() +try: + row = db.execute("SELECT COUNT(*) AS n FROM delivery_locks WHERE artifact_id=?", + (tid2,)).fetchone() + check("сбой TG: лок освобождён (не ждём TTL)", row["n"] == 0, row["n"]) +finally: + db.close() +# TG ожил: демон досылает на следующем проходе +tg_inbox.notify_chat = lambda text: True +try: + delivered = deliver_pending() +finally: + tg_inbox.notify_chat = _orig_notify +check("сбой TG: демон дослал на след. проходе", tid2 in delivered, delivered) +check("сбой TG: DeliverySent появился", + (tid2, "DeliverySent") in tg_sends()) + +# ═══════════════ 6. requeue: сброс ledger и флагов → новая доставка ═══════════════ +commands.requeue_task(tid, "доработка по просьбе Alex") +check("requeue: ledger сброшен", es.delivery_recorded(tid) is False) +model = es.run_projections() +check("requeue: флаг модели сброшен в await (не 'sent' — досылка не порезана)", + model["tasks"][tid].get("delivery_status") == "await", + model["tasks"][tid].get("delivery_status")) +check("requeue: задача снова ready", model["tasks"][tid]["status"] == "ready") +# доработали: новый результат должен уйти (мок) +es.append_event("task", tid, "TaskClaimed", {"level": "hour", "session_id": "s-rq"}) +es.append_event("task", tid, "TaskCompleted", {"answer": "ответ после доработки"}) +tg_inbox.notify_chat = lambda text: True +try: + redone = commands.deliver_result(tid, "ответ после доработки", owner="tick:hour") +finally: + tg_inbox.notify_chat = _orig_notify +check("requeue: новый результат доставлен", redone is True) +check("requeue: после доработки ровно +1 DeliverySent (2 всего)", + len([1 for a, t in tg_sends() if a == tid and t == "DeliverySent"]) == 2) + +# ═══════════════ 7. deliver_pending: pre-filter без дублей ═══════════════ +tg_inbox.notify_chat = lambda text: True +try: + delivered2 = deliver_pending() +finally: + tg_inbox.notify_chat = _orig_notify +check("pending: уже доставленное не досылается", delivered2 == [], f"got={delivered2}") +# 2 задачи: tid дважды (гонка + доработка после requeue) и tid2 раз → 3 +check("pending: DeliverySent не размножены (3 всего: 2×tid + 1×tid2)", + len([1 for a, t in tg_sends() if t == "DeliverySent"]) == 3) + +# ═══════════════ 8. CLI: tg-notify -k / tg-deliver-file ═══════════════ +tg_inbox.notify_chat = lambda text: True +tg_inbox.send_document_to_chat = lambda path, caption="": True +try: + r1 = run_cli("tg-notify", "тестовое уведомление", "-k", "cli-key-1") + r2 = run_cli("tg-notify", "тестовое уведомление", "-k", "cli-key-1") +finally: + tg_inbox.notify_chat = _orig_notify +check("cli: tg-notify -k первая — OK", "OK" in r1, r1) +check("cli: tg-notify -k повтор — SKIP", "SKIP" in r2, r2) +check("cli: ledger по ключу записан", + es.delivery_recorded("msg:cli-key-1") is True) + +open("/tmp/simbiot-test-dl.txt", "w").write("файл-артефакт") +_orig_send_doc = tg_inbox.send_document_to_chat +tg_inbox.notify_chat = lambda text: True +tg_inbox.send_document_to_chat = lambda path, caption="": True +try: + r3 = run_cli("tg-deliver-file", "/tmp/simbiot-test-dl.txt", "-c", "подпись") + r4 = run_cli("tg-deliver-file", "/tmp/simbiot-test-dl.txt", "-c", "подпись") + r5 = run_cli("tg-deliver-file", "/tmp/simbiot-test-dl.txt", "-c", "повтор", + "--force") +finally: + tg_inbox.notify_chat = _orig_notify + tg_inbox.send_document_to_chat = _orig_send_doc +expected_aid = f"file:{os.path.abspath('/tmp/simbiot-test-dl.txt')}" +check("cli: tg-deliver-file первая — OK", "OK" in r3, r3) +check("cli: tg-deliver-file повтор — SKIP", "SKIP" in r4, r4) +check("cli: --force пробивает SKIP", "OK" in r5, r5) +check("cli: ledger по файлу записан", + es.delivery_recorded(expected_aid) is True) + +# ═══════════════ итог ═══════════════ +print() +if fails: + print(f"FAILED: {len(fails)} проверок провалено: {fails}") + sys.exit(1) +print("OK: идемпотентная доставка — все проверки прошли") \ No newline at end of file