claude-werkzeuge/.vault-tools/auto-push/auto_push.py

"""PostToolUse-Hook: committet und pusht jede Änderung an der veröffentlichten Anweisungsschicht.

Ersteller: Pascal Michel
Erstellt:  2026-09-17

GELTUNGSBEREICH
  Ausgelöst wird nur, wenn tool_input.file_path innerhalb der Vault-Wurzel auf
  CLAUDE.md, README.md, AI/Regelwerk/**, .claude/agents/**, .claude/skills/**,
  .claude/settings.json oder .vault-tools/** zeigt. Die Wurzel ist
  ausschließlich der Ordner zwei Ebenen über dieser Datei; weder
  CLAUDE_PROJECT_DIR noch das Feld cwd der Hook-Eingabe werden gelesen, weil
  cwd einem cd im Bash-Werkzeug folgt (dieselbe Begründung wie in
  regelwerk_nach_edit.py).

  Was tatsächlich gepusht wird, entscheidet aber nicht diese Liste, sondern die
  .gitignore der Vault-Wurzel. Sie arbeitet als Positivliste: ausgeschlossen
  ist alles, eingeschlossen nur, was dort ausdrücklich wieder aufgenommen ist.
  Die Liste hier bestimmt nur, ob der Hook überhaupt anläuft — sie ist eine
  Abkürzung, kein zweiter Filter. Wer den Umfang ändern will, ändert die
  .gitignore.

  Still bleibt der Hook in drei Lagen: die Eingabe taugt nicht (leer, kein
  JSON-Objekt, kein file_path als Zeichenkette); der Pfad liegt außerhalb des
  Geltungsbereichs; oder es gibt nichts zu committen. Jede andere Lage —
  fehlendes Token, fehlendes Remote, gescheiterter Commit, gescheiterter Push,
  Zeitüberschreitung — ist ein Befund und wird gemeldet.

VERHALTEN BEI GESCHEITERTEM PUSH
  Der Commit bleibt lokal stehen und der Befund geht in den Kontext. Der
  nächste Lauf schiebt ihn mit, weil `git push` alles bis HEAD überträgt. Ein
  gescheiterter Push verliert also keine Arbeit, er verzögert sie.

NEBENLÄUFIGKEIT
  Zwei schnell aufeinanderfolgende Edits starten zwei Hook-Läufe. Ohne Sperre
  kollidieren sie an der index.lock von git. Deshalb eine eigene Sperrdatei mit
  O_EXCL; wer sie nicht bekommt, wartet bis SPERRE_WARTE_S und läuft danach
  selbst. Findet er nichts mehr zu committen, bleibt er still — der andere Lauf
  hat seine Änderung dann bereits mitgenommen.

BENANNTE LÜCKEN
  * Ob tool_name Edit, Write oder MultiEdit ist, prüft der Matcher in
    settings.json, nicht dieses Skript.
  * Der Hook prüft den Inhalt nicht. Er hat keinen Geheimnis-Scan und keine
    Freigabe. Was in einer eingeschlossenen Datei steht, geht heraus, sobald
    sie geschrieben wird — auch halbfertig. Das ist die ausdrückliche Vorgabe
    (Pascal, 2026-09-17), und es ist der Grund, warum die .gitignore eine
    Positivliste ist und keine Ausschlussliste.
  * Pfade werden ueber realpath aufgeloest. Ein Symlink von ausserhalb auf
    eine eingeschlossene Datei loest den Hook deshalb aus, ein Symlink von
    innen nach aussen nicht.
  * Ein Push wird nicht wiederholt. Nachgeholt wird er beim nächsten Lauf.

Exit ist immer 0. Gemeldet wird über additionalContext plus systemMessage, wie
beim Nachbar-Hook regelwerk_nach_edit.py.
"""

from __future__ import annotations

import errno
import json
import os
import subprocess
import sys
import time
from pathlib import Path

SKRIPT = Path(__file__).resolve()

# Unter dem timeout von 60 s, den der settings.json-Eintrag setzen soll.
TIMEOUT_GIT_S = 15
TIMEOUT_PUSH_S = 40
SPERRE_WARTE_S = 25
SPERRE_TAKT_S = 0.4
MAX_ZEILEN = 30
MAX_ZEICHEN = 3000
MAX_DATEIEN_IN_MELDUNG = 12

# Kleinbuchstaben und Schrägstrich vorwärts, verglichen gegen normiere().
GELTUNG_DATEIEN = ("claude.md", "readme.md", ".gitignore", ".gitattributes")
GELTUNG_ORDNER = (
    "ai/regelwerk/",
    ".claude/agents/",
    ".claude/skills/",
    ".vault-tools/",
)
GELTUNG_EINZELN = (".claude/settings.json",)

ASKPASS_REL = Path(".vault-tools") / "auto-push" / "askpass.sh"
SPERRE_REL = Path(".vault-tools") / "auto-push" / ".sperre"


