radar: „Jetzt testen“ – Kandidaten auf Knopfdruck prüfen, wenn der Speicher reicht

User 25.09.: „Warum kann ich Modell-Tests nicht selbst anstoßen, sondern muss
immer auf die Box warten?“ Nachts testet das Radar weiter selbst. Tagsüber jetzt
auch per Knopf – aber nur, wenn Kandidat, Warm-Set UND der heutige Coder
zusammen unter die 115-GB-Grenze passen (der Test-Wächter stoppt den Kandidaten
erst bis zu 10 s, nachdem der Coder zu laden beginnt), und nicht 00:00–03:30.
Sonst steht der Grund beim Kandidaten. Test als Auftrag (radar_lauf.py --test),
höchstens 2 h, dieselben Leitplanken wie nachts; jedes Ergebnis kommt als Meldung.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
2026-09-25 13:16:39 +02:00
co-authored by Claude Opus 5.5
parent c31ef827fc
commit 44f33c31d3
31 changed files with 278 additions and 36 deletions
+64 -6
View File
@@ -86,6 +86,10 @@ MIN_PARAMS_B = float(os.environ.get("MC_RADAR_MIN_PARAMS_B", "20"))
HIRN_MAX_AKTIV_B = float(os.environ.get("MC_RADAR_HIRN_MAX_AKTIV_B", "12")) # Verdikt: dichte Modelle nie als Hirn
ABGELEHNT_TAGE = 30
TESTS_MAX = 100
# „Jetzt testen“ (User 25.09.2026: „Warum kann ich Modell-Tests nicht selbst anstoßen?“): so lange darf ein Test auf
# Knopfdruck dauern, und in dieser Zeit ist er gesperrt (Nachtlauf des Radars, um 03:00 NerdQuiz).
TAGTEST_H = float(os.environ.get("MC_RADAR_TAGTEST_H", "2"))
KNOPF_GESPERRT = (os.environ.get("MC_RADAR_KNOPF_SPERRE_VON", "00:00"), os.environ.get("MC_RADAR_KNOPF_SPERRE_BIS", "03:30"))
STORE_PATH = Path(os.environ.get("MC_RADAR_STORE", str(MODELS_DIR / "mc2-radar.json")))
BASELINE_PATH = Path(os.environ.get("MC_RADAR_BASELINE", str(MODELS_DIR / "mc2-radar-baseline.json")))
@@ -1191,9 +1195,11 @@ def _draft_text(wahl: dict) -> str:
f" (13k: {'; '.join(tempi)})." if len(tempi) > 1 else ".")
def teste(kid: str, frist: float) -> dict | None:
def teste(kid: str, frist: float, auf_knopfdruck: bool = False) -> dict | None:
"""Einen Kandidaten prüfen: Download → Baseline → nur Warm-Set in llama-swap → Kandidat auf :5899 →
Urteil. Gibt den Test-Eintrag zurück (None, wenn nur gemessen wurde oder es den Kandidaten nicht gibt)."""
Urteil. Gibt den Test-Eintrag zurück (None, wenn nur gemessen wurde oder es den Kandidaten nicht gibt).
auf_knopfdruck: gestartet mit „Jetzt testen“ (test_starten) statt im Nachtlauf — nur der Text zum Zeitende
unterscheidet sich; die Leitplanken (Warm-Set, Test-Wächter, Frist) sind dieselben."""
ps = _pruefstand()
k = _lade()["kandidaten"].get(kid)
if not k or k.get("status") not in ("neu", "wartet"): # inzwischen verworfen o. Ä.
@@ -1226,7 +1232,8 @@ def teste(kid: str, frist: float) -> dict | None:
werte["baseline"] = {"modell": (basis.get("werte") or {}).get("modell"), "gemessen": _iso(basis.get("zeit"))}
ergebnis, satz = lauf["ergebnis"], lauf["zusammenfassung"] + _draft_text(wahl)
except ps.Zeitende:
satz = f"Das Nachtfenster endete um {FENSTER_ENDE}, bevor der Test fertig war."
satz = (f"Die {TAGTEST_H:g} Stunden für den Test auf Knopfdruck waren um, bevor er fertig war."
if auf_knopfdruck else f"Das Nachtfenster endete um {FENSTER_ENDE}, bevor der Test fertig war.")
except ps.ServerFehler as e:
if "unknown model architecture" in str(e).lower():
ergebnis, satz = "durchgefallen", f"Durchgefallen: läuft nicht im offiziellen llama.cpp ({str(e)[:160]})."
@@ -1446,8 +1453,59 @@ def _tausche_ein(k: dict) -> dict:
return {"modell_id": modell_id, "pfad": str(ziel), "alt": (alt or {}).get("name"), "hinweis": hinweis}
def _oeffentlich(k: dict) -> dict:
return {f: k.get(f) for f in KANDIDAT_FELDER}
def _oeffentlich(k: dict, zeit: datetime | None = None) -> dict:
testbar, grund = jetzt_testbar(k, zeit)
return {**{f: k.get(f) for f in KANDIDAT_FELDER}, "testbar_jetzt": testbar, "testbar_grund": grund}
def _in_knopf_sperre(zeit: datetime) -> bool:
von, bis = (uhrzeit(*(int(x) for x in t.split(":"))) for t in KNOPF_GESPERRT)
uhr = zeit.astimezone(LOCAL_TZ).time()
return von <= uhr < bis if von <= bis else (uhr >= von or uhr < bis)
def jetzt_testbar(k: dict, zeit: datetime | None = None) -> tuple[bool, str | None]:
"""Darf dieser Kandidat jetzt mit „Jetzt testen“ geprüft werden — und wenn nicht, warum?
Tagsüber kann jederzeit der Coder geladen werden (OpenChamber). Der Test-Wächter stoppt den Kandidaten dann zwar
(_Nachtwaechter), aber erst nach bis zu zehn Sekunden; so lange stünden Warm-Set, Kandidat UND Coder im Speicher.
Also nur, wenn alle drei zusammen unter die Speichergrenze passen. Größere testet das Radar nachts."""
if k.get("status") not in ("neu", "wartet"):
return False, None
bedarf = float(k.get("bedarf_gb") or 0)
frei = SPEICHER_GRENZE_GB - WARMSET_GB - CODER_HEUTE_GB
if not k.get("passt") or not bedarf:
return False, "Ob er in den Speicher passt, ist nicht bekannt — das Radar testet ihn nachts."
if bedarf > frei:
return False, (f"Zu groß für einen Test tagsüber: braucht ~{_gb(bedarf)} GB, neben Warm-Set und Coder sind "
f"{_gb(frei)} GB frei. Das Radar testet ihn nachts.")
if _in_knopf_sperre(zeit or jetzt()):
return False, (f"Zwischen {KNOPF_GESPERRT[0]} und {KNOPF_GESPERRT[1]} testet das Radar selbst, und um 03:00 "
"braucht NerdQuiz das Hirn — danach geht es auf Knopfdruck.")
return True, None
def test_starten(kid: str, zeit: datetime | None = None) -> dict:
"""„Jetzt testen“: den Test als Auftrag starten (Gruppe „radar“, höchstens einer zugleich; radar_lauf.py --test).
Antwortet sofort; den Stand zeigen Auftragskarte und Kandidat („Läuft gerade seit …“)."""
stand = _lade()
k = stand["kandidaten"].get(kid)
if not k:
return {"ok": False, "status": 404, "detail": "Diesen Kandidaten gibt es nicht."}
testbar, grund = jetzt_testbar(k, zeit)
if not testbar:
return {"ok": False, "status": 409, "detail": grund or f"{k.get('name')} lässt sich nicht testen "
f"(Status: {k.get('status')})."}
if _aktiver_lauf(stand):
return {"ok": False, "status": 409, "detail": "Es läuft schon ein Test des Radars."}
job_id, laeuft = jobengine.start_job_exklusiv(
"radar", [sys.executable, str(_LAUF_SKRIPT), "--test", kid], f"Modell-Radar: Test von {k.get('name')}",
zeitlimit_s=int(TAGTEST_H * 3600) + 900)
if job_id is None:
return {"ok": False, "status": 409,
"detail": f"Das Radar ist gerade beschäftigt: {(laeuft or {}).get('label') or 'ein Auftrag'}."}
return {"ok": True, "job_id": job_id,
"text": f"Der Test von {k.get('name')} läuft. Er dauert je nach Download etwa eine halbe bis zwei Stunden; "
"das Ergebnis kommt als Meldung."}
def uebernehmen(kid: str) -> dict:
@@ -1523,7 +1581,7 @@ def uebersicht(zeit: datetime | None = None) -> dict:
kandidaten = sorted(stand["kandidaten"].values(),
key=lambda k: (_REIHENFOLGE.get(k.get("status"), 9), 0 if k.get("quelle") == "watchlist" else 1,
k.get("rang", 999)))
oeffentlich = [_oeffentlich(k) for k in kandidaten]
oeffentlich = [_oeffentlich(k, zeit) for k in kandidaten]
if (lauf := _aktiver_lauf(stand)): # den laufenden Test in seiner Begründung zeigen (nicht gespeichert)
for k in oeffentlich:
if k["id"] == lauf.get("kandidat"):