Queue-Steuerung: Stopp/Neuer-Versuch/Prio + Haenger-Warnung (UI+Telegram)

Karten lassen sich gezielt anhalten (generischer Block + Prozessbaum-Kill),
frisch starten (Retry beendet Worker samt Kindern - uvicorn-Falle) und per
Prio-Pfeilen umsortieren (Dispatcher zieht priority DESC nativ). Laufende
Karten ohne Heartbeat >10 min bekommen ein Warn-Badge; der Ketten-Digest
alarmiert nach 15 min per Telegram, inkl. Kontext-Verdichtungs-Hinweis.
Ausloeser: Scaffold-Worker hing 45 min an selbst gestartetem uvicorn.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

backend/services/ideen.py

@@ -126,6 +126,9 @@ def _ketten_anreichern(items: list[dict]) -> list[dict]:
         it.setdefault("wartet_auf", [])
         it.setdefault("notiz", None)
         it.setdefault("projekt", None)
+        it.setdefault("prio", 0)
+        it.setdefault("puls_alter", None)
+        it.setdefault("gestartet", None)
     if not items or not KANBAN_DB.is_file():
         return []
     by_id = {it["id"]: it for it in items if it.get("id")}
@@ -136,6 +139,18 @@ def _ketten_anreichern(items: list[dict]) -> list[dict]:
             links = [(p, c) for p, c in con.execute(
                 "SELECT parent_id, child_id FROM task_links")
                 if p in by_id or c in by_id]
+            # Prio (für die Tausch-Knöpfe) + Lebenszeichen (Hänger-Warnung): ein Rutsch.
+            ph = ",".join("?" * len(by_id))
+            jetzt = time.time()
+            for tid, prio, hb, started, pid in con.execute(
+                    f"SELECT id, priority, last_heartbeat_at, started_at, worker_pid"
+                    f" FROM tasks WHERE id IN ({ph})", list(by_id)):
+                it = by_id[tid]
+                it["prio"] = prio or 0
+                if it.get("status") == "running":
+                    it["gestartet"] = started
+                    puls = hb or started
+                    it["puls_alter"] = int(jetzt - puls) if puls else None
             for it in items:
                 if it.get("status") != "running":
                     continue
@@ -369,6 +384,161 @@ def ergebnis_of(task_id: str) -> dict:
     return data
 
 
