fix(queue): зависший определитель пропускает остаток своего пакета, очередь идёт дальше (R-13, I-4)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014Typ5sXWNMFirr2va2uL8o

src/jira_hub/queue.py

@@ -108,10 +108,21 @@ def _definer_status(conn, batch_id, temp_key):
     return r["status"] if r else "failed"
 
 
+def _definer_stale(conn, batch_id, temp_key) -> bool:
+    """Определитель взят агентом и давно не отвечает (старше STALE_MINUTES)."""
+    cutoff = (db.utcnow() - timedelta(minutes=STALE_MINUTES)).isoformat()
+    return conn.execute("SELECT 1 FROM command WHERE batch_id = ? AND type = 'ensure_issue' AND status = 'claimed' "
+                        "AND claimed_at <= ? AND json_extract(params, '$.temp_key') = ?",
+                        (batch_id, cutoff, temp_key)).fetchone() is not None
+
+
 def _next(conn) -> dict | None:
     rows = conn.execute("SELECT c.* FROM command c JOIN batch b ON b.id = c.batch_id "
                         "WHERE c.status = 'queued' AND b.cancelled_at IS NULL ORDER BY b.created_at, b.id, c.seq").fetchall()
+    blocked_batches = set()           # пакеты, ждущие зависшего определителя (R-13)
     for row in rows:
+        if row["batch_id"] in blocked_batches:
+            continue
         cmd = _cmd(row)
         needed = C.temp_keys_used(cmd)
         if needed:
@@ -119,7 +130,10 @@ def _next(conn) -> dict | None:
             unresolved = [k for k in needed if k not in mapping]
             pending = [k for k in unresolved if _definer_status(conn, cmd["batch_id"], k) not in TERMINAL]
             if pending:
-                return None                                    # создание ещё не завершено — ждём
+                if all(_definer_stale(conn, cmd["batch_id"], k) for k in pending):
+                    blocked_batches.add(cmd["batch_id"])       # определитель завис: остаток пакета ждёт человека,
+                    continue                                   # очередь идёт к следующему пакету
+                return None                                    # создание ещё не завершено — ждём (одна в полёте)
             cmd["params"] = C.substitute(cmd["type"], cmd["params"], mapping)
             if cmd["type"] == "refresh":
                 cmd["params"]["keys"] = [k for k in cmd["params"].get("keys", []) if not C.is_temp(k)]

tests/test_queue.py

@@ -265,3 +265,33 @@ def test_reconcile_does_not_overwrite_finished_command(monkeypatch):
     assert queue.reconcile(conn) == []
     assert queue.get_command(conn, c["id"])["status"] == "cancelled"
     assert mirror.get_note(conn, "new:1") is not None          # rename_note не выполнялся
+
+
+def _two_batches_with_dependent(conn):
+    a = queue.create_batch(conn, "A", "claude", [
+        {"type": "ensure_issue", "params": {"temp_key": "new:1", "issue_type": "task", "summary": "Тема"}},
+        {"type": "ensure_link", "params": {"source": "new:1", "target": "A-1", "link_type": "связанные"}}])
+    first = queue.claim(conn)
+    assert first["type"] == "ensure_issue"
+    b = queue.create_batch(conn, "B", "claude", [{"type": "refresh", "params": {"jql": "project = X"}}])
+    return a, b, first
+
+
+def test_stale_definer_does_not_block_other_batches():
+    """R-13: зависший ensure_issue с зависимыми не держит очередь целиком."""
+    conn = fresh()
+    a, b, first = _two_batches_with_dependent(conn)
+    conn.execute("UPDATE command SET claimed_at = '2020-01-01T00:00:00+00:00' WHERE id = ?", (first["id"],))
+    got = queue.claim(conn)
+    assert got is not None and got["batch_id"] == b["id"] and got["type"] == "refresh"
+    # остаток пакета A остаётся в очереди, не выдаётся и не пропускается
+    assert [c["status"] for c in queue.batch_view(conn, a["id"])["commands"]][1:] == ["queued", "queued"]
+
+
+def test_fresh_definer_still_blocks_queue():
+    conn = fresh()
+    a, b, first = _two_batches_with_dependent(conn)
+    assert queue.claim(conn) is None                           # одна в полёте
+    conn.execute("UPDATE command SET status = 'done', result = '{\"key\": \"A-9\"}' WHERE id = ?", (first["id"],))
+    nxt = queue.claim(conn)
+    assert nxt["batch_id"] == a["id"] and nxt["params"]["source"] == "A-9"