Hitonabi
1216 lines
57 KiB
Python
1216 lines
57 KiB
Python
"""
|
||
Ideen-Queue — DIE eine Sammelstelle für Ideen des Commanders (Entscheid 10.07.2026).
|
||
|
||
Die Queue selbst ist das NATIVE Hermes-Kanban (SQLite-Board, Dispatcher im Gateway):
|
||
Idee rein (triage) → der Specifier arbeitet sie aus → ein Worker-Profil setzt sie um →
|
||
Code-Ergebnisse kommen als Vorschlags-Branch zurück und erscheinen als Karte im
|
||
Auftragsbuch. Dieser Service ist nur die MC2-Tür dazu: anlegen + anzeigen + aufräumen,
|
||
alles über die Hermes-CLI (stabile --json-Schnittstelle, kein Griff in die kanban.db).
|
||
|
||
Türen insgesamt: Telegram/Desktop (natives kanban_create-Tool), Lucy-Voice
|
||
(mcp_voice.idee_notieren → POST /api/ideen) und die Zentrale (Auftragsbuch-Tab → hier).
|
||
|
||
Lokal (Windows-Dev) ist alles harmlos: available=False, Aktionen geben Fehler statt zu crashen.
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import subprocess
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import httpx
|
||
from config import GATEWAY_URL, V1_UPSTREAM
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
HERMES = Path(os.environ.get("MC_HERMES_BIN", "~/.local/bin/hermes")).expanduser()
|
||
# Ketten-Sicht (20.07.): Eltern-Links + Heartbeat-Notizen stehen NICHT im `list --json`
|
||
# der CLI — dafür (und NUR dafür) ein read-only-Blick in die kanban.db. Das bricht die
|
||
# „alles über die CLI"-Regel bewusst minimal: zwei stabile Tabellen (task_links 2 Spalten,
|
||
# task_events), Verbindung strikt mode=ro, jeder Fehler fällt lautlos auf die flache
|
||
# Sicht zurück. Schreiben läuft weiterhin ausschließlich über die CLI.
|
||
KANBAN_DB = Path(os.environ.get("MC_KANBAN_DB", "~/.hermes/kanban.db")).expanduser()
|
||
|
||
_TASK_ID_RX = re.compile(r"^t_[0-9a-f]{4,16}$")
|
||
|
||
# Die CLI kostet pro Aufruf ein paar Sekunden Python-Start — die UI pollt aber.
|
||
_list_cache: dict = {"ts": 0.0, "data": None}
|
||
_LIST_EVERY = 15.0 # s
|
||
|
||
# Cache für worker log
|
||
_log_cache: dict = {"ts": 0.0, "task_id": None, "data": None}
|
||
_LOG_EVERY = 4.0 # s
|
||
|
||
# Cache für Ergebnis (Abschluss-Zusammenfassung) — ändert sich nach done nie mehr
|
||
_erg_cache: dict = {}
|
||
_ERG_MAX = 40 # simple Deckelung gegen unbegrenztes Wachsen
|
||
|
||
|
||
def _available() -> bool:
|
||
return os.name == "posix" and HERMES.is_file()
|
||
|
||
|
||
def _hermes(args: list[str], timeout: int = 60) -> subprocess.CompletedProcess:
|
||
return subprocess.run([str(HERMES), "kanban", *args],
|
||
capture_output=True, text=True, timeout=timeout)
|
||
|
||
|
||
def _parse_json(stdout: str):
|
||
"""CLI-Ausgabe kann Log-Zeilen vor dem JSON enthalten — ab der ersten Klammer parsen."""
|
||
text = stdout or ""
|
||
for opener in ("[", "{"):
|
||
idx = text.find(opener)
|
||
if idx >= 0:
|
||
try:
|
||
return json.loads(text[idx:])
|
||
except json.JSONDecodeError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _blocker_frage(task_id: str) -> dict | None:
|
||
"""Für eine blockierte Aufgabe: die Frage/den Grund des Workers holen (needs_input etc.).
|
||
|
||
Die Wahrheit steckt im jüngsten `blocked`-Event (payload.reason/kind); fällt das leer aus,
|
||
greifen wir auf den letzten „BLOCKED:"-Kommentar zurück. Ein Extra-`show`-Aufruf pro
|
||
hängender Karte ist billig — die gibt es fast nie und wenn, dann nur eine Handvoll.
|
||
"""
|
||
try:
|
||
r = _hermes(["show", task_id, "--json"], timeout=30)
|
||
d = _parse_json(r.stdout)
|
||
except Exception:
|
||
log.warning("ideen: kanban show %s fehlgeschlagen", task_id, exc_info=True)
|
||
return None
|
||
if not isinstance(d, dict):
|
||
return None
|
||
|
||
frage, kind = "", ""
|
||
for ev in d.get("events") or []: # chronologisch → letztes blocked = aktive Frage
|
||
if isinstance(ev, dict) and ev.get("kind") == "blocked":
|
||
p = ev.get("payload") or {}
|
||
frage = (p.get("reason") or "").strip()
|
||
kind = (p.get("kind") or "").strip()
|
||
if not frage: # Fallback: letzter „BLOCKED:"-Kommentar
|
||
for c in d.get("comments") or []:
|
||
body = (c.get("body") or "").strip()
|
||
if body.startswith("BLOCKED:"):
|
||
frage = body[len("BLOCKED:"):].strip()
|
||
return {"frage": frage[:800], "kind": kind}
|
||
|
||
|
||
def _kurz(text: str, max_len: int = 80) -> str:
|
||
"""Auf max_len kürzen — aber an der WORTGRENZE, nicht mitten im Wort (User 21.07.:
|
||
„Idee: Recherchiere zu ARM … und lass uns überlegen wie")."""
|
||
t = " ".join((text or "").split())
|
||
if len(t) <= max_len:
|
||
return t
|
||
schnitt = t[:max_len].rsplit(" ", 1)[0].rstrip(" ,;:–-")
|
||
return (schnitt or t[:max_len]) + "…"
|
||
|
||
|
||
_TITEL_PRAEFIX_RX = re.compile(r"^\s*(idee|projekt|bitte)\s*[:\-–]?\s*", re.IGNORECASE)
|
||
|
||
|
||
def _projekt_titel(titles: list[str]) -> str:
|
||
"""Anzeigename einer Karten-Familie: gemeinsamer Titel-Präfix (z. B. „Homelab-Dashboard 2.0"),
|
||
sonst der AUFGERÄUMTESTE Mitglieds-Titel. Reine Anzeige-Heuristik, nichts hängt davon ab.
|
||
|
||
Der kürzeste Titel gewinnt, weil die gepflegten Karten (aus einem Konzept abgeleitet)
|
||
kurz sind und der rohe Zuruf des Commanders lang — vorher nahm die Gruppe stur die
|
||
älteste Karte und schnitt deren Textwüste bei 80 Zeichen mitten im Wort ab.
|
||
"""
|
||
if not titles:
|
||
return "Projekt"
|
||
prefix = titles[0]
|
||
for t in titles[1:]:
|
||
while prefix and not t.startswith(prefix):
|
||
prefix = prefix[:-1]
|
||
prefix = prefix.strip(" –-:·")
|
||
if len(prefix) >= 10:
|
||
return _kurz(prefix)
|
||
return _kurz(_TITEL_PRAEFIX_RX.sub("", min(titles, key=len)))
|
||
|
||
|
||
def _ketten_anreichern(items: list[dict]) -> list[dict]:
|
||
"""Ketten-Sicht über die Queue legen (read-only, best-effort).
|
||
|
||
Ergänzt pro Karte: `eltern` (offene Vorgänger-IDs), `wartet_auf` (deren Titel),
|
||
`notiz` (jüngste Heartbeat-Notiz des Workers bei running) und `projekt`
|
||
(Familien-Schlüssel). Liefert die Projekt-Gruppen für die UI: zusammenhängende
|
||
Karten (task_links) = ein Projekt; Einzelkarten bleiben ohne `projekt`.
|
||
Jeder Fehler (Windows-Dev, Schema-Drift, Lock) → unveränderte flache Sicht.
|
||
"""
|
||
for it in items:
|
||
it.setdefault("eltern", [])
|
||
it.setdefault("wartet_auf", [])
|
||
it.setdefault("notiz", None)
|
||
it.setdefault("projekt", None)
|
||
it.setdefault("prio", 0)
|
||
it.setdefault("puls_alter", None)
|
||
it.setdefault("gestartet", None)
|
||
if not items or not KANBAN_DB.is_file():
|
||
return []
|
||
by_id = {it["id"]: it for it in items if it.get("id")}
|
||
try:
|
||
con = sqlite3.connect(f"file:{KANBAN_DB}?mode=ro", uri=True, timeout=3)
|
||
try:
|
||
con.execute("PRAGMA busy_timeout=2000")
|
||
links = [(p, c) for p, c in con.execute(
|
||
"SELECT parent_id, child_id FROM task_links")
|
||
if p in by_id or c in by_id]
|
||
# Prio (für die Tausch-Knöpfe) + Lebenszeichen (Hänger-Warnung): ein Rutsch.
|
||
ph = ",".join("?" * len(by_id))
|
||
jetzt = time.time()
|
||
for tid, prio, hb, started, pid in con.execute(
|
||
f"SELECT id, priority, last_heartbeat_at, started_at, worker_pid"
|
||
f" FROM tasks WHERE id IN ({ph})", list(by_id)):
|
||
it = by_id[tid]
|
||
it["prio"] = prio or 0
|
||
if it.get("status") == "running":
|
||
it["gestartet"] = started
|
||
puls = hb or started
|
||
it["puls_alter"] = int(jetzt - puls) if puls else None
|
||
for it in items:
|
||
if it.get("status") != "running":
|
||
continue
|
||
for (payload,) in con.execute(
|
||
"SELECT payload FROM task_events WHERE task_id=? AND kind='heartbeat' "
|
||
"AND payload IS NOT NULL ORDER BY created_at DESC LIMIT 5", (it["id"],)):
|
||
try:
|
||
note = (json.loads(payload) or {}).get("note")
|
||
except Exception:
|
||
note = None
|
||
if note:
|
||
it["notiz"] = str(note)[:300]
|
||
break
|
||
finally:
|
||
con.close()
|
||
except Exception:
|
||
log.debug("ideen: Ketten-Anreicherung fehlgeschlagen", exc_info=True)
|
||
return []
|
||
|
||
# Eltern/wartet_auf: nur Vorgänger zählen, die selbst noch offen in der Liste stehen —
|
||
# archivierte/fremde Parents gelten als erledigt.
|
||
ERLEDIGT = ("done",)
|
||
for parent, child in links:
|
||
it = by_id.get(child)
|
||
p = by_id.get(parent)
|
||
if it and p and p.get("status") not in ERLEDIGT:
|
||
it["eltern"].append(parent)
|
||
if len(it["wartet_auf"]) < 3:
|
||
it["wartet_auf"].append((p.get("titel") or parent)[:90])
|
||
|
||
# Familien = zusammenhängende Komponenten über die Links (Union-Find, klein genug).
|
||
chef: dict[str, str] = {}
|
||
|
||
def boss(x: str) -> str:
|
||
while chef.get(x, x) != x:
|
||
chef[x] = chef.get(chef[x], chef[x])
|
||
x = chef[x]
|
||
return x
|
||
|
||
for parent, child in links:
|
||
if parent in by_id and child in by_id:
|
||
a, b = boss(parent), boss(child)
|
||
if a != b:
|
||
chef[a] = b
|
||
|
||
familien: dict[str, list[dict]] = {}
|
||
for it in items:
|
||
familien.setdefault(boss(it["id"]), []).append(it)
|
||
projekte = []
|
||
for mitglieder in familien.values():
|
||
if len(mitglieder) < 2:
|
||
continue
|
||
mitglieder.sort(key=lambda i: i.get("erstellt") or 0)
|
||
key = mitglieder[0]["id"]
|
||
for it in mitglieder:
|
||
it["projekt"] = key
|
||
fertig = sum(1 for i in mitglieder if i["status"] in ERLEDIGT)
|
||
projekte.append({
|
||
"key": key,
|
||
"titel": _projekt_titel([i["titel"] for i in mitglieder]),
|
||
"gesamt": len(mitglieder),
|
||
"fertig": fertig,
|
||
"laeuft": sum(1 for i in mitglieder if i["status"] == "running"),
|
||
"haengt": sum(1 for i in mitglieder if i["status"] == "blocked"),
|
||
})
|
||
# Aktive Projekte zuerst (hängend > laufend > Rest), dann nach Größe.
|
||
projekte.sort(key=lambda p: (-p["haengt"], -p["laeuft"], -p["gesamt"]))
|
||
return projekte
|
||
|
||
|
||
def list_queue() -> dict:
|
||
"""Alle nicht archivierten Aufgaben des Boards, jüngste zuerst — die Queue-Sicht der UI."""
|
||
if not _available():
|
||
return {"available": False, "items": [], "offen": 0}
|
||
|
||
now = time.time()
|
||
if _list_cache["data"] is not None and now - _list_cache["ts"] < _LIST_EVERY:
|
||
return _list_cache["data"]
|
||
|
||
try:
|
||
r = _hermes(["list", "--json", "--sort", "created-desc"])
|
||
raw = _parse_json(r.stdout)
|
||
except Exception:
|
||
log.warning("ideen: kanban list fehlgeschlagen", exc_info=True)
|
||
raw = None
|
||
if not isinstance(raw, list):
|
||
# CLI kaputt/Timeout → ehrlich leer melden, aber nicht cachen (nächster Poll versucht's neu)
|
||
return {"available": True, "items": [], "offen": 0, "fehler": "Queue nicht lesbar"}
|
||
|
||
items = []
|
||
for t in raw[:60]:
|
||
try:
|
||
body = t.get("body") or ""
|
||
ide = t.get("assignee") == "projektstart"
|
||
items.append({
|
||
"id": t.get("id", ""),
|
||
"titel": (t.get("title") or "(ohne Titel)")[:200],
|
||
"body": body[:600],
|
||
"status": t.get("status", "?"),
|
||
"assignee": t.get("assignee"),
|
||
# IDE-Welt = das Vorbereiter-Profil (Konzept + Repo + Doku, kein Bau).
|
||
"welt": "ide" if ide else "box",
|
||
# Auftrags-Art innerhalb der IDE-Welt: nur durchdenken oder Projekt
|
||
# vorbereiten? Steht nirgends als Feld — der Marker im Body ist die
|
||
# Wahrheit (schreibt add_idea, siehe _ART_IDEE).
|
||
"art": ("idee" if _ART_MARKER in body else "projekt") if ide else "box",
|
||
"erstellt": t.get("created_at"),
|
||
"fertig": t.get("completed_at"),
|
||
"von": t.get("created_by"),
|
||
"frage": None, # bei blocked die Worker-Frage (unten nachgeladen)
|
||
"frage_kind": None,
|
||
})
|
||
except Exception:
|
||
continue
|
||
|
||
# Hängende Karten anreichern: die Frage des Workers holen, damit die Zentrale sie
|
||
# beantworten kann (Sackgasse-Fix). Deckel gegen Ausreißer, damit ein Poll nie hängt.
|
||
for it in [i for i in items if i["status"] == "blocked"][:6]:
|
||
info = _blocker_frage(it["id"])
|
||
if info:
|
||
it["frage"] = info["frage"]
|
||
it["frage_kind"] = info["kind"]
|
||
|
||
projekte = _ketten_anreichern(items)
|
||
offen = sum(1 for i in items if i["status"] not in ("done",))
|
||
data = {"available": True, "items": items, "offen": offen, "projekte": projekte}
|
||
_list_cache.update(ts=now, data=data)
|
||
return data
|
||
|
||
|
||
# Aufwand-Text je Tiefe, den der Commander per Knopf waehlt — wird dem Auftrag
|
||
# vorangestellt und ueberstimmt die Groessen-Selbsteinschaetzung von konzept-fliessband.
|
||
_AUFWAND = {
|
||
"simpel": ("AUFWAND (vom Commander gewaehlt): SIMPEL — nutze die KLEINE konzept-fliessband-"
|
||
"Variante: 2 Fach-Rollen, 1 Denk-Runde + 1 Haertetest, frueh raus bei WASSERDICHT. "
|
||
"Das ueberstimmt deine eigene Groessen-Einschaetzung."),
|
||
"gruendlich": ("AUFWAND (vom Commander gewaehlt): GRUENDLICH — nutze die VOLLE konzept-fliessband-"
|
||
"Kaskade: 3-5 Fach-Rollen, bis 3 Denk-Runden, mehrfacher Advocatus-Diaboli-Haertetest. "
|
||
"Das ueberstimmt deine eigene Groessen-Einschaetzung, auch wenn das Projekt klein wirkt."),
|
||
}
|
||
|
||
|
||
# Infrastruktur des Commanders — ohne das plant jedes Modell nach Industrie-Default
|
||
# (Docker/Cloud) statt für die reale Umgebung.
|
||
# Umfeld-Kontext je Ziel: Das Heimlabor-Wissen (Box vs. pve, LXCs, PBS) ist NUR fuer
|
||
# LXC-Projekte relevant. Ein Docker-Projekt ist bewusst portabel/extern gedacht — dort
|
||
# wuerde der Proxmox-Kram das Konzept nur in die falsche Richtung schubsen.
|
||
_INFRA_DOCKER = (
|
||
"UMFELD: Dieses Projekt ist bewusst PORTABEL und fuer den Einsatz AUSSERHALB des Heimlabors "
|
||
"gedacht. Plane es umgebungsunabhaengig nach Standard-Praxis — NICHT auf die Proxmox-/LXC-Welt "
|
||
"des Commanders zugeschnitten, keine Annahmen ueber seine Hosts. Der Code liegt in seiner "
|
||
"self-hosted Gitea (CI-Pipelines dort moeglich).")
|
||
|
||
_INFRA_LXC = (
|
||
"UMFELD des Commanders — WICHTIG, WO WAS LAEUFT (nicht raten, das sind zwei Maschinen!):\n"
|
||
"· AI-Box 192.168.178.151: hier laeufst DU (Agent/Lucy/MC2/Modelle). Das ist nur die WERKBANK "
|
||
"— NICHT das Ziel-System. Projekt-Dienste werden hier NIEMALS installiert.\n"
|
||
"· Proxmox-Host 'pve' 192.168.178.108: DAS Ziel-System fuers Deployment, ein SEPARATER Rechner. "
|
||
"Dort laufen die LXC-Container; ein Proxmox Backup Server sichert sie automatisch. Von der Box "
|
||
"aus per 'ssh pve' erreichbar (root, Key), Steuerung mit 'pct'. Was dort schon laeuft (Reverse-"
|
||
"Proxy, DNS, Gitea usw.), schaust du bei Bedarf mit 'ssh pve pct list' nach — nicht annehmen.\n"
|
||
"· Gitea ist self-hosted im Heimnetz (CI-Pipelines moeglich).\n"
|
||
"· Alles laeuft lokal — KEINE Cloud, kein fremder Hoster, keine Bezahl-Dienste.")
|
||
|
||
_ZIEL_HINWEIS = ("Richte den Tech-Vorschlag an der gewaehlten ZIEL-PLATTFORM aus, "
|
||
"nicht am Industrie-Default.")
|
||
|
||
# Auftrags-Art „Idee pruefen": Produkt-Blick statt Umsetzungs-Blick, KEIN Projektstart.
|
||
# Langer Zuruf → kurzer Kartentitel; der volle Wortlaut steht dann hier im Auftrag.
|
||
# Der Worker MUSS wissen, dass der Titel nur eine Kurzfassung ist, sonst arbeitet er
|
||
# am abgeschnittenen Satz statt an der ganzen Idee.
|
||
_WORTLAUT_KOPF = ("DIE IDEE IM WORTLAUT (der Kartentitel ist nur eine Kurzfassung fuer die "
|
||
"Anzeige — DAS hier ist der Auftrag):\n")
|
||
|
||
# Die erste Zeile ist zugleich der MARKER, an dem die Queue-Sicht die Art wiedererkennt
|
||
# (das Kanban kennt kein eigenes Feld dafuer) — beim Aendern _ART_MARKER mitziehen.
|
||
_ART_MARKER = "AUFTRAGS-ART: IDEE PRUEFEN"
|
||
_ART_IDEE = (
|
||
"AUFTRAGS-ART: IDEE PRUEFEN (KEIN Projektstart!). Der Commander will diese Produkt-Idee erst "
|
||
"einmal DURCHDACHT haben — er hat sich NICHT entschieden, sie zu bauen.\n"
|
||
"· Blickwinkel zuerst PRODUKT, nicht Technik: Was ist es genau? Welches Problem, fuer wen? Was "
|
||
"gibt es dafuer schon und warum reicht das nicht? Was waere der kleinste sinnvolle Wurf? Was "
|
||
"spricht DAGEGEN (Aufwand, Betrieb, Wartung, Risiken)? Technik nur so tief, wie es fuer die "
|
||
"Machbarkeit noetig ist.\n"
|
||
"· Rollen-Cast entsprechend waehlen — NICHT nur Entwickler-Rollen, sondern auch Zielgruppen-, "
|
||
"Abgrenzungs-/Markt- und Betriebs-Sicht. Volle Denk-Kaskade mit Advocatus Diaboli.\n"
|
||
"· Ergebnis: NUR ein KONZEPT.md im neuen Repo (plus knappes README). KEINE ROADMAP.md, KEINE "
|
||
"AGENTS.md, KEIN SAVEPOINT.md, KEIN Geruest — das entsteht erst, wenn der Commander sich fuers "
|
||
"Bauen entscheidet.\n"
|
||
"· Uebergabe NICHT als Bau-Aufforderung formulieren: sag, was die Idee waere, was sie braeuchte "
|
||
"und was dagegen spricht — die Entscheidung liegt beim Commander.")
|
||
|
||
# Ziel-Plattform je Projekt (vom Commander gewaehlt) — ueberstimmt die Annahmen des Modells.
|
||
_ZIEL = {
|
||
"lxc": ("ZIEL-PLATTFORM (vom Commander gewaehlt): LXC — plane die Umsetzung als Dienst in einem "
|
||
"eigenen Proxmox-LXC-Container: systemd-Unit, Dateien im Container, KEIN Docker und KEIN "
|
||
"docker compose. Der LXC wird vom Proxmox Backup Server automatisch mitgesichert."),
|
||
"docker": ("ZIEL-PLATTFORM (vom Commander gewaehlt): DOCKER — plane NATIVES Docker nach "
|
||
"Standard-Praxis: Dockerfile, docker compose, sauber gebaute Images, portabel und "
|
||
"CI-tauglich (Build/Deploy ueber Gitea-/GitLab-Pipelines). Nimm eine normale "
|
||
"Docker-Umgebung an; KEINE Verrenkungen wie 'Docker im LXC mit Nesting' und keine "
|
||
"Proxmox-Sonderwege — dieses Projekt darf bewusst ausserhalb der LXC-Welt leben."),
|
||
}
|
||
_ZIEL_EHRLICH = ("Traegt die gewaehlte Ziel-Plattform fuer dieses Projekt NICHT (z. B. eine noetige "
|
||
"Komponente gibt es nur als Docker-Image), dann sag das im Konzept AUSDRUECKLICH mit "
|
||
"Begruendung — ignoriere die Wahl nicht still und scheitere nicht daran.")
|
||
|
||
|
||
def add_idea(titel: str, notiz: str = "", created_by: str = "mc2-ui",
|
||
welt: str = "box", tiefe: str = "", ziel: str = "",
|
||
art: str = "projekt") -> dict:
|
||
"""Idee in die Queue legen.
|
||
|
||
welt='box' (Default): triage — der Specifier arbeitet sie aus, die Werkstatt baut,
|
||
Ergebnis ist ein Patch-Vorschlag im Auftragsbuch (Box-Welt).
|
||
welt='ide': geht DIREKT an das Profil `projektstart` (am Triage/Auto-Zerleger vorbei) —
|
||
die Box härtet ein Konzept (konzept-fliessband), legt ein leeres Gitea-Repo an und
|
||
füllt es NUR mit Doku. Ergebnis ist ein vorbereitetes Projekt, das der Commander in
|
||
Agentic IDE selbst baut (IDE-Welt). Bewusst KEIN --triage: so fasst der Auto-Zerleger die
|
||
Karte nie an und die Werkstatt baut sie nie (der Fan-out-Fehler vom 20.07.).
|
||
tiefe (nur ide): 'simpel' | 'gruendlich' — vom Knopf gewaehlte Konzept-Tiefe; leer =
|
||
die Box schaetzt selbst (Fallback fuer Telegram/Lucy).
|
||
ziel (nur ide-Projekt): 'lxc' | 'docker' — Ziel-Plattform; leer = die Box entscheidet.
|
||
art (nur ide): 'projekt' (Default, Repo+Doku+Roadmap, Bau in IDE) oder 'idee'
|
||
(Produkt-Idee nur durchdenken: NUR Konzept, keine Roadmap/Geruest, kein Bau-Druck)."""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
titel = (titel or "").strip()
|
||
if not (3 <= len(titel) <= 200):
|
||
return {"ok": False, "error": "Titel bitte zwischen 3 und 200 Zeichen."}
|
||
welt = "ide" if str(welt).lower() == "ide" else "box"
|
||
notiz = (notiz or "").strip()
|
||
# Der Commander schreibt roh — sein Zuruf ist oft ein ganzer Absatz und wurde bisher
|
||
# komplett zum Kartentitel (in jeder Liste abgeschnitten, unlesbar). Also: kurzer
|
||
# Titel für die Anzeige, der WORTLAUT wandert vollständig in den Auftrag, damit der
|
||
# Worker nichts verliert.
|
||
wortlaut = ""
|
||
if len(titel) > 90:
|
||
satz = re.split(r"(?<=[.!?])\s", titel)[0].strip()
|
||
wortlaut = titel
|
||
titel = satz if 10 <= len(satz) <= 90 else _kurz(titel, 80)
|
||
if welt == "ide":
|
||
args = ["create", titel, "--assignee", "projektstart", "--created-by", created_by, "--json"]
|
||
# Vorspann: Aufwand (Tiefe) + Ziel-Plattform + Infrastruktur — überstimmt die
|
||
# Annahmen des Modells und macht das Konzept auf die reale Umgebung passend.
|
||
if str(art).lower() == "idee":
|
||
# Idee pruefen: Produkt-Blick, volle Kaskade, KEIN Ziel/Geruest/Bau-Druck.
|
||
# Sein Heimlabor bleibt Kontext (dort liefe es ja), aber ohne Plattform-Festlegung.
|
||
vorspann = [_ART_IDEE, _AUFWAND["gruendlich"], _INFRA_LXC]
|
||
else:
|
||
zielwahl = _ZIEL.get(str(ziel).lower())
|
||
vorspann = [p for p in (_AUFWAND.get(str(tiefe).lower()), zielwahl) if p]
|
||
# Heimlabor-Kontext nur beim LXC-Pfad (und ohne Wahl = seine Standard-Welt);
|
||
# Docker-Projekte bekommen bewusst den portablen Kontext.
|
||
infra = _INFRA_DOCKER if str(ziel).lower() == "docker" else _INFRA_LXC
|
||
vorspann.append(infra + " " + _ZIEL_HINWEIS
|
||
+ (" " + _ZIEL_EHRLICH if zielwahl else ""))
|
||
if wortlaut:
|
||
vorspann.insert(0, _WORTLAUT_KOPF + wortlaut)
|
||
notiz = "\n\n".join(vorspann) + ("\n\n" + notiz if notiz else "")
|
||
else:
|
||
args = ["create", titel, "--triage", "--created-by", created_by, "--json"]
|
||
if wortlaut:
|
||
notiz = _WORTLAUT_KOPF + wortlaut + ("\n\n" + notiz if notiz else "")
|
||
if notiz:
|
||
args[2:2] = ["--body", notiz[:4000]]
|
||
try:
|
||
r = _hermes(args)
|
||
except Exception as exc:
|
||
return {"ok": False, "error": f"Queue nicht erreichbar: {exc}"}
|
||
task = _parse_json(r.stdout)
|
||
if r.returncode != 0 or not isinstance(task, dict) or not task.get("id"):
|
||
return {"ok": False, "error": (r.stderr or r.stdout or "kanban create fehlgeschlagen").strip()[:300]}
|
||
_list_cache["ts"] = 0.0 # nächster Poll zeigt die neue Idee sofort
|
||
try:
|
||
from services import announce
|
||
if welt == "ide" and str(art).lower() == "idee":
|
||
announce.add(f"Idee wird durchdacht: „{titel}“ ({task['id']}) — die Box arbeitet ein "
|
||
"Konzept aus (Produkt-Blick). Danach entscheidest du, ob was draus wird.",
|
||
"[Ideen-Queue]", "ideen-queue", "silent")
|
||
elif welt == "ide":
|
||
announce.add(f"Neues IDE-Projekt in Vorbereitung: „{titel}“ ({task['id']}) — die Box "
|
||
"härtet das Konzept und legt das Repo an, dann baust du in der Agentic IDE.",
|
||
"[Ideen-Queue]", "ideen-queue", "silent")
|
||
else:
|
||
announce.add(f"Neue Idee in der Queue: „{titel}“ ({task['id']}) — die Box arbeitet sie aus.",
|
||
"[Ideen-Queue]", "ideen-queue", "silent")
|
||
except Exception:
|
||
pass
|
||
return {"ok": True, "id": task["id"], "status": task.get("status"), "welt": welt}
|
||
|
||
|
||
def answer_task(task_id: str, antwort: str) -> dict:
|
||
"""Eine hängende Aufgabe beantworten: Antwort als Kommentar protokollieren + entsperren.
|
||
|
||
`unblock --reason` schreibt die Antwort als „UNBLOCK:"-Kommentar und stellt die Karte auf
|
||
ready — der Dispatcher spawnt den Worker neu, der die Antwort im Task-Kontext vorfindet und
|
||
weitermacht. Genau der Ausweg aus der Sackgasse „Aufgabe fragt, aber niemand kann antworten".
|
||
"""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
|
||
antwort = (antwort or "").strip()
|
||
if not (1 <= len(antwort) <= 2000):
|
||
return {"ok": False, "error": "Bitte eine Antwort zwischen 1 und 2000 Zeichen."}
|
||
try:
|
||
r = _hermes(["unblock", task_id, "--reason", antwort])
|
||
except Exception as exc:
|
||
return {"ok": False, "error": f"Queue nicht erreichbar: {exc}"}
|
||
if r.returncode != 0:
|
||
return {"ok": False, "error": (r.stderr or r.stdout or "Entsperren fehlgeschlagen").strip()[:300]}
|
||
_list_cache["ts"] = 0.0 # nächster Poll zeigt den neuen Status (ready/running) sofort
|
||
try:
|
||
from services import announce
|
||
announce.add(f"Deine Antwort ging an die hängende Aufgabe {task_id} — die Box macht weiter.",
|
||
"[Ideen-Queue]", "ideen-queue", "silent")
|
||
except Exception:
|
||
pass
|
||
return {"ok": True}
|
||
|
||
|
||
def archive_task(task_id: str) -> dict:
|
||
"""Erledigtes/Verworfenes aus der Sicht räumen (Kanban-Archiv, nichts wird gelöscht)."""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
|
||
try:
|
||
r = _hermes(["archive", task_id])
|
||
except Exception as exc:
|
||
return {"ok": False, "error": f"Queue nicht erreichbar: {exc}"}
|
||
if r.returncode != 0:
|
||
return {"ok": False, "error": (r.stderr or r.stdout or "Archivieren fehlgeschlagen").strip()[:300]}
|
||
_list_cache["ts"] = 0.0
|
||
return {"ok": True}
|
||
|
||
|
||
_EINORDNEN_SYS = (
|
||
"Du ordnest eine frisch getippte Projekt-Idee ein. Antworte AUSSCHLIESSLICH mit einem "
|
||
"JSON-Objekt, ohne Prosa, ohne Codezaun:\n"
|
||
'{"wer":"box"|"ide","tiefe":"schlank"|"gruendlich","nur_denken":true|false}\n\n'
|
||
"wer=box: Aufgabe am eigenen Heimlabor des Nutzers — seine Server, seine bestehenden "
|
||
"Dienste (Mission Control, Lucy, Rippy), Wartung, Doku, Automatisierung. Die Box baut selbst.\n"
|
||
"wer=ide: ein NEUES eigenstaendiges Programm/Werkzeug/App, das der Nutzer danach selbst "
|
||
"programmieren will. Er bekommt nur ein vorbereitetes Repo.\n"
|
||
"tiefe=gruendlich: gross, vage oder mit vielen offenen Entwurfsfragen. "
|
||
"tiefe=schlank: klein und klar umrissen.\n"
|
||
"nur_denken=true: der Text stellt eine FRAGE oder waegt ab (\"lohnt sich\", \"waere es sinnvoll\", "
|
||
"\"brauche ich\") statt einen Auftrag zu geben."
|
||
)
|
||
|
||
|
||
def einordnen(text: str) -> dict:
|
||
"""Grober Einordnungs-VORSCHLAG für eine frisch getippte Idee.
|
||
|
||
Nur eine Vorauswahl in der Oberfläche — der Mensch sieht und überschreibt sie. Deshalb
|
||
bewusst anspruchslos: ein einziger kurzer Aufruf ans dauerwarme Hirn (69,6 t/s gemessen),
|
||
Schnellspur-Marker gegen das Nachdenken, hartes Timeout. Scheitert irgendetwas, kommt
|
||
`{"ok": False}` zurück und die Oberfläche behält schlicht ihre Voreinstellung — eine
|
||
Einordnung darf das Tippen NIE blockieren oder eine Fehlermeldung erzeugen.
|
||
"""
|
||
text = (text or "").strip()
|
||
if len(text) < 12:
|
||
return {"ok": False}
|
||
ziel_url = (V1_UPSTREAM or GATEWAY_URL).rstrip("/")
|
||
marker = os.environ.get("MC_NO_THINK_MARKER", "[[MC:NO_THINK]]")
|
||
try:
|
||
r = httpx.post(f"{ziel_url}/v1/chat/completions", timeout=12.0, json={
|
||
"model": "hermes", "temperature": 0, "max_tokens": 60, "stream": False,
|
||
"messages": [
|
||
{"role": "system", "content": _EINORDNEN_SYS},
|
||
{"role": "user", "content": f"{marker} {text[:600]}"},
|
||
],
|
||
})
|
||
r.raise_for_status()
|
||
inhalt = (r.json()["choices"][0]["message"]["content"] or "").strip()
|
||
except Exception:
|
||
log.debug("ideen: einordnen fehlgeschlagen", exc_info=True)
|
||
return {"ok": False}
|
||
|
||
# Modelle packen JSON gern in einen Codezaun oder schreiben einen Satz davor.
|
||
m = re.search(r"\{.*\}", inhalt, re.DOTALL)
|
||
if not m:
|
||
return {"ok": False}
|
||
try:
|
||
d = json.loads(m.group(0))
|
||
except Exception:
|
||
return {"ok": False}
|
||
|
||
wer = d.get("wer") if d.get("wer") in ("box", "ide") else None
|
||
tiefe = d.get("tiefe") if d.get("tiefe") in ("schlank", "gruendlich") else None
|
||
if wer is None and tiefe is None:
|
||
return {"ok": False}
|
||
return {"ok": True, "wer": wer or "ide", "tiefe": tiefe or "schlank",
|
||
"nur_denken": bool(d.get("nur_denken"))}
|
||
|
||
|
||
def archive_done(projekt: str = "") -> dict:
|
||
"""Alle FERTIGEN Karten wegräumen — optional nur die eines Projekts.
|
||
|
||
Sammel-Aufräumen (25.07.2026): Bei fünf abgeschlossenen Projekt-Ketten wären das
|
||
sonst ~18 Einzelklicks. Wie beim Einzel-Archivieren wird nichts gelöscht — die
|
||
Karten wandern ins Kanban-Archiv und bleiben dort lesbar.
|
||
|
||
‼️ Bewusst NUR `done`. Laufende, hängende, wartende und in Ausarbeitung befindliche
|
||
Karten werden nie angefasst, auch nicht bei leerem `projekt`-Filter — ein Sammel-
|
||
Knopf darf niemals versehentlich Arbeit abräumen.
|
||
"""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
q = list_queue()
|
||
if not q.get("available"):
|
||
return {"ok": False, "error": q.get("fehler") or "Queue nicht lesbar."}
|
||
|
||
ziel = [i for i in q.get("items", [])
|
||
if i.get("status") == "done" and (not projekt or i.get("projekt") == projekt)]
|
||
if not ziel:
|
||
return {"ok": True, "archiviert": 0, "fehlgeschlagen": []}
|
||
|
||
ok, fehler = 0, []
|
||
for item in ziel:
|
||
r = archive_task(item.get("id", ""))
|
||
if r.get("ok"):
|
||
ok += 1
|
||
else:
|
||
# Die Items aus list_queue tragen "titel" (deutsch), nicht das rohe "title".
|
||
fehler.append({"id": item.get("id"), "titel": _kurz(item.get("titel") or "", 60),
|
||
"error": r.get("error", "")})
|
||
_list_cache["ts"] = 0.0
|
||
return {"ok": True, "archiviert": ok, "fehlgeschlagen": fehler}
|
||
|
||
|
||
def ergebnis_of(task_id: str) -> dict:
|
||
"""Das ERGEBNIS einer fertigen Aufgabe — die Abschluss-Zusammenfassung des Workers.
|
||
|
||
User-Wunsch 15.07.: „im Auftragsbuch werden Dinge getestet, ich sehe aber das
|
||
Ergebnis nicht." Die Wahrheit liegt beim kanban_complete-Aufruf (summary/result)
|
||
im jüngsten `completed`-Event; `list --json` trägt sie nicht → lazy per `show`,
|
||
nur wenn der User die Karte aufklappt. Fertige Ergebnisse ändern sich nie → Cache.
|
||
"""
|
||
if not _available():
|
||
return {"available": False, "summary": "", "result": ""}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"available": True, "summary": "", "result": ""}
|
||
if task_id in _erg_cache:
|
||
return _erg_cache[task_id]
|
||
|
||
try:
|
||
r = _hermes(["show", task_id, "--json"], timeout=30)
|
||
d = _parse_json(r.stdout)
|
||
except Exception:
|
||
log.warning("ideen: kanban show %s (ergebnis) fehlgeschlagen", task_id, exc_info=True)
|
||
return {"available": True, "summary": "", "result": ""}
|
||
summary, result = "", ""
|
||
if isinstance(d, dict):
|
||
if isinstance(d.get("result"), str):
|
||
result = d["result"].strip()
|
||
for ev in d.get("events") or []: # chronologisch → letztes completed gewinnt
|
||
if isinstance(ev, dict) and ev.get("kind") == "completed":
|
||
s = ((ev.get("payload") or {}).get("summary") or "").strip()
|
||
if s:
|
||
summary = s
|
||
data = {"available": True, "summary": summary[:4000], "result": result[:4000]}
|
||
if len(_erg_cache) >= _ERG_MAX:
|
||
_erg_cache.pop(next(iter(_erg_cache)))
|
||
_erg_cache[task_id] = data
|
||
return data
|
||
|
||
|
||
# ── Karten-Steuerung (Stopp / Neuer Versuch / Prio) — User-Wunsch 20.07. ─────
|
||
# Die CLI kann blocken/entsperren, aber weder haengende Worker beenden noch die
|
||
# Prioritaet aendern → fuer genau diese zwei Faelle eine eng begrenzte Direkt-
|
||
# Schreibstelle in die kanban.db (BEGIN IMMEDIATE, nur tasks-Spalten). Alles
|
||
# andere laeuft weiter ueber die CLI.
|
||
|
||
|
||
def _task_row(task_id: str) -> dict | None:
|
||
try:
|
||
con = sqlite3.connect(f"file:{KANBAN_DB}?mode=ro", uri=True, timeout=3)
|
||
try:
|
||
con.row_factory = sqlite3.Row
|
||
r = con.execute("SELECT id, status, worker_pid, priority FROM tasks WHERE id=?",
|
||
(task_id,)).fetchone()
|
||
return dict(r) if r else None
|
||
finally:
|
||
con.close()
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _kill_tree(pid: int) -> None:
|
||
"""Worker samt Kind-Prozessen beenden (TERM, dann KILL). Ohne das bleiben
|
||
Test-Server (uvicorn & Co.) als Waisen zurueck — live gesehen 20.07."""
|
||
import signal
|
||
|
||
def kinder(p: int) -> list[int]:
|
||
try:
|
||
out = subprocess.run(["pgrep", "-P", str(p)], capture_output=True, text=True, timeout=5)
|
||
return [int(x) for x in out.stdout.split()]
|
||
except Exception:
|
||
return []
|
||
|
||
def baum(p: int) -> list[int]:
|
||
alle = [p]
|
||
for k in kinder(p):
|
||
alle.extend(baum(k))
|
||
return alle
|
||
|
||
pids = baum(pid)
|
||
for p in reversed(pids):
|
||
try:
|
||
os.kill(p, signal.SIGTERM)
|
||
except OSError:
|
||
pass
|
||
time.sleep(2)
|
||
for p in reversed(pids):
|
||
try:
|
||
os.kill(p, signal.SIGKILL)
|
||
except OSError:
|
||
pass
|
||
|
||
|
||
def _requeue(task_id: str) -> None:
|
||
"""Karte sauber zurueck auf todo (Claim/PID/Zaehler leeren) — der Dispatcher
|
||
spawnt sie beim naechsten Tick frisch (Caps gelten)."""
|
||
con = sqlite3.connect(str(KANBAN_DB), timeout=15)
|
||
try:
|
||
con.execute("PRAGMA busy_timeout=10000")
|
||
con.execute("BEGIN IMMEDIATE")
|
||
con.execute(
|
||
"UPDATE tasks SET status='todo', claim_lock=NULL, claim_expires=NULL,"
|
||
" worker_pid=NULL, current_run_id=NULL, last_heartbeat_at=NULL,"
|
||
" started_at=NULL, consecutive_failures=0 WHERE id=?", (task_id,))
|
||
con.commit()
|
||
finally:
|
||
con.close()
|
||
|
||
|
||
def stop_task(task_id: str) -> dict:
|
||
"""Karte gezielt anhalten: generischer Block (bleibt liegen, kein Auto-Unblock-
|
||
Kind) + laufenden Worker-Prozessbaum beenden."""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
|
||
row = _task_row(task_id)
|
||
if not row:
|
||
return {"ok": False, "error": "Karte nicht gefunden."}
|
||
try:
|
||
r = _hermes(["block", task_id, "Vom", "Commander", "gestoppt"])
|
||
except Exception as exc:
|
||
return {"ok": False, "error": f"Queue nicht erreichbar: {exc}"}
|
||
if r.returncode != 0:
|
||
return {"ok": False, "error": (r.stderr or r.stdout or "Blocken fehlgeschlagen").strip()[:300]}
|
||
if row.get("status") == "running" and row.get("worker_pid"):
|
||
_kill_tree(int(row["worker_pid"]))
|
||
_list_cache["ts"] = 0.0
|
||
return {"ok": True}
|
||
|
||
|
||
def retry_task(task_id: str) -> dict:
|
||
"""Neuer Versuch: haengende/laufende Karte → Worker-Baum beenden + frisch auf todo;
|
||
blockierte Karte → entsperren. Der Dispatcher uebernimmt den Rest (Caps gelten)."""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
|
||
row = _task_row(task_id)
|
||
if not row:
|
||
return {"ok": False, "error": "Karte nicht gefunden."}
|
||
status = row.get("status")
|
||
try:
|
||
if status == "blocked":
|
||
r = _hermes(["unblock", task_id, "--reason", "Neuer Versuch vom Commander"])
|
||
if r.returncode != 0:
|
||
return {"ok": False, "error": (r.stderr or r.stdout or "Entsperren fehlgeschlagen").strip()[:300]}
|
||
else:
|
||
if status == "running" and row.get("worker_pid"):
|
||
_kill_tree(int(row["worker_pid"]))
|
||
_requeue(task_id)
|
||
try:
|
||
_hermes(["comment", task_id, "Neuer", "Versuch", "vom", "Commander", "(Karte zurueckgesetzt)"], timeout=20)
|
||
except Exception:
|
||
pass
|
||
except Exception as exc:
|
||
return {"ok": False, "error": f"Neuer Versuch fehlgeschlagen: {exc}"}
|
||
_list_cache["ts"] = 0.0
|
||
try:
|
||
from services import announce
|
||
announce.add(f"Karte {task_id} wurde auf Commander-Wunsch neu angestoßen.",
|
||
"[Ideen-Queue]", "ideen-queue", "silent")
|
||
except Exception:
|
||
pass
|
||
return {"ok": True}
|
||
|
||
|
||
def prio_task(task_id: str, richtung: str) -> dict:
|
||
"""Prio hoch/runter (Dispatcher zieht ready-Karten nach `priority DESC, created ASC`
|
||
→ hoch = frueher dran). Direkt-Schreibstelle, die CLI kennt keine Prio-Aenderung."""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
|
||
if richtung not in ("hoch", "runter"):
|
||
return {"ok": False, "error": "richtung muss 'hoch' oder 'runter' sein."}
|
||
delta = 1 if richtung == "hoch" else -1
|
||
try:
|
||
con = sqlite3.connect(str(KANBAN_DB), timeout=15)
|
||
try:
|
||
con.execute("PRAGMA busy_timeout=10000")
|
||
con.execute("BEGIN IMMEDIATE")
|
||
cur = con.execute("UPDATE tasks SET priority = COALESCE(priority,0) + ? WHERE id=?",
|
||
(delta, task_id))
|
||
con.commit()
|
||
if cur.rowcount == 0:
|
||
return {"ok": False, "error": "Karte nicht gefunden."}
|
||
finally:
|
||
con.close()
|
||
except Exception as exc:
|
||
return {"ok": False, "error": f"Prio-Änderung fehlgeschlagen: {exc}"}
|
||
_list_cache["ts"] = 0.0
|
||
return {"ok": True}
|
||
|
||
|
||
# ── Konzept-Ansicht (IDE-Welt) — KONZEPT.md aus dem vorbereiteten Repo holen ──
|
||
# Der projektstart-Worker legt ein Gitea-Repo an und schreibt das wasserdichte
|
||
# Konzept als KONZEPT.md hinein. Hier holen wir es (Gitea-API + Auth aus
|
||
# ~/.git-credentials), damit der Commander es DIREKT im Auftragsbuch liest,
|
||
# ohne das Repo aufzumachen (User-Wunsch 20.07.).
|
||
_GITEA_HOST = "git.tobisniceshomelab.ddnsfree.com"
|
||
# Kein Raten am Zeilenende: greedy bis zum ersten Nicht-Repo-Zeichen matchen und den
|
||
# Treffer HINTERHER säubern. (Der Worker schreibt die URL gern in Markdown-Backticks —
|
||
# eine Lookahead-Liste erwischt sowas nie zuverlässig, Bug-Fund 20.07.)
|
||
_GITEA_REPO_RX = re.compile(
|
||
re.escape(_GITEA_HOST) + r"/([A-Za-z0-9_.-]+/[A-Za-z0-9_][A-Za-z0-9_.-]*)"
|
||
)
|
||
_REPO_FALLBACK_RX = re.compile(r"[Rr]epo\S*[:\s]+[`'\"]?([A-Za-z0-9_.-]+/[A-Za-z0-9_][A-Za-z0-9_.-]*)")
|
||
|
||
|
||
def _clean_repo(raw: str) -> str | None:
|
||
"""Trailing Markdown-/Satzzeichen und .git vom Treffer abschneiden."""
|
||
r = (raw or "").strip().strip("`'\"")
|
||
r = r.rstrip(".,;:)»>")
|
||
r = re.sub(r"\.git$", "", r)
|
||
r = r.rstrip(".,;:")
|
||
return r or None
|
||
_KONZEPT_NAMES = ("KONZEPT.md", "Konzept.md", "konzept.md", "CONCEPT.md")
|
||
_konzept_cache: dict = {}
|
||
|
||
# Zweiter Ablage-Ort (Bug-Fund 21.07.): der `konzept-fliessband`-Skill endet mit
|
||
# „cp … ~/konzepte/<slug>-konzept.md" — Worker folgen dieser Skill-Anweisung auch dann,
|
||
# wenn ihre Profil-SOUL eigentlich ein Repo verlangt. Das Konzept war dann fertig, aber
|
||
# im Auftragsbuch unsichtbar („noch kein Repo hinterlegt"). Also: liegt kein Repo vor,
|
||
# hier nachsehen. Bewusst eng — nur ~/konzepte, nur .md.
|
||
_KONZEPT_DIR = Path("~/konzepte").expanduser()
|
||
_KONZEPT_PFAD_RX = re.compile(r"(?:~|/home/[A-Za-z0-9_.-]+)?/?konzepte/([A-Za-z0-9_.-]+\.md)")
|
||
|
||
|
||
def _lokales_konzept(text: str, titel: str = "") -> tuple | None:
|
||
"""(pfad, inhalt) einer Konzept-Datei aus ~/konzepte — aus dem Text oder per Titel geraten.
|
||
|
||
`titel` nur setzen, wenn die Karte FERTIG ist: sonst hängt die Raterei einer gerade
|
||
laufenden Karte das Konzept ihres Vorgängers an (live gesehen 21.07. — eine neue
|
||
ARM-Karte zeigte das alte ARM-Konzept, als wäre es ihr Ergebnis).
|
||
"""
|
||
kandidaten: list[str] = []
|
||
m = _KONZEPT_PFAD_RX.search(text or "")
|
||
if m:
|
||
kandidaten.append(m.group(1))
|
||
# Kein Pfad genannt (aeltere Karten): ueber den Titel raten — Wortschnittmenge zwischen
|
||
# Dateiname und Kartentitel. Streng: EIN Treffer, und ein einzelnes kurzes Allerweltswort
|
||
# („arm") reicht nicht — sonst passt irgendeine Datei auf irgendeine Karte.
|
||
if not kandidaten and titel and _KONZEPT_DIR.is_dir():
|
||
worte = {w for w in re.findall(r"[a-z0-9]{3,}", titel.lower())}
|
||
treffer = [p.name for p in _KONZEPT_DIR.glob("*.md")
|
||
if len(g := worte & set(re.findall(r"[a-z0-9]{3,}", p.stem.lower()))) >= 2
|
||
or any(len(w) >= 6 for w in g)]
|
||
if len(treffer) == 1:
|
||
kandidaten.append(treffer[0])
|
||
for name in kandidaten:
|
||
pfad = (_KONZEPT_DIR / name).resolve()
|
||
if pfad.parent != _KONZEPT_DIR.resolve() or not pfad.is_file():
|
||
continue
|
||
try:
|
||
return str(pfad), pfad.read_text(encoding="utf-8", errors="replace")
|
||
except Exception:
|
||
log.warning("ideen: Konzept-Datei %s nicht lesbar", pfad, exc_info=True)
|
||
return None
|
||
|
||
|
||
def _gitea_creds() -> tuple | None:
|
||
"""(user, token) fuer den oeffentlichen Gitea-Host aus ~/.git-credentials."""
|
||
try:
|
||
cred = Path("~/.git-credentials").expanduser()
|
||
for line in cred.read_text(encoding="utf-8").splitlines():
|
||
m = re.match(r"https://([^:]+):([^@]+)@" + re.escape(_GITEA_HOST), line.strip())
|
||
if m:
|
||
return m.group(1), m.group(2)
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
|
||
def _extract_repo(text: str) -> str | None:
|
||
"""Ziel-Repo (owner/name) aus dem Ergebnis-/Body-Text der Karte ziehen."""
|
||
if not text:
|
||
return None
|
||
m = _GITEA_REPO_RX.search(text)
|
||
if m:
|
||
return _clean_repo(m.group(1))
|
||
m = _REPO_FALLBACK_RX.search(text)
|
||
return _clean_repo(m.group(1)) if m else None
|
||
|
||
|
||
def konzept_of(task_id: str) -> dict:
|
||
"""Das ausgearbeitete Konzept einer IDE-Karte holen — aus dem Gitea-Repo ODER,
|
||
falls der Worker keins angelegt hat, aus seiner Datei in ~/konzepte."""
|
||
if not _available():
|
||
return {"available": False}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"available": True, "error": "Ungültige Aufgaben-Nummer."}
|
||
if task_id in _konzept_cache:
|
||
return _konzept_cache[task_id]
|
||
|
||
# Repo aus Ergebnis + Body der Karte ziehen (show --json trägt beides).
|
||
try:
|
||
r = _hermes(["show", task_id, "--json"], timeout=30)
|
||
d = _parse_json(r.stdout)
|
||
except Exception:
|
||
return {"available": True, "error": "Karte nicht lesbar."}
|
||
# `kanban show --json` legt die Task-Felder verschachtelt unter "task" ab und
|
||
# liefert die jüngste Abschluss-Zusammenfassung als "latest_summary".
|
||
haystack, task_status, titel = "", None, ""
|
||
if isinstance(d, dict):
|
||
t = d.get("task") if isinstance(d.get("task"), dict) else d
|
||
task_status = t.get("status")
|
||
titel = str(t.get("title") or "")
|
||
haystack = "\n".join(str(x or "") for x in (
|
||
t.get("result"), t.get("body"), d.get("latest_summary")))
|
||
for ev in d.get("events") or []:
|
||
if isinstance(ev, dict) and ev.get("kind") == "completed":
|
||
haystack += "\n" + str((ev.get("payload") or {}).get("summary") or "")
|
||
|
||
def _datei_antwort(repo: str | None = None) -> dict | None:
|
||
"""Konzept aus ~/konzepte als vollwertige Antwort — der zweite Ablage-Ort."""
|
||
# Titel-Raterei nur bei fertigen Karten (sonst erbt eine laufende Karte das
|
||
# Konzept ihres Vorgängers); läuft sie noch, zählt allein ein genannter Pfad.
|
||
lokal = _lokales_konzept(haystack, titel if task_status == "done" else "")
|
||
if not lokal:
|
||
return None
|
||
pfad, inhalt = lokal
|
||
return {"available": True, "repo": repo, "quelle": "datei", "pfad": pfad,
|
||
"datei": Path(pfad).name, "konzept": inhalt[:60000]}
|
||
|
||
repo = _extract_repo(haystack)
|
||
if not repo:
|
||
data = _datei_antwort()
|
||
if data:
|
||
if task_status == "done":
|
||
if len(_konzept_cache) >= 20:
|
||
_konzept_cache.pop(next(iter(_konzept_cache)))
|
||
_konzept_cache[task_id] = data
|
||
return data
|
||
return {"available": True, "repo": None,
|
||
"error": "Noch kein Konzept hinterlegt — die Box arbeitet noch daran."}
|
||
|
||
creds = _gitea_creds()
|
||
auth = creds if creds else None
|
||
konzept, used_name = "", None
|
||
try:
|
||
import httpx
|
||
with httpx.Client(timeout=10.0, follow_redirects=True) as c:
|
||
for name in _KONZEPT_NAMES:
|
||
url = f"https://{_GITEA_HOST}/{repo}/raw/branch/main/{name}"
|
||
resp = c.get(url, auth=auth)
|
||
if resp.status_code == 200 and resp.text.strip():
|
||
konzept, used_name = resp.text, name
|
||
break
|
||
except Exception:
|
||
log.warning("ideen: KONZEPT-Abruf für %s (%s) fehlgeschlagen", task_id, repo, exc_info=True)
|
||
return (_datei_antwort(repo)
|
||
or {"available": True, "repo": repo, "error": "Konzept konnte nicht geladen werden."})
|
||
|
||
if not konzept:
|
||
# Repo genannt, aber (noch) ohne KONZEPT.md → die lokale Ablage kann trotzdem stehen.
|
||
return (_datei_antwort(repo)
|
||
or {"available": True, "repo": repo,
|
||
"error": "KONZEPT.md im Repo (noch) nicht gefunden."})
|
||
clone_url = f"https://{_GITEA_HOST}/{repo}.git"
|
||
data = {"available": True, "repo": repo, "clone_url": clone_url, "quelle": "repo",
|
||
"datei": used_name, "konzept": konzept[:60000]}
|
||
# Fertige Konzepte ändern sich selten → cachen (aber nur, wenn die Karte done ist).
|
||
if task_status == "done":
|
||
if len(_konzept_cache) >= 20:
|
||
_konzept_cache.pop(next(iter(_konzept_cache)))
|
||
_konzept_cache[task_id] = data
|
||
return data
|
||
|
||
|
||
# ── „Gefaellt mir" — aus der geprueften Idee ein IDE-Projekt machen (21.07.) ──
|
||
# Ohne das war die Idee eine Sackgasse: der Commander haette den Titel neu eintippen
|
||
# muessen und die Box haette das Konzept von vorn gedacht. Hier bekommt die neue Karte
|
||
# das FERTIGE Konzept als Vorlage mit — plus die Kette zur Idee-Karte.
|
||
_AUS_IDEE_KOPF = (
|
||
"AUFTRAGS-ART: PROJEKT AUS GEPRUEFTER IDEE. Der Commander hat diese Idee erst durchdenken "
|
||
"lassen — das Konzept gefaellt ihm, jetzt soll daraus ein startklares Repo werden, das er "
|
||
"SELBST in der Agentic IDE baut.")
|
||
_AUS_IDEE_FUSS = (
|
||
"· Denk das Konzept NICHT neu. Es ist bereits mit voller Kaskade + Advocatus Diaboli "
|
||
"gehaertet. Lies es vollstaendig und uebernimm es; eine kurze Nachschaerf-Runde fuer die "
|
||
"Bau-Sicht (Etappen, Ziel-Plattform) ist erlaubt, eine komplette neue Denk-Kaskade ist "
|
||
"Verschwendung.\n"
|
||
"· Der Rest ist deine normale Projektstart-Rolle: das Repo enthaelt am Ende NUR Doku "
|
||
"(KONZEPT.md, ROADMAP.md, AGENTS.md, README.md, SAVEPOINT.md, .aiexclude) — KEIN Code.\n"
|
||
"· Uebergabe wie ueblich: Repo-URL nennen, damit der Commander in der Agentic IDE loslegen kann.")
|
||
|
||
|
||
def _wo_liegt(konz: dict) -> str:
|
||
"""Eine Zeile: WO das Konzept wirklich liegt — für den Auftrag an den Worker.
|
||
|
||
Wichtig ist der Unterschied zwischen „Repo bekannt" und „Konzept KOMMT aus dem Repo":
|
||
findet die Repo-Datei nicht statt (Push verpufft, Datei anders benannt), liest MC2 die
|
||
lokale Fassung. Wer dann trotzdem „liegt im Repo als <lokaler Dateiname>" in den Auftrag
|
||
schreibt, schickt den Worker auf eine Datei, die es dort nie gab.
|
||
"""
|
||
repo, pfad = konz.get("repo"), konz.get("pfad")
|
||
url = konz.get("clone_url") or (f"https://{_GITEA_HOST}/{repo}.git" if repo else "")
|
||
if konz.get("quelle") == "repo":
|
||
return (f"Das gepruefte Konzept liegt im Gitea-Repo `{repo}` ({url}) "
|
||
f"als `{konz.get('datei') or 'KONZEPT.md'}`.")
|
||
if repo:
|
||
return (f"Das gepruefte Konzept liegt als Datei `{pfad}` auf dieser Box. Zur Idee gehoert "
|
||
f"bereits das Gitea-Repo `{repo}` ({url}) — dort ist das Konzept aber NICHT (mehr) "
|
||
"abrufbar. Pruefe das Repo, bevor du etwas anlegst.")
|
||
return f"Das gepruefte Konzept liegt als Datei `{pfad}` auf dieser Box."
|
||
|
||
|
||
def _konzept_und_name(task_id: str) -> tuple:
|
||
"""(Konzept-Daten, kurzer Anzeige-Name, Fehlertext) einer Idee-Karte.
|
||
|
||
Der Kartentitel ist der ROHE Zuruf des Commanders („Recherchiere zu … Lass ma…") —
|
||
als Name unbrauchbar. Die Ueberschrift des Konzepts ist die aufgeraeumte Fassung
|
||
derselben Sache; nur wenn es keine gibt, den Zuruf kappen.
|
||
"""
|
||
konz = konzept_of(task_id)
|
||
if not konz.get("konzept"):
|
||
return {}, "", (konz.get("error")
|
||
or "Zu dieser Idee gibt es noch kein Konzept — erst durchdenken lassen.")
|
||
try:
|
||
r = _hermes(["show", task_id, "--json"], timeout=30)
|
||
d = _parse_json(r.stdout)
|
||
except Exception as exc:
|
||
return {}, "", f"Karte nicht lesbar: {exc}"
|
||
t = (d.get("task") if isinstance(d, dict) and isinstance(d.get("task"), dict)
|
||
else d if isinstance(d, dict) else {})
|
||
quell_titel = _TITEL_PRAEFIX_RX.sub("", str(t.get("title") or "").strip())
|
||
if not quell_titel:
|
||
return {}, "", "Quell-Karte hat keinen Titel."
|
||
h1 = re.search(r"^#\s+(.+)$", konz["konzept"], re.MULTILINE)
|
||
kurz = h1.group(1).strip() if h1 else re.split(r"(?<=[.!?])\s", quell_titel)[0]
|
||
return konz, _kurz(kurz, 85), ""
|
||
|
||
|
||
def projekt_aus_idee(task_id: str, ziel: str = "", tiefe: str = "") -> dict:
|
||
"""Aus einer fertig durchdachten Idee eine IDE-Projekt-Karte machen.
|
||
|
||
Das Konzept der Quell-Karte (Repo oder ~/konzepte-Datei) wird als Vorlage in den
|
||
Auftrag geschrieben; die neue Karte haengt als Kind an der Idee (task_links), damit
|
||
das Auftragsbuch beide als eine Kette zeigt.
|
||
"""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
|
||
|
||
konz, kurz, fehler = _konzept_und_name(task_id)
|
||
if fehler:
|
||
return {"ok": False, "error": fehler}
|
||
titel = f"Projekt: {kurz}"[:200]
|
||
|
||
if konz.get("repo"):
|
||
# Liegt das Konzept schon in einem Gitea-Repo, ist DAS das Projekt-Repo —
|
||
# ein zweites anzulegen wuerde den Commander nur mit Dubletten beschenken.
|
||
quelle = (f"· {_wo_liegt(konz)}\n"
|
||
f"· NUTZE GENAU DIESES REPO als Projekt-Repo — lege KEIN neues an (Schritt 2 "
|
||
"entfaellt). Klone es voll, ergaenze die fehlenden Doku-Dateien und pushe auf main.")
|
||
else:
|
||
quelle = (f"· {_wo_liegt(konz)}\n"
|
||
"· Lies es vollstaendig, lege dann wie in Schritt 2 ein frisches Repo an und "
|
||
"uebernimm den Inhalt als `KONZEPT.md`.")
|
||
|
||
zielwahl = _ZIEL.get(str(ziel).lower())
|
||
infra = _INFRA_DOCKER if str(ziel).lower() == "docker" else _INFRA_LXC
|
||
teile = [f"{_AUS_IDEE_KOPF}\n{quelle}\n{_AUS_IDEE_FUSS}"]
|
||
if _AUFWAND.get(str(tiefe).lower()):
|
||
teile.append(_AUFWAND[str(tiefe).lower()])
|
||
if zielwahl:
|
||
teile.append(zielwahl)
|
||
teile.append(infra + " " + _ZIEL_HINWEIS + (" " + _ZIEL_EHRLICH if zielwahl else ""))
|
||
notiz = "\n\n".join(teile)
|
||
|
||
args = ["create", titel, "--body", notiz[:4000], "--assignee", "projektstart",
|
||
"--parent", task_id, "--created-by", "mc2-ui", "--json"]
|
||
try:
|
||
r = _hermes(args)
|
||
except Exception as exc:
|
||
return {"ok": False, "error": f"Queue nicht erreichbar: {exc}"}
|
||
task = _parse_json(r.stdout)
|
||
if r.returncode != 0 or not isinstance(task, dict) or not task.get("id"):
|
||
return {"ok": False,
|
||
"error": (r.stderr or r.stdout or "kanban create fehlgeschlagen").strip()[:300]}
|
||
_list_cache["ts"] = 0.0
|
||
try:
|
||
from services import announce
|
||
announce.add(f"Aus der Idee „{_kurz(kurz, 60)}“ wird ein Projekt: {task['id']} — die Box "
|
||
"bereitet das Repo vor, gebaut wird in der Agentic IDE.",
|
||
"[Ideen-Queue]", "ideen-queue", "silent")
|
||
except Exception:
|
||
pass
|
||
return {"ok": True, "id": task["id"], "titel": titel}
|
||
|
||
|
||
# ── „So nicht" — Konzept mit Rueckmeldung ueberarbeiten lassen (21.07.) ──────
|
||
# Vorher gab es nach dem Lesen nur annehmen oder wegwerfen. Ein Konzept wird aber
|
||
# selten beim ersten Wurf richtig: der Commander sieht EINEN Punkt, der nicht passt.
|
||
# Die Ueberarbeitung ist wieder eine Idee-Karte (kein Bau) und schreibt die neue
|
||
# Fassung an DIESELBE Stelle — so zeigt „Konzept" ueberall den frischen Stand.
|
||
_UEBERARBEITEN_KOPF = (
|
||
"AUFTRAGS-ART: IDEE PRUEFEN — UEBERARBEITUNG. Es gibt bereits ein durchdachtes, "
|
||
"gehaertetes Konzept. Der Commander hat es GELESEN und will gezielte Aenderungen. "
|
||
"KEIN Projektstart, KEIN Bau, KEIN Geruest.")
|
||
_UEBERARBEITEN_FUSS = (
|
||
"· Arbeite die Rueckmeldung EIN: lies das bestehende Konzept vollstaendig, aendere die "
|
||
"genannten Punkte und alles, was dadurch nicht mehr stimmt (Folgewirkungen!). Denk das "
|
||
"Ganze NICHT von vorn — was er nicht angesprochen hat, bleibt.\n"
|
||
"· Nimm seinen Vorschlag ernst, aber sei ehrlich: schafft er ein Problem, dann bau ihn ein "
|
||
"UND schreib die Folge sichtbar ins Konzept (Abschnitt Risiken/offene Punkte). Setz dich "
|
||
"nicht stillschweigend darueber hinweg und rede es auch nicht schoen.\n"
|
||
"· Ein KURZER Haertetest (Advocatus Diaboli) NUR auf die geaenderten Teile reicht — die "
|
||
"volle Kaskade ist Verschwendung, der Rest ist bereits geprueft.\n"
|
||
"· Uebergabe: was du geaendert hast, was daraus folgte und was du bewusst gelassen hast.")
|
||
|
||
|
||
def konzept_ueberarbeiten(task_id: str, hinweis: str) -> dict:
|
||
"""Rueckmeldung zum Konzept → neue Idee-Karte, die das Konzept ueberarbeitet.
|
||
|
||
Haengt als Kind an der Quell-Karte (Kette im Auftragsbuch) und schreibt die neue
|
||
Fassung an dieselbe Stelle wie die alte (Repo bzw. Datei).
|
||
"""
|
||
if not _available():
|
||
return {"ok": False, "error": "Die Ideen-Queue lebt auf der Box."}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"ok": False, "error": f"Keine gültige Aufgaben-Nummer: {task_id!r}"}
|
||
hinweis = (hinweis or "").strip()
|
||
if not (3 <= len(hinweis) <= 2000):
|
||
return {"ok": False, "error": "Bitte schreib zwischen 3 und 2000 Zeichen, was anders werden soll."}
|
||
|
||
konz, kurz, fehler = _konzept_und_name(task_id)
|
||
if fehler:
|
||
return {"ok": False, "error": fehler}
|
||
|
||
if konz.get("quelle") == "repo":
|
||
quelle = (f"· {_wo_liegt(konz)}\n"
|
||
"· Die NEUE FASSUNG gehoert an dieselbe Stelle: Repo voll klonen, die Datei "
|
||
"ueberschreiben, committen und auf main pushen (Push mit `git ls-remote origin main` "
|
||
"BEWEISEN, nicht annehmen). KEIN neues Repo anlegen — der Commander vergleicht die "
|
||
"Fassungen ueber die Git-Historie.")
|
||
else:
|
||
quelle = (f"· {_wo_liegt(konz)}\n"
|
||
"· Die NEUE FASSUNG gehoert an dieselbe Stelle. Sichere die alte Fassung vorher "
|
||
"als `<datei>.bak`, dann schreib die ueberarbeitete Fassung in die Originaldatei."
|
||
+ (" Gibt es das genannte Repo wirklich, gehoert die neue Fassung ZUSAETZLICH "
|
||
"als `KONZEPT.md` hinein (klonen, schreiben, pushen, Push beweisen)."
|
||
if konz.get("repo") else ""))
|
||
|
||
teile = [(f"{_UEBERARBEITEN_KOPF}\n{quelle}\n"
|
||
f"· SO SOLL ES ANDERS WERDEN — Wortlaut des Commanders:\n „{hinweis}“\n"
|
||
f"{_UEBERARBEITEN_FUSS}"),
|
||
_INFRA_LXC]
|
||
titel = f"Konzept nachschärfen: {kurz}"[:200]
|
||
args = ["create", titel, "--body", "\n\n".join(teile)[:4000], "--assignee", "projektstart",
|
||
"--parent", task_id, "--created-by", "mc2-ui", "--json"]
|
||
try:
|
||
r = _hermes(args)
|
||
except Exception as exc:
|
||
return {"ok": False, "error": f"Queue nicht erreichbar: {exc}"}
|
||
task = _parse_json(r.stdout)
|
||
if r.returncode != 0 or not isinstance(task, dict) or not task.get("id"):
|
||
return {"ok": False,
|
||
"error": (r.stderr or r.stdout or "kanban create fehlgeschlagen").strip()[:300]}
|
||
# Die alte Fassung wird ueberschrieben → gecachtes Konzept der Quell-Karte verwerfen,
|
||
# sonst zeigt die alte Karte nach der Ueberarbeitung noch tagelang den alten Stand.
|
||
_konzept_cache.pop(task_id, None)
|
||
_list_cache["ts"] = 0.0
|
||
try:
|
||
from services import announce
|
||
announce.add(f"Konzept „{_kurz(kurz, 60)}“ wird nachgeschärft ({task['id']}): "
|
||
f"„{_kurz(hinweis, 120)}“",
|
||
"[Ideen-Queue]", "ideen-queue", "silent")
|
||
except Exception:
|
||
pass
|
||
return {"ok": True, "id": task["id"], "titel": titel}
|
||
|
||
|
||
def log_of(task_id: str) -> dict:
|
||
"""Worker-Log einer Idee (hermes kanban log <id>), ANSI entfernt, letzte ~120 Zeilen."""
|
||
if not _available():
|
||
return {"available": False, "lines": []}
|
||
if not _TASK_ID_RX.match(task_id or ""):
|
||
return {"available": True, "lines": []}
|
||
|
||
now = time.time()
|
||
if (
|
||
_log_cache["data"] is not None
|
||
and _log_cache["task_id"] == task_id
|
||
and now - _log_cache["ts"] < _LOG_EVERY
|
||
):
|
||
return _log_cache["data"]
|
||
|
||
try:
|
||
r = _hermes(["log", task_id])
|
||
except Exception:
|
||
log.warning("ideen: kanban log fehlgeschlagen", exc_info=True)
|
||
return {"available": True, "lines": []}
|
||
if r.returncode != 0:
|
||
log.warning("ideen: kanban log exit=%s stderr=%s", r.returncode, r.stderr)
|
||
return {"available": True, "lines": []}
|
||
|
||
raw = r.stdout or ""
|
||
# ANSI-Steuerzeichen entfernen
|
||
ansi = re.compile(r"\x1b\[[0-9;]*m")
|
||
cleaned = ansi.sub("", raw)
|
||
|
||
# Letzte ~120 Zeilen
|
||
lines = cleaned.strip().splitlines()[-120:]
|
||
data = {"available": True, "lines": lines}
|
||
_log_cache.update(ts=now, task_id=task_id, data=data)
|
||
return data
|