claude-regelwerk/.vault-tools/workflow-pruefung/verbrauch_reihe.py

"""Wertet die Verbrauchsablesungen der Workflows aus: wie viel Prozent des 5-Stunden-Fensters und der Woche zwischen zwei Ablesungen verbraucht wurden und welche Agenten dazwischen liefen.

Koordinator-Werkzeug (entstanden 2026-09-24). Es misst und beurteilt nichts; die Stufen in
.claude/workflows/verbrauch.js stellt der Mensch nach dieser Messreihe ein.
Quelle der Daten ist der Session-Speicher von Claude Code. Je Workflow-Lauf liegen unter
<speicher>/<sitzung>/subagents/workflows/<lauf>/ die Datei journal.jsonl (Ereignisse "started"
mit agentId und label, "result" mit agentId und result), je Agent agent-<id>.jsonl (Zeilen mit
timestamp) und agent-<id>.meta.json (model). Ablesungen sind Agenten, deren label mit
"verbrauch:" beginnt; ihr result traegt fenster mit label, percentUsed und resetsAt.
Die Fenster werden wie in verbrauch.js erkannt: "5-hour ..." und "Weekly ... all models".
Was das Skript nicht sieht: den Verbrauch anderer Sitzungen und der Hauptsitzung im selben
Zeitraum (er steckt in der Differenz mit drin), Ablesungen ueber probe (sie starten keinen
Agenten) und Laeufe, deren Protokolle geloescht sind. Die Kennzahl je Opus-Agentenstunde ist eine
Naeherung, kein Nachweis.
Aufruf: python verbrauch_reihe.py [--speicher <pfad>] [--seit JJJJ-MM-TT]
Exit 0 mit Auswertung (auch bei null Ablesungen), 2 wenn der Speicher fehlt oder --seit kein
Datum ist. Die Ausgabe ist reines ASCII, damit sie auch auf einer cp1252-Konsole durchlaeuft.
"""

import argparse
import json
import math
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

STANDARD_SPEICHER = Path.home() / ".claude" / "projects" / "D--Claude"
RESET_TOLERANZ = timedelta(minutes=5)
FUENF = re.compile(r"^\s*5[- ]?hour", re.I)
WOCHE = re.compile(r"^\s*weekly\b.*\ball models\b", re.I)
FAMILIEN = ("opus", "sonnet", "haiku", "fable")


def zeit(text):
    return datetime.fromisoformat(text.replace("Z", "+00:00"))


def lies_jsonl(pfad, befunde):
    """Liest eine JSONL-Datei zeilenweise; kaputte Zeilen sind ein Befund, kein Absturz."""
    zeilen = []
    try:
        with pfad.open(encoding="utf-8") as datei:
            for nr, zeile in enumerate(datei, 1):
                if not zeile.strip():
                    continue
                try:
                    zeilen.append(json.loads(zeile))
                except ValueError:
                    befunde.append(f"{pfad.parent.name}/{pfad.name}: Zeile {nr} ist kein JSON, uebergangen")
    except OSError as e:
        befunde.append(f"{pfad.parent.name}/{pfad.name}: nicht lesbar ({type(e).__name__})")
    return zeilen


def einordnen(fenster):
    """(prozent, resetsAt) fuer 5 Stunden und Woche; None, wo das Fenster fehlt oder kaputt ist."""
    fuenf = woche = None
    if not isinstance(fenster, list):
        return None, None
    for f in fenster:
        if not isinstance(f, dict) or not isinstance(f.get("label"), str):
            return None, None
        art = "fuenf" if FUENF.match(f["label"]) else "woche" if WOCHE.match(f["label"]) else None
        if art is None:
            continue
        p = f.get("percentUsed")
        if isinstance(p, bool) or not isinstance(p, (int, float)) or not math.isfinite(p) or p < 0:
            return None, None
        eintrag = (float(p), f.get("resetsAt"))
        if art == "fuenf" and (fuenf is None or p > fuenf[0]):
            fuenf = eintrag
        elif art == "woche" and (woche is None or p > woche[0]):
            woche = eintrag
    return fuenf, woche


def zeitspanne(ordner, agent_id, befunde):
    stempel = []
    for o in lies_jsonl(ordner / f"agent-{agent_id}.jsonl", befunde):
        t = o.get("timestamp") if isinstance(o, dict) else None
        if isinstance(t, str):
            try:
                stempel.append(zeit(t))
            except ValueError:
                pass
    return (min(stempel), max(stempel)) if stempel else None


def familie(ordner, agent_id):
    try:
        modell = json.loads((ordner / f"agent-{agent_id}.meta.json").read_text(encoding="utf-8")).get("model")
    except (OSError, ValueError, AttributeError):
        return "?"
    if not isinstance(modell, str):
        return "?"
    for f in FAMILIEN:
        if f in modell.lower():
            return f
    return modell


def lauf_auswerten(ordner, befunde):
    labels, ergebnisse = {}, {}
    for o in lies_jsonl(ordner / "journal.jsonl", befunde):
        if not isinstance(o, dict) or not isinstance(o.get("agentId"), str):
            continue
        if o.get("type") == "started":
            labels[o["agentId"]] = o.get("label") if isinstance(o.get("label"), str) else ""
        elif o.get("type") == "result":
            ergebnisse[o["agentId"]] = o.get("result")
    ablesungen, agenten = [], []
    for agent_id, label in labels.items():
        spanne = zeitspanne(ordner, agent_id, befunde)
        if label.startswith("verbrauch:"):
            if spanne is None:
                befunde.append(f"{ordner.name}: Ablesung {label} ohne Zeitstempel, uebergangen")
                continue
            ergebnis = ergebnisse.get(agent_id)
            fuenf, woche = einordnen(ergebnis.get("fenster") if isinstance(ergebnis, dict) else None)
            # Ablesezeit ist das Ende des Ableseagenten: get_usage laeuft kurz vor seinem Schluss.
            ablesungen.append({"anlass": label[len("verbrauch:"):], "zeit": spanne[1], "fuenf": fuenf, "woche": woche})
        elif spanne is not None:
            agenten.append({"familie": familie(ordner, agent_id), "start": spanne[0], "ende": spanne[1]})
    ablesungen.sort(key=lambda a: a["zeit"])
    return ablesungen, agenten


