"""Point d'entrée unique du Space (port 7860) : FastAPI, interface Gradio UNIQUE montée à la racine, actifs Symbioza. Interface du Space UNIQUE (v4.0, mode local) : un seul Space, aucune inférence en ligne. Trois onglets : pilotage de l'entraînement local (courbes, historique, console), dépôt de fichiers JSONL vérifiés dans le bac à sable, et vérification de code (boucle M.O.U.H. : AST, pytest, Z3). """ import hmac import json import os import subprocess import sys import threading import time from contextlib import asynccontextmanager from pathlib import Path from mouh.learning.local_store import hf_home_for, resolve_root ROOT = resolve_root() _HF_HOME = hf_home_for(ROOT) # v4.1.2 : None = cache des poids sur le disque local (le montage /data faisait dépasser la RAM) if _HF_HOME is not None: os.environ.setdefault("HF_HOME", str(_HF_HOME)) import gradio as gr # noqa: E402 from fastapi import APIRouter, FastAPI # noqa: E402 from fastapi.staticfiles import StaticFiles # noqa: E402 import mouh.service as service # noqa: E402 import mouh.lab as lab # noqa: E402 - bancs d'essai du laboratoire (v4.5) import mouh.world as world # noqa: E402 - ce que l'essaim bâtit dans l'arène (v4.4, secteurs en v4.5) from agents.core import ( # noqa: E402 - pipeline d'agents, journal de travail, auto-jeu AGENTS, EMPTY_STATEMENT, AgentPipeline, WorkJournal, generate_experiences, precheck, read_sessions, ) from engine import trainer as rl # noqa: E402 - n'importe ni torch ni le modèle : aucune VRAM réservée ici from engine.trainer import is_out_of_memory # noqa: E402 from mouh.geometry.api import make_router # noqa: E402 from mouh.geometry.sync import SyncUnavailable, dataset_repo, push_pending # noqa: E402 from mouh.learning import dashboard, recorder # noqa: E402 from mouh.learning.local_ingest import ingest_jsonl # noqa: E402 from mouh.learning.local_store import LocalStore # noqa: E402 from mouh.learning.local_trainer import ( # noqa: E402 LocalTrainer, gpu_status, import_from_hub, suspend_remote_jobs, ) from mouh.lab.api import make_lab_router # noqa: E402 - monde et laboratoire en lecture (v4.5) from mouh.webui import StaticCache, make_page_router, make_ui_router # noqa: E402 - page unique et ses routes JSON from mouh.service import ( # noqa: E402 GENERATOR_CHOICES, load_model_generator, load_unique_generator, stream_session, ) STORE = LocalStore(ROOT) TRAINER = LocalTrainer(STORE) JOURNAL = WorkJournal(STORE.root / "work_journal.jsonl") # journal de travail daté, partagé par tout le Space SKY_QUEUE_NAME = "sky-trajectories" # file d'attente locale ; hors de data/, donc jamais lue par l'entraînement def world_queue() -> Path: return STORE.root / SKY_QUEUE_NAME def world_dir() -> Path: """Magasin du monde (v4.5) : à part de la file de géométries Symbioza.""" return STORE.root / "world" def build_world(deposit, source: str): """Fait entrer une expérience (certifiée, ou rompue par son banc d'essai) dans le monde, et le journalise. ``deposit`` est un appel sans argument vers ``mouh.world``. Une panne est journalisée (« monde_erreur ») et ne remonte jamais : le verdict de l'Arbitre est déjà rendu, et le monde ne peut que s'y ajouter. """ try: result = deposit() except Exception as exc: # noqa: BLE001 JOURNAL.append("monde_erreur", agent="Arbitre", source=source, erreur=f"{type(exc).__name__}: {exc}"[:300]) return None if not result.get("ok"): return result # un échec sans artefact simulé ne bâtit rien : rien à journaliser JOURNAL.append("monde", agent="Arbitre", source=source, ok=True, etat=result.get("status", ""), matiere=result.get("kind", ""), domaine=result.get("domain", ""), slot=result.get("slot", ""), detail=world.describe(result)) return result def world_hook(episode) -> None: """Crochet de fin d'épisode : l'état mesuré (validé, ou rupture chiffrée) entre dans le monde.""" world.building_end(episode.task_hash) build_world(lambda: world.deposit_episode(episode, world_dir()), episode.source) def world_hook_verify(task, outcome) -> None: """Même crochet pour la boucle de l'onglet « Vérifier » : le code final certifié passe au banc d'essai.""" from mouh.learning.records import task_hash files = dict(outcome.final.files) tests = dict(task.test_files) build_world(lambda: world.deposit(world_dir(), instruction=task.instruction, files=files, task_hash=task_hash(task.instruction, tests), certified=True, report=lab.evaluate(files, tests), episode_id=str(outcome.session_id), attempts=int(outcome.attempts or 1), source="interface"), "interface") def announce_building(instruction: str, tests: dict, candidate: str) -> str: """Marque l'emplacement « chantier » pendant l'expérience (domaine : spécification, sinon code fourni).""" from mouh.learning.records import task_hash th = task_hash((instruction or "").strip(), tests) domain = "algorithmique" try: spec = lab.read_spec(tests) or {} fn = lab.artifact_function({"solution.py": candidate or ""}, spec) domain = spec.get("domain") or (lab.ARTIFACTS[fn][0] if fn else domain) world.building_start(world_dir(), th, instruction, domain) except Exception: # noqa: BLE001, S110 — un chantier mal annoncé ne bloque jamais l'expérience pass return th PIPELINE = AgentPipeline(STORE.root, journal=JOURNAL, on_finished=world_hook) REMOTE_ADAPTER = os.environ.get("UNIQUE_ADAPTER_REPO", "").strip() # pour l'import unique depuis le Hub VERIFY_ONLY = GENERATOR_CHOICES[0] UI_CHOICES = [c for c in GENERATOR_CHOICES if c != GENERATOR_CHOICES[1]] # sans le nano-modèle non entraîné _GPU = {"value": None} _STAMP = {"value": None} EXAMPLES = { "Bug : add retourne a - b (rejeté)": ( "Écrire add(a, b) qui renvoie la somme.", "def add(a, b):\n return a - b\n", "from solution import add\n\n\ndef test_add():\n assert add(2, 3) == 5\n\n\ndef test_zero():\n assert add(0, 0) == 0\n", "", ), "Correct : add retourne a + b (certifié)": ( "Écrire add(a, b) qui renvoie la somme.", "def add(a, b):\n return a + b\n", "from solution import add\n\n\ndef test_add():\n assert add(2, 3) == 5\n\n\ndef test_zero():\n assert add(0, 0) == 0\n", '[{"name": "croissance", "declarations": {"x": "Int"}, "assumptions": [[">", "x", 0]], "goal": [">", ["+", "x", 1], 1]}]', ), } INTRO = """# UNIQUE — fabrique locale Un seul Space, **aucune inférence en ligne** : les données vivent dans le dossier local, l'entraînement tourne ici (sur le GPU du Space), et rien n'est certifié sans passer par le bac à sable (AST, pytest, Z3).""" DATA_HELP = """Formats acceptés, une ligne JSON par exemple : - **tâche simple** : `{"instruction": "...", "tests": "from solution import f\\n\\ndef test_f():\\n assert ...", "solution": "def f(...): ...", "level": "extrême", "rejected": "code faux (facultatif)"}` ; - **session** `mouh-session/1` certifiée (format du Space et de l'entraînement) ; - **échec réel** `{task_hash, instruction, tests, files, pytest_output}`. Chaque solution est exécutée contre ses tests ; un `rejected` qui échoue vraiment devient un échec réel (paire DPO et exemple de réparation). Les paires `{prompt, chosen, rejected}` sans tests sont refusées : elles sont invérifiables.""" def authorized(key) -> bool: expected = os.environ.get("UNIQUE_ADMIN_KEY", "") return bool(expected) and hmac.compare_digest(str(key or "").encode(), expected.encode()) REFUSED = "Commande refusée : clé d'administration absente ou incorrecte (secret UNIQUE_ADMIN_KEY)." def gpu() -> dict: if _GPU["value"] is None: _GPU["value"] = gpu_status() return _GPU["value"] def sync_adapter() -> None: """Le Space sert l'adaptateur local promu ; un nouvel adaptateur est rechargé à la génération suivante.""" if STORE.has_adapter(): os.environ["UNIQUE_ADAPTER_REPO"] = str(STORE.adapter_dir) stamp = STORE.adapter_stamp() if stamp != _STAMP["value"]: with service._MODEL_LOCK: service._MODEL_CACHE.pop("unique", None) _STAMP["value"] = stamp def refresh(): sync_adapter() state, progress = STORE.read_json("state.json"), STORE.read_json("progress.json") rows = dashboard.cycle_rows(state, progress) status = dashboard.status_markdown(store=STORE, trainer=TRAINER, gpu=gpu(), state=state, progress=progress, inventory=STORE.inventory()) return status, dashboard.pass_frame(rows), dashboard.loss_frame(rows), dashboard.table(rows), TRAINER.log_tail() def start_cmd(key, dpo): if not authorized(key): return REFUSED TRAINER.dpo = bool(dpo) msg = TRAINER.start() TRAINER.note(msg) return msg def stop_cmd(key): if not authorized(key): return REFUSED msg = TRAINER.stop() TRAINER.note(msg) return msg def suspend_cmd(key): if not authorized(key): return REFUSED try: done = suspend_remote_jobs(os.environ.get("HF_TOKEN", "").strip() or None, cancel_running=True) except Exception as exc: # noqa: BLE001 return f"Jobs Hugging Face illisibles : {exc}" msg = "Jobs Hugging Face : " + (", ".join(done) if done else "aucun job actif à arrêter") TRAINER.note(msg) return msg def import_cmd(key): if not authorized(key): return REFUSED token = os.environ.get("HF_TOKEN", "").strip() or None repo = os.environ.get("MOUH_DATASET_REPO", "").strip() def work(): try: TRAINER.note("import depuis le Hub en cours…") done = import_from_hub(STORE, repo, REMOTE_ADAPTER or None, token) TRAINER.note("import terminé : " + (", ".join(done) or "rien à importer")) except Exception as exc: # noqa: BLE001 TRAINER.note(f"import impossible : {exc}") threading.Thread(target=work, daemon=True).start() return "Import lancé (une seule fois) : suivez la console." def upload_texts(items, verify) -> str: """Dépose des fichiers .jsonl déjà lus en mémoire (page unique) ; même ingestion que l'onglet Données.""" reports = [] for raw_name, text in items: name = os.path.basename(str(raw_name)) if not name.endswith(".jsonl"): reports.append(f"**{name}** : refusé (seuls les fichiers .jsonl sont acceptés)") continue rep = ingest_jsonl(text, name, STORE.data_dir, STORE.failures_dir, verify=bool(verify), log=TRAINER.note) reports.append(rep.markdown()) return "\n\n".join(reports) + "\n\nCes exemples seront lus au prochain passage d'entraînement." def upload_cmd(files, verify, key): if not authorized(key): return REFUSED if not files: return "Aucun fichier choisi." items = [] for f in files if isinstance(files, list) else [files]: path = f if isinstance(f, str) else getattr(f, "name", None) if not path: continue with open(path, encoding="utf-8", errors="replace") as fh: items.append((os.path.basename(path), fh.read())) return upload_texts(items, verify) def run(instruction, code, tests, proofs, max_attempts, generator, record=True): if generator != VERIFY_ONLY and TRAINER.running: yield [], ("GPU occupé par l'entraînement : arrêtez-le, ou choisissez « Vérification seule »."), "", "" return sync_adapter() last = None for s in stream_session(instruction, code, tests, proofs, int(max_attempts), generator, bool(record), on_certified=world_hook_verify): last = s yield s["rows"], s["status"], s["final_code"], s["report"] if last is None: yield [], "aucun résultat", "", "" def load_example(name): return EXAMPLES[name] # -- pipeline d'agents : le monde où UNIQUE fait ses expériences -------------------------------- AGENT_HELP = """Une **expérience** complète : l'Architecte lit le contrat dans les tests, le Codeur propose, le Testeur exécute (pytest, puis propriétés Hypothesis, puis preuves Z3), l'Arbitre accepte ou renvoie les diagnostics au Codeur. Chaque étape est écrite, datée, dans `work_journal.jsonl`. Une expérience certifiée part dans `data/`, sa première tentative ratée dans `failures/` : l'entraînement les lit sans rien changer. L'**auto-jeu** fabrique des variantes fautives de solutions déjà certifiées et ne garde que celles qui échouent réellement aux tests.""" def agent_generator(choice: str): """Relie un générateur du Space (UNIQUE, M.O.U.H.) au Codeur du pipeline. ``None`` = vérification seule.""" if choice == VERIFY_ONLY: return None, "vérification seule : aucun code n'est proposé" from mouh.jev.types import JevDiagnostic, JevErrorType, JevMessageType, JevSeverity, TaskPayload from mouh.verification.generators import GenerationRequest gen, note = (load_model_generator() if choice == GENERATOR_CHOICES[1] else load_unique_generator()) if choice in GENERATOR_CHOICES[2:]: gen = gen.with_options(adaptive=choice == GENERATOR_CHOICES[3]) def generate(instruction, tests, plan, attempt, previous_files, diagnostics): task = TaskPayload(instruction=instruction or "(sans énoncé)", candidate_files={}, test_files=dict(tests)) diags = tuple(JevDiagnostic(error_type=JevErrorType.TEST_FAILURE, severity=JevSeverity.ERROR, message=str(d)[:4000]) for d in (diagnostics or ())) kind = JevMessageType.TASK_SUBMISSION if attempt == 1 else JevMessageType.CORRECTION_REQUEST return gen.generate(GenerationRequest(task=task, attempt=int(attempt), message_type=kind, experts=(1,), previous_files=previous_files, diagnostics=diags)) return generate, note GPU_BUSY = ("GPU occupé par l'entraînement continu : le modèle ne tient pas à côté (un 7B en bf16 demande " "environ 15 Go). Arrêtez l'entraînement dans Pilotage, ou choisissez « Vérification seule » pour " "faire tourner l'expérience sur un code que vous fournissez.") def out_of_memory_message(exc: BaseException) -> str: """Message court et utile quand la VRAM manque, au lieu de la trace CUDA brute.""" return ("Mémoire GPU insuffisante pour charger le modèle. Arrêtez l'entraînement continu dans Pilotage " "(il réserve la carte), puis réessayez ; sinon, utilisez « Vérification seule ». " f"Détail : {str(exc).strip().splitlines()[0][:200]}") def refusal_message(why: str) -> str: """Phrase claire pour un refus décidé avant toute exécution (aucun modèle n'a été chargé).""" if why == EMPTY_STATEMENT: return ("**REFUSÉ** · énoncé vide : décrivez la tâche dans « Énoncé de la tâche », puis relancez. " "Les tests sont lus, mais aucun modèle n'a été chargé.") return f"**REFUSÉ** · {why}. Aucun modèle n'a été chargé." def agent_cmd(instruction, tests, properties, proofs_json, max_attempts, choice, candidate, lab_spec=""): """Lance une expérience et renvoie (résumé, étapes, code final, journal). ``lab_spec`` (v4.5) : spécification de laboratoire (JSON) ; elle voyage avec les tests sous ``lab_spec.json`` et fait du banc d'essai de son domaine un oracle de l'Arbitre.""" tests_text = (tests or "").strip() if not tests_text: return "Aucun test : rien ne peut être vérifié, donc rien n'est lancé.", [], "", journal_rows() task_tests = {"test_solution.py": tests_text} if (lab_spec or "").strip(): task_tests[lab.SPEC_FILE] = lab_spec.strip() # Refus avant tout chargement : un énoncé vide ou des tests inexploitables ne doivent pas coûter ~15 s # de chargement d'un 7B (ni de VRAM). L'épisode refusé est tout de même journalisé, comme avant. why = precheck(instruction or "", task_tests) if why: PIPELINE.run_episode(instruction or "", task_tests, source="interface") return refusal_message(why), [], "", journal_rows() if choice != VERIFY_ONLY and TRAINER.running: # même garde que l'onglet « Vérifier du code » return GPU_BUSY, [], "", journal_rows() try: gen, note = agent_generator(choice) except Exception as exc: # noqa: BLE001 if is_out_of_memory(exc): return out_of_memory_message(exc), [], "", journal_rows() return f"Générateur indisponible : {exc}", [], "", journal_rows() proofs = [] if (proofs_json or "").strip(): try: raw = json.loads(proofs_json) proofs = raw if isinstance(raw, list) else [raw] except ValueError as exc: return f"Preuves illisibles : {exc}", [], "", journal_rows() props = {"test_properties.py": properties} if (properties or "").strip() else None PIPELINE.arbiter.max_attempts = max(1, int(max_attempts)) th = announce_building(instruction or "", task_tests, candidate or "") try: episode = PIPELINE.run_episode(instruction or "", task_tests, candidate_files={"solution.py": candidate} if (candidate or "").strip() else None, generator=gen, properties=props, proofs=proofs, source="interface") except Exception as exc: # noqa: BLE001 — une panne du générateur ne doit pas casser l'onglet world.building_end(th) JOURNAL.append("episode_error", source="interface", erreur=f"{type(exc).__name__}: {exc}"[:400]) if is_out_of_memory(exc): return out_of_memory_message(exc), [], "", journal_rows() return f"Expérience interrompue : {type(exc).__name__} : {str(exc)[:300]}", [], "", journal_rows() v = episode.verdict verdict_line = (f"**{episode.status.upper()}** en {episode.attempts} tentative(s) · " f"tests {'OK' if v.get('tests_ok') else 'échec'}" + ("" if v.get("properties_ok") is None else f" · propriétés {'OK' if v['properties_ok'] else 'échec'}") + ("" if v.get("proofs_ok") is None else f" · preuves {'OK' if v['proofs_ok'] else 'échec'}") + ("" if v.get("simulation_ok") is None else f" · simulation {'OK' if v['simulation_ok'] else 'RUPTURE'}") + f" · score {v.get('score', 0):.2f} · {episode.duration_s:.1f} s" + (f" · {note}" if note else "")) rows = [[s["attempt"], s["agent"], s["action"], "oui" if s["ok"] else "non", s["detail"]] for s in episode.steps] code = "\n\n".join(f"# === {n} ===\n{c}" for n, c in sorted(episode.files.items())) or "(aucun code produit)" return verdict_line, rows, code, journal_rows() def selfplay_cmd(key, limit, per_task): if not authorized(key): return REFUSED, journal_rows() try: report = generate_experiences(STORE.root, journal=JOURNAL, limit=int(limit), per_task=int(per_task)) except Exception as exc: # noqa: BLE001 return f"Auto-jeu impossible : {exc}", journal_rows() msg = (f"{report['echecs_produits']} échec(s) vérifié(s) à partir de {report['sessions_examinees']} session(s). " f"{report['tests_sans_mutation_detectee']} session(s) dont aucune variante fautive n'a été détectée " f"(couverture de tests faible).") return msg, journal_rows() def world_rebuild_cmd(key, limit=60): """Matérialise dans le monde les sessions déjà certifiées (les plus récentes d'abord retenues, une par tâche).""" if not authorized(key): return REFUSED limit = max(1, min(int(limit), 200)) try: report = world.backfill(read_sessions(STORE.data_dir, limit=limit * 3), world_dir(), limit=limit) except Exception as exc: # noqa: BLE001 JOURNAL.append("monde_erreur", agent="Arbitre", source="reconstruction", erreur=f"{type(exc).__name__}: {exc}"[:300]) return f"Reconstruction impossible : {type(exc).__name__} : {str(exc)[:200]}" JOURNAL.append("monde", agent="Arbitre", source="reconstruction", ok=True, matiere="reconstruction", slot="", detail=(f"monde : reconstruction — {report['deposees']} objet(s) déposé(s) dont {report['ruptures']} " f"en rupture, {report['inchangees']} déjà présent(s), {report['refusees']} refusé(s) sur " f"{report['examinees']} tâche(s) certifiée(s)")) msg = (f"{report['examinees']} tâche(s) certifiée(s) examinée(s) : {report['deposees']} nouvel(s) objet(s) dans le monde " f"(dont {report['ruptures']} en rupture au banc d'essai), {report['inchangees']} déjà présent(s), " f"{report['refusees']} refusé(s).") if report["raisons"]: msg += " Raisons : " + " | ".join(report["raisons"]) return msg def world_state() -> dict: """Aperçu du monde pour la console : objets par secteur et par état, chantiers en cours.""" return world.summary(world_dir()) def journal_rows(limit: int = 40): return [[e.get("at", "")[11:19], e.get("kind", ""), e.get("agent") or e.get("phase") or "", (e.get("detail") or e.get("instruction") or e.get("resultat") or json.dumps({k: v for k, v in e.items() if k not in ("seq", "at", "kind")}, ensure_ascii=False))[:160]] for e in JOURNAL.tail(limit)] def agents_refresh(): s = JOURNAL.stats() line = (f"**Épisodes** {s['episodes']} · certifiés {s['certifiees']} · réparés {s['reparees']} · " f"échecs {s['echecs']} · compétences {s['competences']} · " f"expériences d'auto-jeu {s['experiences_autojeu']} · cache de verdicts {PIPELINE.cache.info()['entrees']}") if TRAINER.running: # dit la contrainte avant le clic, pas après l'erreur line += ("\n\n⚠ Entraînement continu en cours : il réserve le GPU. Dans cet onglet, seul " "« Vérification seule » fonctionne tant qu'il tourne.") return line, journal_rows() # -- moteur GRPO/DPO lancé hors du processus de l'interface ------------------------------------- RL_STATE: dict = {"proc": None, "log": None} def rl_plan_text() -> str: try: p = rl.plan() except Exception as exc: # noqa: BLE001 return f"Profil de VRAM illisible : {exc}" return f"**Profil** · {p.describe()}" + ("\n\n" + " · ".join(p.notes) if p.notes else "") def rl_start(key, phases): if not authorized(key): return REFUSED if TRAINER.running: return "L'entraînement continu occupe le GPU : arrêtez-le dans Pilotage avant de lancer GRPO/DPO." proc = RL_STATE.get("proc") if proc is not None and proc.poll() is None: return "Une passe GRPO/DPO tourne déjà." log_path = STORE.logs_dir / f"rl-{time.strftime('%Y%m%dT%H%M%S')}.log" cmd = [sys.executable, "-m", "engine.trainer", "--root", str(STORE.root), "--phases", str(phases).strip()] env = {**os.environ, "PYTHONUNBUFFERED": "1"} with open(log_path, "ab") as fh: RL_STATE["proc"] = subprocess.Popen(cmd, stdout=fh, stderr=subprocess.STDOUT, env=env, cwd=str(Path(__file__).resolve().parent)) RL_STATE["log"] = log_path JOURNAL.append("entrainement", phase="lancement", phases=str(phases), journal=log_path.name) return f"Passe GRPO/DPO lancée ({phases}) · journal {log_path.name} · adaptateur écrit dans adapter_rl/" def rl_stop(key): if not authorized(key): return REFUSED proc = RL_STATE.get("proc") if proc is None or proc.poll() is not None: return "Aucune passe GRPO/DPO en cours." proc.terminate() return "Arrêt demandé." def rl_log(n: int = 120) -> str: path = RL_STATE.get("log") if path is None or not Path(path).exists(): return "aucune passe GRPO/DPO lancée dans cette session" with open(path, "rb") as fh: fh.seek(0, os.SEEK_END) fh.seek(max(0, fh.tell() - 48_000)) text = fh.read().decode("utf-8", "replace") return "\n".join(text.splitlines()[-n:]) with gr.Blocks(title="UNIQUE") as demo: gr.Markdown(INTRO) with gr.Tab("Pilotage"): status = gr.Markdown("Lecture de l'état…") with gr.Row(): admin_key = gr.Textbox(label="Clé d'administration (UNIQUE_ADMIN_KEY)", type="password", scale=2) dpo = gr.Checkbox(value=TRAINER.dpo, label="Phase DPO (paires data/ + failures/)") with gr.Row(): start = gr.Button("Démarrer l'entraînement continu", variant="primary") stop = gr.Button("Arrêter", variant="stop") suspend = gr.Button("Arrêter les anciens jobs Hugging Face") importer = gr.Button("Importer une fois depuis le Hub") command_msg = gr.Markdown() with gr.Row(): pass_plot = gr.LinePlot(x="n", y="taux", color="série", title="Réussite à l'examen (%) par phase évaluée") loss_plot = gr.LinePlot(x="n", y="loss", color="phase", title="Loss d'entraînement par phase") history = gr.Dataframe(headers=dashboard.TABLE_HEADERS, interactive=False, wrap=True, label="Historique des cycles (plus récent en haut)") console = gr.Textbox(label="Console", lines=16, max_lines=16, interactive=False, autoscroll=True) start.click(start_cmd, [admin_key, dpo], command_msg) stop.click(stop_cmd, admin_key, command_msg) suspend.click(suspend_cmd, admin_key, command_msg) importer.click(import_cmd, admin_key, command_msg) with gr.Tab("Données"): gr.Markdown(DATA_HELP) uploads = gr.File(label="Fichiers .jsonl", file_count="multiple", file_types=[".jsonl"]) with gr.Row(): verify = gr.Checkbox(value=True, label="Réexécuter aussi les sessions déjà certifiées (recommandé)") data_key = gr.Textbox(label="Clé d'administration", type="password") add = gr.Button("Ajouter au dataset local", variant="primary") upload_msg = gr.Markdown() add.click(upload_cmd, [uploads, verify, data_key], upload_msg) with gr.Tab("Vérifier du code"): with gr.Row(): with gr.Column(): example = gr.Dropdown(list(EXAMPLES), label="Exemple", value=None) instruction = gr.Textbox(label="Énoncé de la tâche", lines=2) code = gr.Code(label="Code candidat (solution.py)", language="python", lines=10) tests = gr.Code(label="Tests pytest (test_solution.py)", language="python", lines=10) proofs = gr.Code(label="Obligations de preuve Z3 (JSON, optionnel)", language="json", lines=6) with gr.Row(): attempts = gr.Slider(1, 8, value=4, step=1, label="Tentatives max") generator = gr.Dropdown(UI_CHOICES, value=VERIFY_ONLY, label="Générateur de corrections") record = gr.Checkbox(value=os.environ.get("MOUH_RECORD_DEFAULT", "1") == "1", label="Garder cette session pour l'entraînement (dossier local data/)") btn = gr.Button("Vérifier", variant="primary") with gr.Column(): out_status = gr.Textbox(label="Statut", interactive=False) bus = gr.Dataframe(headers=["#", "type", "émetteur → destinataire", "résumé"], interactive=False, wrap=True, label="Bus JEV (temps réel)") final = gr.Code(label="Sortie certifiée", language="python", interactive=False) report = gr.Code(label="Rapport", language="json", interactive=False) example.change(load_example, example, [instruction, code, tests, proofs]) btn.click(run, [instruction, code, tests, proofs, attempts, generator, record], [bus, out_status, final, report]) with gr.Tab("Agents"): gr.Markdown(AGENT_HELP) agents_status = gr.Markdown("Lecture du journal…") with gr.Row(): with gr.Column(): a_instruction = gr.Textbox(label="Énoncé de la tâche", lines=2) a_tests = gr.Code(label="Tests pytest (test_solution.py) — la vérité terrain", language="python", lines=9) a_candidate = gr.Code(label="Code de départ (optionnel)", language="python", lines=6) with gr.Accordion("Oracles supplémentaires (optionnels)", open=False): a_properties = gr.Code(label="Tests de propriétés Hypothesis (test_properties.py) — vide : " "un contrôle de déterminisme est engendré automatiquement", language="python", lines=6) a_proofs = gr.Code(label="Obligations de preuve Z3 (JSON)", language="json", lines=4) with gr.Row(): a_attempts = gr.Slider(1, 8, value=4, step=1, label="Tentatives max") a_generator = gr.Dropdown(UI_CHOICES, value=VERIFY_ONLY, label="Codeur") a_run = gr.Button("Lancer l'expérience", variant="primary") with gr.Column(): a_verdict = gr.Markdown() a_steps = gr.Dataframe(headers=["tentative", "agent", "action", "ok", "détail"], interactive=False, wrap=True, label=f"Chaîne {' → '.join(AGENTS)}") a_code = gr.Code(label="Code retenu", language="python", interactive=False) with gr.Accordion("Auto-jeu : fabriquer des échecs vérifiés", open=False): gr.Markdown("Mute une solution certifiée, exécute les tests, et ne garde que les variantes qui " "échouent réellement. Une mutation que les tests ne voient pas signale un trou de " "couverture, et elle est signalée comme telle.") with gr.Row(): sp_limit = gr.Slider(1, 200, value=20, step=1, label="Échecs à produire au plus") sp_per = gr.Slider(1, 5, value=2, step=1, label="Par tâche") sp_key = gr.Textbox(label="Clé d'administration", type="password") sp_btn = gr.Button("Lancer l'auto-jeu") sp_msg = gr.Markdown() with gr.Accordion("Moteur GRPO / DPO (récompenses vérifiables)", open=False): rl_profile = gr.Markdown("Lecture du profil de VRAM…") gr.Markdown("L'adaptateur entraîné est écrit dans `adapter_rl/` : l'adaptateur promu en service " "dans `adapter/` n'est jamais modifié.") with gr.Row(): rl_phases = gr.Textbox(value=os.environ.get("UNIQUE_RL_PHASES", "grpo:60,dpo:200"), label="Phases (ex. grpo:60,dpo:200)") rl_key = gr.Textbox(label="Clé d'administration", type="password") rl_go = gr.Button("Lancer une passe", variant="primary") rl_halt = gr.Button("Arrêter", variant="stop") rl_msg = gr.Markdown() rl_console = gr.Textbox(label="Journal GRPO/DPO", lines=12, max_lines=12, interactive=False, autoscroll=True) a_journal = gr.Dataframe(headers=["heure", "type", "agent/phase", "détail"], interactive=False, wrap=True, label="Journal de travail (work_journal.jsonl, plus récent en haut)") a_run.click(agent_cmd, [a_instruction, a_tests, a_properties, a_proofs, a_attempts, a_generator, a_candidate], [a_verdict, a_steps, a_code, a_journal]) sp_btn.click(selfplay_cmd, [sp_key, sp_limit, sp_per], [sp_msg, a_journal]) rl_go.click(rl_start, [rl_key, rl_phases], rl_msg) rl_halt.click(rl_stop, rl_key, rl_msg) timer = gr.Timer(10) timer.tick(refresh, None, [status, pass_plot, loss_plot, history, console]) timer.tick(agents_refresh, None, [agents_status, a_journal]) timer.tick(rl_log, None, rl_console) demo.load(refresh, None, [status, pass_plot, loss_plot, history, console]) demo.load(agents_refresh, None, [agents_status, a_journal]) demo.load(rl_plan_text, None, rl_profile) def startup() -> None: """Au lancement : mise à jour par zip si demandée, arrêt des anciens jobs planifiés, entraînement automatique.""" sync_adapter() token = os.environ.get("HF_TOKEN", "").strip() or None if os.environ.get("UNIQUE_EXT_INSTALL", "").strip() or os.environ.get("UNIQUE_EXT_ROLLBACK", "0") == "1": try: from huggingface_hub import HfApi from unique_ext.updater import ensure_update msg, _ = ensure_update(HfApi(token=token), {"token": token, "space": os.environ.get("SPACE_ID", ""), "dataset": os.environ.get("MOUH_DATASET_REPO", "")}) TRAINER.note(msg) except Exception as exc: # noqa: BLE001 TRAINER.note(f"mise à jour impossible : {exc}") if token and os.environ.get("UNIQUE_SUSPEND_REMOTE_JOBS", "1") == "1": try: done = suspend_remote_jobs(token) TRAINER.note("anciens jobs planifiés : " + (", ".join(done) if done else "aucun à suspendre")) except Exception as exc: # noqa: BLE001 TRAINER.note(f"anciens jobs planifiés illisibles : {exc}") if os.environ.get("UNIQUE_AUTOTRAIN", "0") == "1": TRAINER.note(TRAINER.start()) SYMBIOZA_DIR = Path(os.environ.get("SYMBIOZA_DIR") or Path(__file__).resolve().parent / "web" / "symbioza") @asynccontextmanager async def lifespan(_app: FastAPI): """Le thread d'arrière-plan (mise à jour, jobs, entraînement automatique) suit la durée de vie du serveur.""" threading.Thread(target=startup, name="unique-startup", daemon=True).start() yield def sync_sky() -> dict: """Envoie la file sky-trajectories vers le dépôt Dataset privé (jeton lu ici, jamais renvoyé ni journalisé).""" from huggingface_hub import HfApi token = os.environ.get("HF_TOKEN", "").strip() if not token: raise SyncUnavailable("HF_TOKEN absent : synchronisation désactivée") return push_pending(STORE.root / SKY_QUEUE_NAME, HfApi(token=token), dataset_repo()) MAX_JOURNAL_PAGE = 200 def agents_router() -> APIRouter: """Lecture seule du journal de travail : ce que Symbioza affiche comme activité réelle de l'essaim. Aucune de ces routes n'expose de code ni de secret : seulement des compteurs, des libellés et des états. """ router = APIRouter(prefix="/api/agents", tags=["agents"]) @router.get("/stats") def stats() -> dict: return {**JOURNAL.stats(), "cache": PIPELINE.cache.info(), "agents": list(AGENTS)} @router.get("/journal") def journal(limit: int = 50) -> dict: return {"events": JOURNAL.tail(max(1, min(limit, MAX_JOURNAL_PAGE)))} @router.get("/episodes") def episodes(limit: int = 20, status: str = "") -> dict: return {"episodes": JOURNAL.episodes(max(1, min(limit, MAX_JOURNAL_PAGE)), status=status.strip() or None)} return router def build_app() -> FastAPI: """Application ASGI unique (v4.4). La page unique (arène 3D + console UNIQUE) est servie à la racine ; l'ancienne interface Gradio reste intacte sur ``/gradio/`` et l'ancienne adresse ``/symbioza/`` sert toujours la même page. Les routes d'API passent avant les montages statiques. Sans le dossier ``web/symbioza`` (installation partielle), Gradio reprend la racine : le Space ne se retrouve jamais sans interface. """ demo.queue() api = FastAPI(title="Space unifié : UNIQUE + Symbioza", lifespan=lifespan) api.add_middleware(StaticCache) # v4.5.1 : une mise à jour du Space se voit au prochain chargement de la page api.include_router(make_router(STORE.root / SKY_QUEUE_NAME, authorized, sync=sync_sky)) api.include_router(agents_router()) api.include_router(make_ui_router(sys.modules[__name__])) api.include_router(make_lab_router(world_dir)) if not SYMBIOZA_DIR.is_dir(): return gr.mount_gradio_app(api, demo, path="/") api.include_router(make_page_router(SYMBIOZA_DIR)) # v4.5.4 : page versionnée, avant les montages statiques api.mount("/symbioza", StaticFiles(directory=str(SYMBIOZA_DIR), html=True), name="symbioza") gr.mount_gradio_app(api, demo, path="/gradio") api.mount("/", StaticFiles(directory=str(SYMBIOZA_DIR), html=True), name="hub") # en dernier : capte le reste return api if __name__ == "__main__": import uvicorn recorder.use_folder(STORE.root) # sessions de l'interface → data/ local, jamais vers le Hub uvicorn.run(build_app(), host="0.0.0.0", port=int(os.environ.get("PORT", "7860")), proxy_headers=True)