t-261015000631: атомарный клейм задач (claim_task_if_ready) — гонка двойного claim hour/day закрыта на уровне стора

harness/es.py

@@ -336,6 +336,95 @@ def append_event(aggregate: str, aggregate_id: str, event_type: str, data: dict)
         return (eid, version)
 
 
+def claim_task_if_ready(task_id: str, claim: dict):
+    """Атомарный условный клейм задачи: TaskClaimed ТОЛЬКО если она сейчас ready.
+
+    t-261015000631-13aef0 (гонка hour/day 2026-10-15 00:00): оба уровня читали
+    read model ДО клейма друг друга и каждый по своему снапшоту писал TaskClaimed
+    → L-задача исполнялась дважды (двойной LLM-burn 9router). Read-check-write
+    в scheduler таймингом не закрыть: снапшот модели к моменту записи — вчерашний.
+
+    Атомарность гарантирует СТОР, не тайминг: проверка статуса и INSERT — в одной
+    BEGIN IMMEDIATE транзакции. Статус считаем не из read model (она может
+    отставать), а прямой свёрткой хвоста событий агрегата из events — истина
+    на момент транзакции. Побеждает ровно один писатель; проигравший получает
+    False и молча пропускает задачу (её уже ведёт другой уровень).
+
+    Возвращает (event_id, version) при успехе, False — задача не ready
+    (клеймнута/закрыта/не существует), None — дубль записи (интegrity).
+    """
+    with _LOCK:
+        _ensure_signature_schema()
+        db = _conn()
+        try:
+            db.execute("BEGIN IMMEDIATE")
+        except sqlite3.OperationalError:
+            pass
+        row = db.execute(
+            "SELECT event_type, version FROM events "
+            "WHERE aggregate='task' AND aggregate_id=? ORDER BY version",
+            (task_id,),
+        ).fetchall()
+        # свёртка статуса по потоку событий задачи (та же логика, что
+        # _apply_event, но без полного read model — только этот агрегат)
+        status = None
+        for r in row:
+            et = r["event_type"]
+            if et == "TaskSubmitted":
+                status = "ready"
+            elif et == "TaskClaimed":
+                status = "running"
+            elif et in ("TaskCompleted",):
+                status = "done"
+            elif et == "TaskFailed":
+                status = "failed"
+            elif et == "TaskClosed":
+                if status == "failed":
+                    status = "closed"
+            elif et in ("TaskPromoted", "TaskRequeued"):
+                status = "ready"
+            elif et == "TaskBlocked":
+                status = "blocked"
+        if status != "ready":
+            db.rollback()
+            db.close()
+            return False
+        version = (row[-1]["version"] if row else 0) + 1
+        ts = datetime.now(timezone.utc).isoformat()
+        data_str = json.dumps(claim, ensure_ascii=False)
+        sig_f = {}
+        try:
+            from . import audit as _audit
+            prev = db.execute(
+                "SELECT sig FROM events ORDER BY id DESC LIMIT 1").fetchone()
+            sig_f = _audit.sign_event_row("task", task_id, "TaskClaimed",
+                                          data_str, ts, version,
+                                          prev["sig"] if prev else None)
+            if sig_f.get("key_id"):
+                dom = db.execute(
+                    "SELECT key_id FROM events WHERE key_id IS NOT NULL "
+                    "ORDER BY id DESC LIMIT 1").fetchone()
+                if dom and dom["key_id"] and dom["key_id"] != sig_f["key_id"]:
+                    sig_f = {}
+        except Exception:
+            sig_f = {}
+        try:
+            cur = db.execute(
+                "INSERT INTO events (aggregate,aggregate_id,event_type,data,timestamp,version,"
+                "sig,key_id,prev_sig) VALUES (?,?,?,?,?,?,?,?,?)",
+                ("task", task_id, "TaskClaimed", data_str, ts, version,
+                 sig_f.get("sig"), sig_f.get("key_id"), sig_f.get("prev_sig")),
+            )
+            db.commit()
+            eid = cur.lastrowid
+        except sqlite3.IntegrityError:
+            db.rollback()
+            db.close()
+            return None
+        db.close()
+        return (eid, version)
+
+
 def append_tick(level: str, seq: int):
     """CycleTicked идемпотентно: aggregate_id = 'level:seq' → UNIQUE даёт защиту."""
     return append_event("cycle", f"{level}:{seq}", "CycleTicked", {"level": level, "seq": seq})

