symbiot/simbiot-harness
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

fix(delivery): идемпотентная автодоставка артефактов в TG — три атомарных барьера против дублей (t-260930083937)

Browse source 
Инцидент: 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 тестов ОК.
This commit is contained in:
Simbiot Harness 2026-09-30 13:41:34 +00:00
parent 9a214d6ac3
commit 1bc33d5678
8 changed files with 720 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
README.md
View file

@ -54,6 +54,9 @@ Strands Harness SDK как исполнительное ядро, модели
./venv/bin/python cli.py tg-reply <msg-id> "текст" # ответ на входящее 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
доработка доставляется снова. Формат: «✅ Готово: <title>» + выжимка ≤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`

35
cli.py
View file

@ -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
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

110
harness/commands.py
View file

@ -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,29 +161,59 @@ 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 # доставка только готового результата
# барьер 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:
@ -187,11 +221,20 @@ def deliver_result(task_id: str, answer: str) -> bool:
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 # доставка зафиксирована событием (ok или not 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()

154
harness/es.py
View file

@ -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] = {

15
harness/levels.py
View file

@ -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,
),
}

10
harness/scheduler.py
View file

@ -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:

74
harness/tg_inbox.py
View file

@ -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:

301
tests/test_delivery_idempotency.py Normal file
View file

@ -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: идемпотентная доставка — все проверки прошли")