claude-regelwerk/.vault-tools/verbrauch-auswertung/faelle.py

"""Regressionstreiber für analyse_transkripte.py: fährt die Fälle unter fixture/ und die Mutanten, die jeder Fall fangen soll.

Ersteller: Pascal Michel
Erstellt:  2026-10-30
Doku:      LIESMICH.md in diesem Ordner

Was er belegt: die drei Zählregeln (größter Wert je message.id, Deduplizierung über Dateigrenzen,
Haiku-5.5-Stufe über 100k Prompt), die Ablehnung einer Ausgabe im Vault, den fehlenden
projects-Ordner, kaputte Zeilen ohne Absturz und die Auswahl nach Titel. Jeder Fall läuft einzeln
gegen einen eigenen Fixture-Ordner und einen frischen Ausgabeordner im System-Temp; zuerst die
Grundlinie, scheitert sie, messen die übrigen Fälle nichts.
Rotnachweis: Je Mutante wird eine Kopie des Skripts mit genau einer Ersetzung gebaut; die Mutante
gilt als gefangen, wenn der ihr zugeordnete Fall rot wird. Eine Ersetzung, die im Skript nicht
(genau einmal) vorkommt, bricht mit Exit 2 ab, statt still nichts zu prüfen.
Was er nicht sieht: echte Transkripte, Preise (PREISE wird nicht gegen eine Quelle geprüft), die
Berichtstabellen außer den hier abgefragten Zahlen, claudedata_wurzel.py und scratchpads.py.
Ein Fangnetz, kein Nachweis.
Aufruf: python -I -X utf8 faelle.py [--ohne-mutanten]
Exit 0 alle Fälle grün und alle Mutanten gefangen, 1 sonst, 2 bei veraltetem Mutantenmuster.
"""
import json
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

HIER = Path(__file__).resolve().parent
SKRIPT = HIER / "analyse_transkripte.py"
FIXTURE = HIER / "fixture"
VAULT = HIER.parents[1]


def lauf(skript, fixture, ausgabe, *extra):
    r = subprocess.run([sys.executable, "-I", "-X", "utf8", str(skript), str(fixture), "--ausgabe", str(ausgabe), *extra],
                       capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=120)
    daten = None
    j = Path(ausgabe) / "auswertung.json"
    if r.returncode == 0 and j.exists():
        daten = json.loads(j.read_text(encoding="utf-8"))
    return r, daten


def sitzung(d, sid):
    return next((s for s in d["sessions"] if s["sid"] == sid), None)


def nahe(a, b):
    return abs(a - b) < 1e-9


def f_grundlinie(skript, tmp):
    r, d = lauf(skript, FIXTURE / "grundlinie", tmp)
    if d is None:
        return f"Exit {r.returncode}: {r.stdout[-200:]}{r.stderr[-200:]}"
    s = sitzung(d, "sess-g")
    if not s or s["turns"] != 1 or s["out"] != 100 or not nahe(s["usd"], 0.006):
        return f"erwartet 1 Runde, 100 Ausgabe, 0.006 USD; gesehen {s and (s['turns'], s['out'], s['usd'])}"
    if d["kennzahlen"]["doppelt_dateiuebergreifend"] or d["kennzahlen"]["nachrichten_mit_teilausgabe"] or d["befunde"]:
        return f"Grundlinie meldet Kopie, Teilausgabe oder Befund: {d['kennzahlen']} {d['befunde']}"
    return None


def teilausgabe(ordner, agent, erwartet_teil):
    def fall(skript, tmp):
        r, d = lauf(skript, FIXTURE / ordner, tmp)
        if d is None:
            return f"Exit {r.returncode}"
        sa = next((x for x in d["subagents"] if x["agent"] == agent), None)
        if not sa or sa["turns"] != 1 or sa["out"] != 900:
            return f"Subagent: erwartet 1 Runde mit 900 Ausgabe, gesehen {sa and (sa['turns'], sa['out'])}"
        teil = d["kennzahlen"]["nachrichten_mit_teilausgabe"]
        if teil != erwartet_teil:
            return f"Teilausgabe gezählt {teil}, erwartet {erwartet_teil}"
        if erwartet_teil and d["teilausgabe_version"] != {"2.1.300": 1}:
            return f"Verteilung nach Version falsch: {d['teilausgabe_version']}"
        return None
    return fall