harness/scheduler.py

@@ -522,16 +522,33 @@ def run_level_tick(level_cfg, seq: int, model: dict):
     batch = [model["tasks"].get(tid) for tid in batch_ids]
     batch = [t for t in batch if t]
     claimed = []
+    stale = []  # заняты другим уровнем после снапшота модели (гонка клейма)
     for t in batch:
-        es.append_event("task", t["id"], "TaskClaimed", {
+        # t-261015000631-13aef0: атомарный условный клейм НА УРОВНЕ СТОРА
+        # (проверка статуса + INSERT в одной BEGIN IMMEDIATE транзакции).
+        # Гонка 2026-10-15 00:00: hour и day читали модель ДО клейма друг
+        # друга → L-задача t-261014221908 клеймнута дважды (TaskClaimed
+        # id=2767823/2767828) и исполнена дважды = двойной burn 9router.
+        # Снапшот модели здесь — вчерашний; истина только в сторе.
+        res = es.claim_task_if_ready(t["id"], {
             "level": level_cfg.name, "session_id": f"s-{level_cfg.name}-{seq}",
         })
+        if not res:
+            # False — уже не ready (взята/закрыта), None — конфликт записи:
+            # события нет, клейма нет → пропускаем, задача не наша.
+            stale.append(t["id"])
+            continue
         claimed.append(t["id"])
         # TG-visibility: взяли задачу — сообщаем в чат (анти-спам: не чаще 15м на уровень)
         try:
             _notify_busy(level_cfg, t)
         except Exception:
             pass  # уведомление — best effort, не валить тик
+    if stale:
+        _log("claim-race: пропущены задачи, занятые другим уровнем/не ready: "
+             + ", ".join(stale))
+    # в промпт — только задачи, чей клейм реально наш
+    batch = [t for t in batch if t["id"] in set(claimed)]
     if not claimed:
         claimed = None
     prompt = build_context(level_cfg, model, batch)

tests/test_seen_indication.py

@@ -134,6 +134,12 @@ class _FakeES:
     def append_event(self, aggregate, aid, evtype, data={}):
         written.append((evtype, aid))
         return "x"
+
+    # t-261015000631-13aef0: условный клейм в стабе всегда успешен
+    def claim_task_if_ready(self, task_id, claim):
+        written.append(("TaskClaimed", task_id))
+        return ("x", 1)
+
 class _FakeRunner:
     @staticmethod
     def run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None):

tests/test_task_batch.py

@@ -162,6 +162,12 @@ class _FakeES:
         written.append((evtype, aid))
         return "x"
 
+    # t-261015000631-13aef0: клейм теперь условный и атомарный (es.claim_task_if_ready);
+    # стаб без __getattr__-заглушки: всегда успешен, как готовая задача
+    def claim_task_if_ready(self, task_id, claim):
+        written.append(("TaskClaimed", task_id))
+        return ("x", 1)
+
 
 class _FakeRunner:
     @staticmethod

tests/test_task_claim_race.py

@@ -0,0 +1,185 @@
+# -*- coding: utf-8 -*-
+"""t-261015000631-13aef0: атомарный клейм — регресс гонки двойного claim.
+
+Инцидент 2026-10-15 00:00: L-задача t-261014221908-d20347 клеймнута ДВАЖДЫ
+(TaskClaimed id=2767823 hour 00:00:00 и id=2767828 day 00:00:01) — оба уровня
+читали read model до клейма друг друга, а старый цикл писал TaskClaimed по
+снапшоту без перепроверки → двойное исполнение = двойной burn 9router.
+
+Проверяет:
+  1) es.claim_task_if_ready: ready → успех; повторно/не-ready/несуществующая
+     → False; после requeue (TaskPromoted) → снова успех (обратимость);
+  2) атомарность СТОРА: 8 потоков параллельно клеймят одну ready-задачу —
+     ровно один truthy-результат, ровно одно событие TaskClaimed;
+  3) интеграция — точный сценарий прод-инцидента: hour- и day-тик с ОДНИМ
+     и тем же устаревшим снапшотом модели (оба видят задачу ready) →
+     ровно один TaskClaimed и один TaskCompleted, тик-проигравший
+     отрабатывает вхолостую без падения;
+  4) свёртка статуса в сторе согласована с проекциями (done не ре-клеймится).
+
+Запуск: ./venv/bin/python tests/test_task_claim_race.py
+"""
+import os
+import sys
+import threading
+from concurrent.futures import ThreadPoolExecutor
+
+_DB = "/tmp/simbiot-test-claim-race.db"
+for _p in (_DB, _DB + "-wal", _DB + "-shm"):
+    if os.path.exists(_p):
+        os.unlink(_p)
+os.environ["SIMBIOT_DB"] = _DB
+sys.path.insert(0, "/root/simbiot-harness")
+
+from harness import es
+from harness import commands
+
+es.init_db()
+
+fails = []
+
+
+def check(name, cond, extra=""):
+    print(("PASS" if cond else "FAIL"), name, ("" if cond else " | " + str(extra)))
+    if not cond:
+        fails.append(name)
+
+
+def events_of(tid, etype=None):
+    db = es._conn()
+    try:
+        if etype:
+            rows = db.execute(
+                "SELECT * FROM events WHERE aggregate='task' AND aggregate_id=? "
+                "AND event_type=?", (tid, etype)).fetchall()
+        else:
+            rows = db.execute(
+                "SELECT * FROM events WHERE aggregate='task' AND aggregate_id=? "
+                "ORDER BY version", (tid,)).fetchall()
+        return [dict(r) for r in rows]
+    finally:
+        db.close()
+
+
+def status_of(tid):
+    t = (es.run_projections().get("tasks") or {}).get(tid)
+    return (t or {}).get("status")
+
+
+# ── 1. Семантика claim_task_if_ready ──────────────────────────────────────
+tid, _ = commands.submit_task(title="гонка клейма: атомарный claim",
+                              minutes=60, details="фикс t-261015000631")
+check("1.1: ready → клейм успешен", bool(es.claim_task_if_ready(
+    tid, {"level": "hour", "session_id": "s-hour-1"})))
+check("1.2: running → клейм ОТКАЗАН (второй уровень не клеймнет)",
+      es.claim_task_if_ready(tid, {"level": "day", "session_id": "s-day-1"}) is False)
+check("1.3: событий TaskClaimed ровно одно",
+      len(events_of(tid, "TaskClaimed")) == 1, str(len(events_of(tid, "TaskClaimed"))))
+check("1.4: несуществующая задача → False",
+      es.claim_task_if_ready("t-nope-000000", {}) is False)
+
+# закрыли → клейм невозможен; requeue вернул в очередь → снова возможен
+es.append_event("task", tid, "TaskCompleted", {"answer": "done"})
+check("1.5: done → клейм ОТКАЗАН",
+      es.claim_task_if_ready(tid, {"level": "hour", "session_id": "s-hour-2"}) is False)
+commands.requeue_task(tid, reason="тест обратимости")
+check("1.6: после requeue (TaskPromoted → ready) клейм снова успешен",
+      bool(es.claim_task_if_ready(tid, {"level": "min", "session_id": "s-min-1"})))
+check("1.7: свёртка стора == проекции (running)",
+      status_of(tid) == "running", status_of(tid))
+
+# ── 2. Атомарность стора: 8 потоков, одна ready-задача ────────────────────
+tid2, _ = commands.submit_task(title="гонка клейма: 8 потоков",
+                               minutes=60, details="один победитель")
+barrier = threading.Barrier(8)
+
+
+def _claim(_i):
+    barrier.wait()  # максимальная конкуренция на одной ready-задаче
+    return es.claim_task_if_ready(tid2, {"level": "race", "session_id": f"s-{_i}"})
+
+
+with ThreadPoolExecutor(max_workers=8) as ex:
+    results = list(ex.map(_claim, range(8)))
+winners = [r for r in results if r]
+losers = [r for r in results if r is False]
+check("2.1: ровно ОДИН победитель из 8 параллельных", len(winners) == 1, str(results))
+check("2.2: остальные 7 — чистый False (не None-дубль)",
+      len(losers) == 7, str(results))
+check("2.3: TaskClaimed в сторе ровно один",
+      len(events_of(tid2, "TaskClaimed")) == 1,
+      str(len(events_of(tid2, "TaskClaimed"))))
+check("2.4: проекция видит running", status_of(tid2) == "running", status_of(tid2))
+
+# ── 3. Интеграция: прод-сценарий hour+day со stale-снапшотом ───────────────
+from harness import scheduler
+from harness.levels import LEVELS
+import harness.isolated_runner as _ir
+
+tid3, _ = commands.submit_task(title="гонка клейма: hour vs day",
+                               minutes=60, details="сценарий 2026-10-15 00:00")
+# ОДИН снапшот модели для обоих тиков — как в проде: оба уровня читали модель
+# ДО клейма друг друга. Снапшот устаревает сразу после первого клейма —
+# именно это раньше и рождало второй TaskClaimed.
+stale_model = es.run_projections()
+check("3.1: подготовка — в снапшоте задача ready",
+      (stale_model["tasks"].get(tid3) or {}).get("status") == "ready")
+
+_orig_run_llm_tick = _ir.run_llm_tick
+
+
+def _fake_run_llm_tick(level_cfg, prompt, session_id=None, trace_prefix=None, timeout=None):
+    return (f"дайджест {level_cfg.name}: отработана пачка",
+            {"model_calls": 1}, f"trace-{level_cfg.name}")
+
+
+ticks_out = {}
+tick_err = {}
+
+
+def _run_tick(level_name, seq):
+    try:
+        # fresh-снапшот имитировать НЕ будем: оба тика получают ОДИН stale —
+        # худший случай. Внутри тика claim пойдёт в стор атомарно.
+        ticks_out[level_name] = scheduler.run_level_tick(
+            LEVELS[level_name], seq, stale_model)
+    except Exception as e:  # noqa: BLE001
+        tick_err[level_name] = repr(e)
+
+
+try:
+    _ir.run_llm_tick = staticmethod(_fake_run_llm_tick)
+    barrier2 = threading.Barrier(2)
+
+    def _run(level_name, seq):
+        barrier2.wait()
+        _run_tick(level_name, seq)
+
+    th_hour = threading.Thread(target=_run, args=("hour", 4242))
+    th_day = threading.Thread(target=_run, args=("day", 4243))
+    th_hour.start(), th_day.start()
+    th_hour.join(), th_day.join()
+finally:
+    _ir.run_llm_tick = _orig_run_llm_tick
+
+check("3.2: оба тика завершились без исключений", not tick_err, str(tick_err))
+claims3 = events_of(tid3, "TaskClaimed")
+comps3 = events_of(tid3, "TaskCompleted")
+check("3.3: TaskClaimed ровно ОДИН (регресс: было два)",
+      len(claims3) == 1, f"n={len(claims3)}")
+check("3.4: TaskCompleted ровно ОДИН (исполнение не задвоилось)",
+      len(comps3) == 1, f"n={len(comps3)}")
+winner_level = (claims3[0]["data"] if claims3 else "").find('"level"') >= 0
+check("3.5: клейм содержит level победителя", bool(winner_level),
+      claims3[0]["data"][:120] if claims3 else "нет событий")
+loser = [lv for lv in ("hour", "day")
+         if lv != __import__("json").loads(claims3[0]["data"])["level"]][0] \
+    if claims3 else "?"
+check("3.6: проигравший тик отработал вхолостую (пустой батч, без падения)",
+      loser in ticks_out and ticks_out[loser][1] in ([], None),
+      str({k: v[1] for k, v in ticks_out.items()}))
+check("3.7: финальный статус задачи done", status_of(tid3) == "done", status_of(tid3))
+
+print()
+print("FAILS:", len(fails))
+sys.exit(1 if fails else 0)