phase1b: Backend entruempelt und robuster (35 tote Routen raus, Sperren, ehrliche Update-Pruefung)
Ampel / ampel (push) Successful in 26s
Ampel / ampel (push) Successful in 26s
Ballast raus: - 35 Routen ohne Nutzer entfernt (agent/*, fit, roles, ctx, drafts, groups, routing/policy, system/history, system/self-update, maintenance/reboot, zeitmaschine/inhalt, zeitplan, voice/health|metrics|trace|voices|reference|tts). Von 95 auf 60. - Tote Module geloescht: agent-Router, roles, agent_aktivitaet, metrics_history (samt 10-s-Sampler), voice_metrics, migrate_config, parse_mc2_timeout, scripts/. - Unbenutzte Funktionen und Konstanten entfernt (Modell-Upgrade-Empfehlung, Draft-/Kontext- Setzer, Konsole, PC-Ausfuehrer-Probe, Routing-Policy-Editor ...). Robuster: - Jobs in eigener Prozessgruppe (Abbrechen beendet wirklich alles), Zeitlimit je Job-Art, start_job_exklusiv: zwei Klicks starten kein doppeltes Update mehr; alte Jobs raeumen sich auf. - Update-Pruefung meldet Fehler (pruef_fehler, Lampe "Pruefung unklar") statt "aktuell". - Nach jedem Update sofort neu pruefen (update_stand) statt 10 Minuten alten Stand zeigen. - llama-swap-Config: Sperre (RLock + flock) fuer UI, Radar, Aufraeumen und Hirn-Umstellung. - Hermes-Config: bei Lesefehler nichts schreiben, atomar, mit Sicherung. - Live-Strom und Gateway-Warnung blockieren den Event-Loop nicht mehr (Lucy, OpenChamber). - Gateway antwortet bei Engine-Ausfall im OpenAI-Fehlerformat (502) statt nacktem 500. - Abgestuerzte Waechter-Pruefung wird ein gelber Hinweis statt still zu verschwinden. - Download laedt nur den gewuenschten Quant (vorher bei Fehlen alle Teile aller Varianten), Download-Jobs in Gruppe "download"; HF-Suche kodiert den Suchbegriff. - Herkunftspruefung: schreibende /api-Aufrufe fremder Webseiten werden abgelehnt (keine Anmeldung, User-Entscheid); Skripte, Desktop-Lucy und /v1 unveraendert. - Modellpfade: Eintragen und Loeschen nur innerhalb von MODELS_DIR. - Dienste-Liste fragt keine abgebauten Dienste mehr ab (PC-Ausfuehrer haette 3 s gekostet). - SSE-Fehlerzeilen von /api/voice/chat als gueltiges JSON. - mission-control-2.service: --timeout-graceful-shutdown 3 (Neustart ohne 10-s-Haenger). Tests: 92 gruen (neu: Herkunft, Quant-Auswahl, abgestuerzte Pruefung). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
c8718fcde0
commit
e9f488b56c
@@ -1,11 +1,22 @@
|
||||
"""
|
||||
Mini-Job-System: Hintergrund-Prozesse mit Live-Log + Download-Fortschritt.
|
||||
Portiert aus Mission Control v1 (jobengine.py). In-Memory, ein Daemon-Thread je Job.
|
||||
|
||||
Seit 24.09.2026:
|
||||
- Jobs laufen in einer eigenen Prozessgruppe; „Abbrechen“ beendet die ganze Gruppe (vorher nur
|
||||
die bash-Hülle — sudo, apt und hf liefen weiter, während der Job schon „abgebrochen“ hieß).
|
||||
- Jede Job-Art hat ein Zeitlimit; ein hängendes apt hält den Wartungs-Riegel nicht mehr ewig.
|
||||
- `start_job_exklusiv` prüft und trägt unter EINER Sperre ein: zwei Klicks starten nicht mehr
|
||||
zwei Updates derselben Gruppe.
|
||||
- Beendete Jobs werden nach einem Tag bzw. ab 40 Stück aufgeräumt.
|
||||
|
||||
Noch offen (Phase 2): Jobs überleben keinen Neustart von MC2, sie gehören in einen eigenen Worker.
|
||||
"""
|
||||
|
||||
import glob
|
||||
import os
|
||||
import shlex
|
||||
import signal
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
@@ -14,6 +25,11 @@ import uuid
|
||||
JOBS: dict[str, dict] = {}
|
||||
_PROCS: dict[str, subprocess.Popen] = {}
|
||||
_LOG_CAP = 400
|
||||
_LOCK = threading.Lock()
|
||||
BEHALTEN_MAX = 40 # so viele beendete Jobs bleiben sichtbar
|
||||
BEHALTEN_S = 24 * 3600 # beendete Jobs verschwinden nach einem Tag
|
||||
STANDARD_ZEITLIMIT_S = 3 * 3600
|
||||
_ENDE = ("done", "failed", "canceled")
|
||||
|
||||
|
||||
def _append_log(job: dict, line: str) -> None:
|
||||
@@ -57,34 +73,64 @@ def _pump_output(job: dict, stream) -> None:
|
||||
commit()
|
||||
|
||||
|
||||
def _run_job(job_id: str, args: list[str], env: dict | None = None):
|
||||
def _beende_gruppe(proc: subprocess.Popen) -> None:
|
||||
"""Den Job-Prozess samt Kindern beenden (posix: ganze Prozessgruppe, sonst nur den Prozess)."""
|
||||
try:
|
||||
if os.name == "posix":
|
||||
os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
|
||||
else:
|
||||
proc.terminate()
|
||||
except (ProcessLookupError, PermissionError, OSError):
|
||||
pass
|
||||
|
||||
|
||||
def _run_job(job_id: str, args: list[str], env: dict | None = None, zeitlimit_s: int | None = None):
|
||||
"""Job-Prozess starten und mitschreiben.
|
||||
|
||||
v3-Umbau P1 (28.08.2026): Hier wurde frueher ein Sudo-Passwort aus dem Browser an
|
||||
stdin gefuettert (und dafuer `sudo -n` in den Argumenten zu `sudo -S` umgeschrieben).
|
||||
Auf der Box laeuft sudo passwortlos (`NOPASSWD: ALL`), der Pfad war tot. Ohne ihn
|
||||
braucht der Prozess auch keine stdin-Pipe mehr: DEVNULL sorgt dafuer, dass ein Job,
|
||||
der wider Erwarten nach einem Passwort fragt, sofort scheitert statt still zu haengen."""
|
||||
v3-Umbau P1 (28.08.2026): Auf der Box läuft sudo passwortlos (`NOPASSWD: ALL`), der Job
|
||||
braucht keine stdin-Pipe. DEVNULL sorgt dafür, dass ein Job, der wider Erwarten nach einem
|
||||
Passwort fragt, sofort scheitert statt still zu hängen."""
|
||||
job = JOBS[job_id]
|
||||
job["state"] = "running"
|
||||
wecker: threading.Timer | None = None
|
||||
try:
|
||||
proc = subprocess.Popen(
|
||||
list(args), stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||||
stdin=subprocess.DEVNULL,
|
||||
bufsize=0,
|
||||
env={**os.environ, **(env or {})},
|
||||
start_new_session=(os.name == "posix"),
|
||||
)
|
||||
_PROCS[job_id] = proc
|
||||
grenze = zeitlimit_s or STANDARD_ZEITLIMIT_S
|
||||
|
||||
def _zu_lang() -> None:
|
||||
job["zeitlimit"] = True
|
||||
_append_log(job, f"[mc] Zeitlimit ({grenze // 60} min) überschritten, Job wird beendet.")
|
||||
_beende_gruppe(proc)
|
||||
|
||||
wecker = threading.Timer(grenze, _zu_lang)
|
||||
wecker.daemon = True
|
||||
wecker.start()
|
||||
|
||||
_pump_output(job, proc.stdout)
|
||||
proc.wait()
|
||||
job["returncode"] = proc.returncode
|
||||
job["state"] = "canceled" if job.get("canceled") else ("done" if proc.returncode == 0 else "failed")
|
||||
if job.get("canceled"):
|
||||
job["state"] = "canceled"
|
||||
elif job.get("zeitlimit"):
|
||||
job["state"] = "failed"
|
||||
job["error"] = "Zeitlimit überschritten"
|
||||
else:
|
||||
job["state"] = "done" if proc.returncode == 0 else "failed"
|
||||
except Exception as exc:
|
||||
_append_log(job, f"[mc] Fehler: {exc}")
|
||||
job["state"] = "failed"
|
||||
job["error"] = str(exc)
|
||||
job["returncode"] = -1
|
||||
finally:
|
||||
if wecker is not None:
|
||||
wecker.cancel()
|
||||
_PROCS.pop(job_id, None)
|
||||
job["finished_at"] = time.time()
|
||||
cb = job.pop("_on_done", None)
|
||||
@@ -110,7 +156,7 @@ def attach_download_progress(job_id: str, local_dir: str, total_bytes: int) -> N
|
||||
rate = 0.0
|
||||
while True:
|
||||
j = JOBS.get(job_id)
|
||||
if not j or j["state"] in ("done", "failed", "canceled"):
|
||||
if not j or j["state"] in _ENDE:
|
||||
break
|
||||
try:
|
||||
inc = glob.glob(pat, recursive=True)
|
||||
@@ -137,25 +183,54 @@ def attach_download_progress(job_id: str, local_dir: str, total_bytes: int) -> N
|
||||
threading.Thread(target=_watch, daemon=True).start()
|
||||
|
||||
|
||||
def start_job(args: list[str], label: str, env: dict | None = None, on_done=None,
|
||||
group: str | None = None) -> str:
|
||||
def _aufraeumen() -> None:
|
||||
"""Alte beendete Jobs entfernen (unter _LOCK aufrufen)."""
|
||||
jetzt = time.time()
|
||||
fertig = sorted((j for j in JOBS.values() if j["state"] in _ENDE),
|
||||
key=lambda j: j.get("finished_at") or 0, reverse=True)
|
||||
for i, j in enumerate(fertig):
|
||||
if i >= BEHALTEN_MAX or jetzt - (j.get("finished_at") or jetzt) > BEHALTEN_S:
|
||||
JOBS.pop(j["id"], None)
|
||||
|
||||
|
||||
def _eintragen(args: list[str], label: str, on_done, group: str | None) -> str:
|
||||
job_id = uuid.uuid4().hex[:12]
|
||||
log_args = list(args)
|
||||
JOBS[job_id] = {
|
||||
"id": job_id, "label": label, "state": "queued", "group": group,
|
||||
"log": ["$ " + " ".join(shlex.quote(a) for a in log_args)],
|
||||
"log": ["$ " + " ".join(shlex.quote(a) for a in args)],
|
||||
"returncode": None, "started_at": time.time(), "finished_at": None,
|
||||
}
|
||||
if on_done:
|
||||
JOBS[job_id]["_on_done"] = on_done
|
||||
threading.Thread(target=_run_job, args=(job_id, args, env), daemon=True).start()
|
||||
return job_id
|
||||
|
||||
|
||||
def start_job(args: list[str], label: str, env: dict | None = None, on_done=None,
|
||||
group: str | None = None, zeitlimit_s: int | None = None) -> str:
|
||||
with _LOCK:
|
||||
_aufraeumen()
|
||||
job_id = _eintragen(list(args), label, on_done, group)
|
||||
threading.Thread(target=_run_job, args=(job_id, list(args), env, zeitlimit_s), daemon=True).start()
|
||||
return job_id
|
||||
|
||||
|
||||
def start_job_exklusiv(group: str, args: list[str], label: str, env: dict | None = None, on_done=None,
|
||||
zeitlimit_s: int | None = None) -> tuple[str | None, dict | None]:
|
||||
"""Wie start_job, aber nur, wenn in der Gruppe nichts läuft — Prüfen und Eintragen unter einer
|
||||
Sperre. Rückgabe: (neue Job-ID, None) oder (None, der schon laufende Job)."""
|
||||
with _LOCK:
|
||||
if (laeuft := active_in_group(group)) is not None:
|
||||
return None, laeuft
|
||||
_aufraeumen()
|
||||
job_id = _eintragen(list(args), label, on_done, group)
|
||||
threading.Thread(target=_run_job, args=(job_id, list(args), env, zeitlimit_s), daemon=True).start()
|
||||
return job_id, None
|
||||
|
||||
|
||||
def active_in_group(group: str) -> dict | None:
|
||||
"""Erster laufender/wartender Job einer Gruppe (z.B. 'maintenance'), sonst None.
|
||||
Basis für den Wartungs-Riegel: nur EIN System-Update gleichzeitig."""
|
||||
for j in JOBS.values():
|
||||
for j in list(JOBS.values()):
|
||||
if j.get("group") == group and j.get("state") in ("running", "queued"):
|
||||
return j
|
||||
return None
|
||||
@@ -163,16 +238,13 @@ def active_in_group(group: str) -> dict | None:
|
||||
|
||||
def cancel_job(job_id: str) -> bool:
|
||||
job = JOBS.get(job_id)
|
||||
if not job or job["state"] in ("done", "failed", "canceled"):
|
||||
if not job or job["state"] in _ENDE:
|
||||
return False
|
||||
job["canceled"] = True
|
||||
_append_log(job, "[mc] Abbruch angefordert…")
|
||||
proc = _PROCS.get(job_id)
|
||||
if proc is not None:
|
||||
try:
|
||||
proc.terminate()
|
||||
except Exception:
|
||||
pass
|
||||
_beende_gruppe(proc)
|
||||
else:
|
||||
job["state"] = "canceled"
|
||||
job["finished_at"] = time.time()
|
||||
@@ -181,4 +253,6 @@ def cancel_job(job_id: str) -> bool:
|
||||
|
||||
def public_jobs() -> list[dict]:
|
||||
"""Jobs ohne interne Felder (_on_done) für die API."""
|
||||
return [{k: v for k, v in j.items() if not k.startswith("_")} for j in JOBS.values()]
|
||||
with _LOCK:
|
||||
_aufraeumen()
|
||||
return [{k: v for k, v in j.items() if not k.startswith("_")} for j in JOBS.values()]
|
||||
|
||||
Reference in New Issue
Block a user