320 lines
13 KiB
Python
320 lines
13 KiB
Python
"""Die Planung an einen Agenten übergeben.
|
|
|
|
Claude implementiert die geplanten Änderungen im Arbeitsverzeichnis — lokal über
|
|
die Claude-CLI mit dem angemeldeten Konto. Solange er arbeitet, bleibt die
|
|
Planung stehen: die Übersicht sagt, was das Projekt kann, und das stimmt erst,
|
|
wenn der Code da ist. Erst nach Erfolg wird die Planung übernommen. Ein
|
|
Fehlschlag übernimmt nichts; die Planung steht noch und kann erneut los.
|
|
|
|
Der Agent meldet jeden erledigten Schritt als eigene Zeile; der Server liest den
|
|
Stream mit und hakt die Schritte ab. Zustand nur im Speicher — er überlebt keinen
|
|
Serverneustart, der laufende Prozess arbeitet trotzdem zu Ende, übernimmt dann
|
|
aber nichts mehr sichtbar.
|
|
"""
|
|
|
|
import asyncio
|
|
import difflib
|
|
import json
|
|
import re
|
|
|
|
import parser
|
|
import planung
|
|
import projekte
|
|
import verzeichnis
|
|
|
|
# stream-json liefert je Zeile ein Ereignis: sichtbar wird, woran der Agent
|
|
# gerade arbeitet, und die ERLEDIGT-Zeilen kommen an, sobald sie fallen.
|
|
KOMMANDO = ("claude", "-p", "--model", "opus", "--permission-mode", "acceptEdits",
|
|
"--output-format", "stream-json", "--verbose")
|
|
MARKE = "ERLEDIGT:"
|
|
# Einzelne Stream-Zeilen tragen ganze Dateien — die Standardgrenze ist zu klein
|
|
PUFFER = 10 * 1024 * 1024
|
|
|
|
# projektname -> {"laeuft", "fehler", "text", "schritte", "aktivitaet", "uebernommen"}
|
|
_laeufe = {}
|
|
_aufgaben = set()
|
|
|
|
PRUEFEN = "PRÜFEN"
|
|
# Eine neue Funktion ist immer unbewiesen — es gibt noch keinen Code, der sie leistet.
|
|
ART = {"weg": "GESTRICHEN", "geaendert": "GEÄNDERT", "neu": PRUEFEN}
|
|
|
|
AUFTRAG = """Du bekommst die Feature-Karten eines Projekts und die geplanten Änderungen.
|
|
Eine Karte ist ein Auftrag: Regeln sind Zusicherungen, eine Funktion ist etwas,
|
|
das man für sich testen kann. Setze den Code in diesem Verzeichnis so um, dass er
|
|
den geänderten Karten entspricht.
|
|
|
|
- Lies zuerst den bestehenden Code und folge seinen Konventionen.
|
|
- Ändere nur, was die Arbeitsschritte verlangen.
|
|
- {pruefen} heißt: die Funktion gilt als unbewiesen. Prüfe, ob der Code genau das
|
|
leistet, was ihre Testschritte beschreiben. Fehlt sie, baue sie. Weicht sie ab,
|
|
bring sie in Ordnung. Sie kann schon einmal gebaut worden sein und trotzdem falsch
|
|
arbeiten — glaube dem Code nicht, weil er existiert.
|
|
- GEÄNDERT heißt: gebaut, aber die Karte sagt jetzt etwas anderes.
|
|
- GESTRICHEN heißt: die Fähigkeit wird entfernt.
|
|
- Ein Kommando in Anführungszeichen muss danach genau so aufrufbar sein.
|
|
- Führe vorhandene Tests aus, wenn es welche gibt.
|
|
|
|
Die Arbeitsschritte:
|
|
|
|
{schritte}
|
|
|
|
Gib nach jedem abgeschlossenen Schritt sofort eine eigene Textzeile aus,
|
|
mit dem Schritt wörtlich wie oben:
|
|
{marke} <Schritt>
|
|
|
|
Hake nur ab, was du geprüft hast. Konntest du einen Ablauf nicht ausführen — weil
|
|
kein Fenster, kein Gerät oder kein Dienst da ist — dann melde ihn NICHT als erledigt.
|
|
Schreib am Ende in einer eigenen Zeile, was du nicht prüfen konntest und warum.
|
|
|
|
Die betroffenen Karten, wie sie danach gelten:
|
|
|
|
{karten}"""
|
|
|
|
|
|
def _kartentexte(projekt: str, titel: set) -> str:
|
|
"""Der volle Text der geplanten Karten — der Sollzustand, nicht der Diff."""
|
|
texte = []
|
|
for datei in sorted(projekte.wurzel(projekt).glob("*.md")):
|
|
roh = datei.read_text()
|
|
if parser.zerlege(roh)["titel"] in titel:
|
|
texte.append(roh.strip())
|
|
return "\n\n=====\n\n".join(texte) or "(keine — nur Streichungen)"
|
|
|
|
|
|
def _offene(projekt: str) -> set:
|
|
"""(Karte, Funktion) für alles, was noch nicht bewiesen ist.
|
|
|
|
Auch in Karten, die sich gegenüber dem Stand nicht geändert haben: eine Funktion
|
|
zurück auf „offen“ zu setzen ist die Art, eine kaputte Umsetzung zu melden. Sie
|
|
muss dann in die Arbeitsliste, sonst hätte der Klick keine Folge.
|
|
"""
|
|
raus = set()
|
|
for datei in projekte.wurzel(projekt).glob("*.md"):
|
|
s = parser.zerlege(datei.read_text())
|
|
if s["art"] == parser.VISION:
|
|
continue
|
|
for f in s["funktionen"]:
|
|
if f["stand"] == parser.OFFEN:
|
|
raus.add((s["titel"], f["text"]))
|
|
return raus
|
|
|
|
|
|
def _schritte(projekt: str, aenderungen: dict) -> list[dict]:
|
|
"""Die Arbeitsliste: ein Eintrag je Streichung, Kopfänderung und Funktion.
|
|
|
|
Karte und Funktion stehen daneben, nicht nur im Text. Beim Abhaken wird die
|
|
Funktion auf „gebaut“ gesetzt — den Namen aus dem Satz zurückzuparsen wäre
|
|
eine zweite Wahrheit über dieselbe Sache.
|
|
"""
|
|
offen = _offene(projekt)
|
|
raus = []
|
|
erledigt = set()
|
|
|
|
def funktionsschritt(titel: str, name: str, art: str):
|
|
erledigt.add((titel, name))
|
|
# Offen schlägt geändert: „prüfen und herstellen“ deckt beides ab, und der
|
|
# Agent soll dem vorhandenen Code nicht glauben.
|
|
marke = PRUEFEN if (titel, name) in offen else ART[art]
|
|
raus.append({"text": f"{titel} / {name}: {marke}", "karte": titel,
|
|
# Eine gestrichene Funktion gibt es hinterher nicht mehr
|
|
"funktion": "" if art == "weg" else name})
|
|
|
|
for titel in sorted(aenderungen):
|
|
e = aenderungen[titel]
|
|
if e["art"] == "weg":
|
|
raus.append({"text": f"Karte „{titel}“ streichen",
|
|
"karte": titel, "funktion": ""})
|
|
erledigt |= {(titel, n) for n in e.get("funktionen", {})}
|
|
continue
|
|
if e.get("kopf"):
|
|
raus.append({"text": f"Karte „{titel}“: Kartenkopf anpassen",
|
|
"karte": titel, "funktion": ""})
|
|
for name, art in sorted(e.get("funktionen", {}).items()):
|
|
funktionsschritt(titel, name, art)
|
|
|
|
for titel, name in sorted(offen - erledigt):
|
|
funktionsschritt(titel, name, "geaendert")
|
|
return raus
|
|
|
|
|
|
def _visionen(projekt: str) -> set:
|
|
"""Titel der Vision-Karten, in Stand und Arbeitskopie."""
|
|
titel = set()
|
|
for ordner in (projekte.stand(projekt), projekte.wurzel(projekt)):
|
|
for datei in ordner.glob("*.md"):
|
|
s = parser.zerlege(datei.read_text())
|
|
if s["art"] == parser.VISION:
|
|
titel.add(s["titel"])
|
|
return titel
|
|
|
|
|
|
def bau_auftrag(projekt: str) -> tuple[str, list[dict]]:
|
|
"""Prompt und Arbeitsliste, aus dem Stand der laufenden Planung.
|
|
|
|
Die Vision bleibt außen vor: sie beschreibt Absicht, keinen Auftrag —
|
|
eine geänderte Vision ist nichts, was ein Agent bauen soll.
|
|
"""
|
|
if not planung.laeuft(projekt):
|
|
return "", []
|
|
visionen = _visionen(projekt)
|
|
aenderungen = {t: e for t, e in planung.aenderungen(projekt).items()
|
|
if t not in visionen}
|
|
schritte = _schritte(projekt, aenderungen)
|
|
if not schritte:
|
|
return "", []
|
|
# Der Sollzustand jeder Karte, die Arbeit trägt — auch der unveränderten mit
|
|
# einer offenen Funktion. Ohne ihren Text wüsste der Agent nicht, wogegen er prüft.
|
|
bleibend = {s["karte"] for s in schritte if s["karte"] not in {
|
|
t for t, e in aenderungen.items() if e["art"] == "weg"}}
|
|
text = AUFTRAG.format(
|
|
schritte="\n".join(f'- {s["text"]}' for s in schritte),
|
|
marke=MARKE,
|
|
pruefen=PRUEFEN,
|
|
karten=_kartentexte(projekt, bleibend))
|
|
return text, schritte
|
|
|
|
|
|
def status(projekt: str) -> dict:
|
|
return _laeufe.get(projekt) or {
|
|
"laeuft": False, "fehler": "", "text": "",
|
|
"schritte": [], "aktivitaet": "", "uebernommen": False,
|
|
}
|
|
|
|
|
|
# „offen“ steht als eigener Block in der Datei; fehlt er, gilt die Funktion als
|
|
# gebaut. Umstellen heißt darum: den Block streichen. Ein Umbau über den Parser
|
|
# würde die Datei neu schreiben und dabei unbekannte Blöcke verlieren.
|
|
_FUNKTIONSMARKE = re.compile(r"^<!--\s*funktion\s*-->[ \t]*$", re.M | re.I)
|
|
_OFFENBLOCK = re.compile(
|
|
r"^<!--\s*stand\s*-->[ \t]*\n[ \t]*" + parser.OFFEN + r"[ \t]*\n", re.M | re.I)
|
|
|
|
|
|
def _als_gebaut(roh: str, funktion: str) -> str:
|
|
"""Streicht den offen-Vermerk genau dieser Funktion."""
|
|
marken = [m.start() for m in _FUNKTIONSMARKE.finditer(roh)]
|
|
for i, start in enumerate(marken):
|
|
ende = marken[i + 1] if i + 1 < len(marken) else len(roh)
|
|
block = roh[start:ende]
|
|
zeilen = [z.strip() for z in block.splitlines()[1:] if z.strip()]
|
|
titel = next((z for z in zeilen
|
|
if not z.startswith("<!--") and not z.startswith("---")), "")
|
|
if titel != funktion:
|
|
continue
|
|
return roh[:start] + _OFFENBLOCK.sub("", block, count=1) + roh[ende:]
|
|
return roh
|
|
|
|
|
|
def _setze_gebaut(projekt: str, schritt: dict):
|
|
"""Ein abgehakter Schritt heißt: der Code ist da, die Funktion ist gebaut.
|
|
|
|
Geschrieben wird in die Arbeitskopie; `planung.setze_um` trägt sie danach in den
|
|
Projektstand. Scheitert der Agent später, bleibt der Haken stehen — was gebaut
|
|
ist, ist gebaut, und die Planung läuft von dort weiter.
|
|
"""
|
|
if not schritt.get("funktion"):
|
|
return
|
|
for datei in projekte.wurzel(projekt).glob("*.md"):
|
|
roh = datei.read_text()
|
|
if parser.zerlege(roh)["titel"] != schritt.get("karte"):
|
|
continue
|
|
neu = _als_gebaut(roh, schritt["funktion"])
|
|
if neu != roh:
|
|
datei.write_text(neu)
|
|
return
|
|
|
|
|
|
def _hake_ab(projekt: str, lauf: dict, zeile: str):
|
|
"""Eine ERLEDIGT-Zeile gegen die offenen Schritte halten.
|
|
|
|
Wörtlich wäre schön, aber ein Modell zitiert nicht immer wörtlich —
|
|
deshalb genügt große Ähnlichkeit mit genau einem offenen Schritt.
|
|
"""
|
|
gemeldet = zeile.split(MARKE, 1)[1].strip()
|
|
offen = [s for s in lauf["schritte"] if not s["fertig"]]
|
|
for s in offen:
|
|
if s["text"] == gemeldet:
|
|
s["fertig"] = True
|
|
_setze_gebaut(projekt, s)
|
|
return
|
|
naechster = max(
|
|
offen,
|
|
key=lambda s: difflib.SequenceMatcher(None, s["text"], gemeldet).ratio(),
|
|
default=None)
|
|
if naechster and difflib.SequenceMatcher(
|
|
None, naechster["text"], gemeldet).ratio() >= 0.6:
|
|
naechster["fertig"] = True
|
|
_setze_gebaut(projekt, naechster)
|
|
|
|
|
|
def _verarbeite(projekt: str, lauf: dict, roh: bytes):
|
|
"""Ein Stream-Ereignis: Haken setzen, Aktivität zeigen, Ergebnis merken."""
|
|
try:
|
|
ereignis = json.loads(roh)
|
|
except ValueError:
|
|
return
|
|
if ereignis.get("type") == "result":
|
|
lauf["text"] = str(ereignis.get("result") or "").strip()
|
|
return
|
|
if ereignis.get("type") != "assistant":
|
|
return
|
|
for block in (ereignis.get("message") or {}).get("content") or []:
|
|
if block.get("type") == "text":
|
|
for zeile in str(block.get("text") or "").splitlines():
|
|
if MARKE in zeile:
|
|
_hake_ab(projekt, lauf, zeile)
|
|
elif block.get("type") == "tool_use":
|
|
eingabe = block.get("input") or {}
|
|
ziel = (eingabe.get("file_path") or eingabe.get("path")
|
|
or eingabe.get("command") or eingabe.get("pattern") or "")
|
|
lauf["aktivitaet"] = f'{block.get("name", "")} {str(ziel)[:80]}'.strip()
|
|
|
|
|
|
async def _fuehre_aus(projekt: str, auftrag: str, ordner: str):
|
|
lauf = _laeufe[projekt]
|
|
try:
|
|
prozess = await asyncio.create_subprocess_exec(
|
|
*KOMMANDO, cwd=ordner, limit=PUFFER,
|
|
stdin=asyncio.subprocess.PIPE,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE)
|
|
prozess.stdin.write(auftrag.encode())
|
|
await prozess.stdin.drain()
|
|
prozess.stdin.close()
|
|
while zeile := await prozess.stdout.readline():
|
|
_verarbeite(projekt, lauf, zeile)
|
|
fehler = (await prozess.stderr.read()).decode(errors="replace").strip()
|
|
if await prozess.wait() != 0:
|
|
lauf["fehler"] = fehler or f"Agent endete mit Code {prozess.returncode}"
|
|
return
|
|
# Erst jetzt wird die Planung wahr: der Code entspricht den Karten.
|
|
planung.setze_um(projekt)
|
|
lauf["uebernommen"] = True
|
|
except FileNotFoundError:
|
|
lauf["fehler"] = "claude-CLI nicht gefunden — ist Claude Code installiert?"
|
|
except Exception as e: # sichtbar statt still verschluckt
|
|
lauf["fehler"] = str(e)
|
|
finally:
|
|
lauf["aktivitaet"] = ""
|
|
lauf["laeuft"] = False
|
|
|
|
|
|
def starte(projekt: str, auftrag: str, schritte: list[dict]) -> bool:
|
|
"""Startet den Agenten im Hintergrund. False, wenn nichts zu tun ist."""
|
|
if not auftrag:
|
|
return False
|
|
ordner = verzeichnis.lies(projekt)
|
|
if not ordner:
|
|
return False
|
|
if status(projekt)["laeuft"]:
|
|
raise ValueError("Für dieses Projekt arbeitet schon ein Agent")
|
|
_laeufe[projekt] = {
|
|
"laeuft": True, "fehler": "", "text": "",
|
|
"schritte": [{**s, "fertig": False} for s in schritte],
|
|
"aktivitaet": "", "uebernommen": False,
|
|
}
|
|
aufgabe = asyncio.get_event_loop().create_task(
|
|
_fuehre_aus(projekt, auftrag, ordner))
|
|
_aufgaben.add(aufgabe)
|
|
aufgabe.add_done_callback(_aufgaben.discard)
|
|
return True
|