Auftragsbuch/Chronik/Wissen/Zeitmaschine: Backend + Annahme-Runner (Ein-Klick-Gate)

Der letzte Meter des propose-only-Kreislaufs: Vorschlags-Branches (Werkstatt/
Orchestrator) und Traum-Skill-Kandidaten werden als API sichtbar; Annehmen laeuft
als detached systemd-Unit (Merge im Worktree -> Push main -> Deploy -> Health ->
Auto-Revert bei Rot). Dazu: Chronik-Endpoint (Announce-Store als Timeline),
Wissens-Vault-Reader (read-only, Traversal-Guard), Zeitmaschine (Snapshots +
detached Restore) und Morgenlage-Spiegel im Chef-Gutachter-Feed (priority=silent).
python-multipart explizit in requirements (voice braucht es, war implizit).

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

backend/app.py

@@ -22,6 +22,8 @@ from starlette.requests import Request
 from config import FRONTEND_DIST, VERSION
 from routers import (
     agent,
+    auftragsbuch,
+    chronik,
     connect,
     console,
     gateway_proxy,
@@ -33,6 +35,8 @@ from routers import (
     routing,
     system,
     voice,
+    wissen,
+    zeitmaschine,
 )
 from routers import reminders as reminders_router
 from services import memory as memory_svc
@@ -118,6 +122,10 @@ app.include_router(
 )  # Erinnerungen/Routinen (A3) — feuern in den Briefkasten
 app.include_router(gateway_proxy.router)  # OpenAI-kompatibler /v1-Gateway (model:auto)
 app.include_router(maintenance.router)
+app.include_router(auftragsbuch.router)   # Vorschlags-Inbox (Mensch-Gate als Klick)
+app.include_router(chronik.router)        # Timeline der autonomen Taten (Announce-Store)
+app.include_router(wissen.router)         # Wissens-Vault (Traum-Notizen) read-only
+app.include_router(zeitmaschine.router)   # Snapshots ansehen + Ein-Klick-Restore (detached)
 app.include_router(
     console.router
 )  # Box-Konsole (ttyd) same-origin durchreichen — VOR dem SPA-Catch-all

backend/requirements.txt

@@ -1,4 +1,5 @@
 fastapi>=0.115
+python-multipart>=0.0.9
 uvicorn[standard]>=0.30
 httpx>=0.27
 ruamel.yaml>=0.18

backend/routers/auftragsbuch.py

@@ -0,0 +1,62 @@
+"""Auftragsbuch-Endpoints (Vorschlags-Inbox) — dünner REST-Layer über services/auftragsbuch.py.
+LAN-only wie alle MC2-Endpoints; das Gate ist der Klick des Commanders."""
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+
+from services import auftragsbuch
+
+router = APIRouter(prefix="/api")
+
+
+class BranchIn(BaseModel):
+    branch: str
+
+
+class KandidatIn(BaseModel):
+    file: str
+
+
+@router.get("/auftragsbuch")
+def list_proposals() -> dict:
+    return auftragsbuch.list_proposals()
+
+
+@router.get("/auftragsbuch/diff")
+def diff(branch: str) -> dict:
+    res = auftragsbuch.diff_of(branch)
+    if not res.get("ok"):
+        raise HTTPException(400, res.get("error", "Diff nicht verfügbar."))
+    return res
+
+
+@router.post("/auftragsbuch/annehmen")
+def accept(body: BranchIn) -> dict:
+    res = auftragsbuch.accept(body.branch)
+    if not res.get("ok"):
+        raise HTTPException(400, res.get("error", "Annehmen fehlgeschlagen."))
+    return res
+
+
+@router.post("/auftragsbuch/ablehnen")
+def reject(body: BranchIn) -> dict:
+    res = auftragsbuch.reject(body.branch)
+    if not res.get("ok"):
+        raise HTTPException(400, res.get("error", "Ablehnen fehlgeschlagen."))
+    return res
+
+
+@router.post("/auftragsbuch/skill/annehmen")
+def skill_accept(body: KandidatIn) -> dict:
+    res = auftragsbuch.skill_accept(body.file)
+    if not res.get("ok"):
+        raise HTTPException(400, res.get("error", "Beauftragen fehlgeschlagen."))
+    return res
+
+
+@router.post("/auftragsbuch/skill/ablehnen")
+def skill_reject(body: KandidatIn) -> dict:
+    res = auftragsbuch.skill_reject(body.file)
+    if not res.get("ok"):
+        raise HTTPException(400, res.get("error", "Verwerfen fehlgeschlagen."))
+    return res

backend/routers/chronik.py

@@ -0,0 +1,15 @@
+"""Chronik — die lesbare Timeline dessen, was die Box von allein getan und gemeldet hat.
+Quelle ist der persistente Melde-Briefkasten (services/announce.py): Health-Wächter,
+Auto-Updates, Erinnerungen, Radar/Traum/Chef-Gutachter-Crons, Auftragsbuch — alle
+autonomen Kanäle laufen dort bereits durch. Hier wird nichts Neues erhoben, nur erzählt."""
+
+from fastapi import APIRouter
+
+from services import announce
+
+router = APIRouter(prefix="/api")
+
+
+@router.get("/chronik")
+def chronik(limit: int = 150) -> dict:
+    return {"items": announce.list_recent(limit)}

backend/routers/wissen.py

@@ -0,0 +1,71 @@
+"""Wissens-Vault-Reader: die nächtlichen Traum-Notizen (~/wissens-vault) als klickbares
+Wiki in der Zentrale. Bewusst READ-ONLY — geschrieben wird der Vault nur vom Traum-Cron
+(und via Auftragsbuch-Entscheidungen); hier wird nur gelesen und navigiert."""
+
+import os
+import time
+from pathlib import Path
+
+from fastapi import APIRouter, HTTPException
+
+router = APIRouter(prefix="/api")
+
+VAULT = Path(os.environ.get("MC_VAULT_DIR", "~/wissens-vault")).expanduser()
+
+
+def _available() -> bool:
+    return VAULT.is_dir()
+
+
+def _title_of(p: Path) -> str:
+    """Erste nicht-leere Zeile (ohne Markdown-#) als Anzeigename."""
+    try:
+        with p.open(encoding="utf-8", errors="replace") as fh:
+            for line in fh:
+                s = line.strip()
+                if s:
+                    return s.lstrip("# ").strip()[:160]
+    except OSError:
+        pass
+    return p.stem
+
+
+@router.get("/wissen")
+def list_vault() -> dict:
+    """Alle Markdown-Notizen des Vaults (relativer Pfad, Titel, Ordner, Alter)."""
+    if not _available():
+        return {"available": False, "files": []}
+    files = []
+    now = time.time()
+    for p in sorted(VAULT.rglob("*.md")):
+        if ".git" in p.parts:
+            continue
+        rel = p.relative_to(VAULT).as_posix()
+        st = p.stat()
+        files.append({
+            "path": rel,
+            "name": p.stem,
+            "dir": p.parent.relative_to(VAULT).as_posix() if p.parent != VAULT else "",
+            "title": _title_of(p),
+            "mtime": st.st_mtime,
+            "neu": (now - st.st_mtime) < 36 * 3600,  # „neu seit gestern Nacht"
+        })
+    files.sort(key=lambda f: f["mtime"], reverse=True)
+    return {"available": True, "files": files}
+
+
+@router.get("/wissen/datei")
+def read_file(pfad: str) -> dict:
+    """Inhalt einer Vault-Notiz — Traversal hart geblockt (resolve + is_relative_to)."""
+    if not _available():
+        raise HTTPException(404, "Wissens-Vault liegt auf der Box (hier nicht verfügbar).")
+    try:
+        target = (VAULT / pfad).resolve()
+        if not target.is_relative_to(VAULT.resolve()) or target.suffix != ".md" or not target.is_file():
+            raise HTTPException(404, "Notiz nicht gefunden.")
+        return {"path": pfad, "content": target.read_text(encoding="utf-8", errors="replace")[:400_000],
+                "mtime": target.stat().st_mtime}
+    except HTTPException:
+        raise
+    except (ValueError, OSError):
+        raise HTTPException(404, "Notiz nicht lesbar.")

backend/routers/zeitmaschine.py

@@ -0,0 +1,71 @@
+"""Zeitmaschine — Snapshots ansehen und per Klick zu einem Stand zurückkehren.
+
+Die Maschinerie existiert komplett (deploy/backup.sh Timer 03:30, Off-Box-Kopie,
+deploy/restore.sh mit Pre-Restore-Sicherung); hier kommt nur der letzte Meter dazu:
+Liste + Inhalt in der UI und ein Restore-Start als EIGENE systemd-Unit — restore.sh
+stoppt mission-control-2 selbst, als Kind des Backends stürbe der Lauf mittendrin."""
+
+import os
+import re
+import subprocess
+import time
+from pathlib import Path
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel
+
+from services import backup as backup_svc
+
+router = APIRouter(prefix="/api")
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+RESTORE_SH = _REPO_ROOT / "deploy" / "restore.sh"
+_FILE_RX = re.compile(r"^mc2-state-[0-9T:_-]+\.tar\.gz$")
+
+
+class RestoreIn(BaseModel):
+    file: str
+
+
+@router.get("/zeitmaschine")
+def list_snapshots() -> dict:
+    return {"available": os.name == "posix", "backups": backup_svc.list_backups()}
+
+
+@router.get("/zeitmaschine/inhalt")
+def snapshot_contents(file: str) -> dict:
+    """Top-Level-Komponenten eines Snapshots (mem0, hermes, llama-swap, MANIFEST …)."""
+    if not _FILE_RX.match(file):
+        raise HTTPException(400, "Ungültiger Snapshot-Name.")
+    p = backup_svc.BACKUP_DIR / file
+    if not p.is_file():
+        raise HTTPException(404, "Snapshot nicht gefunden.")
+    return {"file": file, "components": backup_svc.snapshot_components(p)}
+
+
+@router.post("/zeitmaschine/restore")
+def restore(body: RestoreIn) -> dict:
+    """Wiederherstellung starten (detached). restore.sh macht VORHER selbst ein
+    Sicherheits-Backup des aktuellen Zustands — der Schritt ist also reversibel."""
+    if os.name != "posix":
+        raise HTTPException(400, "Wiederherstellen geht nur auf der Box.")
+    if not _FILE_RX.match(body.file):
+        raise HTTPException(400, "Ungültiger Snapshot-Name.")
+    if not (backup_svc.BACKUP_DIR / body.file).is_file():
+        raise HTTPException(404, "Snapshot nicht gefunden.")
+
+    try:
+        from services import announce
+        announce.add(f"Zeitmaschine: Wiederherstellung von {body.file} gestartet — "
+                     "die Dienste starten gleich neu.", "[Zeitmaschine]", "zeitmaschine", "silent")
+    except Exception:
+        pass
+
+    unit = f"mc2-restore-{int(time.time())}"
+    r = subprocess.run(
+        ["systemd-run", "--user", "--collect", f"--unit={unit}",
+         "/bin/bash", str(RESTORE_SH), "--yes", body.file],
+        capture_output=True, text=True, timeout=20)
+    if r.returncode != 0:
+        raise HTTPException(500, f"Start fehlgeschlagen: {(r.stderr or r.stdout).strip()[:300]}")
+    return {"ok": True, "unit": unit}

backend/services/announce.py

@@ -89,6 +89,14 @@ def notify_telegram(subject: str, text: str) -> None:
         log.warning("notify_telegram: notify.sh fehlgeschlagen", exc_info=True)
 
 
+def list_recent(limit: int = 150) -> list[dict]:
+    """Die jüngsten Einträge (neueste zuerst) — Datenquelle der Chronik: alles, was die Box
+    dem Commander je aktiv gemeldet hat (Health-Wächter, Updates, Radar, Träume, Alarme)."""
+    with _lock:
+        state = _load()
+        return list(reversed(state["items"][-max(1, min(limit, MAX_ITEMS)):]))
+
+
 def list_after(after: int | None, limit: int = 20) -> dict:
     """Einträge NACH Cursor `after` (aufsteigend). Ohne Cursor nur den aktuellen
     Stand liefern (latest) — so initialisiert Lucy ihren Cursor, ohne Altes nachzuplappern."""

backend/services/auftragsbuch.py

@@ -0,0 +1,316 @@
+"""
+Auftragsbuch — die Vorschlags-Inbox der Box (das Mensch-Gate als Klick statt Git-Handarbeit).
+
+Quellen der Karten:
+  • Vorschlags-Branches auf Gitea (wartung/*, orchestrator/*, doku/*) — die Werkstatt und der
+    Orchestrator arbeiten propose-only und lassen ihre Ergebnisse dort liegen.
+  • Skill-Kandidaten aus dem Wissens-Vault (~/wissens-vault/skill-kandidaten/) — Vorschläge des
+    nächtlichen Traum-Crons, Gate war bisher „Commander sagt mach".
+
+Annehmen (Branch) startet deploy/auftrag-annehmen.sh als EIGENE systemd-Unit (detached):
+Merge im isolierten Worktree → Push origin/main → Deploy → Health → bei Rot Revert+Alarm.
+Detached, weil deploy.sh mission-control-2 neu startet — ein Kind des Backends stürbe mittendrin.
+Der Fortschritt landet in STATUS_PATH (JSON), das Skript schreibt, die API liest nur.
+
+Lokal (Windows-Dev) ist alles harmlos: available=False, Aktionen geben Fehler statt zu crashen.
+"""
+
+import json
+import logging
+import os
+import re
+import shutil
+import subprocess
+import time
+from pathlib import Path
+
+from config import MODELS_DIR
+
+log = logging.getLogger(__name__)
+
+REPO = Path(os.environ.get("MC2_SOURCE_DIR", "~/mission-control-v2")).expanduser()
+VAULT = Path(os.environ.get("MC_VAULT_DIR", "~/wissens-vault")).expanduser()
+STATUS_PATH = Path(os.environ.get("MC2_AUFTRAG_STATUS", str(MODELS_DIR / "mc2-auftragsbuch.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).
+_BRANCH_RX = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._/-]{0,120}$")
+_KANDIDAT_RX = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._ -]{0,120}\.md$")
+
+_fetch_cache: dict = {"ts": 0.0}
+_FETCH_EVERY = 30.0  # s — Gitea nicht bei jedem UI-Poll anfragen
+
+
+def _available() -> bool:
+    return os.name == "posix" and (REPO / ".git").exists()
+
+
+def _git(args: list[str], timeout: int = 20) -> subprocess.CompletedProcess:
+    return subprocess.run(["git", "-C", str(REPO), *args],
+                          capture_output=True, text=True, timeout=timeout)
+
+
+def _fetch_throttled() -> None:
+    now = time.time()
+    if now - _fetch_cache["ts"] < _FETCH_EVERY:
+        return
+    _fetch_cache["ts"] = now
+    try:
+        _git(["fetch", "-q", "--prune", "origin"], timeout=30)
+    except Exception:
+        log.warning("auftragsbuch: git fetch fehlgeschlagen", exc_info=True)
+
+
+def _statuses() -> dict:
+    try:
+        return (json.loads(STATUS_PATH.read_text(encoding="utf-8")) or {}).get("branches", {})
+    except Exception:
+        return {}
+
+
+def _set_status(branch: str, state: str, detail: str) -> None:
+    try:
+        try:
+            data = json.loads(STATUS_PATH.read_text(encoding="utf-8"))
+        except Exception:
+            data = {}
+        data.setdefault("branches", {})[branch] = {"state": state, "detail": detail, "ts": time.time()}
+        tmp = STATUS_PATH.with_suffix(".tmp")
+        tmp.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+        tmp.replace(STATUS_PATH)
+    except OSError:
+        log.warning("auftragsbuch: Status %s nicht schreibbar", STATUS_PATH, exc_info=True)
+
+
+def _remote_branches() -> list[str]:
+    r = _git(["for-each-ref", "--format=%(refname:short)", "refs/remotes/origin"])
+    out = []
+    for line in (r.stdout or "").splitlines():
+        name = line.strip().removeprefix("origin/")
+        if name and name != "HEAD" and name.startswith(PREFIXES):
+            out.append(name)
+    return out
+
+
+def _valid_branch(branch: str) -> bool:
+    return bool(_BRANCH_RX.match(branch)) and ".." not in branch and branch.startswith(PREFIXES)
+
+
+def list_proposals() -> dict:
+    """Alle offenen Vorschläge: Branches (mit Commit-/Diff-Zusammenfassung + Status) und
+    Skill-Kandidaten aus dem Vault. Eine Antwort für die ganze Auftragsbuch-Seite."""
+    if not _available():
+        return {"available": False, "items": [], "skill_kandidaten": [], "open_count": 0}
+
+    _fetch_throttled()
+    statuses = _statuses()
+    items = []
+    for branch in _remote_branches():
+        ref = f"origin/{branch}"
+        try:
+            ahead = int((_git(["rev-list", "--count", f"origin/main..{ref}"]).stdout or "0").strip() or 0)
+            if ahead == 0 and (statuses.get(branch, {}).get("state") not in ("laeuft", "rollback")):
+                continue  # bereits in main enthalten → keine offene Entscheidung mehr
+            behind = int((_git(["rev-list", "--count", f"{ref}..origin/main"]).stdout or "0").strip() or 0)
+            show = _git(["show", "-s", "--format=%s%x1f%b%x1f%ct%x1f%an", ref])
+            subject, body, cts, author = ((show.stdout or "").split("\x1f") + ["", "", "", ""])[:4]
+            stat = (_git(["diff", "--shortstat", f"origin/main...{ref}"]).stdout or "").strip()
+            files_raw = (_git(["diff", "--name-status", f"origin/main...{ref}"]).stdout or "").splitlines()
+            files = []
+            for line in files_raw[:60]:
+                parts = line.split("\t")
+                if len(parts) >= 2:
+                    files.append({"status": parts[0][:2], "path": parts[-1]})
+            paths = [f["path"] for f in files]
+            frontend_src = any(p.startswith("frontend/src") for p in paths)
+            frontend_dist = any(p.startswith("frontend/dist") for p in paths)
+            items.append({
+                "branch": branch,
+                "kind": branch.split("/", 1)[0],
+                "subject": subject.strip(),
+                "body": body.strip()[:2000],
+                "author": author.strip(),
+                "ts": int(cts) if cts.strip().isdigit() else None,
+                "ahead": ahead,
+                "behind": behind,
+                "shortstat": stat,
+                "files": files,
+                "files_truncated": len(files_raw) > 60,
+                # Frontend-Quelltext ohne gebautes Bundle: Box kann nicht bauen (kein Node) —
+                # nach dem Annehmen bliebe die Oberfläche alt, bis am PC gebaut wird.
+                "frontend_ohne_build": frontend_src and not frontend_dist,
+                "status": statuses.get(branch),
+            })
+        except Exception:
+            log.warning("auftragsbuch: Branch %s nicht lesbar", branch, exc_info=True)
+    items.sort(key=lambda i: i.get("ts") or 0, reverse=True)
+
+    kandidaten = list_skill_kandidaten()
+    open_count = sum(1 for i in items
+                     if (i.get("status") or {}).get("state") not in ("eingespielt",)) + len(kandidaten)
+    return {"available": True, "items": items, "skill_kandidaten": kandidaten, "open_count": open_count}
+
+
+def diff_of(branch: str) -> dict:
+    if not _available():
+        return {"ok": False, "error": "Nur auf der Box verfügbar."}
+    if not _valid_branch(branch) or branch not in _remote_branches():
+        return {"ok": False, "error": f"Unbekannter Vorschlags-Branch: {branch}"}
+    r = _git(["diff", f"origin/main...origin/{branch}"], timeout=30)
+    text = r.stdout or ""
+    truncated = len(text) > 200_000
+    return {"ok": True, "diff": text[:200_000], "truncated": truncated}
+
+
+def accept(branch: str) -> dict:
+    """Annehmen: detached Runner starten (Merge→Push→Deploy→Health→ggf. Rollback)."""
+    if not _available():
+        return {"ok": False, "error": "Annehmen geht nur auf der Box."}
+    if not _valid_branch(branch) or branch not in _remote_branches():
+        return {"ok": False, "error": f"Unbekannter Vorschlags-Branch: {branch}"}
+    state = (_statuses().get(branch) or {}).get("state")
+    if state in ("laeuft", "rollback"):
+        return {"ok": False, "error": "Für diesen Vorschlag läuft bereits ein Annahme-Lauf."}
+
+    # Runner als /tmp-Kopie starten: deploy.sh resettet das Repo hart — das Original-Skript
+    # würde einem laufenden bash unter den Füßen getauscht (bekannte Selbst-Reset-Falle).
+    src = REPO / "deploy" / "auftrag-annehmen.sh"
+    runner = Path(f"/tmp/mc2-auftrag-runner-{int(time.time())}.sh")
+    try:
+        shutil.copyfile(src, runner)
+    except OSError as exc:
+        return {"ok": False, "error": f"Runner nicht kopierbar: {exc}"}
+
+    slug = re.sub(r"[^a-z0-9-]+", "-", branch.lower())[:40].strip("-")
+    unit = f"mc2-auftrag-{slug}-{int(time.time())}"
+    r = subprocess.run(
+        ["systemd-run", "--user", "--collect", f"--unit={unit}",
+         "/bin/bash", str(runner), branch],
+        capture_output=True, text=True, timeout=20)
+    if r.returncode != 0:
+        return {"ok": False, "error": f"Start fehlgeschlagen: {(r.stderr or r.stdout).strip()[:300]}"}
+    _set_status(branch, "laeuft", "Annahme-Lauf gestartet")
+    return {"ok": True, "unit": unit}
+
+
+def reject(branch: str) -> dict:
+    """Ablehnen: Remote-Branch löschen (die Arbeit bleibt in der Gitea-Historie referenzierbar,
+    aber die Entscheidung ist gefallen). Meldung in den Briefkasten — ehrlich und sichtbar."""
+    if not _available():
+        return {"ok": False, "error": "Ablehnen geht nur auf der Box."}
+    if not _valid_branch(branch) or branch not in _remote_branches():
+        return {"ok": False, "error": f"Unbekannter Vorschlags-Branch: {branch}"}
+    r = _git(["push", "origin", "--delete", branch], timeout=30)
+    if r.returncode != 0:
+        return {"ok": False, "error": f"Löschen fehlgeschlagen: {(r.stderr or '').strip()[:300]}"}
+    _git(["fetch", "-q", "--prune", "origin"], timeout=30)
+    _set_status(branch, "abgelehnt", "Vom Commander abgelehnt — Branch gelöscht")
+    try:
+        from services import announce
+        announce.add(f"Vorschlag '{branch}' wurde abgelehnt und der Branch gelöscht.",
+                     "[Auftragsbuch]", "auftragsbuch", "silent")
+    except Exception:
+        pass
+    return {"ok": True}
+
+
+# ── Skill-Kandidaten (Wissens-Vault) ─────────────────────────────────────────
+
+def _kandidaten_dir() -> Path:
+    return VAULT / "skill-kandidaten"
+
+
+def list_skill_kandidaten() -> list[dict]:
+    d = _kandidaten_dir()
+    if os.name != "posix" or not d.is_dir():
+        return []
+    out = []
+    for p in sorted(d.glob("*.md"), key=lambda x: x.stat().st_mtime, reverse=True):
+        try:
+            text = p.read_text(encoding="utf-8", errors="replace")
+            first = next((ln.strip().lstrip("# ") for ln in text.splitlines() if ln.strip()), p.stem)
+            out.append({"file": p.name, "title": first[:160],
+                        "preview": text[:3000], "mtime": p.stat().st_mtime})
+        except OSError:
+            continue
+    return out
+
+
+def _vault_git(args: list[str]) -> None:
+    try:
+        subprocess.run(["git", "-C", str(VAULT), *args], capture_output=True, text=True, timeout=15)
+    except Exception:
+        pass
+
+
+def _kandidat_path(fname: str) -> Path | None:
+    if not _KANDIDAT_RX.match(fname):
+        return None
+    p = (_kandidaten_dir() / fname).resolve()
+    if not p.is_relative_to(_kandidaten_dir().resolve()) or not p.is_file():
+        return None
+    return p
+
+
+def skill_accept(fname: str) -> dict:
+    """Skill-Kandidat beauftragen: Inhalt als Werkstatt-Auftrag an die bewährte
+    `hermes -z`-CLI-Lane (detached — der Lauf dauert Minuten und braucht das Backend nicht).
+    Ergebnis ist wieder propose-only: ein neuer Branch, der hier als Karte auftaucht."""
+    if not _available():
+        return {"ok": False, "error": "Beauftragen geht nur auf der Box."}
+    p = _kandidat_path(fname)
+    if not p:
+        return {"ok": False, "error": f"Kandidat nicht gefunden: {fname}"}
+
+    beauftragt = _kandidaten_dir() / "beauftragt"
+    beauftragt.mkdir(parents=True, exist_ok=True)
+    target = beauftragt / p.name
+    try:
+        content = p.read_text(encoding="utf-8", errors="replace")
+        p.rename(target)
+    except OSError as exc:
+        return {"ok": False, "error": f"Kandidat nicht verschiebbar: {exc}"}
+    _vault_git(["add", "-A"])
+    _vault_git(["commit", "-q", "-m", f"Skill-Kandidat beauftragt: {p.name}"])
+
+    order = ("Nutze den orchestrator-Skill (propose-only, Zwei-Kritiker-Gate). "
+             "Auftrag aus dem Wissens-Vault (vom Commander im Auftragsbuch freigegeben):\n\n"
+             + content)
+    order_file = Path(f"/tmp/mc2-skill-auftrag-{int(time.time())}.txt")
+    try:
+        order_file.write_text(order, encoding="utf-8")
+    except OSError as exc:
+        return {"ok": False, "error": f"Auftrag nicht schreibbar: {exc}"}
+    unit = f"mc2-skill-auftrag-{int(time.time())}"
+    r = subprocess.run(
+        ["systemd-run", "--user", "--collect", f"--unit={unit}",
+         "/bin/bash", "-lc", f'hermes -z "$(cat {order_file})"'],
+        capture_output=True, text=True, timeout=20)
+    if r.returncode != 0:
+        return {"ok": False, "error": f"Start fehlgeschlagen: {(r.stderr or r.stdout).strip()[:300]}"}
+    try:
+        from services import announce
+        announce.add(f"Skill-Kandidat '{fname}' wurde beauftragt — die Werkstatt arbeitet, "
+                     "das Ergebnis erscheint als neuer Vorschlag im Auftragsbuch.",
+                     "[Auftragsbuch]", "auftragsbuch", "silent")
+    except Exception:
+        pass
+    return {"ok": True, "unit": unit}
+
+
+def skill_reject(fname: str) -> dict:
+    if not _available():
+        return {"ok": False, "error": "Verwerfen geht nur auf der Box."}
+    p = _kandidat_path(fname)
+    if not p:
+        return {"ok": False, "error": f"Kandidat nicht gefunden: {fname}"}
+    verworfen = _kandidaten_dir() / "verworfen"
+    verworfen.mkdir(parents=True, exist_ok=True)
+    try:
+        p.rename(verworfen / p.name)
+    except OSError as exc:
+        return {"ok": False, "error": f"Kandidat nicht verschiebbar: {exc}"}
+    _vault_git(["add", "-A"])
+    _vault_git(["commit", "-q", "-m", f"Skill-Kandidat verworfen: {p.name}"])
+    return {"ok": True}

backend/services/backup.py

@@ -27,6 +27,11 @@ def _latest() -> Path | None:
     return snaps[0] if snaps else None
 
 
+def snapshot_components(tarball: Path) -> list[str]:
+    """Öffentliche Sicht auf die Snapshot-Komponenten (Zeitmaschine-Detail in der UI)."""
+    return _components(tarball)
+
+
 def _components(tarball: Path) -> list[str]:
     """Top-Level-Einträge im Tarball (zur Anzeige im UI)."""
     try: