From f2da893f7137e59a01e4acf3bb9b6c085f0795e6 Mon Sep 17 00:00:00 2001 From: Robert Rapp Date: Mon, 12 Oct 2026 00:43:26 +0200 Subject: [PATCH] core: Policy, Pflege-Agent (HITL), Healer, Lifecycle-Pipeline, Graduierung + Konfiguration Co-Authored-By: Claude Opus 5.5 --- config/healer-rules.yaml | 53 +++ config/lifecycle.yaml | 6 + config/maintenance.yaml | 11 + config/policy.yaml | 20 + core/src/knaster/graduation.py | 57 +++ core/src/knaster/healer/__init__.py | 1 + core/src/knaster/healer/drivers.py | 124 +++++++ core/src/knaster/healer/engine.py | 165 +++++++++ core/src/knaster/lifecycle.py | 246 +++++++++++++ core/src/knaster/maintenance/__init__.py | 1 + core/src/knaster/maintenance/actions.py | 127 +++++++ core/src/knaster/maintenance/agent.py | 448 +++++++++++++++++++++++ core/src/knaster/metrics.py | 39 ++ core/src/knaster/policy.py | 48 +++ core/src/knaster/registry.py | 11 + core/src/knaster/router.py | 20 +- core/src/knaster/runner.py | 6 +- core/src/knaster/store.py | 8 +- core/src/knaster/system.py | 121 ++++++ core/tests/conftest.py | 43 +++ core/tests/test_lifecycle.py | 119 ++++++ core/tests/test_maintenance.py | 210 +++++++++++ core/tests/test_policy_healer.py | 124 +++++++ docs/spec/SPEC-maintenance.md | 14 +- sites/knaster/registry.yaml | 76 ++++ skills/rechnung-extrahieren/SKILL.md | 47 +++ skills/ticket-kategorisieren/SKILL.md | 22 ++ skills/zusammenfassung/SKILL.md | 22 ++ 28 files changed, 2179 insertions(+), 10 deletions(-) create mode 100644 config/healer-rules.yaml create mode 100644 config/lifecycle.yaml create mode 100644 config/maintenance.yaml create mode 100644 config/policy.yaml create mode 100644 core/src/knaster/graduation.py create mode 100644 core/src/knaster/healer/__init__.py create mode 100644 core/src/knaster/healer/drivers.py create mode 100644 core/src/knaster/healer/engine.py create mode 100644 core/src/knaster/lifecycle.py create mode 100644 core/src/knaster/maintenance/__init__.py create mode 100644 core/src/knaster/maintenance/actions.py create mode 100644 core/src/knaster/maintenance/agent.py create mode 100644 core/src/knaster/metrics.py create mode 100644 core/src/knaster/policy.py create mode 100644 core/src/knaster/system.py create mode 100644 core/tests/test_lifecycle.py create mode 100644 core/tests/test_maintenance.py create mode 100644 core/tests/test_policy_healer.py create mode 100644 sites/knaster/registry.yaml create mode 100644 skills/rechnung-extrahieren/SKILL.md create mode 100644 skills/ticket-kategorisieren/SKILL.md create mode 100644 skills/zusammenfassung/SKILL.md diff --git a/config/healer-rules.yaml b/config/healer-rules.yaml new file mode 100644 index 0000000..b29bc77 --- /dev/null +++ b/config/healer-rules.yaml @@ -0,0 +1,53 @@ +# Deterministisches Autohealing (SPEC-healer). {label} wird aus den Alert-Labels ersetzt. +max_actions_per_hour: 10 +# Infrastruktur-Aktionen laufen nur, wenn KNASTER_DRY_RUN=false UND sie hier stehen. +enabled_actions: [] +rules: + - id: slot-endpoint-down + match: {alertname: SlotEndpointDown} + ladder: + - {action: restart_container, params: {host: "{docker_host}", container: "{container}"}, attempts: 2, cooldown: 5m} + - {action: restart_vm, params: {pve_node: "{pve_node}", vmid: "{vmid}"}, attempts: 1, cooldown: 15m} + - {action: mark_slot, params: {slot: "{slot}", status: degraded}, attempts: 1} + on_resolved: + - {action: mark_slot, params: {slot: "{slot}", status: ok}, only_if_set_by_healer: true} + + - id: slot-canary-failing + match: {alertname: SlotCanaryFailing} + ladder: + - {action: restart_container, params: {host: "{docker_host}", container: "{container}"}, attempts: 1, cooldown: 10m} + - {action: mark_slot, params: {slot: "{slot}", status: degraded}, attempts: 1} + on_resolved: + - {action: mark_slot, params: {slot: "{slot}", status: ok}, only_if_set_by_healer: true} + + - id: gpu-xid + match: {alertname: GpuXidError} + ladder: + - {action: mark_slot, params: {slot: "{slot}", status: degraded}, attempts: 1} + - {action: restart_vm, params: {pve_node: "{pve_node}", vmid: "{vmid}"}, attempts: 1, cooldown: 20m} + # ältere Karten ohne FLR brauchen ggf. einen Host-Neustart – nur mit Freigabe/Wartungsfenster + - {action: reboot_host, params: {pve_node: "{pve_node}"}, attempts: 1} + on_resolved: + - {action: mark_slot, params: {slot: "{slot}", status: ok}, only_if_set_by_healer: true} + + - id: gpu-hot + match: {alertname: GpuTemperatureHigh} + ladder: + - {action: mark_slot, params: {slot: "{slot}", status: degraded}, attempts: 1} + on_resolved: + - {action: mark_slot, params: {slot: "{slot}", status: ok}, only_if_set_by_healer: true} + + - id: kv-cache-saturated + match: {alertname: VllmKvCacheSaturated} + ladder: + - {action: notify, params: {text: "KV-Cache von {slot} dauerhaft voll – Last verteilen oder max_model_len senken", severity: warning}, attempts: 1} + + - id: node-down + match: {alertname: NodeDown} + ladder: + - {action: notify, params: {text: "Node {instance} nicht erreichbar – Proxmox HA prüfen", severity: high}, attempts: 1} + + - id: disk-full + match: {alertname: DiskAlmostFull} + ladder: + - {action: notify, params: {text: "Platte auf {instance} ({mountpoint}) fast voll", severity: warning}, attempts: 1} diff --git a/config/lifecycle.yaml b/config/lifecycle.yaml new file mode 100644 index 0000000..84b3fc9 --- /dev/null +++ b/config/lifecycle.yaml @@ -0,0 +1,6 @@ +# Grenzwerte der Lifecycle-Pipeline (SPEC-lifecycle §2) +tolerance: 0.02 +min_pass_rate: 0.85 +min_agreement: 0.85 +min_n: 50 +canary_share: 0.1 diff --git a/config/maintenance.yaml b/config/maintenance.yaml new file mode 100644 index 0000000..bbb4d93 --- /dev/null +++ b/config/maintenance.yaml @@ -0,0 +1,11 @@ +# Grenzwerte der Service-Routine (SPEC-maintenance §3) +demote_below: 0.80 +min_cases: 20 +window_cases: 50 +promote_min_rate: 0.95 +merge_similarity: 0.92 +degraded_hours: 1 +context_signal_limit: 10 +adapter_min_cases: 300 +adapter_target_rate: 0.90 +lookback_days: 7 diff --git a/config/policy.yaml b/config/policy.yaml new file mode 100644 index 0000000..b68f3b2 --- /dev/null +++ b/config/policy.yaml @@ -0,0 +1,20 @@ +# Freigabe-Richtlinie (ADR-0004): sichere Richtung automatisch, Qualität/Kosten nach Freigabe. +default: approve +actions: + # automatisch + demote_skill: {mode: auto, risk: low} + mark_slot: {mode: auto, risk: low} + restart_container: {mode: auto, risk: low} + restart_vm: {mode: auto, risk: medium} + recalibrate_context: {mode: auto, risk: low} + notify: {mode: auto, risk: low} + # mit Freigabe + request_graduation: {mode: approve, risk: low} + promote_skill: {mode: approve, risk: medium} + merge_skills: {mode: approve, risk: low} + retire_skill: {mode: approve, risk: medium} + train_adapter: {mode: approve, risk: low} + create_candidate: {mode: approve, risk: medium} + promote_candidate: {mode: approve, risk: high} + rollback_candidate: {mode: approve, risk: medium} + reboot_host: {mode: approve, risk: high, window: "02:00-05:00"} diff --git a/core/src/knaster/graduation.py b/core/src/knaster/graduation.py new file mode 100644 index 0000000..a649895 --- /dev/null +++ b/core/src/knaster/graduation.py @@ -0,0 +1,57 @@ +"""Graduierung: Schafft das schwache Modell einen Skill zuverlässig? Dann Vorschlag für Stufe 1.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + +from .cases import split_for +from .executor import Executor +from .router import Router +from .skills import SkillLibrary +from .store import Store +from .verifier import Verifier + + +class GraduationReport(BaseModel): + skill: str + n: int + passed_n: int + pass_rate: float + threshold: float + graduated: bool + failures: list[dict] = Field(default_factory=list) + + +def graduate(skill_name: str, variants: list[str], *, router: Router, executor: Executor, verifier: Verifier, + library: SkillLibrary, store: Store, threshold: float = 0.9, min_variants: int = 10 + ) -> GraduationReport: + skill = library.get(skill_name) + passed_n = 0 + failures: list[dict] = [] + for text in variants: + decision = router.plan_stage1(text, skill_name) + if decision.stage != 1 or decision.status != "routed": + failures.append({"input": text, "reason": "kein Stufe-1-Slot", "reasons": decision.reasons}) + continue + execution = executor.execute(decision, text) + ok = False + detail = [s.value for s in execution.signals] + verification = None + if not execution.signals: + verification = verifier.verify(skill, text, execution.output) + ok = verification.passed and not verification.inconclusive + detail = [c.detail for c in verification.failed_checks] or (["nicht entscheidbar"] if not ok else []) + store.add_case(input=text, skill=skill.name, skill_version=skill.meta.version, cluster=skill.cluster, + stage=1, slot=decision.slot, model=decision.model, adapter=decision.adapter, + messages=execution.messages, output=execution.output, + passed=None if verification and verification.inconclusive else ok, + signals=[s.value for s in execution.signals], prompt_tokens=decision.estimated_tokens, + split=split_for(text), origin="graduation") + if ok: + passed_n += 1 + else: + failures.append({"input": text, "output": execution.output[:500], "detail": detail}) + n = len(variants) + rate = passed_n / n if n else 0.0 + return GraduationReport(skill=skill_name, n=n, passed_n=passed_n, pass_rate=rate, threshold=threshold, + graduated=n >= min_variants and rate >= threshold, failures=failures) diff --git a/core/src/knaster/healer/__init__.py b/core/src/knaster/healer/__init__.py new file mode 100644 index 0000000..31455d1 --- /dev/null +++ b/core/src/knaster/healer/__init__.py @@ -0,0 +1 @@ +"""Deterministisches Autohealing.""" diff --git a/core/src/knaster/healer/drivers.py b/core/src/knaster/healer/drivers.py new file mode 100644 index 0000000..284d822 --- /dev/null +++ b/core/src/knaster/healer/drivers.py @@ -0,0 +1,124 @@ +"""Treiber für Infrastruktur-Aktionen (Portainer, Proxmox, Webhook) mit Dry-Run.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Protocol + +import httpx + + +class InfraDriver(Protocol): + def restart_container(self, host: str, container: str) -> str: ... + def restart_vm(self, pve_node: str, vmid: str) -> str: ... + def reboot_host(self, pve_node: str) -> str: ... + + +class Notifier(Protocol): + def notify(self, text: str, severity: str = "info", data: dict[str, Any] | None = None) -> str: ... + + +@dataclass +class DryRunDriver: + """Protokolliert nur. Standard bis zur bewussten Freischaltung.""" + + calls: list[tuple[str, dict[str, Any]]] = field(default_factory=list) + + def _log(self, action: str, **kw: Any) -> str: + self.calls.append((action, kw)) + args = ", ".join(f"{k}={v}" for k, v in kw.items()) + return f"[dry-run] {action}({args})" + + def restart_container(self, host: str, container: str) -> str: + return self._log("restart_container", host=host, container=container) + + def restart_vm(self, pve_node: str, vmid: str) -> str: + return self._log("restart_vm", pve_node=pve_node, vmid=vmid) + + def reboot_host(self, pve_node: str) -> str: + return self._log("reboot_host", pve_node=pve_node) + + def notify(self, text: str, severity: str = "info", data: dict[str, Any] | None = None) -> str: + return self._log("notify", severity=severity, text=text) + + +class PortainerDriver: + """Container-Neustart über die Portainer-API. `host` = Endpoint-ID oder Endpoint-Name.""" + + def __init__(self, base_url: str, api_key: str, client: httpx.Client | None = None): + self.base_url = base_url.rstrip("/") + self.client = client or httpx.Client(timeout=30.0, follow_redirects=True) + self.headers = {"X-API-Key": api_key} + + def _endpoint_id(self, host: str) -> str: + if host.isdigit(): + return host + resp = self.client.get(f"{self.base_url}/api/endpoints", headers=self.headers) + resp.raise_for_status() + for ep in resp.json(): + if ep.get("Name") == host: + return str(ep["Id"]) + raise LookupError(f"Portainer-Endpoint {host} nicht gefunden") + + def restart_container(self, host: str, container: str) -> str: + ep = self._endpoint_id(host) + resp = self.client.post( + f"{self.base_url}/api/endpoints/{ep}/docker/containers/{container}/restart", headers=self.headers + ) + resp.raise_for_status() + return f"Container {container} auf {host} neu gestartet" + + +class ProxmoxDriver: + def __init__(self, base_url: str, token_id: str, token_secret: str, verify_tls: bool = True, + client: httpx.Client | None = None): + self.base_url = base_url.rstrip("/") + self.client = client or httpx.Client(timeout=30.0, verify=verify_tls) + self.headers = {"Authorization": f"PVEAPIToken={token_id}={token_secret}"} + + def _post(self, path: str, data: dict[str, Any] | None = None) -> Any: + resp = self.client.post(f"{self.base_url}/api2/json{path}", headers=self.headers, data=data) + resp.raise_for_status() + return resp.json().get("data") + + def restart_vm(self, pve_node: str, vmid: str) -> str: + task = self._post(f"/nodes/{pve_node}/qemu/{vmid}/status/reboot") + return f"VM {vmid} auf {pve_node} Neustart angestoßen ({task})" + + def reboot_host(self, pve_node: str) -> str: + self._post(f"/nodes/{pve_node}/status", {"command": "reboot"}) + return f"Host {pve_node} Neustart angestoßen" + + +class WebhookNotifier: + def __init__(self, url: str, client: httpx.Client | None = None): + self.url = url + self.client = client or httpx.Client(timeout=15.0) + + def notify(self, text: str, severity: str = "info", data: dict[str, Any] | None = None) -> str: + resp = self.client.post(self.url, json={"text": text, "severity": severity, "data": data or {}}) + resp.raise_for_status() + return "benachrichtigt" + + +class CompositeInfra: + """Verteilt auf Portainer (Container) und Proxmox (VM/Host).""" + + def __init__(self, portainer: PortainerDriver | None, proxmox: ProxmoxDriver | None): + self.portainer = portainer + self.proxmox = proxmox + + def restart_container(self, host: str, container: str) -> str: + if not self.portainer: + raise RuntimeError("Portainer nicht konfiguriert") + return self.portainer.restart_container(host, container) + + def restart_vm(self, pve_node: str, vmid: str) -> str: + if not self.proxmox: + raise RuntimeError("Proxmox nicht konfiguriert") + return self.proxmox.restart_vm(pve_node, vmid) + + def reboot_host(self, pve_node: str) -> str: + if not self.proxmox: + raise RuntimeError("Proxmox nicht konfiguriert") + return self.proxmox.reboot_host(pve_node) diff --git a/core/src/knaster/healer/engine.py b/core/src/knaster/healer/engine.py new file mode 100644 index 0000000..8dcc61c --- /dev/null +++ b/core/src/knaster/healer/engine.py @@ -0,0 +1,165 @@ +"""Regel-Engine: Alert → Eskalationsleiter von Aktionen mit Versuchsbudget, Cooldown und globaler Bremse.""" + +from __future__ import annotations + +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from pydantic import BaseModel, Field + +from ..maintenance.actions import ActionError, ActionExecutor +from ..policy import Policy +from ..store import Store +from ..util import MissingLabel, fill_template, load_yaml, now, parse_duration + + +class Step(BaseModel): + action: str + params: dict[str, Any] = Field(default_factory=dict) + attempts: int = 1 + cooldown: str | float = 0 + only_if_set_by_healer: bool = False + + +class Rule(BaseModel): + id: str + match: dict[str, str] + ladder: list[Step] + on_exhausted: str = "notify" + on_resolved: list[Step] = Field(default_factory=list) + + def matches(self, labels: dict[str, str]) -> bool: + return all(labels.get(k) == v for k, v in self.match.items()) + + +class HealerConfig(BaseModel): + rules: list[Rule] = Field(default_factory=list) + max_actions_per_hour: int = 10 + enabled_actions: list[str] = Field(default_factory=list) + + @classmethod + def from_file(cls, path: str | Path) -> HealerConfig: + return cls(**load_yaml(path)) + + +class Outcome(BaseModel): + fingerprint: str + rule: str | None + status: str # executed | proposed | cooldown | exhausted | unmatched | throttled | resolved | skipped | failed + action: str | None = None + message: str = "" + + +class Healer: + def __init__(self, config: HealerConfig, store: Store, actions: ActionExecutor, policy: Policy, + propose: Callable[..., Any] | None = None, clock: Callable[[], float] = now): + self.config = config + self.store = store + self.actions = actions + self.policy = policy + self.propose = propose + self.clock = clock + + def handle_alertmanager(self, payload: dict[str, Any]) -> list[Outcome]: + return [self.handle_alert(a) for a in payload.get("alerts", [])] + + def handle_alert(self, alert: dict[str, Any]) -> Outcome: + labels = {k: str(v) for k, v in (alert.get("labels") or {}).items()} + fp = alert.get("fingerprint") or "|".join(f"{k}={v}" for k, v in sorted(labels.items())) + rule = next((r for r in self.config.rules if r.matches(labels)), None) + if alert.get("status") == "resolved": + return self._resolved(rule, fp, labels) + if rule is None: + msg = self._notify(f"Unbekannter Alert {labels.get('alertname')}: {labels}", "warning") + return Outcome(fingerprint=fp, rule=None, status="unmatched", message=msg) + return self._firing(rule, fp, labels) + + # --- intern ----------------------------------------------------------------------------- + def _key(self, rule: Rule, fp: str) -> str: + return f"{rule.id}|{fp}" + + def _throttled(self) -> bool: + return self.store.count_events("healer_action", since=self.clock() - 3600) >= self.config.max_actions_per_hour + + def _notify(self, text: str, severity: str) -> str: + try: + return self.actions.execute("notify", {"text": text, "severity": severity})["message"] + except ActionError as exc: + return f"Benachrichtigung fehlgeschlagen: {exc}" + + def _firing(self, rule: Rule, fp: str, labels: dict[str, str]) -> Outcome: + key = self._key(rule, fp) + state = self.store.get_healer_state(key) or {"used": [0] * len(rule.ladder), "last": 0.0, + "cooldown": 0.0, "notified": False, "set_slot": None} + if self.clock() - state["last"] < state["cooldown"]: + return Outcome(fingerprint=fp, rule=rule.id, status="cooldown") + + for i, step in enumerate(rule.ladder): + if state["used"][i] >= step.attempts: + continue + state["used"][i] += 1 + state["last"] = self.clock() + state["cooldown"] = parse_duration(step.cooldown) + try: + params = fill_template(step.params, labels) + except MissingLabel as exc: + self.store.set_healer_state(key, state) + return Outcome(fingerprint=fp, rule=rule.id, status="skipped", action=step.action, + message=f"Label {exc} fehlt") + mode = self.policy.decide(step.action) + if mode == "deny": + self.store.set_healer_state(key, state) + return Outcome(fingerprint=fp, rule=rule.id, status="skipped", action=step.action, + message="durch Richtlinie verboten") + if mode == "approve" and self.propose is not None: + p = self.propose(step.action, params, f"Healer: {step.action} für {labels.get('alertname')}", + reason=f"Regel {rule.id}", source="healer", force_approval=True) + self.store.set_healer_state(key, state) + return Outcome(fingerprint=fp, rule=rule.id, status="proposed", action=step.action, + message=f"Vorschlag {p.id}") + if self._throttled(): + self.store.set_healer_state(key, state) + msg = self._notify(f"Healer gebremst (>{self.config.max_actions_per_hour}/h): {rule.id}", "high") + return Outcome(fingerprint=fp, rule=rule.id, status="throttled", action=step.action, message=msg) + if step.action == "mark_slot": + params = {**params, "set_by": "healer", "reason": f"Healer-Regel {rule.id}"} + state["set_slot"] = params.get("slot") + try: + result = self.actions.execute(step.action, params) + status, message = "executed", result["message"] + except ActionError as exc: + status, message = "failed", str(exc) + self.store.add_event("healer_action", subject=rule.id, data={"action": step.action, "status": status}, + ts=self.clock()) + self.store.set_healer_state(key, state) + return Outcome(fingerprint=fp, rule=rule.id, status=status, action=step.action, message=message) + + if not state["notified"]: + state["notified"] = True + self.store.set_healer_state(key, state) + msg = self._notify(f"Healer-Regel {rule.id} erschöpft – bitte eingreifen. Labels: {labels}", "high") + return Outcome(fingerprint=fp, rule=rule.id, status="exhausted", message=msg) + return Outcome(fingerprint=fp, rule=rule.id, status="exhausted", message="bereits gemeldet") + + def _resolved(self, rule: Rule | None, fp: str, labels: dict[str, str]) -> Outcome: + if rule is None: + return Outcome(fingerprint=fp, rule=None, status="resolved") + key = self._key(rule, fp) + state = self.store.get_healer_state(key) or {} + messages = [] + for step in rule.on_resolved: + try: + params = fill_template(step.params, labels) + except MissingLabel: + continue + if step.only_if_set_by_healer and step.action == "mark_slot": + if state.get("set_slot") != params.get("slot"): + continue + params = {**params, "set_by": "healer", "reason": "Alert behoben"} + try: + messages.append(self.actions.execute(step.action, params)["message"]) + except ActionError as exc: + messages.append(str(exc)) + self.store.delete_healer_state(key) + return Outcome(fingerprint=fp, rule=rule.id, status="resolved", message="; ".join(messages)) diff --git a/core/src/knaster/lifecycle.py b/core/src/knaster/lifecycle.py new file mode 100644 index 0000000..26d5683 --- /dev/null +++ b/core/src/knaster/lifecycle.py @@ -0,0 +1,246 @@ +"""Lifecycle-Pipeline: ein Ablauf für Basismodell, Adapter, Quantisierung, Embedding-Modell, Engine.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, Literal, Protocol + +from pydantic import BaseModel + +from .cases import pass_rate +from .registry import AdapterSpec, Registry +from .store import Store +from .util import load_yaml, now + +Kind = Literal["base_model", "adapter", "quantization", "embedding_model", "engine"] +TERMINAL = {"promoted", "rejected", "rolled_back", "failed"} + +# erlaubte Übergänge durch Job-Reports +REPORT_FLOW = { + "build": ("building", "evaluating"), + "eval": ("evaluating", "shadow"), + "shadow": ("shadow", "canary"), + "canary": ("canary", "awaiting_promotion"), +} + + +class LifecycleConfig(BaseModel): + tolerance: float = 0.02 + min_pass_rate: float = 0.85 # wenn keine Baseline bekannt ist + min_agreement: float = 0.85 + min_n: int = 50 + canary_share: float = 0.1 + + @classmethod + def from_file(cls, path: str | Path) -> LifecycleConfig: + return cls(**load_yaml(path)) + + +class JobRunner(Protocol): + def submit(self, job: str, candidate: dict[str, Any]) -> str: ... + + +class LoggingJobRunner: + """Dry-Run: merkt sich Aufträge; im Betrieb ersetzt durch Webhook an n8n/Runner.""" + + def __init__(self) -> None: + self.jobs: list[tuple[str, dict[str, Any]]] = [] + + def submit(self, job: str, candidate: dict[str, Any]) -> str: + self.jobs.append((job, candidate)) + return f"[dry-run] Job {job} für Kandidat {candidate['id']}" + + +class WebhookJobRunner: + def __init__(self, url: str, client: Any | None = None): + import httpx + + self.url = url + self.client = client or httpx.Client(timeout=15.0) + + def submit(self, job: str, candidate: dict[str, Any]) -> str: + resp = self.client.post(self.url, json={"job": job, "candidate": candidate}) + resp.raise_for_status() + return f"Job {job} übergeben" + + +class LifecycleError(Exception): + pass + + +class Lifecycle: + def __init__(self, store: Store, registry: Registry, jobs: JobRunner, config: LifecycleConfig | None = None): + self.store = store + self.registry = registry + self.jobs = jobs + self.config = config or LifecycleConfig() + + # --- Anlegen / Starten ------------------------------------------------------------------ + def create(self, kind: Kind, slot: str, target: dict[str, Any], baseline: dict[str, Any] | None = None) -> str: + if not self.registry.has_slot(slot): + raise LifecycleError(f"Unbekannter Slot {slot}") + if kind == "adapter" and not target.get("cluster"): + raise LifecycleError("Adapter-Kandidat braucht target.cluster") + if kind == "base_model" and not target.get("base_model"): + raise LifecycleError("Basismodell-Kandidat braucht target.base_model") + if baseline is None: + baseline = {"pass_rate": self._baseline(kind, slot, target)} + return self.store.add_candidate(kind=kind, slot=slot, target=target, baseline=baseline, state="proposed", + metrics={}, history=[], snapshot=None) + + def _baseline(self, kind: str, slot: str, target: dict[str, Any]) -> float | None: + rows = [r for r in self.store.list_cases(stage=1, split="eval") if r["slot"] == slot] + if kind == "adapter": + rows = [r for r in rows if r["cluster"] == target["cluster"]] + return pass_rate(rows) + + def get(self, cid: str) -> dict[str, Any]: + cand = self.store.get_candidate(cid) + if cand is None: + raise LifecycleError(f"Kandidat {cid} unbekannt") + return cand + + def _move(self, cand: dict[str, Any], to: str, note: str = "", **extra: Any) -> dict[str, Any]: + history = list(cand.get("history") or []) + history.append({"ts": now(), "from": cand["state"], "to": to, "note": note}) + self.store.update_candidate(cand["id"], state=to, history=history, **extra) + return self.get(cand["id"]) + + def start(self, cid: str) -> dict[str, Any]: + cand = self.get(cid) + if cand["state"] != "proposed": + raise LifecycleError(f"Kandidat ist {cand['state']}, nicht proposed") + cand = self._move(cand, "building", "freigegeben") + self.jobs.submit("build", cand) + return cand + + # --- Job-Reports ------------------------------------------------------------------------ + def report(self, cid: str, step: str, payload: dict[str, Any]) -> dict[str, Any]: + cand = self.get(cid) + if payload.get("error"): + self._cleanup_canary(cand) + return self._move(cand, "failed", f"{step}: {payload['error']}") + if step not in REPORT_FLOW: + raise LifecycleError(f"Unbekannter Schritt {step}") + expected, nxt = REPORT_FLOW[step] + if cand["state"] != expected: + raise LifecycleError(f"Report {step} passt nicht zum Zustand {cand['state']}") + metrics = {**(cand.get("metrics") or {}), step: payload} + ok, why = self._gate(cand, step, payload) + if not ok: + self._cleanup_canary(cand) + return self._move(cand, "rejected", why, metrics=metrics) + target = dict(cand["target"]) + if step == "build" and payload.get("model"): + target["model"] = payload["model"] # Name, unter dem der Kandidat bereitsteht + cand = self._move(cand, nxt, why, metrics=metrics, target=target) + if nxt == "canary": + self._start_canary(cand) + elif nxt in {"evaluating", "shadow"}: + self.jobs.submit({"evaluating": "eval", "shadow": "shadow"}[nxt], cand) + elif nxt == "awaiting_promotion": + self._cleanup_canary(cand) + return cand + + def _gate(self, cand: dict[str, Any], step: str, p: dict[str, Any]) -> tuple[bool, str]: + c = self.config + if step == "build": + return bool(p.get("ok", True)), "Build ok" + n = int(p.get("n", 0)) + if step == "shadow": + agreement = float(p.get("agreement", 0)) + ok = agreement >= c.min_agreement and n >= c.min_n + return ok, f"Shadow-Übereinstimmung {agreement:.2f} (n={n})" + rate = float(p.get("pass_rate", 0)) + base = (cand.get("baseline") or {}).get("pass_rate") + needed = c.min_pass_rate if base is None else base - c.tolerance + min_n = c.min_n if step == "canary" else 1 + ok = rate >= needed and n >= min_n + return ok, f"{step}: Pass-Rate {rate:.2f} (benötigt {needed:.2f}, n={n})" + + # --- Canary ----------------------------------------------------------------------------- + def _start_canary(self, cand: dict[str, Any]) -> None: + model = cand["target"].get("model") + if not model: + raise LifecycleError("Kandidat ohne bereitgestelltes Modell kann nicht in den Canary") + cluster = cand["target"].get("cluster") if cand["kind"] == "adapter" else None + self.registry.set_canary(cand["slot"], model, self.config.canary_share, cluster) + + def _cleanup_canary(self, cand: dict[str, Any]) -> None: + slot = self.registry.slot(cand["slot"]) + if slot.canary_model and slot.canary_model == cand["target"].get("model"): + self.registry.set_canary(cand["slot"], None) + + def canary_metrics(self, cid: str) -> dict[str, Any]: + """Pass-Rate der Fälle, die über das Canary-Modell liefen.""" + cand = self.get(cid) + model = cand["target"].get("model") + since = next((h["ts"] for h in cand.get("history") or [] if h["to"] == "canary"), None) + rows = [r for r in self.store.list_cases(stage=1, since=since) if r["model"] == model] + return {"pass_rate": pass_rate(rows), "n": len([r for r in rows if r["passed"] is not None])} + + # --- Promote / Rollback ----------------------------------------------------------------- + def promote(self, cid: str) -> dict[str, Any]: + cand = self.get(cid) + if cand["state"] != "awaiting_promotion": + raise LifecycleError(f"Kandidat ist {cand['state']}, nicht awaiting_promotion") + slot = self.registry.slot(cand["slot"]) + snapshot: dict[str, Any] = { + "base_model": slot.base_model_override, + "adapters": {a.name: a.status for a in self.registry.adapters()}, + } + followups: list[str] = [] + kind, target = cand["kind"], cand["target"] + if kind == "base_model": + old_base = slot.base_model + self.registry.set_base_model(slot.name, target["base_model"]) + for a in self.registry.adapters(): + if a.base_model == old_base and a.status == "active": + self.registry.set_adapter(a.model_copy(update={"status": "retired"})) + followups.append(self.create("adapter", slot.name, {"cluster": a.cluster, + "base_model": target["base_model"]})) + elif kind == "adapter": + name = target.get("model") or target.get("adapter") + if not name: + raise LifecycleError("Adapter-Kandidat ohne Adapter-Namen") + base = target.get("base_model") or slot.base_model + current = self.registry.active_adapter(base, target["cluster"]) + version = (current.version + 1) if current else 1 + if current: + self.registry.set_adapter(current.model_copy(update={"status": "retired"})) + self.registry.set_adapter(AdapterSpec(name=name, base_model=base, cluster=target["cluster"], + version=version, status="active")) + self.jobs.submit("promote", cand) + return self._move(cand, "promoted", "freigegeben", snapshot={**snapshot, "followups": followups}) + + def rollback(self, cid: str) -> dict[str, Any]: + cand = self.get(cid) + if cand["state"] != "promoted": + raise LifecycleError(f"Kandidat ist {cand['state']}, nicht promoted") + snap = cand.get("snapshot") or {} + if cand["kind"] == "base_model": + self.registry.set_base_model(cand["slot"], snap.get("base_model")) + statuses = snap.get("adapters") or {} + for a in self.registry.adapters(): + old = statuses.get(a.name) + if old is None and a.status == "active": + self.registry.set_adapter(a.model_copy(update={"status": "retired"})) + elif old is not None and old != a.status: + self.registry.set_adapter(a.model_copy(update={"status": old})) + for follow in snap.get("followups") or []: + f = self.store.get_candidate(follow) + if f and f["state"] not in TERMINAL: + self._move(f, "rejected", "Rollback des Basismodells") + self.jobs.submit("rollback", cand) + return self._move(cand, "rolled_back", "Rollback") + + def latest_promoted(self, slot: str) -> dict[str, Any] | None: + rows = [c for c in self.store.list_candidates(state="promoted") if c["slot"] == slot] + return rows[-1] if rows else None + + def reject(self, cid: str, note: str = "abgelehnt") -> dict[str, Any]: + cand = self.get(cid) + if cand["state"] in TERMINAL: + raise LifecycleError(f"Kandidat ist bereits {cand['state']}") + self._cleanup_canary(cand) + return self._move(cand, "rejected", note) diff --git a/core/src/knaster/maintenance/__init__.py b/core/src/knaster/maintenance/__init__.py new file mode 100644 index 0000000..978aae5 --- /dev/null +++ b/core/src/knaster/maintenance/__init__.py @@ -0,0 +1 @@ +"""Pflege-Agent mit Human-in-the-Loop.""" diff --git a/core/src/knaster/maintenance/actions.py b/core/src/knaster/maintenance/actions.py new file mode 100644 index 0000000..cfc7d9e --- /dev/null +++ b/core/src/knaster/maintenance/actions.py @@ -0,0 +1,127 @@ +"""Ausführung von Aktionen (geteilt von Pflege-Agent und Healer).""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from ..calibration import effective_context +from ..healer.drivers import DryRunDriver, InfraDriver, Notifier +from ..lifecycle import Lifecycle +from ..registry import Registry +from ..skills import SkillLibrary +from ..store import Store + +INFRA_ACTIONS = {"restart_container", "restart_vm", "reboot_host"} + + +class ActionError(Exception): + pass + + +class ActionExecutor: + """Interne Zustandsänderungen laufen direkt; Infrastruktur-Aktionen nur, wenn ausdrücklich freigeschaltet.""" + + def __init__(self, *, store: Store, registry: Registry, library: SkillLibrary, lifecycle: Lifecycle, + infra: InfraDriver | None = None, notifier: Notifier | None = None, dry_run: bool = True, + enabled_infra_actions: set[str] | None = None, on_skills_changed: Callable[[], None] | None = None): + self.store = store + self.registry = registry + self.library = library + self.lifecycle = lifecycle + self.dry = DryRunDriver() + self.infra = infra + self.notifier = notifier or self.dry + self.dry_run = dry_run + self.enabled_infra_actions = enabled_infra_actions or set() + self.on_skills_changed = on_skills_changed or (lambda: None) + + def infra_driver(self, kind: str) -> InfraDriver: + if self.dry_run or kind not in self.enabled_infra_actions or self.infra is None: + return self.dry + return self.infra + + def execute(self, kind: str, params: dict[str, Any]) -> dict[str, Any]: + handler = getattr(self, f"_do_{kind}", None) + if handler is None: + raise ActionError(f"Unbekannte Aktion {kind}") + try: + return {"ok": True, "message": handler(**params)} + except (KeyError, TypeError, ValueError, LookupError, RuntimeError) as exc: + raise ActionError(f"{kind}: {type(exc).__name__}: {exc}") from exc + + # --- Infrastruktur ---------------------------------------------------------------------- + def _do_restart_container(self, host: str, container: str, **_: Any) -> str: + return self.infra_driver("restart_container").restart_container(host, container) + + def _do_restart_vm(self, pve_node: str, vmid: str, **_: Any) -> str: + return self.infra_driver("restart_vm").restart_vm(pve_node, str(vmid)) + + def _do_reboot_host(self, pve_node: str, **_: Any) -> str: + return self.infra_driver("reboot_host").reboot_host(pve_node) + + def _do_notify(self, text: str, severity: str = "info", **_: Any) -> str: + return self.notifier.notify(text, severity) + + # --- Registry / Skills ------------------------------------------------------------------ + def _do_mark_slot(self, slot: str, status: str, reason: str = "", set_by: str = "maintenance", **_: Any) -> str: + self.registry.set_status(slot, status, reason=reason, set_by=set_by) # type: ignore[arg-type] + return f"Slot {slot} → {status}" + + def _do_demote_skill(self, skill: str, reason: str = "", **_: Any) -> str: + self.library.set_state(skill, status="demoted", reason=reason) + self.on_skills_changed() + return f"Skill {skill} auf Stufe 0 zurückgestuft" + + def _do_promote_skill(self, skill: str, stage: int = 1, **_: Any) -> str: + self.library.set_state(skill, status="active", stage=int(stage), reason="freigegeben") + self.on_skills_changed() + return f"Skill {skill} auf Stufe {stage} freigegeben" + + def _do_retire_skill(self, skill: str, **_: Any) -> str: + self.library.set_state(skill, status="retired", reason="stillgelegt") + self.on_skills_changed() + return f"Skill {skill} stillgelegt" + + def _do_request_graduation(self, skill: str, **_: Any) -> str: + if not self.library.has(skill): + raise KeyError(f"Skill {skill} unbekannt") + return self.lifecycle.jobs.submit("graduate_skill", {"id": f"graduate:{skill}", "skill": skill}) + + def _do_merge_skills(self, a: str, b: str, **_: Any) -> str: + return self.lifecycle.jobs.submit("merge_skills", {"id": f"merge:{a}+{b}", "skills": [a, b]}) + + def _do_recalibrate_context(self, slot: str, target: float = 0.9, min_n: int = 20, **_: Any) -> str: + rows = [r for r in self.store.list_cases(stage=None) if r["slot"] == slot + and r["prompt_tokens"] is not None and r["passed"] is not None] + value = effective_context(((r["prompt_tokens"], bool(r["passed"])) for r in rows), target=target, + min_n=min_n) + if value is None: + return f"Slot {slot}: zu wenige Daten für Kalibrierung ({len(rows)} Fälle)" + self.registry.set_calibrated_context(slot, value) + return f"Slot {slot}: effektiver Kontext = {value} Tokens" + + # --- Lifecycle -------------------------------------------------------------------------- + def _do_create_candidate(self, kind: str, slot: str, target: dict[str, Any], **_: Any) -> str: + cid = self.lifecycle.create(kind, slot, target) # type: ignore[arg-type] + self.lifecycle.start(cid) + return f"Kandidat {cid} ({kind}) angelegt und Build gestartet" + + def _do_train_adapter(self, cluster: str, slot: str = "tier1-general", **_: Any) -> str: + base = self.registry.slot(slot).base_model + return self._do_create_candidate("adapter", slot, {"cluster": cluster, "base_model": base}) + + def _do_promote_candidate(self, candidate: str, **_: Any) -> str: + self.lifecycle.promote(candidate) + return f"Kandidat {candidate} übernommen" + + def _do_rollback_candidate(self, slot: str | None = None, candidate: str | None = None, **_: Any) -> str: + if candidate is None: + if slot is None: + raise ValueError("slot oder candidate nötig") + latest = self.lifecycle.latest_promoted(slot) + if latest is None: + raise LookupError(f"Kein übernommener Kandidat für {slot}") + candidate = latest["id"] + self.lifecycle.rollback(candidate) + return f"Kandidat {candidate} zurückgerollt" diff --git a/core/src/knaster/maintenance/agent.py b/core/src/knaster/maintenance/agent.py new file mode 100644 index 0000000..a6f4a17 --- /dev/null +++ b/core/src/knaster/maintenance/agent.py @@ -0,0 +1,448 @@ +"""Pflege-Agent: Service-Routine, Vorschläge, Freigaben per Prompt.""" + +from __future__ import annotations + +import re +from collections.abc import Callable +from datetime import datetime +from pathlib import Path +from typing import Any, Protocol + +from pydantic import BaseModel, Field + +from ..cases import pass_rate +from ..embeddings import SkillIndex +from ..lifecycle import Lifecycle +from ..llm import ChatClient, LLMError +from ..policy import Policy +from ..registry import Registry +from ..skills import SkillLibrary +from ..store import Store +from ..util import load_yaml, now +from .actions import ActionError, ActionExecutor + +CONTEXT_SIGNALS = ("context_exceeded", "compact_prompt", "larger_slot", "output_truncated", "context_overflow") + + +class MaintenanceConfig(BaseModel): + demote_below: float = 0.80 + min_cases: int = 20 + window_cases: int = 50 + promote_min_rate: float = 0.95 + merge_similarity: float = 0.92 + degraded_hours: float = 1.0 + context_signal_limit: int = 10 + adapter_min_cases: int = 300 + adapter_target_rate: float = 0.90 + lookback_days: float = 7.0 + + @classmethod + def from_file(cls, path: str | Path) -> MaintenanceConfig: + return cls(**load_yaml(path)) + + +class Proposal(BaseModel): + seq: int + kind: str + params: dict[str, Any] = Field(default_factory=dict) + summary: str = "" + reason: str = "" + risk: str = "medium" + state: str = "open" + source: str = "" + created_at: float | None = None + decided_at: float | None = None + decided_by: str | None = None + result: dict[str, Any] | None = None + + @property + def id(self) -> str: + return f"P-{self.seq}" + + def line(self) -> str: + return f"{self.id} [{self.risk}] {self.summary}" + (f" – {self.reason}" if self.reason else "") + + +class RoutineReport(BaseModel): + findings: list[str] = Field(default_factory=list) + proposals: list[Proposal] = Field(default_factory=list) + auto_executed: list[str] = Field(default_factory=list) + text: str = "" + + +class CommandReply(BaseModel): + text: str + proposals: list[Proposal] = Field(default_factory=list) + + +class IntentMapper(Protocol): + def map(self, text: str) -> str | None: ... + + +HELP = """Befehle: +- status · hilfe · routine · vorschläge +- genehmige P-3 [P-5 …] · genehmige alle mit risiko niedrig · lehne P-3 ab: +- wechsle auf · trainiere adapter für · rollback +- stufe zurück · markiere als ok|degraded|unavailable""" + + +class LLMIntentMapper: + """Bildet Freitext auf genau einen bekannten Befehl ab – führt nie selbst etwas aus.""" + + def __init__(self, chat: ChatClient, model: str = "tier1-general"): + self.chat = chat + self.model = model + + def map(self, text: str) -> str | None: + system = ("Übersetze die Nachricht in GENAU einen Befehl aus dieser Liste oder antworte NONE.\n" + + HELP + "\nAntworte nur mit dem Befehl.") + try: + res = self.chat.chat(self.model, [{"role": "system", "content": system}, + {"role": "user", "content": text}], max_tokens=40, temperature=0) + except LLMError: + return None + out = res.content.strip().strip("`").splitlines()[0] if res.content.strip() else "" + return None if not out or out.upper() == "NONE" else out + + +_ID = r"p-?\d+" +COMMANDS: list[tuple[str, re.Pattern[str]]] = [ + (name, re.compile(pattern, re.IGNORECASE)) for name, pattern in [ + ("status", r"^(status|überblick|overview)$"), + ("help", r"^(hilfe|help|\?)$"), + ("routine", r"^((starte\s+(die\s+)?)?service-?routine|routine)$"), + ("list", r"^((zeige\s+)?(offene\s+)?vorschläge|proposals)$"), + ("approve_low", r"^(genehmige|approve)\s+alle\s+(mit\s+)?risiko\s+(niedrig|low)$"), + ("approve", rf"^(genehmige|approve)\s+(?P{_ID}([\s,]+(und\s+)?{_ID})*)$"), + ("reject", rf"^(lehne\s+(?P{_ID})\s+ab|reject\s+(?P{_ID}))(\s*[:,]?\s*(weil\s+)?(?P.+))?$"), + ("switch_base", r"^wechsle\s+(?P[\w.-]+)\s+auf\s+(?P[\w./:@-]+)$"), + ("train_adapter", r"^trainiere\s+(einen\s+)?adapter\s+für\s+(den\s+cluster\s+)?(?P[\w.-]+)$"), + ("rollback", r"^rollback\s+(?P[\w.-]+)$"), + ("demote", r"^(stufe\s+(?P[\w.-]+)\s+zurück|degradiere\s+(?P[\w.-]+))$"), + ("mark", r"^markiere\s+(?P[\w.-]+)\s+(als\s+)?(?Pok|degraded|unavailable)$"), + ("confirm", r"^(ja|yes|j|ok)$"), + ("cancel", r"^(nein|no|n|abbrechen)$"), + ] +] + + +def parse_command(text: str) -> tuple[str, dict[str, str]] | None: + cleaned = text.strip().rstrip(".!") + for name, pattern in COMMANDS: + m = pattern.match(cleaned) + if m: + return name, {k: v for k, v in m.groupdict().items() if v is not None} + return None + + +def _seq(token: str) -> int: + return int(re.sub(r"\D", "", token)) + + +class MaintenanceAgent: + def __init__(self, *, store: Store, registry: Registry, library: SkillLibrary, index: SkillIndex, + policy: Policy, actions: ActionExecutor, lifecycle: Lifecycle, + config: MaintenanceConfig | None = None, intent_mapper: IntentMapper | None = None, + clock: Callable[[], float] = now, wallclock: Callable[[], datetime] = datetime.now): + self.store = store + self.registry = registry + self.library = library + self.index = index + self.policy = policy + self.actions = actions + self.lifecycle = lifecycle + self.config = config or MaintenanceConfig() + self.intent_mapper = intent_mapper + self.clock = clock + self.wallclock = wallclock + self._pending: dict[str, str] = {} # Sitzung → zu bestätigender Befehl + + # --- Vorschläge ------------------------------------------------------------------------- + def _load(self, seq: int) -> Proposal: + row = self.store.get_proposal(seq) + if row is None: + raise KeyError(f"P-{seq} unbekannt") + return Proposal(**row) + + def proposals(self, state: str | None = "open") -> list[Proposal]: + return [Proposal(**r) for r in self.store.list_proposals(state=state)] + + def propose(self, kind: str, params: dict[str, Any], summary: str, *, reason: str = "", + source: str = "routine", dedup_key: str | None = None, force_approval: bool = False) -> Proposal: + key = dedup_key or f"{kind}:{sorted(params.items())}" + existing = self.store.open_proposal_with_key(key) + if existing: + return Proposal(**existing) + mode = self.policy.decide(kind) + if kind == "mark_slot" and params.get("status") == "ok": + force_approval = True # Wieder-Freischalten ist die unsichere Richtung + seq = self.store.add_proposal(kind=kind, params=params, summary=summary, reason=reason, + risk=self.policy.risk(kind), source=source, dedup_key=key) + if mode == "deny": + self.store.update_proposal(seq, state="rejected", decided_by="policy", decided_at=self.clock(), + result={"message": "durch Richtlinie verboten"}) + elif mode == "auto" and not force_approval: + return self.approve(seq, by="policy") + return self._load(seq) + + def approve(self, seq: int, by: str = "human") -> Proposal: + p = self._load(seq) + if p.state != "open": + raise ValueError(f"{p.id} ist {p.state}") + self.store.update_proposal(seq, state="approved", decided_by=by, decided_at=self.clock()) + if not self.policy.in_window(p.kind, self.wallclock()): + self.store.update_proposal(seq, result={"deferred": True, "message": "wartet auf Wartungsfenster"}) + return self._load(seq) + return self._execute(seq) + + def _execute(self, seq: int) -> Proposal: + p = self._load(seq) + try: + result = self.actions.execute(p.kind, p.params) + self.store.update_proposal(seq, state="executed", result=result) + except ActionError as exc: + self.store.update_proposal(seq, state="failed", result={"ok": False, "message": str(exc)}) + return self._load(seq) + + def reject(self, seq: int, by: str = "human", reason: str = "") -> Proposal: + p = self._load(seq) + if p.state != "open": + raise ValueError(f"{p.id} ist {p.state}") + self.store.update_proposal(seq, state="rejected", decided_by=by, decided_at=self.clock(), + result={"message": reason or "abgelehnt"}) + if p.kind == "promote_candidate": + self.lifecycle.reject(p.params["candidate"], f"abgelehnt: {reason}") + return self._load(seq) + + def run_deferred(self) -> list[Proposal]: + done = [] + for p in self.proposals(state="approved"): + if (p.result or {}).get("deferred") and self.policy.in_window(p.kind, self.wallclock()): + done.append(self._execute(p.seq)) + return done + + # --- Service-Routine -------------------------------------------------------------------- + def run_routine(self) -> RoutineReport: + report = RoutineReport() + c = self.config + since = self.clock() - c.lookback_days * 86400 + + def add(p: Proposal) -> None: + if p.state == "executed" and p.decided_by == "policy": + report.auto_executed.append(f"{p.id}: {p.summary} → {(p.result or {}).get('message', '')}") + else: + report.proposals.append(p) + + # 1. Drift / 2. Graduierungskandidaten + for skill in self.library.routable(): + if skill.meta.status == "active" and skill.meta.stage >= 1: + rows = self.store.list_cases(skill=skill.name, min_stage=1, limit=c.window_cases) + rate = pass_rate(rows) + n = len([r for r in rows if r["passed"] is not None]) + if rate is not None and n >= c.min_cases and rate < c.demote_below: + report.findings.append(f"Drift: {skill.name} Pass-Rate {rate:.0%} (n={n})") + add(self.propose("demote_skill", {"skill": skill.name, "reason": f"Pass-Rate {rate:.0%}"}, + f"Skill {skill.name} auf Stufe 0 zurückstufen", + reason=f"Pass-Rate {rate:.0%} < {c.demote_below:.0%} (n={n})")) + else: + rows = self.store.list_cases(skill=skill.name, stage=0, limit=c.window_cases) + rate = pass_rate(rows) + n = len([r for r in rows if r["passed"] is not None]) + if rate is not None and n >= c.min_cases and rate >= c.promote_min_rate: + report.findings.append(f"Graduierungskandidat: {skill.name} ({rate:.0%} auf Stufe 0, n={n})") + add(self.propose("request_graduation", {"skill": skill.name}, + f"Graduierungslauf für {skill.name} anstoßen", + reason=f"stabil auf Stufe 0 ({rate:.0%}, n={n})")) + + # 3. ähnliche Skills + for a, b, sim in self.index.pairwise_similar(c.merge_similarity): + if self.library.has(a) and self.library.has(b) and self.library.get(a).routable \ + and self.library.get(b).routable: + report.findings.append(f"Ähnliche Skills: {a} ~ {b} ({sim:.2f})") + add(self.propose("merge_skills", {"a": a, "b": b}, f"Skills {a} und {b} zusammenführen", + reason=f"Ähnlichkeit {sim:.2f}")) + + # 4. degradierte Slots + for slot in self.registry.slots(): + state = self.store.get_slot_state(slot.name) or {} + age_h = (self.clock() - (state.get("updated_at") or self.clock())) / 3600 + if not slot.available and age_h >= c.degraded_hours: + report.findings.append(f"Slot {slot.name} seit {age_h:.1f} h {slot.status} ({slot.reason or '-'})") + add(self.propose("mark_slot", {"slot": slot.name, "status": "ok", "reason": "nach Prüfung"}, + f"Slot {slot.name} wieder auf ok setzen", + reason=f"{slot.status} seit {age_h:.1f} h – vorher Ursache prüfen")) + + # 5. Kontext-Signale + totals: dict[str, int] = {} + for kind in CONTEXT_SIGNALS: + for subject, count in self.store.event_subjects(kind, since=since).items(): + totals[subject] = totals.get(subject, 0) + count + for subject, count in sorted(totals.items()): + if self.registry.has_slot(subject) and count >= c.context_signal_limit: + report.findings.append(f"Kontext-Signale auf {subject}: {count} in {c.lookback_days:g} Tagen") + add(self.propose("recalibrate_context", {"slot": subject}, + f"Effektiven Kontext von {subject} neu kalibrieren", reason=f"{count} Signale")) + + # 6. Adapter-Bedarf + open_clusters = {c_["target"].get("cluster") for c_ in self.store.list_candidates() + if c_["kind"] == "adapter" and c_["state"] not in {"promoted", "rejected", "failed", + "rolled_back"}} + clusters: dict[str, str] = {} + for skill in self.library.routable(): + clusters.setdefault(skill.cluster, skill.meta.slot) + for cluster, slot in sorted(clusters.items()): + if cluster in open_clusters or not self.registry.has_slot(slot): + continue + if not self.registry.slot(slot).spec.supports_lora: + continue + train = self.store.list_cases(cluster=cluster, split="train", passed=True) + if len(train) < c.adapter_min_cases: + continue + weak = self.store.list_cases(cluster=cluster, min_stage=1, limit=c.window_cases) + rate = pass_rate(weak) + if rate is None or rate < c.adapter_target_rate: + report.findings.append(f"Adapter-Bedarf: {cluster} ({len(train)} Trainingsfälle)") + add(self.propose("train_adapter", {"cluster": cluster, "slot": slot}, + f"LoRA-Adapter für {cluster} trainieren", + reason=f"{len(train)} verifizierte Fälle, Stufe-1-Rate " + f"{'–' if rate is None else f'{rate:.0%}'}")) + + # 7. Lifecycle: Canary auswerten, Freigaben vorschlagen + for cand in self.store.list_candidates(state="canary"): + m = self.lifecycle.canary_metrics(cand["id"]) + if m["n"] >= self.lifecycle.config.min_n and m["pass_rate"] is not None: + after = self.lifecycle.report(cand["id"], "canary", m) + report.auto_executed.append(f"Canary {cand['id'][:8]} ausgewertet → {after['state']}") + for cand in self.store.list_candidates(state="awaiting_promotion"): + add(self.propose("promote_candidate", {"candidate": cand["id"]}, + f"{cand['kind']} für {cand['slot']} übernehmen ({cand['target']})", + reason="alle Prüfstufen bestanden", source="lifecycle")) + + for p in self.run_deferred(): + report.auto_executed.append(f"{p.id} im Wartungsfenster ausgeführt: {p.summary}") + + open_all = self.proposals("open") + report.text = self._render_report(report, open_all) + return report + + @staticmethod + def _render_report(report: RoutineReport, open_all: list[Proposal]) -> str: + lines = ["Service-Routine"] + lines.append("\nBefunde:" if report.findings else "\nKeine Auffälligkeiten.") + lines += [f"- {f}" for f in report.findings] + if report.auto_executed: + lines.append("\nAutomatisch erledigt:") + lines += [f"- {a}" for a in report.auto_executed] + if open_all: + lines.append("\nOffene Vorschläge (genehmige P-x / lehne P-x ab):") + lines += [f"- {p.line()}" for p in open_all] + return "\n".join(lines) + + # --- Befehle ---------------------------------------------------------------------------- + def handle_command(self, text: str, user: str = "human", session: str = "default") -> CommandReply: + parsed = parse_command(text) + if parsed is None: + if self.intent_mapper is not None: + mapped = self.intent_mapper.map(text) + if mapped and parse_command(mapped) and parse_command(mapped)[0] not in {"confirm", "cancel"}: + self._pending[session] = mapped + return CommandReply(text=f"Meinst du: `{mapped}`? Antworte mit „ja“.") + return CommandReply(text="Das habe ich nicht verstanden.\n" + HELP) + + name, args = parsed + if name == "confirm": + pending = self._pending.pop(session, None) + if not pending: + return CommandReply(text="Es gibt nichts zu bestätigen.") + return self.handle_command(pending, user, session) + if name == "cancel": + self._pending.pop(session, None) + return CommandReply(text="Abgebrochen.") + try: + return getattr(self, f"_cmd_{name}")(args, user) + except (KeyError, ValueError) as exc: + return CommandReply(text=f"Fehler: {exc}") + + def _cmd_help(self, args: dict[str, str], user: str) -> CommandReply: + return CommandReply(text=HELP) + + def _cmd_status(self, args: dict[str, str], user: str) -> CommandReply: + lines = ["Slots:"] + lines += [f"- {s.name}: {s.status} (Basis {s.base_model}, Kontext {s.effective_context})" + for s in self.registry.slots()] + by_status: dict[str, int] = {} + for s in self.library.all(): + by_status[s.meta.status] = by_status.get(s.meta.status, 0) + 1 + lines.append("Skills: " + ", ".join(f"{k} {v}" for k, v in sorted(by_status.items()))) + running = [c for c in self.store.list_candidates() if c["state"] not in + {"promoted", "rejected", "failed", "rolled_back"}] + lines.append(f"Kandidaten in Arbeit: {len(running)}") + lines.append(f"Offene Vorschläge: {len(self.proposals('open'))}") + return CommandReply(text="\n".join(lines)) + + def _cmd_routine(self, args: dict[str, str], user: str) -> CommandReply: + report = self.run_routine() + return CommandReply(text=report.text, proposals=report.proposals) + + def _cmd_list(self, args: dict[str, str], user: str) -> CommandReply: + props = self.proposals("open") + if not props: + return CommandReply(text="Keine offenen Vorschläge.") + return CommandReply(text="\n".join(p.line() for p in props), proposals=props) + + def _decided_text(self, p: Proposal) -> str: + msg = (p.result or {}).get("message", "") + return f"{p.id} {p.state}" + (f": {msg}" if msg else "") + + def _cmd_approve(self, args: dict[str, str], user: str) -> CommandReply: + ids = re.findall(_ID, args["ids"], re.IGNORECASE) + done = [self.approve(_seq(i), by=user) for i in ids] + return CommandReply(text="\n".join(self._decided_text(p) for p in done), proposals=done) + + def _cmd_approve_low(self, args: dict[str, str], user: str) -> CommandReply: + done = [self.approve(p.seq, by=user) for p in self.proposals("open") if p.risk == "low"] + if not done: + return CommandReply(text="Keine offenen Vorschläge mit niedrigem Risiko.") + return CommandReply(text="\n".join(self._decided_text(p) for p in done), proposals=done) + + def _cmd_reject(self, args: dict[str, str], user: str) -> CommandReply: + p = self.reject(_seq(args.get("id") or args["id2"]), by=user, reason=args.get("reason", "")) + return CommandReply(text=self._decided_text(p), proposals=[p]) + + def _new(self, kind: str, params: dict[str, Any], summary: str, user: str) -> CommandReply: + p = self.propose(kind, params, summary, reason=f"Auftrag von {user}", source="command", force_approval=True) + return CommandReply(text=f"Vorschlag {p.line()}\nBestätige mit „genehmige {p.id}“.", proposals=[p]) + + def _require_slot(self, slot: str) -> None: + if not self.registry.has_slot(slot): + raise KeyError(f"Slot {slot} unbekannt") + + def _cmd_switch_base(self, args: dict[str, str], user: str) -> CommandReply: + self._require_slot(args["slot"]) + return self._new("create_candidate", + {"kind": "base_model", "slot": args["slot"], "target": {"base_model": args["base"]}}, + f"Basismodell von {args['slot']} auf {args['base']} wechseln (Pipeline, inkl. " + "Neutraining der Adapter)", user) + + def _cmd_train_adapter(self, args: dict[str, str], user: str) -> CommandReply: + slot = next((s.meta.slot for s in self.library.all() if s.cluster == args["cluster"]), "tier1-general") + return self._new("train_adapter", {"cluster": args["cluster"], "slot": slot}, + f"LoRA-Adapter für {args['cluster']} trainieren", user) + + def _cmd_rollback(self, args: dict[str, str], user: str) -> CommandReply: + self._require_slot(args["slot"]) + return self._new("rollback_candidate", {"slot": args["slot"]}, + f"Letzte Übernahme auf {args['slot']} zurückrollen", user) + + def _cmd_demote(self, args: dict[str, str], user: str) -> CommandReply: + skill = args.get("skill") or args["skill2"] + if not self.library.has(skill): + raise KeyError(f"Skill {skill} unbekannt") + return self._new("demote_skill", {"skill": skill, "reason": f"Auftrag von {user}"}, + f"Skill {skill} auf Stufe 0 zurückstufen", user) + + def _cmd_mark(self, args: dict[str, str], user: str) -> CommandReply: + self._require_slot(args["slot"]) + return self._new("mark_slot", {"slot": args["slot"], "status": args["status"].lower(), + "reason": f"Auftrag von {user}", "set_by": user}, + f"Slot {args['slot']} als {args['status'].lower()} markieren", user) diff --git a/core/src/knaster/metrics.py b/core/src/knaster/metrics.py new file mode 100644 index 0000000..2e45b1d --- /dev/null +++ b/core/src/knaster/metrics.py @@ -0,0 +1,39 @@ +"""Prometheus-Metriken des Router-Service (eigene Registry pro Instanz).""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram, generate_latest + +if TYPE_CHECKING: + from .runner import RunResult + + +class Metrics: + def __init__(self) -> None: + self.registry = CollectorRegistry() + self.runs = Counter("knaster_runs_total", "Läufe nach Ergebnis und Endstufe", ["status", "stage"], + registry=self.registry) + self.escalations = Counter("knaster_escalations_total", "Eskalationen nach Signal", ["signal"], + registry=self.registry) + self.verifier = Counter("knaster_verifier_total", "Verifier-Ergebnisse", ["skill", "stage", "passed"], + registry=self.registry) + self.attempts = Histogram("knaster_run_attempts", "Versuche pro Lauf", buckets=(1, 2, 3, 4), + registry=self.registry) + self.slot_status = Gauge("knaster_slot_available", "1 = Slot verfügbar", ["slot"], registry=self.registry) + self.open_proposals = Gauge("knaster_open_proposals", "Offene Pflege-Vorschläge", registry=self.registry) + + def observe_run(self, result: RunResult) -> None: + self.runs.labels(result.status, str(result.stage)).inc() + self.attempts.observe(len(result.attempts)) + for attempt in result.attempts[1:]: + for s in attempt.decision.signals[:1]: + self.escalations.labels(s.value).inc() + for attempt in result.attempts: + if attempt.verification is not None: + self.verifier.labels(attempt.decision.skill or "-", str(attempt.decision.stage), + str(attempt.verification.passed).lower()).inc() + + def render(self) -> bytes: + return generate_latest(self.registry) diff --git a/core/src/knaster/policy.py b/core/src/knaster/policy.py new file mode 100644 index 0000000..41adcb4 --- /dev/null +++ b/core/src/knaster/policy.py @@ -0,0 +1,48 @@ +"""Freigabe-Richtlinie: welche Aktion läuft automatisch, welche braucht einen Menschen.""" + +from __future__ import annotations + +from datetime import datetime, time +from pathlib import Path +from typing import Literal + +from pydantic import BaseModel, Field + +from .util import load_yaml + +Mode = Literal["auto", "approve", "deny"] + + +class ActionPolicy(BaseModel): + mode: Mode = "approve" + window: str | None = None # "HH:MM-HH:MM", lokale Zeit + risk: Literal["low", "medium", "high"] = "medium" + + +class Policy(BaseModel): + default: Mode = "approve" + actions: dict[str, ActionPolicy] = Field(default_factory=dict) + + @classmethod + def from_file(cls, path: str | Path) -> Policy: + return cls(**load_yaml(path)) + + def for_kind(self, kind: str) -> ActionPolicy: + return self.actions.get(kind) or ActionPolicy(mode=self.default) + + def decide(self, kind: str) -> Mode: + return self.for_kind(kind).mode + + def risk(self, kind: str) -> str: + return self.for_kind(kind).risk + + def in_window(self, kind: str, now: datetime | None = None) -> bool: + window = self.for_kind(kind).window + if not window: + return True + start_s, end_s = window.split("-") + start, end = time.fromisoformat(start_s.strip()), time.fromisoformat(end_s.strip()) + current = (now or datetime.now()).time() + if start <= end: + return start <= current < end + return current >= start or current < end # über Mitternacht diff --git a/core/src/knaster/registry.py b/core/src/knaster/registry.py index f096b6e..d83e8db 100644 --- a/core/src/knaster/registry.py +++ b/core/src/knaster/registry.py @@ -45,6 +45,9 @@ class Slot(BaseModel): set_by: str | None = None calibrated_context: int | None = None base_model_override: str | None = None + canary_model: str | None = None + canary_share: float = 0.0 + canary_cluster: str | None = None @property def name(self) -> str: @@ -104,6 +107,9 @@ class Registry: set_by=state.get("set_by"), calibrated_context=state.get("effective_context"), base_model_override=state.get("base_model"), + canary_model=state.get("canary_model"), + canary_share=state.get("canary_share") or 0.0, + canary_cluster=state.get("canary_cluster"), ) def slots(self) -> list[Slot]: @@ -127,6 +133,11 @@ class Registry: def set_base_model(self, name: str, base_model: str | None) -> None: self.store.set_slot_state(name, base_model=base_model) + def set_canary(self, name: str, model: str | None, share: float = 0.0, cluster: str | None = None) -> None: + """Leitet `share` des Stufe-1-Verkehrs (optional nur eines Clusters) auf ein Kandidatenmodell.""" + self.store.set_slot_state(name, canary_model=model, canary_share=share if model else 0.0, + canary_cluster=cluster if model else None) + # --- Adapter ---------------------------------------------------------------------------- def adapters(self) -> list[AdapterSpec]: merged = {n: a.model_copy() for n, a in self._adapters.items()} diff --git a/core/src/knaster/router.py b/core/src/knaster/router.py index 556daac..2243f1a 100644 --- a/core/src/knaster/router.py +++ b/core/src/knaster/router.py @@ -2,6 +2,7 @@ from __future__ import annotations +import random from dataclasses import dataclass from typing import Literal @@ -31,6 +32,7 @@ class RouteDecision(BaseModel): estimated_tokens: int | None = None max_output_tokens: int | None = None attempt: int = 0 + canary: bool = False tried_slots: list[str] = Field(default_factory=list) reasons: list[str] = Field(default_factory=list) signals: list[Signal] = Field(default_factory=list) @@ -62,6 +64,7 @@ class Router: reranker: Reranker | None = None, tokenizer: budget_mod.Tokenizer = budget_mod.estimate_tokens, config: RouterConfig | None = None, + rng: random.Random | None = None, ): self.registry = registry self.library = library @@ -69,6 +72,7 @@ class Router: self.reranker = reranker self.tokenizer = tokenizer self.config = config or RouterConfig() + self.rng = rng or random.Random() self.index = SkillIndex(embedder) self.refresh() @@ -209,7 +213,15 @@ class Router: if found: model, adapter = found.name, found.name reasons.append(f"Adapter {found.name}") - return self._decision(stage=1, slot=slot.name, model=model, adapter=adapter, skill=skill, + canary = False + if (slot.canary_model and slot.canary_cluster in (None, skill.cluster) + and self.rng.random() < slot.canary_share): + canary = True + model = slot.canary_model + if slot.canary_cluster: + adapter = slot.canary_model + reasons.append(f"Canary {slot.canary_model}") + return self._decision(stage=1, slot=slot.name, model=model, adapter=adapter, skill=skill, canary=canary, prompt_variant=variant, similarity=similarity, estimated_tokens=b.prompt_tokens, attempt=attempt, tried_slots=tried + [slot.name], reasons=reasons, signals=signals) @@ -274,3 +286,9 @@ class Router: attempt=attempt) return self._plan_strong(text, skill, reasons, signals, similarity=decision.similarity, attempt=attempt) + + # --- Graduierung ------------------------------------------------------------------------ + def plan_stage1(self, text: str, skill_name: str) -> RouteDecision: + """Plant eine Ausführung auf Stufe 1, unabhängig von der freigegebenen Stufe des Skills.""" + skill = self.library.get(skill_name) + return self._plan_weak(text, skill, [f"Graduierung {skill_name}"], [], similarity=None) diff --git a/core/src/knaster/runner.py b/core/src/knaster/runner.py index 6837cf0..8760f2c 100644 --- a/core/src/knaster/runner.py +++ b/core/src/knaster/runner.py @@ -92,9 +92,13 @@ class Runner: # --- Hilfen ----------------------------------------------------------------------------- def _events(self, decision: RouteDecision) -> None: + # Präventive Signale gehören zum Slot, für den der Skill gedacht ist (nicht zum Ausweichziel) + home = None + if decision.skill and self.library.has(decision.skill): + home = self.library.get(decision.skill).meta.slot for s in decision.signals: if s in PREVENTIVE_EVENT_SIGNALS: - self.store.add_event(s.value, subject=decision.slot or decision.skill) + self.store.add_event(s.value, subject=home or decision.slot) def _record_case(self, run_id: str, text: str, decision: RouteDecision, attempt: Attempt, skill: Any) -> None: diff --git a/core/src/knaster/store.py b/core/src/knaster/store.py index 101189e..691fe37 100644 --- a/core/src/knaster/store.py +++ b/core/src/knaster/store.py @@ -91,6 +91,9 @@ slot_state = Table( Column("status", String(20)), Column("effective_context", Integer), Column("base_model", String(200)), + Column("canary_model", String(200)), + Column("canary_share", Float), + Column("canary_cluster", String(200)), Column("reason", Text), Column("set_by", String(50)), Column("updated_at", Float), @@ -223,9 +226,10 @@ class Store: return [dict(r._mapping) for r in conn.execute(q)] # --- Ereignisse ------------------------------------------------------------------------ - def add_event(self, kind: str, subject: str | None = None, data: dict | None = None) -> None: + def add_event(self, kind: str, subject: str | None = None, data: dict | None = None, + ts: float | None = None) -> None: with self.engine.begin() as conn: - conn.execute(insert(events).values(ts=now(), kind=kind, subject=subject, data=data or {})) + conn.execute(insert(events).values(ts=ts or now(), kind=kind, subject=subject, data=data or {})) def count_events(self, kind: str, *, subject: str | None = None, since: float | None = None) -> int: q = select(func.count()).select_from(events).where(events.c.kind == kind) diff --git a/core/src/knaster/system.py b/core/src/knaster/system.py new file mode 100644 index 0000000..728e5dd --- /dev/null +++ b/core/src/knaster/system.py @@ -0,0 +1,121 @@ +"""Verdrahtung aller Komponenten (einmal pro Prozess, in Tests mit Fakes).""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path + +from .classifier import EmbeddingClassifier +from .embeddings import Embedder, LlamaCppReranker, OpenAIEmbedder, Reranker +from .executor import Executor +from .healer.drivers import CompositeInfra, InfraDriver, Notifier, PortainerDriver, ProxmoxDriver, WebhookNotifier +from .healer.engine import Healer, HealerConfig +from .lifecycle import JobRunner, Lifecycle, LifecycleConfig, LoggingJobRunner, WebhookJobRunner +from .llm import ChatClient, OpenAIChatClient +from .maintenance.actions import ActionExecutor +from .maintenance.agent import IntentMapper, LLMIntentMapper, MaintenanceAgent, MaintenanceConfig +from .metrics import Metrics +from .policy import Policy +from .registry import Registry +from .router import Router +from .runner import Runner +from .settings import Settings +from .skills import SkillLibrary +from .store import Store +from .util import now +from .verifier import Verifier + + +@dataclass +class System: + settings: Settings + store: Store + registry: Registry + library: SkillLibrary + embedder: Embedder + chat: ChatClient + router: Router + executor: Executor + verifier: Verifier + runner: Runner + policy: Policy + lifecycle: Lifecycle + actions: ActionExecutor + maintenance: MaintenanceAgent + healer: Healer + metrics: Metrics + + +def _load_classifiers(directory: Path) -> dict[str, EmbeddingClassifier]: + if not directory.is_dir(): + return {} + return {p.stem: EmbeddingClassifier.load(p) for p in sorted(directory.glob("*.json"))} + + +def _optional(path: Path, loader: Callable, default): # type: ignore[no-untyped-def] + return loader(path) if path.exists() else default + + +def build_system( + settings: Settings, + *, + embedder: Embedder | None = None, + chat: ChatClient | None = None, + reranker: Reranker | None = None, + infra: InfraDriver | None = None, + notifier: Notifier | None = None, + jobs: JobRunner | None = None, + intent_mapper: IntentMapper | None = None, + clock: Callable[[], float] = now, + wallclock: Callable[[], datetime] = datetime.now, +) -> System: + if settings.db_url.startswith("sqlite:///") and not settings.db_url.startswith("sqlite:///:"): + Path(settings.db_url.removeprefix("sqlite:///")).parent.mkdir(parents=True, exist_ok=True) + store = Store(settings.db_url) + registry = Registry.from_file(settings.site_dir / "registry.yaml", store) + library = SkillLibrary.from_dir(settings.skills_dir, store) + + chat = chat or OpenAIChatClient(settings.litellm_base_url, settings.litellm_api_key, settings.request_timeout) + embedder = embedder or OpenAIEmbedder(settings.litellm_base_url, settings.embed_model, settings.litellm_api_key) + if reranker is None and settings.rerank_url: + reranker = LlamaCppReranker(settings.rerank_url, settings.rerank_model) + + cfg = settings.config_dir + policy = _optional(cfg / "policy.yaml", Policy.from_file, Policy()) + healer_cfg = _optional(cfg / "healer-rules.yaml", HealerConfig.from_file, HealerConfig()) + maint_cfg = _optional(cfg / "maintenance.yaml", MaintenanceConfig.from_file, MaintenanceConfig()) + life_cfg = _optional(cfg / "lifecycle.yaml", LifecycleConfig.from_file, LifecycleConfig()) + + metrics = Metrics() + router = Router(registry, library, embedder, reranker) + executor = Executor(library, chat, embedder, _load_classifiers(cfg / "classifiers")) + verifier = Verifier(registry, chat) + runner = Runner(router, executor, verifier, library, store, metrics) + + if jobs is None: + jobs = WebhookJobRunner(settings.job_webhook_url) if settings.job_webhook_url and not settings.dry_run \ + else LoggingJobRunner() + lifecycle = Lifecycle(store, registry, jobs, life_cfg) + + if infra is None and not settings.dry_run: + portainer = PortainerDriver(settings.portainer_host, settings.portainer_key) \ + if settings.portainer_host and settings.portainer_key else None + proxmox = ProxmoxDriver(settings.proxmox_url, settings.proxmox_token_id, settings.proxmox_token_secret) \ + if settings.proxmox_url and settings.proxmox_token_id else None + infra = CompositeInfra(portainer, proxmox) + if notifier is None and settings.notify_webhook_url: + notifier = WebhookNotifier(settings.notify_webhook_url) + + actions = ActionExecutor(store=store, registry=registry, library=library, lifecycle=lifecycle, infra=infra, + notifier=notifier, dry_run=settings.dry_run, + enabled_infra_actions=set(healer_cfg.enabled_actions), on_skills_changed=router.refresh) + if intent_mapper is None and registry.has_slot("tier1-general"): + intent_mapper = LLMIntentMapper(chat, registry.slot("tier1-general").spec.litellm_model) + maintenance = MaintenanceAgent(store=store, registry=registry, library=library, index=router.index, + policy=policy, actions=actions, lifecycle=lifecycle, config=maint_cfg, + intent_mapper=intent_mapper, clock=clock, wallclock=wallclock) + healer = Healer(healer_cfg, store, actions, policy, propose=maintenance.propose, clock=clock) + return System(settings, store, registry, library, embedder, chat, router, executor, verifier, runner, policy, + lifecycle, actions, maintenance, healer, metrics) diff --git a/core/tests/conftest.py b/core/tests/conftest.py index 01ef6e1..24df380 100644 --- a/core/tests/conftest.py +++ b/core/tests/conftest.py @@ -176,3 +176,46 @@ class FakeChat: @pytest.fixture def fake_chat() -> FakeChat: return FakeChat() + + +class Clock: + def __init__(self, t: float | None = None): + import time + + self.t = time.time() if t is None else t + + def __call__(self) -> float: + return self.t + + def advance(self, seconds: float) -> None: + self.t += seconds + + +@pytest.fixture +def clock() -> Clock: + return Clock() + + +@pytest.fixture +def make_system(tmp_path, skills_dir, fake_chat, clock): + import yaml + + from knaster.embeddings import HashingEmbedder + from knaster.healer.drivers import DryRunDriver + from knaster.lifecycle import LoggingJobRunner + from knaster.settings import REPO_ROOT, Settings + from knaster.system import build_system + + def make(**overrides): + site = tmp_path / "site" + site.mkdir(exist_ok=True) + (site / "registry.yaml").write_text(yaml.safe_dump(REGISTRY)) + settings = Settings(db_url="sqlite://", site_dir=site, skills_dir=skills_dir, + config_dir=overrides.pop("config_dir", REPO_ROOT / "config"), + api_token=overrides.pop("api_token", "secret")) + kw = dict(embedder=HashingEmbedder(), chat=fake_chat, jobs=LoggingJobRunner(), notifier=DryRunDriver(), + clock=clock) + kw.update(overrides) + return build_system(settings, **kw) + + return make diff --git a/core/tests/test_lifecycle.py b/core/tests/test_lifecycle.py new file mode 100644 index 0000000..527795f --- /dev/null +++ b/core/tests/test_lifecycle.py @@ -0,0 +1,119 @@ +import pytest + +from knaster.lifecycle import Lifecycle, LifecycleConfig, LifecycleError, LoggingJobRunner +from knaster.registry import AdapterSpec + + +@pytest.fixture +def life(store, registry): + return Lifecycle(store, registry, LoggingJobRunner(), LifecycleConfig(min_n=10)) + + +def run_to_awaiting(life, cid, model="cand-model"): + life.start(cid) + life.report(cid, "build", {"ok": True, "model": model}) + life.report(cid, "eval", {"pass_rate": 0.95, "n": 40}) + life.report(cid, "shadow", {"agreement": 0.9, "n": 20}) + return life.report(cid, "canary", {"pass_rate": 0.93, "n": 12}) + + +def test_full_adapter_flow_and_rollback(life, registry): + cid = life.create("adapter", "tier1-general", {"cluster": "rechnungen"}) + cand = run_to_awaiting(life, cid, model="rechnungen-v2") + assert cand["state"] == "awaiting_promotion" + assert [j for j, _ in life.jobs.jobs] == ["build", "eval", "shadow"] + assert registry.slot("tier1-general").canary_model is None # Canary beendet + + life.promote(cid) + active = registry.active_adapter("base-14b", "rechnungen") + assert active.name == "rechnungen-v2" and active.version == 2 + assert {a.name: a.status for a in registry.adapters()}["rechnungen-v1"] == "retired" + + life.rollback(cid) + assert registry.active_adapter("base-14b", "rechnungen").name == "rechnungen-v1" + assert life.get(cid)["state"] == "rolled_back" + + +def test_canary_is_set_during_canary_phase(life, registry): + cid = life.create("adapter", "tier1-general", {"cluster": "rechnungen"}) + life.start(cid) + life.report(cid, "build", {"model": "rechnungen-v2"}) + life.report(cid, "eval", {"pass_rate": 0.95, "n": 40}) + life.report(cid, "shadow", {"agreement": 0.9, "n": 20}) + slot = registry.slot("tier1-general") + assert (slot.canary_model, slot.canary_cluster, slot.canary_share) == ("rechnungen-v2", "rechnungen", 0.1) + life.report(cid, "canary", {"pass_rate": 0.5, "n": 12}) + assert life.get(cid)["state"] == "rejected" + assert registry.slot("tier1-general").canary_model is None + + +def test_base_model_switch_retires_adapters_and_spawns_retraining(life, registry, store): + cid = life.create("base_model", "tier1-general", {"base_model": "base-32b"}) + run_to_awaiting(life, cid, model="tier1-general-candidate") + life.promote(cid) + assert registry.slot("tier1-general").base_model == "base-32b" + assert registry.active_adapter("base-14b", "rechnungen") is None + follow = [c for c in store.list_candidates() if c["kind"] == "adapter"] + assert len(follow) == 1 and follow[0]["target"] == {"cluster": "rechnungen", "base_model": "base-32b"} + + life.rollback(cid) + assert registry.slot("tier1-general").base_model == "base-14b" + assert registry.active_adapter("base-14b", "rechnungen").name == "rechnungen-v1" + assert life.get(follow[0]["id"])["state"] == "rejected" + + +def test_gates_reject(life): + cid = life.create("adapter", "tier1-general", {"cluster": "rechnungen"}, baseline={"pass_rate": 0.9}) + life.start(cid) + life.report(cid, "build", {"model": "m"}) + assert life.report(cid, "eval", {"pass_rate": 0.85, "n": 40})["state"] == "rejected" + + cid = life.create("adapter", "tier1-general", {"cluster": "rechnungen"}) + life.start(cid) + life.report(cid, "build", {"model": "m"}) + life.report(cid, "eval", {"pass_rate": 0.95, "n": 40}) + assert life.report(cid, "shadow", {"agreement": 0.95, "n": 3})["state"] == "rejected" # zu wenig Daten + + +def test_error_report_and_invalid_transitions(life): + cid = life.create("engine", "tier1-general", {"image": "vllm:v2"}) + with pytest.raises(LifecycleError): + life.report(cid, "build", {"ok": True}) # noch nicht gestartet + life.start(cid) + with pytest.raises(LifecycleError): + life.promote(cid) + assert life.report(cid, "build", {"error": "OOM"})["state"] == "failed" + with pytest.raises(LifecycleError): + life.create("adapter", "tier1-general", {}) + with pytest.raises(LifecycleError): + life.create("base_model", "nope", {"base_model": "x"}) + + +def test_canary_metrics_from_cases(life, store): + cid = life.create("adapter", "tier1-general", {"cluster": "rechnungen"}) + life.start(cid) + life.report(cid, "build", {"model": "rv2"}) + life.report(cid, "eval", {"pass_rate": 0.95, "n": 40}) + life.report(cid, "shadow", {"agreement": 0.9, "n": 20}) + for i in range(4): + store.add_case(input=f"x{i}", stage=1, slot="tier1-general", model="rv2", passed=i < 3, split="train", + origin="route", ts=life.get(cid)["updated_at"] + 1) + store.add_case(input="y", stage=1, slot="tier1-general", model="other", passed=False, split="train", + origin="route") + assert life.canary_metrics(cid) == {"pass_rate": 0.75, "n": 4} + + +def test_router_uses_canary_model(registry, library, embedder): + import random + + from knaster.router import Router + + from .test_router import RECHNUNG + + registry.set_canary("tier1-general", "rechnungen-v2", 1.0, "rechnungen") + r = Router(registry, library, embedder, rng=random.Random(1)) + d = r.route(RECHNUNG) + assert d.canary and d.model == d.adapter == "rechnungen-v2" + registry.set_canary("tier1-general", "rechnungen-v2", 0.0, "rechnungen") + assert not r.route(RECHNUNG).canary + registry.set_adapter(AdapterSpec(name="x", base_model="base-14b", cluster="andere")) diff --git a/core/tests/test_maintenance.py b/core/tests/test_maintenance.py new file mode 100644 index 0000000..65d857e --- /dev/null +++ b/core/tests/test_maintenance.py @@ -0,0 +1,210 @@ +from datetime import datetime + +import pytest + +from knaster.maintenance.agent import parse_command + + +@pytest.mark.parametrize("text,name,args", [ + ("Status", "status", {}), + ("starte die Service-Routine", "routine", {}), + ("zeige vorschläge", "list", {}), + ("genehmige P-3", "approve", {"ids": "P-3"}), + ("genehmige P-3, P-5 und p7", "approve", {"ids": "P-3, P-5 und p7"}), + ("approve p-2", "approve", {"ids": "p-2"}), + ("genehmige alle mit Risiko niedrig", "approve_low", {}), + ("lehne P-4 ab: zu teuer", "reject", {"id": "P-4", "reason": "zu teuer"}), + ("lehne P-4 ab", "reject", {"id": "P-4"}), + ("wechsle tier1-general auf Qwen/Qwen3-32B-AWQ", "switch_base", + {"slot": "tier1-general", "base": "Qwen/Qwen3-32B-AWQ"}), + ("trainiere einen Adapter für den Cluster rechnungen", "train_adapter", {"cluster": "rechnungen"}), + ("rollback tier1-general", "rollback", {"slot": "tier1-general"}), + ("stufe rechnung-extrahieren zurück", "demote", {"skill": "rechnung-extrahieren"}), + ("markiere tier1-flex als unavailable", "mark", {"slot": "tier1-flex", "status": "unavailable"}), + ("ja", "confirm", {}), +]) +def test_parse_command(text, name, args): + parsed = parse_command(text) + assert parsed is not None and parsed[0] == name + for k, v in args.items(): + assert parsed[1][k] == v + + +def test_parse_unknown(): + assert parse_command("mach mal was schönes") is None + + +def test_command_creates_proposal_and_needs_approval(make_system): + sysm = make_system() + m = sysm.maintenance + reply = m.handle_command("stufe rechnung-extrahieren zurück", user="robert") + assert "Bestätige" in reply.text + p = reply.proposals[0] + assert p.state == "open" # obwohl Richtlinie auto sagt: Prompt-Aufträge brauchen Bestätigung + assert sysm.library.get("rechnung-extrahieren").meta.status == "active" + reply = m.handle_command(f"genehmige {p.id}", user="robert") + assert "executed" in reply.text + assert sysm.library.get("rechnung-extrahieren").meta.status == "demoted" + assert m._load(p.seq).decided_by == "robert" + + +def test_reject_and_errors(make_system): + m = make_system().maintenance + p = m.handle_command("markiere tier1-flex als degraded").proposals[0] + assert "rejected" in m.handle_command(f"lehne {p.id} ab: nicht nötig").text + assert "Fehler" in m.handle_command(f"genehmige {p.id}").text + assert "Fehler" in m.handle_command("markiere gibtsnicht als ok").text + assert "Fehler" in m.handle_command("genehmige P-999").text + + +def test_switch_base_runs_pipeline_after_approval(make_system): + sysm = make_system() + m = sysm.maintenance + p = m.handle_command("wechsle tier1-general auf base-32b").proposals[0] + m.handle_command(f"genehmige {p.id}") + cands = sysm.store.list_candidates() + assert cands[0]["kind"] == "base_model" and cands[0]["state"] == "building" + assert sysm.lifecycle.jobs.jobs[0][0] == "build" + + +def test_intent_mapper_requires_confirmation(make_system): + class Mapper: + def map(self, text): + return "stufe vertrag-pruefen zurück" if "vertrag" in text else None + + sysm = make_system(intent_mapper=Mapper()) + m = sysm.maintenance + reply = m.handle_command("der vertrag skill macht nur mist", session="s1") + assert "Meinst du" in reply.text + assert sysm.store.list_proposals() == [] + reply = m.handle_command("ja", session="s1") + assert reply.proposals and reply.proposals[0].state == "open" + assert "nicht verstanden" in m.handle_command("blabla").text + assert "nichts zu bestätigen" in m.handle_command("ja", session="s1").text + + +def test_llm_intent_mapper(fake_chat): + from knaster.maintenance.agent import LLMIntentMapper + + fake_chat.handlers["tier1-general"] = lambda msgs: "rollback tier1-general" + assert LLMIntentMapper(fake_chat).map("mach das letzte update rückgängig") == "rollback tier1-general" + fake_chat.handlers["tier1-general"] = lambda msgs: "NONE" + assert LLMIntentMapper(fake_chat).map("hallo") is None + + +def test_deferred_until_window(make_system): + sysm = make_system(wallclock=lambda: datetime(2026, 1, 1, 12, 0)) + m = sysm.maintenance + p = m.propose("reboot_host", {"pve_node": "pve1"}, "Host neu starten", source="healer") + p = m.approve(p.seq) + assert p.state == "approved" and p.result["deferred"] + assert m.run_deferred() == [] + m.wallclock = lambda: datetime(2026, 1, 2, 3, 0) + done = m.run_deferred() + assert done[0].state == "executed" and "[dry-run] reboot_host" in done[0].result["message"] + + +def test_dedup_of_open_proposals(make_system): + m = make_system().maintenance + a = m.propose("merge_skills", {"a": "x", "b": "y"}, "merge") + b = m.propose("merge_skills", {"a": "x", "b": "y"}, "merge") + assert a.seq == b.seq + + +def _cases(store, n, *, skill, stage, passed_ratio, slot="tier1-general", cluster=None, split="train", + tokens=None): + for i in range(n): + store.add_case(input=f"{skill}-{stage}-{i}", skill=skill, cluster=cluster or skill, stage=stage, slot=slot, + passed=i < n * passed_ratio, split=split, origin="route", prompt_tokens=tokens) + + +def test_routine_drift_auto_demotes(make_system): + sysm = make_system() + _cases(sysm.store, 30, skill="rechnung-extrahieren", cluster="rechnungen", stage=1, passed_ratio=0.5) + report = sysm.maintenance.run_routine() + assert any("Drift" in f for f in report.findings) + assert any("rechnung-extrahieren" in a for a in report.auto_executed) + assert sysm.library.get("rechnung-extrahieren").meta.status == "demoted" + # Router sieht die Änderung sofort + from .test_router import RECHNUNG + assert sysm.router.route(RECHNUNG).stage == 0 + + +def test_routine_graduation_candidate(make_system): + sysm = make_system() + _cases(sysm.store, 25, skill="vertrag-pruefen", stage=0, passed_ratio=1.0, slot="tier0-strong") + report = sysm.maintenance.run_routine() + kinds = {p.kind for p in report.proposals} + assert "request_graduation" in kinds + p = next(p for p in report.proposals if p.kind == "request_graduation") + sysm.maintenance.approve(p.seq) + assert sysm.lifecycle.jobs.jobs[-1][0] == "graduate_skill" + + +def test_routine_degraded_slot_and_context(make_system, clock): + sysm = make_system() + sysm.registry.set_status("tier1-flex", "unavailable", reason="Training") + for _ in range(12): + sysm.store.add_event("compact_prompt", subject="tier1-general", ts=clock()) + clock.advance(2 * 3600) + _cases(sysm.store, 30, skill="rechnung-extrahieren", stage=1, passed_ratio=1.0, cluster="rechnungen", + tokens=1000) + report = sysm.maintenance.run_routine() + assert any("tier1-flex" in f for f in report.findings) + assert any(p.kind == "mark_slot" for p in report.proposals) + assert any("effektiver Kontext = 2048" in a for a in report.auto_executed) + assert sysm.registry.slot("tier1-general").calibrated_context == 2048 + assert "Offene Vorschläge" in report.text + + +def test_routine_adapter_need(make_system): + sysm = make_system() + sysm.maintenance.config.adapter_min_cases = 20 + _cases(sysm.store, 40, skill="rechnung-extrahieren", stage=0, passed_ratio=1.0, cluster="rechnungen", + slot="tier0-strong") + _cases(sysm.store, 10, skill="rechnung-extrahieren", stage=1, passed_ratio=0.7, cluster="rechnungen") + report = sysm.maintenance.run_routine() + p = next(p for p in report.proposals if p.kind == "train_adapter") + assert p.params == {"cluster": "rechnungen", "slot": "tier1-general"} + sysm.maintenance.approve(p.seq) + cand = sysm.store.list_candidates()[0] + assert cand["kind"] == "adapter" and cand["target"]["base_model"] == "base-14b" + # zweite Routine schlägt nicht erneut vor + assert not any(p.kind == "train_adapter" for p in sysm.maintenance.run_routine().proposals) + + +def test_routine_merge_similar_skills(make_system): + sysm = make_system() + sysm.maintenance.config.merge_similarity = 0.0 + report = sysm.maintenance.run_routine() + assert any(p.kind == "merge_skills" for p in report.proposals) + + +def test_routine_evaluates_canary_and_proposes_promotion(make_system): + sysm = make_system() + life = sysm.lifecycle + life.config.min_n = 5 + cid = life.create("adapter", "tier1-general", {"cluster": "rechnungen"}) + life.start(cid) + life.report(cid, "build", {"model": "rv2"}) + life.report(cid, "eval", {"pass_rate": 0.95, "n": 40}) + life.report(cid, "shadow", {"agreement": 0.9, "n": 20}) + for i in range(6): + sysm.store.add_case(input=f"c{i}", stage=1, slot="tier1-general", model="rv2", passed=True, + split="train", origin="route") + report = sysm.maintenance.run_routine() + assert any("Canary" in a for a in report.auto_executed) + p = next(p for p in report.proposals if p.kind == "promote_candidate") + assert p.risk == "high" + sysm.maintenance.approve(p.seq) + assert sysm.registry.active_adapter("base-14b", "rechnungen").name == "rv2" + # rollback per Prompt + m = sysm.maintenance + rp = m.handle_command("rollback tier1-general").proposals[0] + m.approve(rp.seq) + assert sysm.registry.active_adapter("base-14b", "rechnungen").name == "rechnungen-v1" + + +def test_status_command(make_system): + text = make_system().maintenance.handle_command("status").text + assert "tier1-general: ok" in text and "Skills:" in text diff --git a/core/tests/test_policy_healer.py b/core/tests/test_policy_healer.py new file mode 100644 index 0000000..0cefbf3 --- /dev/null +++ b/core/tests/test_policy_healer.py @@ -0,0 +1,124 @@ +from datetime import datetime + +from knaster.healer.engine import HealerConfig +from knaster.policy import Policy +from knaster.util import fill_template, parse_duration + + +def test_policy_modes_and_windows(): + p = Policy(**{"default": "approve", "actions": { + "a": {"mode": "auto"}, "night": {"mode": "approve", "window": "22:00-04:00"}, + "day": {"mode": "auto", "window": "08:00-17:00"}}}) + assert p.decide("a") == "auto" and p.decide("unknown") == "approve" + assert p.in_window("night", datetime(2026, 1, 1, 23, 0)) + assert p.in_window("night", datetime(2026, 1, 1, 3, 0)) + assert not p.in_window("night", datetime(2026, 1, 1, 12, 0)) + assert p.in_window("day", datetime(2026, 1, 1, 9, 0)) + assert p.in_window("a") + + +def test_repo_configs_load(): + from knaster.settings import REPO_ROOT + + cfg = HealerConfig.from_file(REPO_ROOT / "config" / "healer-rules.yaml") + assert {r.id for r in cfg.rules} >= {"slot-endpoint-down", "gpu-xid"} + assert cfg.enabled_actions == [] # sicherer Default + pol = Policy.from_file(REPO_ROOT / "config" / "policy.yaml") + assert pol.decide("promote_candidate") == "approve" and pol.decide("demote_skill") == "auto" + + +def test_util(): + assert parse_duration("5m") == 300 and parse_duration(10) == 10 and parse_duration("2h") == 7200 + assert fill_template({"a": "{x}-{y}", "b": ["{x}"]}, {"x": "1", "y": "2"}) == {"a": "1-2", "b": ["1"]} + + +def alert(status="firing", **labels): + base = {"alertname": "SlotEndpointDown", "slot": "tier1-general", "docker_host": "gpu-main", + "container": "vllm", "pve_node": "pve1", "vmid": "201"} + base.update(labels) + return {"status": status, "labels": base, "fingerprint": "fp1"} + + +def test_healer_ladder_cooldown_and_resolve(make_system, clock): + sysm = make_system() + h = sysm.healer + dry = sysm.actions.dry + out = h.handle_alert(alert()) + assert (out.status, out.action) == ("executed", "restart_container") + assert out.message.startswith("[dry-run]") + assert h.handle_alert(alert()).status == "cooldown" + clock.advance(301) + assert h.handle_alert(alert()).action == "restart_container" + clock.advance(301) + assert h.handle_alert(alert()).action == "restart_vm" + clock.advance(901) + out = h.handle_alert(alert()) + assert out.action == "mark_slot" + slot = sysm.registry.slot("tier1-general") + assert slot.status == "degraded" and slot.set_by == "healer" + out = h.handle_alert(alert()) + assert out.status == "exhausted" + assert h.handle_alert(alert()).message == "bereits gemeldet" + assert [c[0] for c in dry.calls].count("restart_container") == 2 + + out = h.handle_alert(alert(status="resolved")) + assert out.status == "resolved" + assert sysm.registry.slot("tier1-general").status == "ok" + + +def test_healer_does_not_clear_human_degrade(make_system): + sysm = make_system() + sysm.registry.set_status("tier1-general", "degraded", set_by="human") + sysm.healer.handle_alert(alert()) + sysm.healer.handle_alert(alert(status="resolved")) + assert sysm.registry.slot("tier1-general").status == "degraded" + + +def test_healer_approval_step_creates_proposal(make_system, clock): + sysm = make_system() + a = alert(alertname="GpuXidError") + assert sysm.healer.handle_alert(a).action == "mark_slot" + assert sysm.healer.handle_alert(a).action == "restart_vm" + clock.advance(1201) + out = sysm.healer.handle_alert(a) + assert out.status == "proposed" and out.action == "reboot_host" + props = sysm.maintenance.proposals() + assert props[0].kind == "reboot_host" and props[0].source == "healer" + + +def test_healer_missing_label_and_unmatched(make_system): + sysm = make_system() + a = alert() + del a["labels"]["container"] + out = sysm.healer.handle_alert(a) + assert out.status == "skipped" and "container" in out.message + assert sysm.healer.handle_alert({"status": "firing", "labels": {"alertname": "Neu"}}).status == "unmatched" + + +def test_healer_throttle(make_system, clock): + sysm = make_system() + sysm.healer.config.max_actions_per_hour = 2 + for i in range(2): + sysm.healer.handle_alert({**alert(), "fingerprint": f"fp{i}"}) + out = sysm.healer.handle_alert({**alert(), "fingerprint": "fp9"}) + assert out.status == "throttled" + + +def test_infra_actions_need_explicit_enable(make_system): + from knaster.healer.drivers import DryRunDriver + + real = DryRunDriver() + sysm = make_system(infra=real) + sysm.actions.dry_run = False + sysm.healer.handle_alert(alert()) + assert real.calls == [] # nicht in enabled_actions + sysm.actions.enabled_infra_actions = {"restart_container"} + sysm.healer.handle_alert({**alert(), "fingerprint": "other"}) + assert real.calls[0][0] == "restart_container" + + +def test_alertmanager_payload(make_system): + sysm = make_system() + outs = sysm.healer.handle_alertmanager({"version": "4", "alerts": [alert(), alert(alertname="NodeDown", + instance="pve2")]}) + assert [o.status for o in outs] == ["executed", "executed"] diff --git a/docs/spec/SPEC-maintenance.md b/docs/spec/SPEC-maintenance.md index ff6a48f..18fa29b 100644 --- a/docs/spec/SPEC-maintenance.md +++ b/docs/spec/SPEC-maintenance.md @@ -6,9 +6,9 @@ Paket: `knaster.maintenance`, `knaster.policy`. ``` id: "P-" -kind: promote_skill | demote_skill | merge_skills | retire_skill | mark_slot | restart_container | - restart_vm | reboot_host | create_candidate | promote_candidate | rollback_candidate | - train_adapter | recalibrate_context +kind: request_graduation | promote_skill | demote_skill | merge_skills | retire_skill | mark_slot | + restart_container | restart_vm | reboot_host | notify | create_candidate | promote_candidate | + rollback_candidate | train_adapter | recalibrate_context params: {...} summary: str # menschenlesbar reason: str @@ -33,7 +33,8 @@ actions: reboot_host: {mode: approve, window: "02:00-05:00"} ``` -`Policy.decide(kind, now) -> auto | approve | deny`. Liegt eine Aktion mit `window` außerhalb des Zeitfensters, wird sie zwar freigegeben, aber erst im nächsten Fenster ausgeführt (`deferred`). Ein Vorschlag mit Modus `auto` wird sofort ausgeführt und mit `decided_by = policy` protokolliert. +`Policy.decide(kind) -> auto | approve | deny`, dazu `Policy.in_window(kind, now)`. +Ausnahme unabhängig von der Richtlinie: `mark_slot` mit `status: ok` (Slot wieder freischalten) braucht immer eine Freigabe, weil es die unsichere Richtung ist. Liegt eine Aktion mit `window` außerhalb des Zeitfensters, wird sie zwar freigegeben, aber erst im nächsten Fenster ausgeführt (`deferred`). Ein Vorschlag mit Modus `auto` wird sofort ausgeführt und mit `decided_by = policy` protokolliert. ## 3. Service-Routine (`MaintenanceAgent.run_routine`) @@ -42,12 +43,13 @@ Befunde werden aus diesen Quellen gesammelt (alle Grenzwerte in `config/maintena | Befund | Bedingung | Vorschlag | |---|---|---| | Skill-Drift | Pass-Rate der letzten N Fälle auf Stufe ≥ 1 < `demote_below` (0,80), n ≥ `min_cases` (20) | `demote_skill` (auto) | -| Graduierungs-Kandidat | Skill auf Stufe 0, ≥ `min_cases` Fälle, Pass-Rate auf Stufe 0 ≥ 0,95 | `promote_skill` → Graduierungslauf (approve) | +| Graduierungs-Kandidat | Skill auf Stufe 0, ≥ `min_cases` Fälle, Pass-Rate auf Stufe 0 ≥ `promote_min_rate` (0,95) | `request_graduation` (approve): Auftrag an Hermes, Varianten zu erzeugen und `POST /v1/skills/{name}/graduate` aufzurufen. Ein erfolgreicher Lauf erzeugt `promote_skill` (approve) | | Ähnliche Skills | cos(description_a, description_b) ≥ `merge_similarity` (0,92) | `merge_skills` (approve) | | Slot degradiert | Laufzeitzustand ≠ ok seit > 1 h | Hinweis + `mark_slot ok` nach Prüfung (approve) | | Kontext-Ausweicher | Signal `context_exceeded` oder `output_truncated` ≥ `context_signal_limit` (10) pro Slot/Woche | `recalibrate_context` (auto) | | Adapter-Bedarf | Cluster mit ≥ `adapter_min_cases` (300) verifizierten Train-Fällen und Pass-Rate auf Stufe 1 < 0,9 | `train_adapter` (approve) | -| Offene Kandidaten | Lifecycle-Kandidat wartet auf Freigabe | Verweis auf `promote_candidate` | +| Canary auswerten | Kandidat im Zustand `canary` mit ≥ `min_n` Fällen über das Canary-Modell | automatischer Report `canary` an die Pipeline | +| Offene Kandidaten | Lifecycle-Kandidat wartet auf Freigabe | `promote_candidate` (approve, Risiko hoch) | Das Ergebnis ist ein `RoutineReport{findings[], proposals[], auto_executed[], text}`. `text` ist ein deutscher Bericht für den Chat. diff --git a/sites/knaster/registry.yaml b/sites/knaster/registry.yaml new file mode 100644 index 0000000..0bd0613 --- /dev/null +++ b/sites/knaster/registry.yaml @@ -0,0 +1,76 @@ +# Registry für den Cluster "knaster" (SPEC-core §1). +# Clients sprechen nur Slot-Namen an; litellm_model ist der Alias in deploy/litellm/config.yaml. +slots: + tier0-strong: + tier: 0 + device: external + engine: openrouter + litellm_model: tier0-strong + base_model: openrouter-strong # konkretes Modell steht nur in der LiteLLM-Konfiguration + max_context: 200000 + effective_context: 120000 + + tier1-general: + tier: 1 + device: gpu + engine: vllm + node: gpu-main # RTX 4000 Ada #1 + litellm_model: tier1-general + base_model: qwen3-14b-awq # Startkandidat, Entscheidung per Kalibrierungslauf + max_context: 32768 + effective_context: 16384 + supports_lora: true + fallbacks: [tier1-flex] + + tier1-flex: + tier: 1 + device: gpu + engine: vllm + node: gpu-flex # RTX 4000 Ada #2 (nachts Training → unavailable) + litellm_model: tier1-flex + base_model: qwen3-14b-awq + max_context: 32768 + effective_context: 16384 + supports_lora: true + + verifier-small: + tier: 1 + device: gpu + engine: llamacpp + node: gpu-small # Quadro M4000 (Nutzen per Benchmark prüfen) + litellm_model: verifier-small + base_model: qwen3-4b-q4km + max_context: 8192 + effective_context: 6144 + + signature-extractor: + tier: 1 + device: gpu + engine: llamacpp + node: gpu-small + litellm_model: signature-extractor + base_model: qwen3-4b-q4km + max_context: 8192 + effective_context: 6144 + + embed-default: + tier: 1 + device: gpu + engine: llamacpp + node: gpu-embed # Quadro P2000 + litellm_model: embed-default + base_model: qwen3-embedding-0.6b-q8 + max_context: 8192 + kind: embedding + + rerank-default: + tier: 1 + device: gpu + engine: llamacpp + node: gpu-embed + litellm_model: rerank-default + base_model: bge-reranker-v2-m3-q8 + max_context: 8192 + kind: rerank + +adapters: [] diff --git a/skills/rechnung-extrahieren/SKILL.md b/skills/rechnung-extrahieren/SKILL.md new file mode 100644 index 0000000..de48791 --- /dev/null +++ b/skills/rechnung-extrahieren/SKILL.md @@ -0,0 +1,47 @@ +--- +name: rechnung-extrahieren +description: Extrahiert Rechnungsnummer, Rechnungsdatum, Netto-, Steuer- und Bruttobetrag sowie den Rechnungssteller aus dem Text einer Rechnung als JSON. +metadata: + knaster: + version: 1 + status: draft # wird nach erfolgreicher Graduierung "active", stage 1 + stage: 0 + cluster: rechnungen + slot: tier1-general + threshold: 0.80 + triggers: + - "Lies die Rechnungsdaten aus diesem PDF-Text aus" + - "Welche Rechnungsnummer und welcher Betrag stehen auf der Rechnung?" + max_output_tokens: 400 + output: + format: json + schema: + type: object + required: [rechnungsnummer, datum, brutto, waehrung] + properties: + rechnungsnummer: {type: string} + datum: {type: string, pattern: "^\\d{4}-\\d{2}-\\d{2}$"} + netto: {type: [number, "null"]} + steuer: {type: [number, "null"]} + brutto: {type: number} + waehrung: {type: string, minLength: 3, maxLength: 3} + steller: {type: [string, "null"]} + verifier: + - type: judge + slot: verifier-small + rubric: "Stimmen Rechnungsnummer und Bruttobetrag exakt mit dem Text überein? Netto + Steuer = Brutto, falls angegeben." +--- +# Rechnung extrahieren + +Du extrahierst strukturierte Daten aus dem Text einer Rechnung. + +## Regeln +1. Gib nur Werte aus, die wörtlich im Text stehen. Erfinde nichts; fehlende optionale Werte sind `null`. +2. Datum immer als `YYYY-MM-DD`. +3. Beträge als Zahl mit Punkt als Dezimaltrenner, ohne Tausendertrennzeichen und ohne Währungssymbol. +4. Währung als ISO-Code (EUR, CHF, USD). +5. Gibt es mehrere Beträge, ist `brutto` der zu zahlende Gesamtbetrag. + +## Beispiele +Eingabe: „Rechnung Nr. RE-2024-118 vom 03.04.2024 … Nettobetrag 1.000,00 € zzgl. 19 % MwSt. 190,00 € Gesamtbetrag 1.190,00 €“ +Ausgabe: {"rechnungsnummer": "RE-2024-118", "datum": "2024-04-03", "netto": 1000.0, "steuer": 190.0, "brutto": 1190.0, "waehrung": "EUR", "steller": null} diff --git a/skills/ticket-kategorisieren/SKILL.md b/skills/ticket-kategorisieren/SKILL.md new file mode 100644 index 0000000..566ee27 --- /dev/null +++ b/skills/ticket-kategorisieren/SKILL.md @@ -0,0 +1,22 @@ +--- +name: ticket-kategorisieren +description: Ordnet ein IT-Support-Ticket genau einer Kategorie zu (drucker, netzwerk, zugang, hardware, software, sonstiges). +metadata: + knaster: + version: 1 + status: draft + stage: 0 + execution: classifier # Ziel: Stufe 3, sobald genug gelabelte Fälle für den Classifier vorliegen + classifier: ticket-kategorie + min_confidence: 0.7 + llm_fallback: true + slot: tier1-general + threshold: 0.78 + max_output_tokens: 10 + verifier: + - {type: regex, pattern: "^(drucker|netzwerk|zugang|hardware|software|sonstiges)$"} +--- +# Ticket kategorisieren + +Antworte mit genau einem Wort aus: drucker, netzwerk, zugang, hardware, software, sonstiges. +Passwort- und Berechtigungsprobleme sind `zugang`. diff --git a/skills/zusammenfassung/SKILL.md b/skills/zusammenfassung/SKILL.md new file mode 100644 index 0000000..8c415a9 --- /dev/null +++ b/skills/zusammenfassung/SKILL.md @@ -0,0 +1,22 @@ +--- +name: zusammenfassung +description: Fasst einen deutschen Text sachlich in höchstens fünf Stichpunkten zusammen. +metadata: + knaster: + version: 1 + status: draft + stage: 0 + slot: tier1-general + threshold: 0.80 + max_output_tokens: 300 + verifier: + - {type: regex, pattern: "^- "} + - {type: max_chars, value: 1200} + - type: judge + slot: verifier-small + rubric: "Gibt die Zusammenfassung nur Inhalte des Textes wieder (keine Erfindungen) und enthält sie die Kernaussage?" +--- +# Zusammenfassung + +Fasse den Text in höchstens fünf Stichpunkten zusammen. Jeder Stichpunkt beginnt mit „- “. +Keine Einleitung, keine Wertung, nichts erfinden.