claude-werkzeuge/.vault-tools/plan-kopie/welle_vergleich.py

"""Gegenpruefung nach einer Executer-Welle: Menge der geaenderten Dateien und Byte-Vergleich gegen die Kopie.

Koordinator-Werkzeug im Scratchpad. Fragen und Herkunft der Antwort:
  1. Welche Dateien hat die Welle geaendert?      Quelle: `git status --porcelain -z` im Projekt.
  2. Welche durfte sie aendern?                    Quelle: Abschnitt "## Zieldateien" der uebergebenen Taskdateien.
  3. Ist jede Zieldatei byte-gleich mit dem Plan?  Quelle: die Kopie `mat`, auf der plan_materialisiere.py
                                                   alle Codebloecke des Plans angewendet hat.
  4. Welche Zieldateien blendet git aus?           Quelle: `git check-ignore -z --stdin` im Projekt. Getrackte
                                                   Dateien zaehlen nicht, auch wenn ein Muster passt (git-check-ignore(1)).
     Seit 2026-10-17: Eine ignorierte Zieldatei steht nie in `git status`. Frueher hiess das "unveraendert gegenueber
     HEAD" und der Byte-Vergleich fiel aus (Paket R1 im Edge-Stack: 8 und 12 Fehlalarme fuer tests/.work/).
     Jetzt wird sie wie jede andere gegen die Kopie verglichen; ob sie vor der Welle schon so aussah, sieht das
     Werkzeug bei ihr nicht, und eine geloeschte gilt als geloescht, sobald sie im Baum fehlt.
Was es NICHT sieht (Fangnetz, kein Nachweis):
  - Dateien, die .gitignore ausblendet und die keine Zieldatei sind (data/, artifacts/, node_modules/): Ein
    Executer, der dorthin schreibt, bleibt unsichtbar.
  - Ob die Kopie selbst sauber ist. Eine verschmutzte Kopie erzeugt Fehlalarm, kein falsches Gruen.
  - Prosa-Dateien (--prosa): nur Mengenpruefung, kein Byte-Vergleich; die liest der Koordinator selbst.
  - Bereits committete Aenderungen: Verglichen wird der Arbeitsbaum gegen HEAD.
Eine fehlende Voraussetzung (Kopie fehlt, Taskdatei ohne Zieldateien, git laeuft nicht) ist ein Befund.
Aufruf: python welle_vergleich.py --projekt <pfad> --mat <pfad> [--prosa <rel> ...] <taskdatei> [...]
Exit 0 ohne Befund, 2 mit Befund.
"""

import argparse
import re
import subprocess
import sys
import traceback
from pathlib import Path, PureWindowsPath

# Seit 2026-09-18 ist die Klammer hinter dem Pfad wahlfrei, und dahinter darf Text stehen
# ("- `pfad` -- nur die Konstante X"): Plaene nennen Zieldateien nicht immer mit "(neu)".
ZIEL = re.compile(r"^- `(.+?)`(?:\s*\((.*?)\))?")


def zieldateien(taskdatei: Path, projekt: Path, befunde: list[str]) -> dict[str, bool]:
    """Relativer Pfad -> True, wenn die Datei geloescht werden soll."""
    try:
        zeilen = taskdatei.read_text(encoding="utf-8").replace("\r\n", "\n").split("\n")
    except OSError as fehler:
        befunde.append(f"{taskdatei.name}: nicht lesbar ({fehler})")
        return {}
    drin = False
    ziele: dict[str, bool] = {}
    for zeile in zeilen:
        if zeile.startswith("## "):
            drin = zeile.startswith("## Zieldateien")
            continue
        if not drin:
            continue
        treffer = ZIEL.match(zeile)
        if not treffer:
            continue
        absolut = PureWindowsPath(treffer.group(1))
        if absolut.drive == "" and absolut.root == "":
            # Seit 2026-09-18: Ein Pfad ohne Laufwerk und ohne Wurzel gilt relativ zum Projekt
            # (Plaene fuer einen Git-Worktree nennen keine absoluten Pfade). Ein ".." fuehrte
            # aus dem Projekt hinaus und ist ein Befund wie ein fremder absoluter Pfad.
            if ".." in absolut.parts:
                befunde.append(f"{taskdatei.name}: Zieldatei {absolut} liegt nicht unter dem Projekt {projekt}")
                continue
            rel = absolut
        else:
            try:
                rel = absolut.relative_to(PureWindowsPath(str(projekt)))
            except ValueError:
                befunde.append(f"{taskdatei.name}: Zieldatei {absolut} liegt nicht unter dem Projekt {projekt}")
                continue
        klammer = treffer.group(2) or ""
        ziele[rel.as_posix()] = "gelöscht" in klammer or "geloescht" in klammer
    if not ziele:
        befunde.append(f"{taskdatei.name}: keine Zieldateien gefunden -- ohne sie misst der Lauf nichts")
    return ziele


def geaendert(projekt: Path, befunde: list[str]) -> dict[str, str]:
    """Relativer Pfad -> zweistelliger Status aus git."""
    try:
        ergebnis = subprocess.run(
            ["git", "-C", str(projekt), "status", "--porcelain", "-z", "--untracked-files=all"],
            capture_output=True,
            stdin=subprocess.DEVNULL,
        )
    except OSError as fehler:
        befunde.append(f"git laesst sich nicht starten ({fehler})")
        return {}
    if ergebnis.returncode != 0:
        befunde.append(f"git status endet mit {ergebnis.returncode}: {ergebnis.stderr.decode('utf-8', 'replace').strip()}")
        return {}
    teile = ergebnis.stdout.decode("utf-8", "replace").split("\0")
    aus: dict[str, str] = {}
    i = 0
    while i < len(teile):
        eintrag = teile[i]
        i += 1
        if len(eintrag) < 4:
            continue
        status, pfad = eintrag[:2], eintrag[3:]
        if status[0] in "RC":
            alt = teile[i] if i < len(teile) else "?"
            i += 1
            befunde.append(f"{pfad}: git meldet Umbenennung oder Kopie aus {alt} -- im Plan nicht vorgesehen")
        aus[pfad] = status
    return aus