def lies_stdin() -> bytes:
    if sys.stdin is None or sys.stdin.isatty():
        return b""
    return sys.stdin.buffer.read()


def lies_eingabe(roh: bytes) -> object:
    # errors=replace: ein kaputtes Byte im Pfad soll laut enden, nicht still.
    text = roh.decode("utf-8-sig", errors="replace")
    if not text.strip():
        return None
    try:
        return json.loads(text)
    except ValueError:
        return None


def bestimme_wurzel() -> Path:
    return SKRIPT.parents[2]


def normiere(pfad: str) -> str:
    # realpath, nicht abspath: Windows liefert denselben Ort mal als
    # "C:/Users/Pascal Michel/..." und mal als 8.3-Kurznamen
    # "C:/Users/PASCAL~1/...". Ein Vergleich mit abspath haelt die beiden fuer
    # verschiedene Pfade, und der Hook bleibt dann still statt zu pushen --
    # im Fallkatalog vom 2026-09-17 sind daran sieben Faelle gescheitert.
    return os.path.normpath(os.path.realpath(pfad)).replace("\\", "/").lower().rstrip("/")


def im_geltungsbereich(wurzel: Path, datei: str) -> bool:
    kandidat = datei.replace("\\", "/")
    try:
        if not os.path.isabs(kandidat):
            kandidat = os.path.join(str(wurzel), kandidat)
        pfad = normiere(kandidat)
        basis = normiere(str(wurzel))
    except (ValueError, OSError):
        return False
    # Mit Trenner vergleichen, sonst träfe "<wurzel>-CLAUDE.md" neben dem Vault.
    if not pfad.startswith(basis + "/"):
        return False
    rel = pfad[len(basis) + 1:]
    return (
        rel in GELTUNG_DATEIEN
        or rel in GELTUNG_EINZELN
        or rel.startswith(GELTUNG_ORDNER)
    )


def dekodiere(roh: bytes | str | None) -> str:
    if roh is None:
        return ""
    if isinstance(roh, str):
        return roh
    return roh.decode("utf-8", errors="replace")


class GitFehler(Exception):
    """Ein git-Aufruf ist gescheitert; die Meldung ist der Befund."""

    def __init__(self, kopf: str, rumpf: str) -> None:
        super().__init__(kopf)
        self.kopf = kopf
        self.rumpf = rumpf


