Pascal MichelandClaude Opus 5.5
fa1c26b407
.vault-tools (ohne trading-plattform-pruefung und watermarks-remover), .claude/agents, .claude/skills, .claude/workflows, .claude/settings.json. Co-Authored-By: Claude Opus 5.5 <[email protected]>
30 KiB
30 KiB
type, title, status, summary, tags, created, updated, modell, effort, modell_begruendung
| type | title | status | summary | tags | created | updated | modell | effort | modell_begruendung | |||
|---|---|---|---|---|---|---|---|---|---|---|---|---|
| plan-task | Task 07 — verbrauch_reihe.py: Messreihe aus den Workflow-Protokollen | draft | Auswertung der Ablesungen: Differenz im 5-Stunden-Fenster und in der Woche zwischen zwei Ablesungen und die Agentenminuten dazwischen, dazu die Fälle. |
|
2026-09-24 | 2026-09-24 | sonnet | low | Abschrift: jede Zeile steht wörtlich im Plan, Zieldateien und Verbote sind genannt. Sonnet low statt Haiku, weil der Code das Zeichen · (U+00B7) trägt und Haiku in der Messung vom 2026-09-23 typografische Zeichen still ersetzt hat. |
Task 07 — verbrauch_reihe.py: Messreihe aus den Workflow-Protokollen
Die globalen Vorgaben stehen in 00-uebersicht.md (Abschnitt „Globale Vorgaben“). Arbeitsverzeichnis ist der Vault-Wurzelordner D:/Claude; alle Pfade gelten von dort.
Was danach anders ist:
verbrauch_reihe.pyliest aus~/.claude/projects/D--Claude/*/subagents/workflows/wf_*/die Agenten mit Labelverbrauch:und rechnet je Paar die Differenz. Ein Reset ist eine Verschiebung vonresetsAtum mehr als fünf Minuten oder ein fallender Wert; das Zittern um Millisekunden zählt nicht.- Die Ausgabe ist reines ASCII, kaputte Zeilen und Abstürze einzelner Läufe sind Hinweise und kein Abbruch.
mutanten_reihe.pyfährt die Mutantenliste gegen das Skript.
Zieldateien
.vault-tools/workflow-pruefung/verbrauch_reihe.py(neu).vault-tools/workflow-pruefung/faelle_verbrauch_reihe.py(neu).vault-tools/workflow-pruefung/mutanten_reihe.py(neu)
Verbote
- Keine andere Datei, auch nicht die übrigen Workflows unter
.claude/workflows/und nichts unterAI/Regelwerk/,.claude/agents/oderCLAUDE.md. - Kein Commit, kein Push, kein Lauf eines echten Workflows.
- Kein Umformulieren, auch nicht von Kommentaren oder Sonderzeichen; ein Widerspruch zwischen Plan und Baum wird gemeldet.
Umsetzung
.vault-tools/workflow-pruefung/verbrauch_reihe.py
Die ganze Datei ist neu:
"""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())
.vault-tools/workflow-pruefung/faelle_verbrauch_reihe.py
Die ganze Datei ist neu:
"""Faelle fuer verbrauch_reihe.py: baut je Fall einen kuenstlichen Session-Speicher und faehrt das Skript als eigenen Prozess.
Die Grundlinie laeuft zuerst; scheitert sie, messen die uebrigen Faelle nichts. Der Prozess laeuft
ohne PYTHONIOENCODING, damit eine Ausgabe ausserhalb von ASCII auf cp1252 auffaellt.
Aufruf: python faelle_verbrauch_reihe.py
Exit 0, wenn alle Faelle bestehen, sonst 1.
"""
import json
import os
import subprocess
import sys
import tempfile
from pathlib import Path
SKRIPT = Path(__file__).with_name("verbrauch_reihe.py")
def fenster(fuenf, woche, reset5="2026-09-24T18:30:00Z", resetw="2026-09-28T00:00:00Z"):
return [
{"label": "5-hour limit", "percentUsed": fuenf, "resetsAt": reset5},
{"label": "Weekly · all models", "percentUsed": woche, "resetsAt": resetw},
]
def baue(speicher, laeufe):
"""laeufe: {lauf: [agent]}; agent: {id, label, start, ende, modell, ergebnis, ohne_zeit, kaputt}."""
for lauf, agenten in laeufe.items():
ordner = speicher / "sitzung-1" / "subagents" / "workflows" / lauf
ordner.mkdir(parents=True)
with (ordner / "journal.jsonl").open("w", encoding="utf-8") as journal:
journal.write(json.dumps({"type": "launched"}) + "\n")
for a in agenten:
journal.write(json.dumps({"type": "started", "agentId": a["id"], "label": a["label"]}) + "\n")
if a.get("kaputt"):
journal.write("{kein json\n")
if "ergebnis" in a:
journal.write(json.dumps({"type": "result", "agentId": a["id"], "result": a["ergebnis"]}) + "\n")
for a in agenten:
(ordner / f"agent-{a['id']}.meta.json").write_text(json.dumps({"model": a.get("modell", "haiku")}), encoding="utf-8")
if a.get("ohne_zeit"):
continue
with (ordner / f"agent-{a['id']}.jsonl").open("w", encoding="utf-8") as d:
d.write(json.dumps({"timestamp": a["start"], "type": "user"}) + "\n")
d.write(json.dumps({"timestamp": a["ende"], "type": "assistant"}) + "\n")
def ablesung(nr, anlass, zeitpunkt, fuenf, woche, **extra):
ende = extra.pop("bis", None) or zeitpunkt
return {"id": f"v{nr}", "label": "verbrauch:" + anlass, "start": zeitpunkt, "ende": ende, "modell": "haiku",
"ergebnis": {"verfuegbar": True, "status": "ok", "plan": "Pro", "fenster": extra.pop("liste", None) or fenster(fuenf, woche, **extra), "fehler": ""}}
# resetsAt schwankt bei echten Ablesungen um Millisekunden und springt ueber Sekunden- und
# Datumsgrenzen (gemessen 2026-09-24); die Grundlinie traegt deshalb solche Werte.
ZITTER = {"reset5": "2026-09-24T18:29:59.664Z", "resetw": "2026-09-27T23:59:59.664Z"}
def opus(nr, start, ende, modell="opus"):
return {"id": f"o{nr}", "label": f"plankritik:linse{nr}", "start": start, "ende": ende, "modell": modell, "ergebnis": {}}
def fahre(speicher, *zusatz):
umgebung = {k: v for k, v in os.environ.items() if k not in ("PYTHONIOENCODING", "PYTHONUTF8")}
lauf = subprocess.run([sys.executable, str(SKRIPT), "--speicher", str(speicher), *zusatz],
capture_output=True, env=umgebung)
return lauf.returncode, lauf.stdout.decode("ascii", "replace"), lauf.stderr.decode("utf-8", "replace")
GRUND = {
"wf_a": [
ablesung(1, "plankritik:vor Linsen", "2026-09-24T10:00:00Z", 10, 5),
opus(1, "2026-09-24T10:00:10Z", "2026-09-24T10:30:10Z"),
ablesung(2, "plankritik:Ende", "2026-09-24T10:30:10Z", 14, 6, **ZITTER),
]
}
def f_grundlinie(tmp):
baue(tmp, GRUND)
code, aus, err = fahre(tmp)
fehler = []
if code != 0:
fehler.append(f"Exit {code}: {err[-300:]}")
for erwartet in ["2 Ablesungen, 1 Paare", "| wf_a | plankritik:vor Linsen -> plankritik:Ende | 30.2 | opus 30.0 | 4.0 | 1.0 |",
"Naeherung: 8.0 % des 5-Stunden-Fensters je Opus-Agentenstunde (1 Paare, 0.5 Opus-Stunden"]:
if erwartet not in aus:
fehler.append("fehlt: " + erwartet)
if "Hinweise" in aus:
fehler.append("unerwartete Hinweise")
return fehler
def f_reset(tmp):
baue(tmp, {"wf_r": [ablesung(1, "a", "2026-09-24T10:00:00Z", 80, 5), opus(1, "2026-09-24T10:01:00Z", "2026-09-24T10:20:00Z"),
ablesung(2, "b", "2026-09-24T10:30:00Z", 3, 6, reset5="2026-09-24T15:30:00Z")]})
code, aus, _ = fahre(tmp)
fehler = []
if code != 0 or "| Reset | 1.0 |" not in aus:
fehler.append("Reset nicht erkannt: " + aus[-300:])
if "Keine auswertbaren Paare mit Opus-Agenten." not in aus:
fehler.append("Reset-Paar wurde mitgerechnet")
return fehler
def f_speicher_fehlt(tmp):
code, aus, _ = fahre(tmp / "gibt-es-nicht")
return [] if code == 2 and "Speicher fehlt" in aus else [f"Exit {code}, Ausgabe {aus!r}"]
def f_null_ablesungen(tmp):
baue(tmp, {"wf_n": [opus(1, "2026-09-24T10:00:00Z", "2026-09-24T10:10:00Z")]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and "0 Ablesungen, 0 Paare" in aus else [f"Exit {code}, Ausgabe {aus!r}"]
def f_kaputte_zeile(tmp):
agenten = [dict(GRUND["wf_a"][0], kaputt=True), GRUND["wf_a"][1], GRUND["wf_a"][2]]
baue(tmp, {"wf_k": agenten})
code, aus, _ = fahre(tmp)
fehler = []
if code != 0 or "1 Paare" not in aus:
fehler.append("Auswertung trotz kaputter Zeile gescheitert: " + aus[-300:])
if "ist kein JSON" not in aus:
fehler.append("kaputte Zeile nicht als Hinweis gemeldet")
return fehler
def f_ohne_zeitstempel(tmp, position):
agenten = [dict(a) for a in GRUND["wf_a"]]
agenten[position]["ohne_zeit"] = True
baue(tmp, {"wf_z": agenten})
code, aus, _ = fahre(tmp)
fehler = []
if code != 0 or "1 Ablesungen, 0 Paare" not in aus:
fehler.append(f"Exit {code}, Ausgabe {aus[-300:]!r}")
if "ohne Zeitstempel" not in aus:
fehler.append("fehlender Zeitstempel nicht gemeldet")
return fehler
def f_fenster_kaputt(tmp, liste, erwartet="| unbekannt | unbekannt |"):
baue(tmp, {"wf_f": [ablesung(1, "a", "2026-09-24T10:00:00Z", 10, 5, liste=liste), opus(1, "2026-09-24T10:01:00Z", "2026-09-24T10:20:00Z"),
ablesung(2, "b", "2026-09-24T10:30:00Z", 12, 6)]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and erwartet in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}"]
def f_umlaut_im_anlass(tmp):
baue(tmp, {"wf_u": [ablesung(1, "pruefung:vor Prüfung", "2026-09-24T10:00:00Z", 10, 5), ablesung(2, "pruefung:nach Prüfung", "2026-09-24T10:30:00Z", 12, 5)]})
code, aus, err = fahre(tmp)
return [] if code == 0 and "pruefung:vor Pruefung -> pruefung:nach Pruefung" in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}, stderr {err[-200:]!r}"]
def f_seit(tmp):
baue(tmp, {"wf_s": [ablesung(1, "alt", "2026-09-20T10:00:00Z", 10, 5), ablesung(2, "a", "2026-09-24T10:00:00Z", 10, 5),
ablesung(3, "b", "2026-09-24T11:00:00Z", 12, 5)]})
code, aus, _ = fahre(tmp, "--seit", "2026-09-24")
return [] if code == 0 and "2 Ablesungen, 1 Paare" in aus and "alt ->" not in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}"]
def f_seit_ungueltig(tmp):
baue(tmp, GRUND)
code, aus, _ = fahre(tmp, "--seit", "gestern")
return [] if code == 2 and "kein Datum" in aus else [f"Exit {code}, Ausgabe {aus!r}"]
def f_voller_modellname(tmp):
baue(tmp, {"wf_m": [GRUND["wf_a"][0], opus(1, "2026-09-24T10:00:10Z", "2026-09-24T10:30:10Z", modell="claude-opus-5-5"), GRUND["wf_a"][2]]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and "opus 30.0" in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}"]
def f_agent_ausserhalb(tmp):
# Ein Agent, der vor der ersten Ablesung endet, zaehlt nicht mit; einer, der halb hineinreicht, anteilig.
baue(tmp, {"wf_x": [opus(1, "2026-09-24T09:00:00Z", "2026-09-24T09:50:00Z"), GRUND["wf_a"][0],
opus(2, "2026-09-24T09:55:00Z", "2026-09-24T10:10:00Z"), GRUND["wf_a"][2]]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and "| opus 10.0 |" in aus else [f"Exit {code}, Ausgabe {aus[-400:]!r}"]
def f_absturz_ist_befund(tmp):
# Ein Zeitstempel ohne Zeitzone neben einem mit ihr laesst den Vergleich mit TypeError scheitern.
# Der Lauf muss das als Hinweis melden und die anderen Laeufe trotzdem auswerten.
kaputt = [dict(GRUND["wf_a"][0]), dict(GRUND["wf_a"][1], start="2026-09-24T10:00:10"), dict(GRUND["wf_a"][2])]
baue(tmp, {"wf_a": GRUND["wf_a"], "wf_b": kaputt})
code, aus, err = fahre(tmp)
fehler = []
if code != 0:
fehler.append(f"Exit {code}: {err[-300:]}")
if "wf_b: Auswertung abgebrochen (TypeError" not in aus:
fehler.append("Absturz nicht als Hinweis gemeldet")
if "| wf_a |" not in aus:
fehler.append("der gesunde Lauf fehlt")
return fehler
def f_fallender_wert(tmp):
baue(tmp, {"wf_d": [ablesung(1, "a", "2026-09-24T10:00:00Z", 30, 5), ablesung(2, "b", "2026-09-24T10:20:00Z", 5, 5)]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and "| Reset | 0.0 |" in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}"]
def f_reset_unlesbar(tmp):
baue(tmp, {"wf_u": [ablesung(1, "a", "2026-09-24T10:00:00Z", 10, 5, reset5="bald"), ablesung(2, "b", "2026-09-24T10:20:00Z", 12, 6)]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and "| unbekannt | 1.0 |" in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}"]
def f_hoechstes_fenster(tmp):
liste = [{"label": "5-hour limit", "percentUsed": 10, "resetsAt": "2026-09-24T18:30:00Z"},
{"label": "5-hour limit", "percentUsed": 30, "resetsAt": "2026-09-24T18:30:00Z"},
{"label": "Weekly · all models", "percentUsed": 5, "resetsAt": "2026-09-28T00:00:00Z"}]
baue(tmp, {"wf_h": [ablesung(1, "a", "2026-09-24T10:00:00Z", 0, 0, liste=liste), ablesung(2, "b", "2026-09-24T10:20:00Z", 35, 6)]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and "| 5.0 | 1.0 |" in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}"]
def f_ablesezeit_ist_ende(tmp):
# Der Opus-Agent laeuft ganz innerhalb des ersten Ableseagenten, also vor der Ablesung.
baue(tmp, {"wf_t": [ablesung(1, "a", "2026-09-24T10:00:00Z", 10, 5, bis="2026-09-24T10:00:10Z"),
opus(1, "2026-09-24T10:00:02Z", "2026-09-24T10:00:08Z"),
ablesung(2, "b", "2026-09-24T10:30:00Z", 12, 5)]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and "| 29.8 | - | 2.0 | 0.0 |" in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}"]
def f_agent_ueber_das_paar_hinaus(tmp):
baue(tmp, {"wf_b": [ablesung(1, "a", "2026-09-24T10:00:00Z", 10, 5), ablesung(2, "b", "2026-09-24T10:30:00Z", 12, 5),
opus(1, "2026-09-24T10:20:00Z", "2026-09-24T10:50:00Z"), ablesung(3, "c", "2026-09-24T11:00:00Z", 15, 5)]})
code, aus, _ = fahre(tmp)
fehler = []
if code != 0 or "| a -> b | 30.0 | opus 10.0 |" not in aus:
fehler.append("erstes Paar nicht am Ende gekappt: " + aus[-400:])
if "| b -> c | 30.0 | opus 20.0 |" not in aus:
fehler.append("zweites Paar falsch: " + aus[-400:])
return fehler
def f_ueber_hundert(tmp):
baue(tmp, {"wf_g": [ablesung(1, "a", "2026-09-24T10:00:00Z", 98, 5), ablesung(2, "b", "2026-09-24T10:20:00Z", 103, 5)]})
code, aus, _ = fahre(tmp)
return [] if code == 0 and "| 5.0 | 0.0 |" in aus else [f"Exit {code}, Ausgabe {aus[-300:]!r}"]
FAELLE = [
("Grundlinie: zwei Ablesungen, ein Opus-Agent dazwischen", f_grundlinie),
("Reset zwischen zwei Ablesungen", f_reset),
("Speicher fehlt: Exit 2", f_speicher_fehlt),
("null Ablesungen: Exit 0 mit Zaehlung", f_null_ablesungen),
("kaputte Journalzeile: Hinweis, kein Absturz", f_kaputte_zeile),
("erste Ablesung ohne Zeitstempel", lambda t: f_ohne_zeitstempel(t, 0)),
("letzte Ablesung ohne Zeitstempel", lambda t: f_ohne_zeitstempel(t, 2)),
("5-Stunden-Fenster fehlt", lambda t: f_fenster_kaputt(t, [{"label": "Weekly · all models", "percentUsed": 5, "resetsAt": "2026-09-28T00:00:00Z"}], "| unbekannt | 1.0 |")),
("Umlaut im Anlass auf cp1252", f_umlaut_im_anlass),
("Prozent als Text", lambda t: f_fenster_kaputt(t, [{"label": "5-hour limit", "percentUsed": "10", "resetsAt": "2026-09-24T18:30:00Z"}, {"label": "Weekly · all models", "percentUsed": 5, "resetsAt": "2026-09-28T00:00:00Z"}])),
("Prozent als Wahrheitswert", lambda t: f_fenster_kaputt(t, [{"label": "5-hour limit", "percentUsed": True, "resetsAt": "2026-09-24T18:30:00Z"}, {"label": "Weekly · all models", "percentUsed": 5, "resetsAt": "2026-09-28T00:00:00Z"}])),
("fenster kein Array", lambda t: f_fenster_kaputt(t, {"a": 1})),
("--seit filtert aeltere Ablesungen", f_seit),
("--seit ungueltig: Exit 2", f_seit_ungueltig),
("voller Modellname zaehlt als opus", f_voller_modellname),
("Agent ausserhalb und halb innerhalb des Paars", f_agent_ausserhalb),
("Absturz in einem Lauf ist ein Hinweis", f_absturz_ist_befund),
("fallender Wert im selben Fenster ist ein Reset", f_fallender_wert),
("resetsAt ohne Datum: unbekannt", f_reset_unlesbar),
("doppeltes 5-Stunden-Fenster: das hoechste zaehlt", f_hoechstes_fenster),
("Ablesezeit ist das Ende des Ableseagenten", f_ablesezeit_ist_ende),
("Agent ueber das Paar hinaus wird am Ende gekappt", f_agent_ueber_das_paar_hinaus),
("ueber 100 Prozent ist ein Wert, kein Fehler", f_ueber_hundert),
]
def main():
# Fehlertexte koennen Zeichen ausserhalb der Konsolenkodierung tragen; der Treiber darf daran nicht sterben.
sys.stdout.reconfigure(errors="replace")
gescheitert = 0
for name, fall in FAELLE:
with tempfile.TemporaryDirectory() as tmp:
try:
fehler = fall(Path(tmp))
except Exception as e: # ein Absturz im Fall ist selbst ein Befund
fehler = [f"Absturz im Fall: {type(e).__name__}: {e}"]
if fehler:
gescheitert += 1
print(f"FEHLER {name}\n " + "\n ".join(fehler))
else:
print(f"OK {name}")
print(f"{len(FAELLE)} Faelle, {gescheitert} gescheitert")
return 1 if gescheitert else 0
if __name__ == "__main__":
sys.exit(main())
.vault-tools/workflow-pruefung/mutanten_reihe.py
Die ganze Datei ist neu:
"""Mutantenlauf gegen verbrauch_reihe.py: je Mutant genau eine Ersetzung in einer Kopie des Ordners, dann faelle_verbrauch_reihe.py.
Jede Ersetzung muss genau einmal vorkommen, sonst bricht der Lauf mit Exit 2 ab (veralteter Mutant).
Aufruf: python mutanten_reihe.py
Exit 0, wenn jeder Mutant gefangen wird, 1 bei einem Ueberlebenden, 2 bei einem veralteten Muster.
"""
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
HIER = Path(__file__).resolve().parent
MUTANTEN = [
("Reset nur bei gleichem Text", "if abs(rb - ra) > RESET_TOLERANZ or b[0] < a[0]:", "if a[1] != b[1] or b[0] < a[0]:"),
("fallender Wert kein Reset", "if abs(rb - ra) > RESET_TOLERANZ or b[0] < a[0]:", "if abs(rb - ra) > RESET_TOLERANZ:"),
("unlesbares resetsAt als gleich", " if ra is None or rb is None or (ra.tzinfo is None) != (rb.tzinfo is None):\n return \"unbekannt\"\n", ""),
("bool als Zahl", "isinstance(p, bool) or ", ""),
("ueber 100 als Fehler", "not math.isfinite(p) or p < 0", "not math.isfinite(p) or not 0 <= p <= 100"),
("erstes statt hoechstes 5-Stunden-Fenster", "if art == \"fuenf\" and (fuenf is None or p > fuenf[0]):", "if art == \"fuenf\" and fuenf is None:"),
("Ablesezeit Start statt Ende", "\"zeit\": spanne[1]", "\"zeit\": spanne[0]"),
("nur_ascii fehlt", "print(nur_ascii(f\"| {z", "print((f\"| {z"),
("Ende des Paars nicht gekappt", "min(g[\"ende\"], b[\"zeit\"])", "g[\"ende\"]"),
("Anfang des Paars nicht gekappt", "max(g[\"start\"], a[\"zeit\"])", "g[\"start\"]"),
("Absturz nicht gefangen", "except Exception as e: # Ein Absturz", "except ZeroDivisionError as e: # Ein Absturz"),
("kaputte Zeile bricht ab", " except ValueError:\n befunde.append(f\"{pfad.parent.name}/{pfad.name}: Zeile", " except KeyError:\n befunde.append(f\"{pfad.parent.name}/{pfad.name}: Zeile"),
("seit ignoriert", "ablesungen = [x for x in ablesungen if x[\"zeit\"] >= seit]", "pass"),
("Speicher fehlt still", " return 2\n seit = None", " return 0\n seit = None"),
("Familie aus vollem Namen fehlt", " if f in modell.lower():", " if f == modell.lower():"),
]
def main() -> int:
ueberlebt, veraltet = [], []
for name, alt, neu in MUTANTEN:
with tempfile.TemporaryDirectory() as tmp:
ziel = Path(tmp) / "w"
ziel.mkdir()
for datei in ("verbrauch_reihe.py", "faelle_verbrauch_reihe.py"):
shutil.copy2(HIER / datei, ziel / datei)
pfad = ziel / "verbrauch_reihe.py"
text = pfad.read_text(encoding="utf-8")
if text.count(alt) != 1:
veraltet.append(name)
print(f"VERALTET {name} (Muster {text.count(alt)}-mal)")
continue
pfad.write_text(text.replace(alt, neu), encoding="utf-8", newline="\n")
lauf = subprocess.run([sys.executable, str(ziel / "faelle_verbrauch_reihe.py")], capture_output=True, text=True, encoding="utf-8", errors="replace")
rot = [z for z in lauf.stdout.splitlines() if z.startswith("FEHLER")]
if lauf.returncode == 0:
ueberlebt.append(name)
print(f"UEBERLEBT {name}")
else:
print(f"GEFANGEN {name} ({rot[0][7:80] if rot else 'Absturz'})")
print(f"{len(MUTANTEN)} Mutanten, {len(ueberlebt)} ueberlebt, {len(veraltet)} veraltet")
return 2 if veraltet else 1 if ueberlebt else 0
if __name__ == "__main__":
sys.stdout.reconfigure(errors="replace")
sys.exit(main())
Verifikation
Im Vault-Wurzelordner:
python .vault-tools/workflow-pruefung/faelle_verbrauch_reihe.py
Erwartet: Alle übrigen Zeilen beginnen mit OK, die letzte enthält 0 gescheitert, Exit 0.
python .vault-tools/workflow-pruefung/mutanten_reihe.py
Erwartet: Die letzte Zeile enthält 0 ueberlebt, 0 veraltet, Exit 0.
Danach git status --porcelain -- .vault-tools/workflow-pruefung/verbrauch_reihe.py .vault-tools/workflow-pruefung/faelle_verbrauch_reihe.py .vault-tools/workflow-pruefung/mutanten_reihe.py zeigen.