Wartung: Telegram-Ping bei neuer Auftragsbuch-Karte

- Persistenz-Datei ANNOUNCE_BRANCHES_PATH (mc2-announce-branches.json) für gemeldete Branches
- list_proposals() pingt NEUE Branches per notify_telegram() und announce.add()
- Idempotenz: nur EINMAL pro Branch, bei jedem Poll nur wirklich NEUE melden
- notify_telegram ist best-effort/posix (Windows-Dev = No-op), in try/except gekapselt

backend/services/auftragsbuch.py

@@ -48,6 +48,10 @@ REPOS: dict[str, dict] = {
     "lucy": {"path": LUCY, "runner": "lucy-annahme.sh", "key": "lucy:"},
 }
 
+# Persistenz für gemeldete Branches (Idempotenz: nur EINMAL pro Branch pingen).
+# Muster wie mc2-announce.json unter MODELS_DIR.
+ANNOUNCE_BRANCHES_PATH = Path(os.environ.get("MC2_ANNOUNCE_BRANCHES", str(MODELS_DIR / "mc2-announce-branches.json")))
+
 # Nur diese Branch-Familien sind Vorschläge (main/HEAD & Fremdes bleiben draußen).
 PREFIXES = ("wartung/", "orchestrator/", "doku/", "feature/")
 # Branch-Namen kommen vom Client zurück → hart validieren (keine Shell-/Git-Injektion).
@@ -87,6 +91,25 @@ def _fetch_throttled(repo: str) -> None:
         log.warning("auftragsbuch: git fetch (%s) fehlgeschlagen", repo, exc_info=True)
 
 
+def _load_announce_branches() -> set:
+    """Liefert die Menge der bereits gemeldeten Branch-Namen (Idempotenz)."""
+    try:
+        data = json.loads(ANNOUNCE_BRANCHES_PATH.read_text(encoding="utf-8"))
+        return set(data) if isinstance(data, list) else set()
+    except Exception:
+        return set()
+
+
+def _save_announce_branches(branches: set) -> None:
+    """Speichert die Menge gemelder Branch-Namen."""
+    try:
+        tmp = ANNOUNCE_BRANCHES_PATH.with_suffix(".tmp")
+        tmp.write_text(json.dumps(sorted(branches), ensure_ascii=False), encoding="utf-8")
+        tmp.replace(ANNOUNCE_BRANCHES_PATH)
+    except OSError:
+        log.warning("auftragsbuch: Announce-Branches %s nicht schreibbar", ANNOUNCE_BRANCHES_PATH, exc_info=True)
+
+
 def _statuses() -> dict:
     try:
         return (json.loads(STATUS_PATH.read_text(encoding="utf-8")) or {}).get("branches", {})
@@ -194,7 +217,10 @@ def _repo_items(repo: str, statuses: dict, gutachten: dict) -> list[dict]:
 
 def list_proposals() -> dict:
     """Alle offenen Vorschläge: Branches aus beiden Repos (mit Commit-/Diff-Zusammenfassung +
-    Status) und Skill-Kandidaten aus dem Vault. Eine Antwort für die ganze Auftragsbuch-Seite."""
+    Status) und Skill-Kandidaten aus dem Vault. Eine Antwort für die ganze Auftragsbuch-Seite.
+    
+    Nebenbei: Pinge neu auftauchende Vorschlags-Branches per Telegram und füge sie dem
+    Briefkasten hinzu (Idempotenz: nur EINMAL pro Branch)."""
     if not _available():
         return {"available": False, "items": [], "skill_kandidaten": [], "open_count": 0}
 
@@ -208,6 +234,33 @@ def list_proposals() -> dict:
         items.extend(_repo_items(repo, statuses, gutachten))
     items.sort(key=lambda i: i.get("ts") or 0, reverse=True)
 
+    # --- Telegram-Ping für NEUE Vorschlags-Branches (Idempotenz) ---
+    # Alle aktuellen Branch-Namen aus beiden Repos sammeln
+    aktuelle_branches = set()
+    for repo in REPOS:
+        if _repo_ok(repo):
+            aktuelle_branches.update(_remote_branches(repo))
+    
+    # Bereits gemeldete Branches laden
+    gemeldet = _load_announce_branches()
+    
+    # Nur wirklich NEUE Branches pingen
+    neue = aktuelle_branches - gemeldet
+    for branch in neue:
+        # NotifyTelegram best-effort, niemals crashen
+        try:
+            from services import announce
+            subject = "[Auftragsbuch]"
+            text = f"Neue Karte wartet auf deinen Klick: {branch}"
+            announce.notify_telegram(subject, text)
+            # Und ein Eintrag für den Briefkasten (damit Lucy es spricht)
+            announce.add(text, subject, "auftragsbuch", "normal")
+        except Exception:
+            log.warning("auftragsbuch: Telegram-Ping für %s fehlgeschlagen", branch, exc_info=True)
+    
+    # Alle aktuellen Branches als gemeldet speichern
+    _save_announce_branches(aktuelle_branches)
+
     kandidaten = list_skill_kandidaten()
     open_count = sum(1 for i in items
                      if (i.get("status") or {}).get("state") not in ("eingespielt",)) + len(kandidaten)