def kopie(ordner):
    def fall(skript, tmp):
        r, d = lauf(skript, FIXTURE / ordner, tmp)
        if d is None:
            return f"Exit {r.returncode}"
        runden = sum(s["turns"] for s in d["sessions"])
        out = sum(s["out"] for s in d["sessions"])
        if runden != 2 or out != 300 or d["kennzahlen"]["doppelt_dateiuebergreifend"] != 1:
            return f"erwartet 2 Runden, 300 Ausgabe, 1 Kopie; gesehen {runden}, {out}, {d['kennzahlen']['doppelt_dateiuebergreifend']}"
        return None
    return fall


def f_haiku_ueber(skript, tmp):
    r, d = lauf(skript, FIXTURE / "haiku-ueber-100k", tmp)
    if d is None:
        return f"Exit {r.returncode}"
    s = sitzung(d, "sess-h")
    # 150000 x 0.50 + 1000 x 2.50 (Stufe über 100k) plus 50000 x 0.10 + 1000 x 0.50, je /1e6
    if not s or not nahe(s["usd"], 0.0775 + 0.0055):
        return f"erwartet 0.083 USD, gesehen {s and s['usd']}"
    return None


def f_haiku_grenze(skript, tmp):
    r, d = lauf(skript, FIXTURE / "haiku-grenze-100k", tmp)
    if d is None:
        return f"Exit {r.returncode}"
    s = sitzung(d, "sess-k")
    # Genau 100000 Prompt-Tokens liegen noch in der Grundstufe: 100000 x 0.10 / 1e6
    if not s or not nahe(s["usd"], 0.01):
        return f"erwartet 0.01 USD, gesehen {s and s['usd']}"
    return None


def f_kaputte_zeilen(skript, tmp):
    r, d = lauf(skript, FIXTURE / "kaputte-zeilen", tmp)
    if d is None:
        return f"Exit {r.returncode}: {r.stderr[-200:]}"
    s = sitzung(d, "sess-z")
    if not s or s["turns"] != 1 or d["befunde"]:
        return f"erwartet 1 Runde ohne Befund, gesehen {s and s['turns']} und {d['befunde']}"
    return None


def f_auswahl(skript, tmp):
    r, d = lauf(skript, FIXTURE / "auswahl", tmp, "--auswahl", "paket probe")
    a = Path(tmp) / "auswahl.md"
    if d is None or not a.exists():
        return f"Exit {r.returncode}, auswahl.md fehlt"
    text = a.read_text(encoding="utf-8")
    if "Paket Probe eins" not in text or "Anderes Thema" in text or "Treffer: 1 Hauptsitzungen" not in text:
        return "auswahl.md trifft nicht genau die Sitzung mit passendem Titel"
    return None


def f_ausgabe_im_vault(skript, tmp):
    ziel = VAULT / "_verbrauch_auswertung_probe"
    if ziel.exists():
        return f"Vorbedingung: {ziel} existiert schon"
    r, d = lauf(skript, FIXTURE / "grundlinie", ziel)
    angelegt = ziel.exists()
    if angelegt:
        shutil.rmtree(ziel)
    if r.returncode != 2 or angelegt:
        return f"erwartet Exit 2 ohne angelegten Ordner, gesehen Exit {r.returncode}, angelegt {angelegt}"
    return None


def f_projects_fehlt(skript, tmp):
    r, d = lauf(skript, Path(tmp) / "gibt-es-nicht", Path(tmp) / "aus")
    if r.returncode != 2:
        return f"erwartet Exit 2, gesehen {r.returncode}"
    return None


FAELLE = [
    ("grundlinie", f_grundlinie),
    ("teilausgabe-erste-zeile", teilausgabe("teilausgabe-erste-zeile", "agent-t1", 1)),
    ("teilausgabe-letzte-zeile", teilausgabe("teilausgabe-letzte-zeile", "agent-l1", 0)),
    ("kopie-vorn", kopie("kopie-vorn")),
    ("kopie-hinten", kopie("kopie-hinten")),
    ("haiku-ueber-100k", f_haiku_ueber),
    ("haiku-grenze-100k", f_haiku_grenze),
    ("kaputte-zeilen", f_kaputte_zeilen),
    ("auswahl", f_auswahl),
    ("ausgabe-im-vault", f_ausgabe_im_vault),
    ("projects-fehlt", f_projects_fehlt),
]

