"""pipeline.yaml laden + statisch validieren (R8): Normal-Kanten müssen ein DAG sein (graphlib), Rück-/Reparatur-Kanten nur von Gates und stufen-abwärts, Worker registriert, Skill-Dateien vorhanden, genau ein Gate je Stufe (außer letzter). Liefert auch das vorberechnete Layout für die Graphensicht.""" import graphlib from dataclasses import dataclass, field import yaml from . import config, skills class GraphFehler(Exception): pass @dataclass class Knoten: id: str typ: str # system | llm | gate worker: str stufe: str titel: str = "" beschreibung: str = "" skills: list = field(default_factory=list) extra_skills: dict = field(default_factory=dict) # name → Skill-Liste klasse: str = "lang" # lang | mittel | kurz → Bündel-Größe je Call timeout: list | None = None # [basis_s, pro_item_s] runden_max: int = 3 # nur Gates: Runden-Cap max_parallel: int = 4 max_versuche: int = config.MAX_VERSUCHE barriere: bool = False @property def buendel(self) -> int: return config.BUENDEL[self.klasse] @dataclass class Graph: stufen: list[str] knoten: dict[str, Knoten] kanten: list[dict] # {von, nach, art} def stufen_index(self, stufe: str) -> int: return self.stufen.index(stufe) def gate_der_stufe(self, stufe: str) -> Knoten | None: return next((k for k in self.knoten.values() if k.typ == "gate" and k.stufe == stufe), None) def normal_nachfolger(self, knoten_id: str) -> set[str]: return {k["nach"] for k in self.kanten if k["von"] == knoten_id and k["art"] == "normal"} def erlaubte_ziele(self, knoten_id: str) -> set[str]: """Alle Knoten, für die dieser Knoten Tasks erzeugen darf.""" return {k["nach"] for k in self.kanten if k["von"] == knoten_id} def layout(self) -> dict: """Fluss-Layout (R7): Zeile = längster Pfad über Normal-Kanten, Spalte = Spur innerhalb der Zeile. Reine Reparatur-Ziele (ohne Normal-Kanten) rücken neben ihr Gate.""" vor: dict[str, list[str]] = {k: [] for k in self.knoten} for ka in self.kanten: if ka["art"] == "normal": vor[ka["nach"]].append(ka["von"]) ebene: dict[str, int] = {} for kid in graphlib.TopologicalSorter( {k: set(v) for k, v in vor.items()}).static_order(): ebene[kid] = max((ebene[v] + 1 for v in vor[kid]), default=0) an_normal = {ka[e] for ka in self.kanten if ka["art"] == "normal" for e in ("von", "nach")} for ka in self.kanten: if ka["art"] != "normal" and ka["nach"] not in an_normal: ebene[ka["nach"]] = ebene[ka["von"]] pos = {} spuren: dict[int, int] = {} for kid in sorted(ebene, key=lambda i: ebene[i]): k = self.knoten[kid] spur = spuren.get(ebene[kid], 0) spuren[ebene[kid]] = spur + 1 pos[kid] = {"zeile": ebene[kid], "spalte": spur, "typ": k.typ, "barriere": k.barriere, "stufe": k.stufe, "titel": k.titel or kid, "beschreibung": k.beschreibung, "skills": [{"name": s, "art": skills.lade(s)["art"]} for s in k.skills]} return {"stufen": self.stufen, "knoten": pos, "kanten": self.kanten} _graph: Graph | None = None def laden(worker_namen: set[str] | None = None, pfad=None) -> Graph: global _graph daten = yaml.safe_load((pfad or config.PIPELINE_YAML).read_text()) stufen = daten["stufen"] knoten = {} for k in daten["knoten"]: kn = Knoten(**{**k, "skills": k.get("skills", [])}) if kn.id in knoten: raise GraphFehler(f"Knoten-id doppelt: {kn.id}") if kn.stufe not in stufen: raise GraphFehler(f"{kn.id}: unbekannte Stufe {kn.stufe}") if kn.typ not in ("system", "llm", "gate"): raise GraphFehler(f"{kn.id}: unbekannter Typ {kn.typ}") if kn.klasse not in config.BUENDEL: raise GraphFehler(f"{kn.id}: unbekannte Klasse {kn.klasse}") knoten[kn.id] = kn kanten = daten["kanten"] for kante in kanten: for ende in ("von", "nach"): if kante[ende] not in knoten: raise GraphFehler(f"Kante {kante}: unbekannter Knoten {kante[ende]}") if kante["art"] not in ("normal", "reparatur", "ruecklauf"): raise GraphFehler(f"Kante {kante}: unbekannte Art") if kante["art"] != "normal": von = knoten[kante["von"]] if von.typ != "gate": raise GraphFehler(f"{kante['art']}-Kante nur von Gates: {kante}") if stufen.index(knoten[kante["nach"]].stufe) > stufen.index(von.stufe): raise GraphFehler(f"{kante['art']}-Kante muss stufen-abwärts zeigen: {kante}") # Normal-Kanten: DAG-Pflicht (graphlib erwartet Vorgänger-Format) vorgaenger: dict[str, set] = {k: set() for k in knoten} for kante in kanten: if kante["art"] == "normal": vorgaenger[kante["nach"]].add(kante["von"]) try: graphlib.TopologicalSorter(vorgaenger).prepare() except graphlib.CycleError as e: raise GraphFehler(f"Zyklus in Normal-Kanten: {e}") from e for stufe in stufen[:-1]: gates = [k for k in knoten.values() if k.typ == "gate" and k.stufe == stufe] if len(gates) != 1: raise GraphFehler(f"Stufe {stufe}: genau 1 Gate nötig, {len(gates)} gefunden") for kn in knoten.values(): if kn.typ == "llm" and not kn.skills: raise GraphFehler(f"{kn.id}: llm-Knoten ohne Skills") for s in kn.skills + [s for liste in kn.extra_skills.values() for s in liste]: skills.lade(s) # wirft SkillFehler, wenn Datei fehlt/kaputt if worker_namen is not None and kn.worker not in worker_namen: raise GraphFehler(f"{kn.id}: Worker '{kn.worker}' nicht registriert") _graph = Graph(stufen=stufen, knoten=knoten, kanten=kanten) return _graph def get() -> Graph: if _graph is None: raise GraphFehler("Graph nicht geladen") return _graph def skills_von(knoten_id: str, extra: str | None = None) -> list: """EINE Wahrheit für Skill-Kompositionen: pipeline.yaml. Worker fragen hier nach, statt Listen zu hardcoden (Drift bricht sonst erst zur Laufzeit).""" kn = get().knoten[knoten_id] if extra is None: return kn.skills if extra not in kn.extra_skills: raise GraphFehler(f"{knoten_id}: extra_skills '{extra}' nicht deklariert") return kn.extra_skills[extra] def timeout_fuer(knoten_id: str, n: int = 0) -> float: kn = _graph.knoten.get(knoten_id) if _graph else None basis, pro = (kn.timeout if kn and kn.timeout else config.TIMEOUT_DEFAULT) return basis + pro * n