def ignoriert(projekt: Path, pfade: list[str], befunde: list[str]) -> set[str]:
    """Die Zieldateien unter pfade, die git ignoriert. Exit 1 von check-ignore heisst: keine; alles ueber 1 ein Befund."""
    if not pfade:
        return set()
    try:
        ergebnis = subprocess.run(
            ["git", "-C", str(projekt), "check-ignore", "-z", "--stdin"],
            input=("\0".join(pfade) + "\0").encode("utf-8"),
            capture_output=True,
        )
    except OSError as fehler:
        befunde.append(f"git check-ignore laesst sich nicht starten ({fehler})")
        return set()
    if ergebnis.returncode not in (0, 1):
        befunde.append(f"git check-ignore endet mit {ergebnis.returncode}: {ergebnis.stderr.decode('utf-8', 'replace').strip()}")
        return set()
    return {p for p in ergebnis.stdout.decode("utf-8", "replace").split("\0") if p}


def erste_abweichung(echt: bytes, kopie: bytes) -> str:
    if echt.replace(b"\r\n", b"\n") == kopie.replace(b"\r\n", b"\n"):
        return "nur die Zeilenenden weichen ab (CRLF gegen LF)"
    a, b = echt.split(b"\n"), kopie.split(b"\n")
    for nummer, (links, rechts) in enumerate(zip(a, b), start=1):
        if links != rechts:
            return f"Zeile {nummer}: Baum {links[:160]!r} | Plan {rechts[:160]!r}"
    return f"Laenge: Baum {len(a)} Zeilen, Plan {len(b)} Zeilen; die gemeinsamen Zeilen sind gleich"


def pruefe(args: argparse.Namespace, befunde: list[str]) -> None:
    projekt, mat = Path(args.projekt), Path(args.mat)
    if not projekt.is_dir():
        befunde.append(f"Projektordner {projekt} fehlt")
        return
    if not mat.is_dir():
        befunde.append(f"Kopie {mat} fehlt -- ohne sie gibt es keinen Byte-Vergleich")
        return
    prosa = {p.replace("\\", "/") for p in args.prosa}
    ziele: dict[str, bool] = {}
    for name in args.taskdateien:
        ziele.update(zieldateien(Path(name), projekt, befunde))
    status = geaendert(projekt, befunde)
    ign = ignoriert(projekt, sorted(ziele), befunde)

    for pfad in sorted(set(status) - set(ziele)):
        befunde.append(f"{pfad}: geaendert ({status[pfad]!r}), liegt aber ausserhalb der Zieldateien dieser Welle")
    for pfad in sorted(prosa - set(ziele)):
        befunde.append(f"{pfad}: als Prosa uebergeben, steht aber in keiner Taskdatei als Zieldatei")

    gleich = 0
    for pfad, loeschen in sorted(ziele.items()):
        echt, kopie = projekt / pfad, mat / pfad
        if loeschen:
            if echt.exists():
                befunde.append(f"{pfad}: soll geloescht sein, steht aber noch im Baum")
            if kopie.exists():
                befunde.append(f"{pfad}: soll geloescht sein, steht aber noch in der Kopie")
            if not echt.exists() and pfad not in ign and "D" not in status.get(pfad, ""):
                befunde.append(f"{pfad}: fehlt im Baum, git meldet aber keine Loeschung ({status.get(pfad)!r})")
            continue
        if pfad not in status and pfad not in ign:
            befunde.append(f"{pfad}: Zieldatei ist unveraendert gegenueber HEAD")
            continue
        if not echt.is_file():
            befunde.append(f"{pfad}: Zieldatei fehlt im Baum")
            continue
        if pfad in prosa:
            continue
        if not kopie.is_file():
            befunde.append(f"{pfad}: fehlt in der Kopie -- kein Byte-Vergleich moeglich")
            continue
        a, b = echt.read_bytes(), kopie.read_bytes()
        if a == b:
            gleich += 1
        else:
            befunde.append(f"{pfad}: weicht vom Plan ab -- {erste_abweichung(a, b)}")
    print(f"{len(ziele)} Zieldateien, {len(status)} geaenderte Dateien laut git, {gleich} byte-gleich mit der Kopie, {len(prosa & set(ziele))} nur als Menge geprueft, {len(ign)} von git ignoriert.")


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--projekt", required=True)
    parser.add_argument("--mat", required=True)
    parser.add_argument("--prosa", nargs="*", default=[])
    parser.add_argument("taskdateien", nargs="+")
    args = parser.parse_args()
    befunde: list[str] = []
    try:
        pruefe(args, befunde)
    except Exception:  # Ein Absturz ist selbst ein Befund; die gesammelten Befunde gehen trotzdem raus.
        befunde.append("unerwarteter Abbruch: " + traceback.format_exc().strip().splitlines()[-1])
    for befund in befunde:
        print("BEFUND " + befund)
    print(f"{len(befunde)} Befund(e).")
    return 2 if befunde else 0


if __name__ == "__main__":
    sys.exit(main())