"""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"