def git(wurzel: Path, *argumente: str, timeout: int = TIMEOUT_GIT_S) -> str:
    umgebung = dict(
        os.environ,
        PYTHONIOENCODING="utf-8",
        GIT_TERMINAL_PROMPT="0",
        GIT_ASKPASS=str(wurzel / ASKPASS_REL),
        GIT_CONFIG_NOSYSTEM="0",
    )
    befehl = ["git", "-C", str(wurzel), *argumente]
    try:
        lauf = subprocess.run(
            befehl,
            env=umgebung,
            stdin=subprocess.DEVNULL,
            capture_output=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired as abgelaufen:
        teil = (dekodiere(abgelaufen.stdout) + dekodiere(abgelaufen.stderr)).strip()
        raise GitFehler(
            f"git {argumente[0]} nach {timeout} s abgebrochen",
            f"`git {' '.join(argumente)}` lief länger als {timeout} s.\n"
            f"Bis dahin ausgegeben:\n{teil or '(nichts)'}",
        ) from abgelaufen
    except OSError as fehler:
        raise GitFehler(
            f"git nicht startbar: {type(fehler).__name__}",
            f"`git {' '.join(argumente)}` ließ sich nicht starten: "
            f"{type(fehler).__name__}: {fehler}",
        ) from fehler

    if lauf.returncode != 0:
        ausgabe = (dekodiere(lauf.stdout) + "\n" + dekodiere(lauf.stderr)).strip()
        raise GitFehler(
            f"git {argumente[0]}: Exit {lauf.returncode}",
            f"`git {' '.join(argumente)}` endete mit Exit {lauf.returncode}.\n"
            f"{ausgabe or '(keine Ausgabe)'}",
        )
    return dekodiere(lauf.stdout)


def hole_sperre(sperre: Path) -> int | None:
    """Gibt den Dateideskriptor zurück, oder None, wenn die Sperre belegt bleibt."""
    frist = time.monotonic() + SPERRE_WARTE_S
    while True:
        try:
            return os.open(str(sperre), os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        except OSError as fehler:
            if fehler.errno not in (errno.EEXIST, errno.EACCES):
                raise
            # Eine Sperre, die älter ist als die Wartezeit, stammt aus einem
            # abgestürzten Lauf. Sie wird entfernt, nicht ewig respektiert.
            try:
                if time.time() - sperre.stat().st_mtime > SPERRE_WARTE_S * 2:
                    sperre.unlink(missing_ok=True)
                    continue
            except OSError:
                pass
            if time.monotonic() >= frist:
                return None
            time.sleep(SPERRE_TAKT_S)


def gib_sperre_frei(deskriptor: int, sperre: Path) -> None:
    try:
        os.close(deskriptor)
    except OSError:
        pass
    try:
        sperre.unlink(missing_ok=True)
    except OSError:
        pass


def baue_meldung(dateien: list[str]) -> str:
    rest = len(dateien) - MAX_DATEIEN_IN_MELDUNG
    sichtbar = dateien[:MAX_DATEIEN_IN_MELDUNG]
    kopf = (
        f"Automatischer Push: {len(dateien)} Datei(en) geändert"
        if len(dateien) != 1
        else f"Automatischer Push: {dateien[0]}"
    )
    rumpf = "\n".join(f"- {d}" for d in sichtbar)
    if rest > 0:
        rumpf += f"\n- … und {rest} weitere"
    return (
        f"{kopf}\n\n{rumpf}\n\n"
        "Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py.\n"
        "\n"
        "Co-Authored-By: Claude Opus 5 <[email protected]>\n"
    )


def arbeite(wurzel: Path) -> tuple[str, str] | None:
    """None, wenn nichts zu tun war; sonst (systemMessage, Rumpf)."""
    if not (wurzel / ".git").exists():
        raise GitFehler(
            "Auto-Push: kein Git-Repo",
            f"{wurzel} ist kein Git-Repository. Der Hook ist eingetragen, kann "
            "aber nichts pushen. Das ist ein Befund, keine Entwarnung.",
        )
    if not (wurzel / ASKPASS_REL).is_file():
        raise GitFehler(
            "Auto-Push: askpass-Helfer fehlt",
            f"{wurzel / ASKPASS_REL} fehlt. Ohne ihn fragt git nach einem "
            "Passwort und der Push hängt oder scheitert.",
        )
    if not os.environ.get("GITEA_TOKEN", "").strip():
        raise GitFehler(
            "Auto-Push: GITEA_TOKEN nicht gesetzt",
            "Die Umgebungsvariable GITEA_TOKEN ist leer oder fehlt. Der Commit "
            "wird trotzdem gemacht, der Push scheitert. Variable setzen und "
            "eine beliebige Regelwerk-Datei erneut speichern.",
        )

    git(wurzel, "add", "-A")
    gestaged = [z for z in git(wurzel, "diff", "--cached", "--name-only").splitlines() if z.strip()]
    if not gestaged:
        return None

    git(wurzel, "commit", "-m", baue_meldung(gestaged))
    git(wurzel, "push", "origin", "HEAD", timeout=TIMEOUT_PUSH_S)
    return None


def kuerze(text: str) -> str:
    zeilen = text.splitlines()
    rest = len(zeilen) - MAX_ZEILEN
    if rest > 0:
        zeilen = zeilen[:MAX_ZEILEN] + [f"[... {rest} weitere Zeilen gekürzt]"]
    ergebnis = "\n".join(zeilen)
    if len(ergebnis) > MAX_ZEICHEN:
        ergebnis = ergebnis[:MAX_ZEICHEN] + f"\n[... gekürzt auf {MAX_ZEICHEN} Zeichen]"
    return ergebnis


def melde(system: str, kontext: str) -> None:
    ausgabe = {
        "hookSpecificOutput": {
            "hookEventName": "PostToolUse",
            "additionalContext": kuerze(kontext),
        },
        "systemMessage": " ".join(system.split()),
    }
    sys.stdout.write(json.dumps(ausgabe, ensure_ascii=True))
    sys.stdout.flush()


def main(roh: bytes | None = None) -> int:
    try:
        daten = lies_eingabe(lies_stdin() if roh is None else roh)
        if not isinstance(daten, dict):
            return 0
        eingabe = daten.get("tool_input")
        if not isinstance(eingabe, dict):
            return 0
        datei = eingabe.get("file_path")
        if not isinstance(datei, str) or not datei.strip():
            return 0

        wurzel = bestimme_wurzel()
        if not im_geltungsbereich(wurzel, datei):
            return 0

        sperre = wurzel / SPERRE_REL
        deskriptor = hole_sperre(sperre)
        if deskriptor is None:
            # Ein anderer Lauf arbeitet noch. Er nimmt diese Änderung mit,
            # weil er `git add -A` über den ganzen Baum fährt.
            return 0
        try:
            ergebnis = arbeite(wurzel)
        finally:
            gib_sperre_frei(deskriptor, sperre)
        if ergebnis is not None:
            melde(*ergebnis)
        return 0
    except GitFehler as fehler:
        melde(fehler.kopf, fehler.rumpf)
        return 0
    except Exception as fehler:  # noqa: BLE001 - ein Absturz ist selbst ein Befund
        melde(
            f"Auto-Push abgebrochen: {type(fehler).__name__}",
            f"Der Auto-Push-Hook ist unerwartet abgebrochen: "
            f"{type(fehler).__name__}: {fehler}",
        )
        return 0


if __name__ == "__main__":
    raise SystemExit(main())