+# ── Karten-Steuerung (Stopp / Neuer Versuch / Prio) — User-Wunsch 20.07. ─────
+# Die CLI kann blocken/entsperren, aber weder haengende Worker beenden noch die
+# Prioritaet aendern → fuer genau diese zwei Faelle eine eng begrenzte Direkt-
+# Schreibstelle in die kanban.db (BEGIN IMMEDIATE, nur tasks-Spalten). Alles
+# andere laeuft weiter ueber die CLI.
+
+
+def _task_row(task_id: str) -> dict | None:
+    try:
+        con = sqlite3.connect(f"file:{KANBAN_DB}?mode=ro", uri=True, timeout=3)
+        try:
+            con.row_factory = sqlite3.Row
+            r = con.execute("SELECT id, status, worker_pid, priority FROM tasks WHERE id=?",
+                            (task_id,)).fetchone()
+            return dict(r) if r else None
+        finally:
+            con.close()
+    except Exception:
+        return None
+
+
+def _kill_tree(pid: int) -> None:
+    """Worker samt Kind-Prozessen beenden (TERM, dann KILL). Ohne das bleiben
+    Test-Server (uvicorn & Co.) als Waisen zurueck — live gesehen 20.07."""
+    import signal
+
+    def kinder(p: int) -> list[int]:
+        try:
+            out = subprocess.run(["pgrep", "-P", str(p)], capture_output=True, text=True, timeout=5)
+            return [int(x) for x in out.stdout.split()]
+        except Exception:
+            return []
+
+    def baum(p: int) -> list[int]:
+        alle = [p]
+        for k in kinder(p):
+            alle.extend(baum(k))
+        return alle
+
+    pids = baum(pid)
+    for p in reversed(pids):
+        try:
+            os.kill(p, signal.SIGTERM)
+        except OSError:
+            pass
+    time.sleep(2)
+    for p in reversed(pids):
+        try:
+            os.kill(p, signal.SIGKILL)
+        except OSError:
+            pass
+
+
+def _requeue(task_id: str) -> None:
+    """Karte sauber zurueck auf todo (Claim/PID/Zaehler leeren) — der Dispatcher
+    spawnt sie beim naechsten Tick frisch (Caps gelten)."""
+    con = sqlite3.connect(str(KANBAN_DB), timeout=15)
+    try:
+        con.execute("PRAGMA busy_timeout=10000")
+        con.execute("BEGIN IMMEDIATE")
+        con.execute(
+            "UPDATE tasks SET status='todo', claim_lock=NULL, claim_expires=NULL,"
+            " worker_pid=NULL, current_run_id=NULL, last_heartbeat_at=NULL,"
+            " started_at=NULL, consecutive_failures=0 WHERE id=?", (task_id,))
+        con.commit()
+    finally:
+        con.close()
+
+
+def stop_task(task_id: str) -> dict:
+    """Karte gezielt anhalten: generischer Block (bleibt liegen, kein Auto-Unblock-
+    Kind) + laufenden Worker-Prozessbaum beenden."""
+    if not _available():
+        return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
+    if not _TASK_ID_RX.match(task_id or ""):
+        return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
+    row = _task_row(task_id)
+    if not row:
+        return {"ok": False, "error": "Karte nicht gefunden."}
+    try:
+        r = _hermes(["block", task_id, "Vom", "Commander", "gestoppt"])
+    except Exception as exc:
+        return {"ok": False, "error": f"Queue nicht erreichbar: {exc}"}
+    if r.returncode != 0:
+        return {"ok": False, "error": (r.stderr or r.stdout or "Blocken fehlgeschlagen").strip()[:300]}
+    if row.get("status") == "running" and row.get("worker_pid"):
+        _kill_tree(int(row["worker_pid"]))
+    _list_cache["ts"] = 0.0
+    return {"ok": True}
+
+
+def retry_task(task_id: str) -> dict:
+    """Neuer Versuch: haengende/laufende Karte → Worker-Baum beenden + frisch auf todo;
+    blockierte Karte → entsperren. Der Dispatcher uebernimmt den Rest (Caps gelten)."""
+    if not _available():
+        return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
+    if not _TASK_ID_RX.match(task_id or ""):
+        return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
+    row = _task_row(task_id)
+    if not row:
+        return {"ok": False, "error": "Karte nicht gefunden."}
+    status = row.get("status")
+    try:
+        if status == "blocked":
+            r = _hermes(["unblock", task_id, "--reason", "Neuer Versuch vom Commander"])
+            if r.returncode != 0:
+                return {"ok": False, "error": (r.stderr or r.stdout or "Entsperren fehlgeschlagen").strip()[:300]}
+        else:
+            if status == "running" and row.get("worker_pid"):
+                _kill_tree(int(row["worker_pid"]))
+            _requeue(task_id)
+            try:
+                _hermes(["comment", task_id, "Neuer", "Versuch", "vom", "Commander", "(Karte zurueckgesetzt)"], timeout=20)
+            except Exception:
+                pass
+    except Exception as exc:
+        return {"ok": False, "error": f"Neuer Versuch fehlgeschlagen: {exc}"}
+    _list_cache["ts"] = 0.0
+    try:
+        from services import announce
+        announce.add(f"Karte {task_id} wurde auf Commander-Wunsch neu angestoßen.",
+                     "[Ideen-Queue]", "ideen-queue", "silent")
+    except Exception:
+        pass
+    return {"ok": True}
+
+
+def prio_task(task_id: str, richtung: str) -> dict:
+    """Prio hoch/runter (Dispatcher zieht ready-Karten nach `priority DESC, created ASC`
+    → hoch = frueher dran). Direkt-Schreibstelle, die CLI kennt keine Prio-Aenderung."""
+    if not _available():
+        return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
+    if not _TASK_ID_RX.match(task_id or ""):
+        return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
+    if richtung not in ("hoch", "runter"):
+        return {"ok": False, "error": "richtung muss 'hoch' oder 'runter' sein."}
+    delta = 1 if richtung == "hoch" else -1
+    try:
+        con = sqlite3.connect(str(KANBAN_DB), timeout=15)
+        try:
+            con.execute("PRAGMA busy_timeout=10000")
+            con.execute("BEGIN IMMEDIATE")
+            cur = con.execute("UPDATE tasks SET priority = COALESCE(priority,0) + ? WHERE id=?",
+                              (delta, task_id))
+            con.commit()
+            if cur.rowcount == 0:
+                return {"ok": False, "error": "Karte nicht gefunden."}
+        finally:
+            con.close()
+    except Exception as exc:
+        return {"ok": False, "error": f"Prio-Änderung fehlgeschlagen: {exc}"}
+    _list_cache["ts"] = 0.0
+    return {"ok": True}
+
+
 def log_of(task_id: str) -> dict:
     """Worker-Log einer Idee (hermes kanban log <id>), ANSI entfernt, letzte ~120 Zeilen."""
     if not _available():