def zeitpunkt(text):
    if not isinstance(text, str):
        return None
    try:
        return zeit(text)
    except ValueError:
        return None


def differenz(a, b):
    """Prozentpunkte von a nach b im selben Fenster.

    get_usage rechnet resetsAt bei jeder Ablesung neu aus; der Wert schwankt um Millisekunden und
    springt dabei auch ueber eine Sekunden- oder Datumsgrenze (gemessen 2026-09-24: 18:29:59.664Z,
    18:30:00.267Z, 18:30:00.459Z fuer dasselbe Fenster). Als Reset gilt deshalb erst eine
    Verschiebung um mehr als RESET_TOLERANZ oder ein fallender Wert.
    """
    if a is None or b is None:
        return "unbekannt"
    ra, rb = zeitpunkt(a[1]), zeitpunkt(b[1])
    if ra is None or rb is None or (ra.tzinfo is None) != (rb.tzinfo is None):
        return "unbekannt"
    if abs(rb - ra) > RESET_TOLERANZ or b[0] < a[0]:
        return "Reset"
    return round(b[0] - a[0], 1)


def paare(ordner, ablesungen, agenten):
    for a, b in zip(ablesungen, ablesungen[1:]):
        minuten = {}
        for g in agenten:
            anteil = (min(g["ende"], b["zeit"]) - max(g["start"], a["zeit"])).total_seconds()
            if anteil > 0:
                minuten[g["familie"]] = minuten.get(g["familie"], 0.0) + anteil / 60
        yield {
            "lauf": ordner.name,
            "von": a["anlass"],
            "bis": b["anlass"],
            "minuten": round((b["zeit"] - a["zeit"]).total_seconds() / 60, 1),
            "agenten": {k: round(v, 1) for k, v in sorted(minuten.items())},
            "d5": differenz(a["fuenf"], b["fuenf"]),
            "dw": differenz(a["woche"], b["woche"]),
        }


UMLAUTE = str.maketrans({"ä": "ae", "ö": "oe", "ü": "ue", "Ä": "Ae", "Ö": "Oe", "Ü": "Ue", "ß": "ss"})


def nur_ascii(text):
    return str(text).translate(UMLAUTE).encode("ascii", "replace").decode("ascii")


def main(argv=None):
    p = argparse.ArgumentParser(description="Messreihe der Verbrauchsablesungen aus den Workflow-Protokollen")
    p.add_argument("--speicher", type=Path, default=STANDARD_SPEICHER)
    p.add_argument("--seit", default=None, help="nur Ablesungen ab diesem Tag (JJJJ-MM-TT, UTC)")
    a = p.parse_args(argv)
    if not a.speicher.is_dir():
        print(nur_ascii(f"Speicher fehlt: {a.speicher}"))
        return 2
    seit = None
    if a.seit:
        try:
            seit = datetime.fromisoformat(a.seit).replace(tzinfo=timezone.utc)
        except ValueError:
            print(nur_ascii(f"--seit {a.seit!r} ist kein Datum JJJJ-MM-TT"))
            return 2
    befunde, zeilen, anzahl = [], [], 0
    for ordner in sorted(a.speicher.glob("*/subagents/workflows/wf_*")):
        if not ordner.is_dir():
            continue
        try:
            ablesungen, agenten = lauf_auswerten(ordner, befunde)
            if seit is not None:
                ablesungen = [x for x in ablesungen if x["zeit"] >= seit]
            anzahl += len(ablesungen)
            zeilen.extend(paare(ordner, ablesungen, agenten))
        except Exception as e:  # Ein Absturz ist ein Befund (pruefwerkzeuge.md, Abschnitt 4).
            befunde.append(f"{ordner.name}: Auswertung abgebrochen ({type(e).__name__}: {e})")
    print(f"{anzahl} Ablesungen, {len(zeilen)} Paare")
    if zeilen:
        print("| Lauf | von -> bis | Minuten | Agentenminuten | d 5h % | d Woche % |")
        print("|---|---|---|---|---|---|")
        for z in zeilen:
            agenten = ", ".join(f"{k} {v}" for k, v in z["agenten"].items()) or "-"
            print(nur_ascii(f"| {z['lauf']} | {z['von']} -> {z['bis']} | {z['minuten']} | {agenten} | {z['d5']} | {z['dw']} |"))
    werte = [z for z in zeilen if isinstance(z["d5"], float) and z["agenten"].get("opus", 0) > 0]
    if werte:
        prozent = sum(z["d5"] for z in werte)
        stunden = sum(z["agenten"]["opus"] for z in werte) / 60
        print(f"Naeherung: {prozent / stunden:.1f} % des 5-Stunden-Fensters je Opus-Agentenstunde "
              f"({len(werte)} Paare, {stunden:.1f} Opus-Stunden; andere Sitzungen und Nicht-Opus-Agenten stecken in der Differenz)")
    else:
        print("Keine auswertbaren Paare mit Opus-Agenten.")
    if befunde:
        print("Hinweise:")
        for b in befunde:
            print(nur_ascii("- " + b))
    return 0


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