# (Name, alter Text, neuer Text, Fall, der rot werden muss)
MUTANTEN = [
    ("erste Zeile behalten", 'rec["u"][k] = max(rec["u"][k], u.get(k) or 0)', "pass", "teilausgabe-erste-zeile"),
    ("letzte Zeile statt größter Wert", 'rec["u"][k] = max(rec["u"][k], u.get(k) or 0)', 'rec["u"][k] = u.get(k) or 0', "teilausgabe-letzte-zeile"),
    ("keine Deduplizierung über Dateien", "        if mid in GESEHEN:", "        if False:", "kopie-vorn"),
    ("keine Haiku-Stufe", 'if mk == "claude-haiku-5-5" and ctx > 100000:', 'if mk == "claude-haiku-5-5" and ctx > 10**12:', "haiku-ueber-100k"),
    ("Haiku-Grenze einschließlich", 'if mk == "claude-haiku-5-5" and ctx > 100000:', 'if mk == "claude-haiku-5-5" and ctx >= 100000:', "haiku-grenze-100k"),
    ("Teilausgabe nicht nach Version", '            agg["teilausgabe_version"][rec["version"]] += 1\n', "", "teilausgabe-erste-zeile"),
    ("Zeile ohne Typprüfung", "            if not isinstance(o, dict):\n                continue\n", "", "kaputte-zeilen"),
    ("Vault-Sperre fehlt", "    if im_vault(OUT):", "    if False:", "ausgabe-im-vault"),
    ("fehlender Ordner nicht gemeldet", "    if not os.path.isdir(ROOT):", "    if False:", "projects-fehlt"),
    ("Auswahl ohne Filter", "if s[\"turns_seit\"] > 0 and any(rx.search", "if s[\"turns_seit\"] > 0 and True or any(rx.search", "auswahl"),
]


def fahre(skript, nur=None):
    ergebnisse = {}
    for name, fall in FAELLE:
        if nur and name not in nur:
            continue
        tmp = tempfile.mkdtemp(prefix="va-fall-")
        try:
            ergebnisse[name] = fall(skript, tmp)
        except Exception as e:  # Ein Absturz des Falls ist ein roter Fall, kein Abbruch des Treibers.
            ergebnisse[name] = f"Absturz im Fall: {type(e).__name__}: {e}"
        finally:
            shutil.rmtree(tmp, ignore_errors=True)
        if name == "grundlinie" and ergebnisse[name]:
            break
    return ergebnisse


def main(argv):
    rot = 0
    erg = fahre(SKRIPT)
    for name, _ in FAELLE:
        if name not in erg:
            print(f"NICHT GEFAHREN {name} (Grundlinie rot)")
            rot += 1
            continue
        print(("ROT   " if erg[name] else "gruen ") + name + (f": {erg[name]}" if erg[name] else ""))
        rot += bool(erg[name])
    if rot or "--ohne-mutanten" in argv:
        print(f"{len(FAELLE) - rot} von {len(FAELLE)} Fällen grün")
        return 1 if rot else 0
    quelle = SKRIPT.read_text(encoding="utf-8")
    # Die Kopie liegt im Temp; ohne festgenagelten VAULT hielte sie das Temp für den Vault und lehnte jede Ausgabe ab.
    vault_zeile = "VAULT = Path(__file__).resolve().parents[2]"
    if quelle.count(vault_zeile) != 1:
        print("Mutantenmuster veraltet: VAULT-Zeile")
        return 2
    quelle = quelle.replace(vault_zeile, f"VAULT = Path({str(VAULT)!r})")
    entkommen = 0
    for name, alt, neu, ziel in MUTANTEN:
        if quelle.count(alt) != 1:
            print(f"Mutantenmuster veraltet ({quelle.count(alt)} Treffer): {name}")
            return 2
        tmp = Path(tempfile.mkdtemp(prefix="va-mut-"))
        try:
            m = tmp / "analyse_transkripte.py"
            m.write_text(quelle.replace(alt, neu), encoding="utf-8")
            e = fahre(m, nur={"grundlinie", ziel})
        finally:
            shutil.rmtree(tmp, ignore_errors=True)
        if e.get("grundlinie"):
            print(f"UNGUELTIG {name}: Grundlinie der Mutante rot ({e['grundlinie']})")
            entkommen += 1
            continue
        gefangen = bool(e.get(ziel))
        entkommen += not gefangen
        print(("gefangen  " if gefangen else "ENTKOMMEN ") + f"{name} -> {ziel}" + (f" ({e.get(ziel)})" if gefangen else ""))
    print(f"{len(FAELLE)} Fälle grün; {len(MUTANTEN) - entkommen} von {len(MUTANTEN)} Mutanten gefangen")
    return 1 if entkommen else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))