backend/services/ketten_digest.py

@@ -31,9 +31,27 @@ log = logging.getLogger(__name__)
 ENABLED = os.name == "posix" and os.environ.get("MC_KETTEN_DIGEST", "1") != "0"
 INTERVAL = int(os.environ.get("MC_KETTEN_DIGEST_INTERVAL", "300"))       # Prüf-Tick: 5 min
 PULS_SEKUNDEN = int(os.environ.get("MC_KETTEN_DIGEST_PULS", "3600"))     # Zwischenstand: 60 min
+# Hänger-Alarm: laufende Karte ohne Lebenszeichen (Heartbeat) länger als diese Schwelle
+# → sofortige Telegram-Warnung (einmal pro Karten-Lauf). Live-Fall 20.07.: Worker wartete
+# 45 min auf einen selbst gestarteten uvicorn — „pid_alive" hielt den Claim ewig frisch.
+HANG_SEKUNDEN = int(os.environ.get("MC_KETTEN_HANG", "900"))
+KANBAN_LOGS = Path(os.environ.get("MC_KANBAN_LOGS", "~/.hermes/kanban/logs")).expanduser()
 STATE_PATH = Path(os.environ.get("MC_KETTEN_DIGEST_STATE", str(MODELS_DIR / "mc2-ketten-digest.json")))
 
 
+def _kontext_druck(task_id: str) -> int:
+    """Wie oft der Worker zuletzt den Kontext verdichten musste (Log-Marker) —
+    mehrfaches „Compacting context" heißt: die Aufgabe sprengt das Fenster."""
+    try:
+        p = KANBAN_LOGS / f"{task_id}.log"
+        with p.open("rb") as f:
+            f.seek(max(0, p.stat().st_size - 200_000))
+            tail = f.read().decode("utf-8", errors="replace")
+        return tail.count("Compacting context")
+    except Exception:
+        return 0
+
+
 def _load_state() -> dict:
     try:
         d = json.loads(STATE_PATH.read_text(encoding="utf-8"))
@@ -92,6 +110,33 @@ def _tick() -> None:
         gemeldete_fragen[key] = now
         dirty = True
 
+    # 1b) Hänger-Alarm: laufende Karte ohne Lebenszeichen → einmal pro Karten-Lauf warnen.
+    gemeldete_haenger: dict = state.setdefault("haenger", {})
+    for it in items:
+        if it.get("status") != "running":
+            continue
+        puls = it.get("puls_alter")
+        if puls is None or puls < HANG_SEKUNDEN:
+            continue
+        key = f"{it['id']}:{it.get('gestartet') or 0}"
+        if key in gemeldete_haenger:
+            continue
+        minuten = int(puls / 60)
+        text = (f"Karte scheint zu hängen: „{(it.get('titel') or '')[:120]}“ — "
+                f"seit {minuten} min kein Lebenszeichen vom Worker.")
+        druck = _kontext_druck(it["id"])
+        if druck >= 2:
+            text += (f"\nDer Worker musste {druck}× den Kontext verdichten — "
+                     "die Aufgabe ist womöglich zu groß geschnitten.")
+        text += "\nIn der Zentrale: „Neuer Versuch“ startet sie frisch, „Stopp“ hält sie an."
+        _melden("[Werkstatt]", text)
+        gemeldete_haenger[key] = now
+        dirty = True
+    if len(gemeldete_haenger) > 200:
+        for k in sorted(gemeldete_haenger, key=gemeldete_haenger.get)[:100]:
+            gemeldete_haenger.pop(k, None)
+        dirty = True
+
     # Fragen-Gedächtnis klein halten (Karten verschwinden irgendwann ins Archiv).
     if len(gemeldete_fragen) > 200:
         for k in sorted(gemeldete_fragen, key=gemeldete_fragen.get)[:100]: