From 6d615255cdec6cbcc08f0d6177a4928e9caca4fb Mon Sep 17 00:00:00 2001 From: jsboige Date: Sat, 12 Sep 2026 22:59:08 +0200 Subject: [PATCH 1/3] =?UTF-8?q?feat(lean,#15666):=20organe=20d'execution?= =?UTF-8?q?=20confine=20=E2=80=94=20cap=20machine-wide,=20kill-tree,=20zer?= =?UTF-8?q?o-orphelin?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit T1 de l'EPIC #15666 : la seule tranche qui empeche la recidive de l'incident du 12 septembre (~30 lean.exe, ~95 % CPU, DriveFS puis Claudish etouffes, reboot). - Cap strict de population lean/lake machine-wide, etat partage hors de tout worktree (LOCALAPPDATA/XDG_STATE_HOME), admission sous verrou fichier, fail-closed si la population n'est pas mesurable. - Confinement de l'arbre : Job Object Windows kill-on-close + plafond memoire + cap CPU + priorite reduite ; racine lancee CREATE_SUSPENDED, assignee au job, puis reprise (aucun enfant hors du job). POSIX/WSL : setsid + kill du groupe. - Postcondition zero descendant orphelin, verifiee apres fenetre de grace ; survivants = echec visible (exit 126 + pids). - Parallelisme toujours borne (LEAN_NUM_THREADS, -Kjobs=N sur lake build nu). - tree_lock.py n'est pas double : sa logique de peremption (pid_alive tree_lock.py:51-74, host_id:46-48, refus de casser un lock etranger:138-139) est reprise ; le lease par arbre reste le second etage. Rien n'est archive. Tests (scripts/lean/tests/test_lean_exec.py, 10/10) : cap global depuis deux worktrees concurrents, timeout qui tue toute la descendance, controle par faux negatif du detecteur d'orphelins, fail-closed telemetrie, peremption du registre, codes de sortie. Controle positif reel : lake env lean sous le cap (backend windows-job, 0 orphelin). Co-Authored-By: Claude Sonnet 5 --- scripts/lean/README.md | 52 ++ scripts/lean/lean_exec.py | 1055 ++++++++++++++++++++++++++ scripts/lean/tests/test_lean_exec.py | 374 +++++++++ 3 files changed, 1481 insertions(+) create mode 100644 scripts/lean/lean_exec.py create mode 100644 scripts/lean/tests/test_lean_exec.py diff --git a/scripts/lean/README.md b/scripts/lean/README.md index f72119eea4..3de36ff328 100644 --- a/scripts/lean/README.md +++ b/scripts/lean/README.md @@ -13,11 +13,63 @@ Outils pour le cycle de vie des projets Lean 4 du dépôt. | `smoke_test_epita_is.py` | Smoke tests du parcours EPITA-IS (notebooks + preuves) | | `check_public_anchor.py` | Detecte les `sorry` qu'aucune declaration publique n'atteint — l'angle mort residuel du gate `proof-integrity` (voir ci-dessous) | | `count_code_sorry.py` | Compte les `sorry` **hors commentaires** (la vraie dette) et liste les theoremes vacuous (`: True`) — ce que `grep -c sorry` surestime de ~11x (voir ci-dessous) | +| `lean_exec.py` | Organe canonique d'execution Lean : cap de population machine-wide, confinement de l'arbre (Job Object `kill-on-close` / scope POSIX), postcondition zero-orphelin (voir ci-dessous, #15666) | Tests unitaires dans `tests/`. --- +## `lean_exec.py` — organe d'execution confine (T1 de #15666) + +Incident du 2026-09-12 : ~30 `lean.exe` a ~95 % CPU ont etouffe une machine +worker (DriveFS, puis Claudish, puis reboot). Le lease par arbre +`agent_tests/prover/tree_lock.py` ne voit structurellement pas les autres +worktrees ; il **reste** (exclusivite d'un acteur prover par arbre) et devient le +second etage sous l'admission machine-wide. + +T1 livre exactement trois choses : + +1. **Cap machine-wide** de la population `lean`/`lake` — etat partage hors de tout + worktree (`%LOCALAPPDATA%\CoursIA\lean_exec\` / `$XDG_STATE_HOME/coursia/lean_exec/`), + admission sous verrou fichier (la fenetre count->spawn est fermee), fail-closed + si la population n'est pas mesurable. +2. **Confinement de l'arbre** : Job Object Windows cree avec `kill-on-close`, + plafond memoire, cap CPU et priorite reduite ; la racine est lancee + `CREATE_SUSPENDED`, assignee au job, puis reprise — aucun enfant ne peut + naitre hors du job. Le handle vit pendant tout le run : un crash du + superviseur tue l'arbre par le noyau. Cote POSIX/WSL : `setsid` + kill du + groupe (scope systemd quand disponible). +3. **Postcondition zero-orphelin** verifiee apres chaque run, apres une fenetre + de grace : des survivants donnent un **echec visible** (exit `126` + liste des + pids), jamais un « propre » silencieux. + +Le parallelisme est toujours borne : `LEAN_NUM_THREADS` est pose pour les enfants +et `-Kjobs=N` est insere dans un `lake build` nu (jamais le defaut qui prend la +machine). + +```bash +python scripts/lean/lean_exec.py status # population, cap, runs vivants +python scripts/lean/lean_exec.py run --timeout 600 -- lake env lean Fichier.lean +python scripts/lean/lean_exec.py run --json --budget 2 -- lake build +``` + +Codes de sortie stables : `0` succes, `1` echec de la commande enfant (code reel +dans le JSON), `124` timeout, `125` admission refusee (cap atteint / telemetrie +indisponible), `126` cleanup incomplet (orphelins), `127` erreur interne, +`130` interruption. Chaque run publie ses metriques en JSON +(`/last_run.json` : pid, backend, duree, population avant, orphelins). + +Configuration : `LEAN_EXEC_CAP` (defaut `min(8, max(2, nproc/2))`), +`LEAN_EXEC_BUDGET` (defaut 2), `LEAN_EXEC_JOBS` (defaut `nproc/4`), +`LEAN_EXEC_MEM_FRAC` (0.80), `LEAN_EXEC_CPU_PCT` (90), `LEAN_EXEC_STATE_DIR` +(isolation tests), `LEAN_EXEC_WSL=off` (desactive la sonde WSL). + +**Hors T1** (autres tranches de l'EPIC) : admission fine / budget mesure (T2), +politique de backend et coherence de cache (T3), garde CI + migration des 35 +appels directs (T4), procedure operateur et validation de charge bornee (T5). + +--- + ## `check_public_anchor.py` — les `sorry` hors de portee du gate (issue #8782) ### Problème diff --git a/scripts/lean/lean_exec.py b/scripts/lean/lean_exec.py new file mode 100644 index 0000000000..60047b24a8 --- /dev/null +++ b/scripts/lean/lean_exec.py @@ -0,0 +1,1055 @@ +"""Organe canonique d'execution Lean — T1 : cap machine-wide + confinement kill-tree (See #15666). + +Incident du 12 septembre 2026 : ~30 processus ``lean.exe`` a ~95 % CPU ont etouffe +une machine worker (DriveFS tombe, puis Claudish, puis reboot du cluster). Cause +structurelle : chaque appelant lance ``lake``/``lean`` comme il peut, le lease +``.prover.lock`` n'est que par-arbre, et aucun timeout parent ne garantit la mort +de la descendance. + +T1 (cette tranche de l'EPIC #15666) ajoute UNIQUEMENT ce qui empeche la recidive : + +1. **Cap strict machine-wide** de la population ``lean``/``lake`` admise. L'etat + vit hors de tout worktree, dans ``%LOCALAPPDATA%\\CoursIA\\lean_exec\\`` sur + Windows / ``$XDG_STATE_HOME/coursia/lean_exec/`` sinon — meme chaine de + resolution que ``scripts/genai-stack/commands/gpu.py:438`` (LOCALAPPDATA puis + XDG_STATE_HOME puis ``~/.local/state``). Le cap se compte par machine, pas par + arbre : deux worktrees distincts ne peuvent pas depasser ensemble le plafond. +2. **Confinement de l'arbre de processus** : Job Object Windows avec + ``kill-on-close``, plafonds memoire/CPU et priorite reduite. La fermeture du + handle (timeout, interruption, fin du superviseur, crash du parent) tue TOUTE + la descendance. Sous POSIX/WSL : scope systemd si disponible, sinon + ``setsid`` + ``kill(-pgid)``. +3. **Postcondition « zero descendant orphelin »** verifiee apres chaque run et + visible en echec (exit 126 + liste des survivants). Un cleanup non prouve + est un cleanup absent. + +Le lease par arbre ``agent_tests/prover/tree_lock.py`` RESTE (second etage sous +l'admission machine-wide, pour l'exclusivite d'un acteur prover par arbre) ; cet +organe reprend sa logique de detection de peremption (pid_alive + host) plutot +que de la doubler. + +Tout run passe par l'admission sous verrou machine-wide, impose un parallelisme +borne aux enfants (``LEAN_NUM_THREADS``, ``-Kjobs=N`` pour ``lake build``) et +publie ses metriques en JSON. + +Codes de sortie stables : + 0 succes (commande terminee, nettoyage prouve) + 1 echec de la commande enfant (code reel dans le JSON) + 124 timeout (arbre tue, nettoyage prouve) + 125 admission refusee (cap atteint / environnement non mesurable) + 126 cleanup incomplet : orphelins detectes apres termination + 127 erreur interne a l'organe + 130 interruption (SIGINT) : arbre tue, nettoyage prouve + +Env de configuration : ``LEAN_EXEC_STATE_DIR`` (isolation tests), ``LEAN_EXEC_CAP``, +``LEAN_EXEC_BUDGET``, ``LEAN_EXEC_JOBS``, ``LEAN_EXEC_MEM_FRAC``, ``LEAN_EXEC_CPU_PCT``. +""" + +from __future__ import annotations + +import argparse +import ctypes +import json +import os +import platform +import shutil +import signal +import subprocess +import sys +import time +import uuid +from pathlib import Path + +EXIT_OK = 0 +EXIT_CHILD = 1 +EXIT_TIMEOUT = 124 +EXIT_REFUSED = 125 +EXIT_ORPHANS = 126 +EXIT_INTERNAL = 127 +EXIT_INTERRUPTED = 130 + +LEAN_PROC_NAMES = {"lean", "lean.exe", "lake", "lake.exe"} + +_STATE_SUBDIR_WIN = ("CoursIA", "lean_exec") +_STATE_SUBDIR_POSIX = ("coursia", "lean_exec") + + +# --------------------------------------------------------------------------- +# Etat machine-wide — meme chaine de resolution que gpu.py:438 +# --------------------------------------------------------------------------- + +def machine_state_base() -> Path: + """Base d'etat user-scope, hors repo. Chaine identique a gpu.py:438 + (LOCALAPPDATA puis XDG_STATE_HOME puis ~/.local/state) — les deux autres + sites (`scripts/genai-stack/commands/gpu.py`, `scripts/ci/install_prune_task.py:37`) + l'inlinent avec un sous-repertoire different ; leur consolidation sur ce + helper est un suivi hors T1 (perimetre = scripts/lean/lean_exec.py).""" + return Path( + os.environ.get("LOCALAPPDATA") + or os.environ.get("XDG_STATE_HOME") + or str(Path.home() / ".local" / "state") + ) + + +def state_dir() -> Path: + """Repertoire d'etat machine-wide de l'organe, hors de tout worktree.""" + override = os.environ.get("LEAN_EXEC_STATE_DIR") + if override: + return Path(override) + subdir = _STATE_SUBDIR_WIN if os.name == "nt" else _STATE_SUBDIR_POSIX + return machine_state_base().joinpath(*subdir) + + +def runs_dir() -> Path: + return state_dir() / "runs" + + +# --------------------------------------------------------------------------- +# Liveness pid — reprise de tree_lock.py:51-74 (jamais os.kill(pid, 0) sur +# Windows : CPython y TERMINE le processus cible ; probe ctypes a la place). +# --------------------------------------------------------------------------- + +_STILL_ACTIVE = 259 +_PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + + +def host_id() -> str: + """Identite du namespace de pids (tree_lock.py:46-48) — les pids Windows + et WSL vivent dans des namespaces separes.""" + return f"{platform.node()}/{os.name}" + + +def pid_alive(pid: int) -> bool: + """True si ``pid`` est vivant sur CE host (tree_lock.py:51-74).""" + if pid <= 0: + return False + if os.name == "nt": + kernel32 = ctypes.windll.kernel32 + handle = kernel32.OpenProcess( + _PROCESS_QUERY_LIMITED_INFORMATION, False, pid + ) + if not handle: + return False + try: + exit_code = ctypes.c_ulong() + ok = kernel32.GetExitCodeProcess(handle, ctypes.byref(exit_code)) + return bool(ok) and exit_code.value == _STILL_ACTIVE + finally: + kernel32.CloseHandle(handle) + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +# --------------------------------------------------------------------------- +# Scan de population machine-wide lean/lake +# --------------------------------------------------------------------------- + +def _count_from_lines(lines: list[str]) -> int: + n = 0 + for line in lines: + name = line.strip().strip('"').split(",")[0].strip('"') + name = name.rsplit("\\", 1)[-1].rsplit("/", 1)[-1] + if name.lower() in LEAN_PROC_NAMES: + n += 1 + return n + + +def scan_native_population() -> tuple[int, str]: + """Compte les lean/lake vivants cote natif. Echec = mesurable=false.""" + try: + if os.name == "nt": + out = subprocess.run( + ["tasklist", "/FO", "CSV", "/NH"], + capture_output=True, text=True, timeout=15, + encoding="utf-8", errors="replace", + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + names = [ + ln.split('","')[0].strip('"') if ln.strip() else ln + for ln in out.stdout.splitlines() + ] + return _count_from_lines(names), "tasklist" + out = subprocess.run( + ["ps", "-eo", "comm="], capture_output=True, text=True, timeout=15, + encoding="utf-8", errors="replace", + ) + return _count_from_lines(out.stdout.splitlines()), "ps" + except (OSError, subprocess.SubprocessError) as exc: + return -1, f"unavailable: {exc}" + + +def scan_wsl_population() -> tuple[int, str]: + """Best-effort : population lean/lake cote WSL vu depuis Windows. + WSL absent -> (0, "off"). Sonde morte -> (-1, raison) sans bloquer + l'admission native (rapporte dans le JSON).""" + if os.name != "nt": + return 0, "off" + if os.environ.get("LEAN_EXEC_WSL", "").lower() in ("off", "0", "no"): + return 0, "off" + if shutil.which("wsl.exe") is None: + return 0, "off" + try: + out = subprocess.run( + ["wsl.exe", "--", "ps", "-eo", "comm="], + capture_output=True, timeout=10, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + raw = out.stdout + for enc in ("utf-8", "utf-16-le"): + try: + text = raw.decode(enc) + break + except UnicodeDecodeError: + continue + else: + text = raw.decode("utf-8", errors="replace") + return _count_from_lines(text.splitlines()), "wsl ps" + except (OSError, subprocess.SubprocessError) as exc: + return -1, f"unavailable: {exc}" + + +# --------------------------------------------------------------------------- +# Registre des runs — staleness reprise de tree_lock.py:122-151 +# --------------------------------------------------------------------------- + +def read_run(path: Path) -> dict: + try: + return json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return {} + + +def sweep_stale_runs() -> list[str]: + """Retire les runs dont le pid racine est mort sur CE host. Un run d'un + host etranger n'est JAMAIS auto-retire (pids non comparables entre + namespaces, cf tree_lock.py:138) — il ne compte pas non plus comme vivant + pour le budget : seul un pid vivant du meme host est credible.""" + swept = [] + try: + paths = list(runs_dir().glob("*.json")) + except OSError: + return swept + for path in paths: + run = read_run(path) + pid = int(run.get("pid") or -1) + h = run.get("host", "") + if h == host_id() and not pid_alive(pid): + try: + path.unlink() + swept.append(path.stem) + except OSError: + pass + return swept + + +def live_registered_budgets() -> tuple[int, list[dict]]: + """Somme des budgets des runs vivants (meme host, pid vivant).""" + total = 0 + live = [] + try: + paths = list(runs_dir().glob("*.json")) + except OSError: + return 0, live + for path in paths: + run = read_run(path) + if run.get("host") != host_id(): + continue + if not pid_alive(int(run.get("pid") or -1)): + continue + total += max(1, int(run.get("budget") or 1)) + live.append(run) + return total, live + + +# --------------------------------------------------------------------------- +# Verrou d'admission machine-wide (ferme la fenetre TOCTOU count->spawn) +# --------------------------------------------------------------------------- + +class AdmissionLock: + """Verrou fichier exclusif dans le state dir : ouvre, lock non-bloquant + en boucle bornee, unlock/close. msvcrt cote Windows, fcntl cote POSIX.""" + + def __init__(self, timeout_s: float = 10.0): + self.path = state_dir() / "admission.lock" + self.timeout_s = timeout_s + self._fh = None + + def __enter__(self) -> "AdmissionLock": + self.path.parent.mkdir(parents=True, exist_ok=True) + self._fh = open(self.path, "a+b") + deadline = time.monotonic() + self.timeout_s + while True: + if _try_lock(self._fh): + return self + if time.monotonic() >= deadline: + self._fh.close() + self._fh = None + raise TimeoutError( + f"admission lock busy: {self.path}" + ) + time.sleep(0.05) + + def __exit__(self, *exc) -> None: + if self._fh is not None: + _unlock(self._fh) + self._fh.close() + self._fh = None + + +if os.name == "nt": + import msvcrt + + def _try_lock(fh) -> bool: + try: + msvcrt.locking(fh.fileno(), msvcrt.LK_NBLCK, 1) + return True + except OSError: + return False + + def _unlock(fh) -> None: + try: + fh.seek(0) + msvcrt.locking(fh.fileno(), msvcrt.LK_UNLCK, 1) + except OSError: + pass +else: + import fcntl + + def _try_lock(fh) -> bool: + try: + fcntl.flock(fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except OSError: + return False + + def _unlock(fh) -> None: + try: + fcntl.flock(fh.fileno(), fcntl.LOCK_UN) + except OSError: + pass + + +# --------------------------------------------------------------------------- +# Confinement Windows — Job Object kill-on-close + plafonds + priorite +# --------------------------------------------------------------------------- + +if os.name == "nt": + class _JOBOBJECT_BASIC_LIMIT_INFORMATION(ctypes.Structure): + _fields_ = [ + ("PerProcessUserTimeLimit", ctypes.c_longlong), + ("PerJobUserTimeLimit", ctypes.c_longlong), + ("LimitFlags", ctypes.c_ulong), + ("MinimumWorkingSetSize", ctypes.c_size_t), + ("MaximumWorkingSetSize", ctypes.c_size_t), + ("ActiveProcessLimit", ctypes.c_ulong), + ("Affinity", ctypes.c_size_t), + ("PriorityClass", ctypes.c_ulong), + ("SchedulingClass", ctypes.c_ulong), + ] + + class _IO_COUNTERS(ctypes.Structure): + _fields_ = [(n, ctypes.c_ulonglong) for n in ( + "ReadOperationCount", "WriteOperationCount", "OtherOperationCount", + "ReadTransferCount", "WriteTransferCount", "OtherTransferCount", + )] + + class _JOBOBJECT_EXTENDED_LIMIT_INFORMATION(ctypes.Structure): + _fields_ = [ + ("BasicLimitInformation", _JOBOBJECT_BASIC_LIMIT_INFORMATION), + ("IoInfo", _IO_COUNTERS), + ("ProcessMemoryLimit", ctypes.c_size_t), + ("JobMemoryLimit", ctypes.c_size_t), + ("PeakProcessMemoryUsed", ctypes.c_size_t), + ("PeakJobMemoryUsed", ctypes.c_size_t), + ] + + class _JOBOBJECT_CPU_RATE_CONTROL_INFORMATION(ctypes.Structure): + # union { DWORD CpuRate; DWORD Weight; } -> un seul DWORD (8o total) ; + # une Structure imbriquee donnerait 12o et l'API rejetterait. + _fields_ = [ + ("ControlFlags", ctypes.c_ulong), + ("CpuRate", ctypes.c_ulong), + ] + + class _MEMORYSTATUSEX(ctypes.Structure): + _fields_ = [ + ("dwLength", ctypes.c_ulong), + ("dwMemoryLoad", ctypes.c_ulong), + ("ullTotalPhys", ctypes.c_ulonglong), + ("ullAvailPhys", ctypes.c_ulonglong), + ("ullTotalPageFile", ctypes.c_ulonglong), + ("ullAvailPageFile", ctypes.c_ulonglong), + ("ullTotalVirtual", ctypes.c_ulonglong), + ("ullAvailVirtual", ctypes.c_ulonglong), + ("ullAvailExtendedVirtual", ctypes.c_ulonglong), + ] + + JOB_OBJECT_LIMIT_PRIORITY_CLASS = 0x00000020 + JOB_OBJECT_LIMIT_PROCESS_MEMORY = 0x00000100 + JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE = 0x00002000 + BELOW_NORMAL_PRIORITY_CLASS = 0x00004000 + JOB_OBJECT_CPU_RATE_CONTROL_ENABLE = 0x00000001 + JOB_OBJECT_CPU_RATE_CONTROL_HARD_CAP = 0x00000004 + JobObjectExtendedLimitInformation = 9 + JobObjectCpuRateControlInformation = 15 + JobObjectBasicProcessIdList = 3 + PROCESS_TERMINATE = 0x0001 + PROCESS_SET_QUOTA = 0x0100 + + def _total_physical_bytes() -> int: + stat = _MEMORYSTATUSEX() + stat.dwLength = ctypes.sizeof(stat) + ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(stat)) + return int(stat.ullTotalPhys) + + +class WindowsJob: + """Job Object kill-on-close. Le handle reste ouvert pendant tout le run : + si le superviseur meurt (crash, kill -9), Windows tue l'arbre lui-meme.""" + + def __init__(self, mem_frac: float, cpu_pct: int): + self.handle = None + self.ok = False + self.reason = "" + k32 = ctypes.windll.kernel32 + self._k32 = k32 + handle = k32.CreateJobObjectW(None, None) + if not handle: + self.reason = "CreateJobObjectW failed" + return + mem_limit = int(_total_physical_bytes() * mem_frac) + info = _JOBOBJECT_EXTENDED_LIMIT_INFORMATION() + info.BasicLimitInformation.LimitFlags = ( + JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + | JOB_OBJECT_LIMIT_PRIORITY_CLASS + | JOB_OBJECT_LIMIT_PROCESS_MEMORY + ) + info.BasicLimitInformation.PriorityClass = BELOW_NORMAL_PRIORITY_CLASS + info.ProcessMemoryLimit = mem_limit + if not k32.SetInformationJobObject( + handle, JobObjectExtendedLimitInformation, + ctypes.byref(info), ctypes.sizeof(info), + ): + self.reason = "SetInformationJobObject(extended) failed" + k32.CloseHandle(handle) + return + rate = _JOBOBJECT_CPU_RATE_CONTROL_INFORMATION() + rate.ControlFlags = ( + JOB_OBJECT_CPU_RATE_CONTROL_ENABLE + | JOB_OBJECT_CPU_RATE_CONTROL_HARD_CAP + ) + rate.CpuRate = cpu_pct * 100 + if not k32.SetInformationJobObject( + handle, JobObjectCpuRateControlInformation, + ctypes.byref(rate), ctypes.sizeof(rate), + ): + # Plafond CPU non pose : le kill-tree + priorite restent valides. + self.reason = "cpu-rate-cap-unavailable" + self.handle = handle + self.ok = True + + def assign(self, pid: int) -> bool: + k32 = self._k32 + hproc = k32.OpenProcess( + PROCESS_TERMINATE | PROCESS_SET_QUOTA, False, pid + ) + if not hproc: + self.reason = "OpenProcess failed" + return False + try: + if not k32.AssignProcessToJobObject(self.handle, hproc): + # Nesting de jobs incompatible (parent deja dans un job sans + # breakaway) : on continue sans confinement job, le fallback + # PPID-chain prendra le relais a la termination. + self.reason = "assign-failed-nested-job" + return False + return True + finally: + k32.CloseHandle(hproc) + + def member_pids(self) -> list[int]: + k32 = self._k32 + cap = 512 + class _LIST(ctypes.Structure): + _fields_ = [ + ("NumberOfAssignedProcesses", ctypes.c_ulong), + ("NumberOfProcessIdsInList", ctypes.c_ulong), + ("ProcessIdList", ctypes.c_size_t * cap), + ] + buf = _LIST() + buf.NumberOfAssignedProcesses = cap + if not k32.QueryInformationJobObject( + self.handle, JobObjectBasicProcessIdList, + ctypes.byref(buf), ctypes.sizeof(buf), None, + ): + return [] + n = buf.NumberOfProcessIdsInList + return [int(buf.ProcessIdList[i]) for i in range(min(n, cap))] + + def terminate(self) -> None: + self._k32.TerminateJobObject(self.handle, 1) + + def close(self) -> None: + if self.handle: + self._k32.CloseHandle(self.handle) + self.handle = None + + +# --------------------------------------------------------------------------- +# Terminaison par chaine PPID — fallback sans Job Object ET sonde d'orphelins +# --------------------------------------------------------------------------- + +def snapshot_processes() -> dict[int, int]: + """{pid: ppid} de tous les processus. Windows: Toolhelp32Snapshot. + POSIX: ps -o pid=,ppid=.""" + table: dict[int, int] = {} + if os.name == "nt": + k32 = ctypes.windll.kernel32 + TH32CS_SNAPPROCESS = 0x00000002 + + class _PE(ctypes.Structure): + _fields_ = [ + ("dwSize", ctypes.c_ulong), + ("cntUsage", ctypes.c_ulong), + ("th32ProcessID", ctypes.c_ulong), + ("th32DefaultHeapID", ctypes.c_size_t), + ("th32ModuleID", ctypes.c_ulong), + ("cntThreads", ctypes.c_ulong), + ("th32ParentProcessID", ctypes.c_ulong), + ("pcPriClassBase", ctypes.c_long), + ("dwFlags", ctypes.c_ulong), + ("szExeFile", ctypes.c_char * 260), + ] + + snap = k32.CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) + if snap == -1 or not snap: + return table + try: + entry = _PE() + entry.dwSize = ctypes.sizeof(entry) + if k32.Process32First(snap, ctypes.byref(entry)): + while True: + table[int(entry.th32ProcessID)] = int( + entry.th32ParentProcessID + ) + if not k32.Process32Next(snap, ctypes.byref(entry)): + break + finally: + k32.CloseHandle(snap) + else: + out = subprocess.run( + ["ps", "-eo", "pid=,ppid="], + capture_output=True, text=True, timeout=15, + encoding="utf-8", errors="replace", + ) + for line in out.stdout.splitlines(): + parts = line.split() + if len(parts) == 2: + try: + table[int(parts[0])] = int(parts[1]) + except ValueError: + continue + return table + + +def resume_process(pid: int) -> int: + """Reprend un processus cree suspendu (CREATE_SUSPENDED) : resume chaque + thread du pid. Necessaire pour que la racine n'execute rien avant d'etre + assignee au Job Object — c'est ce qui ferme la fenetre ou un enfant + pourrait naitre hors du job.""" + if os.name != "nt": + return 0 + k32 = ctypes.windll.kernel32 + TH32CS_SNAPTHREAD = 0x00000004 + THREAD_SUSPEND_RESUME = 0x0002 + + class _TE(ctypes.Structure): + _fields_ = [ + ("dwSize", ctypes.c_ulong), + ("cntUsage", ctypes.c_ulong), + ("th32ThreadID", ctypes.c_ulong), + ("th32OwnerProcessID", ctypes.c_ulong), + ("tpBasePri", ctypes.c_long), + ("tpDeltaPri", ctypes.c_long), + ("dwFlags", ctypes.c_ulong), + ] + + snap = k32.CreateToolhelp32Snapshot(TH32CS_SNAPTHREAD, 0) + if not snap or snap == -1: + return 0 + resumed = 0 + try: + entry = _TE() + entry.dwSize = ctypes.sizeof(entry) + if k32.Thread32First(snap, ctypes.byref(entry)): + while True: + if entry.th32OwnerProcessID == pid: + h = k32.OpenThread( + THREAD_SUSPEND_RESUME, False, entry.th32ThreadID + ) + if h: + k32.ResumeThread(h) + k32.CloseHandle(h) + resumed += 1 + if not k32.Thread32Next(snap, ctypes.byref(entry)): + break + finally: + k32.CloseHandle(snap) + return resumed + + +def descendants_of(root_pid: int, table: dict[int, int] | None = None) -> set[int]: + """Descendants vivants de ``root_pid``. Sous Windows un orphelin GARDE le + pid de son parent mort comme PPID : la chaine reste donc traicable meme + apres la mort de la racine et des intermediaires.""" + table = table if table is not None else snapshot_processes() + children: dict[int, list[int]] = {} + for pid, ppid in table.items(): + children.setdefault(ppid, []).append(pid) + found: set[int] = set() + frontier = list(children.get(root_pid, [])) + while frontier: + pid = frontier.pop() + if pid in found: + continue + found.add(pid) + frontier.extend(children.get(pid, [])) + return found + + +def kill_pids(pids: list[int]) -> None: + if os.name == "nt": + k32 = ctypes.windll.kernel32 + for pid in pids: # bottom-up fourni par l'appelant + hproc = k32.OpenProcess(PROCESS_TERMINATE, False, pid) + if hproc: + k32.TerminateProcess(hproc, 1) + k32.CloseHandle(hproc) + else: + for pid in pids: + try: + os.kill(pid, signal.SIGKILL) + except OSError: + pass + + +def find_orphans(root_pid: int, job: "WindowsJob | None") -> list[int]: + """Postcondition : orphelins = job non vide apres grace, OU descendants + vivants de la racine (chaine PPID) encore presents.""" + orphans: set[int] = set() + if os.name == "nt" and job is not None and job.handle: + orphans.update(job.member_pids()) + table = snapshot_processes() + orphans.update( + p for p in descendants_of(root_pid, table) if p in table + ) + return sorted(orphans) + + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +def config() -> dict: + cpu = os.cpu_count() or 4 + return { + "cap": int(os.environ.get("LEAN_EXEC_CAP", max(2, min(8, cpu // 2)))), + "budget": int(os.environ.get("LEAN_EXEC_BUDGET", 2)), + "jobs": int(os.environ.get("LEAN_EXEC_JOBS", max(1, cpu // 4))), + "mem_frac": float(os.environ.get("LEAN_EXEC_MEM_FRAC", 0.80)), + "cpu_pct": int(os.environ.get("LEAN_EXEC_CPU_PCT", 90)), + } + + +def bound_command(cmd: list[str], jobs: int) -> list[str]: + """Parallelisme explicitement borne (regle interim de flotte #15666) : + ``-Kjobs=N`` insere pour ``lake build`` nu, le flag reste decidable.""" + if ( + len(cmd) >= 2 + and cmd[0].rsplit("\\", 1)[-1].rsplit("/", 1)[-1].lower() + in ("lake", "lake.exe") + and cmd[1] == "build" + and not any(a.startswith("-K") or a.startswith("--jobs") + for a in cmd[2:]) + ): + return [cmd[0], cmd[1], f"-Kjobs={jobs}", *cmd[2:]] + return cmd + + +# --------------------------------------------------------------------------- +# Le run +# --------------------------------------------------------------------------- + +def _write_run_record(run_id: str, record: dict) -> Path: + runs_dir().mkdir(parents=True, exist_ok=True) + path = runs_dir() / f"{run_id}.json" + path.write_text(json.dumps(record, indent=2), encoding="utf-8") + return path + + +def _remove_run_record(run_id: str) -> None: + try: + (runs_dir() / f"{run_id}.json").unlink() + except OSError: + pass + + +def _emit(result: dict, as_json: bool) -> int: + state_dir().mkdir(parents=True, exist_ok=True) + (state_dir() / "last_run.json").write_text( + json.dumps(result, indent=2), encoding="utf-8" + ) + if as_json: + print(json.dumps(result, indent=2)) + else: + print( + f"[lean_exec] {result['status']} exit={result['exit_code']} " + f"duration={result.get('duration_s', 0):.1f}s " + f"orphans={len(result.get('orphans', []))}", + file=sys.stderr, + ) + for key in ("child_exit_code", "reason", "budget_violation"): + if result.get(key) is not None: + print(f"[lean_exec] {key}: {result[key]}", file=sys.stderr) + return result["exit_code"] + + +def run_command( + cmd: list[str], + timeout_s: float | None = None, + cap_override: int | None = None, + budget_override: int | None = None, + as_json: bool = False, +) -> int: + cfg = config() + cap = cap_override if cap_override is not None else cfg["cap"] + budget = budget_override if budget_override is not None else cfg["budget"] + run_id = uuid.uuid4().hex[:12] + started = time.time() + + result: dict = { + "run_id": run_id, + "cmd": cmd, + "backend": "windows-job" if os.name == "nt" else "posix", + "cap": cap, + "budget": budget, + "jobs": cfg["jobs"], + "orphans": [], + "child_exit_code": None, + "budget_violation": None, + "reason": None, + "duration_s": 0.0, + "population_before": None, + } + + # --- Admission machine-wide sous verrou (ferme count->spawn TOCTOU) --- + try: + with AdmissionLock(): + swept = sweep_stale_runs() + native_pop, native_src = scan_native_population() + if native_pop < 0: + # Fail-closed : impossible de compter = impossible de plafonner. + result.update( + status="refused", exit_code=EXIT_REFUSED, + reason=f"population unmeasurable ({native_src})", + ) + return _emit(result, as_json) + registered, _live = live_registered_budgets() + if native_pop + budget > cap: + result.update( + status="refused", exit_code=EXIT_REFUSED, + reason=( + f"machine-wide cap {cap}: native population " + f"{native_pop} + requested budget {budget} > cap" + ), + population_before={"native": native_pop}, + ) + return _emit(result, as_json) + if registered + budget > cap: + result.update( + status="refused", exit_code=EXIT_REFUSED, + reason=( + f"machine-wide cap {cap}: live registered budgets " + f"{registered} + requested budget {budget} > cap" + ), + population_before={"native": native_pop}, + ) + return _emit(result, as_json) + + env = os.environ.copy() + env.setdefault("LEAN_NUM_THREADS", str(cfg["jobs"])) + spawn_cmd = bound_command(cmd, cfg["jobs"]) + result["cmd_effective"] = spawn_cmd + + job = None + popen_kwargs: dict = {"env": env} + if os.name == "nt": + job = WindowsJob(cfg["mem_frac"], cfg["cpu_pct"]) + # SUSPENDED -> assign -> resume : la racine est dans le job + # AVANT de pouvoir exécuter la moindre instruction, aucun + # enfant ne peut naitre hors du job. + popen_kwargs["creationflags"] = ( + getattr(subprocess, "CREATE_NO_WINDOW", 0) + | getattr(subprocess, "CREATE_SUSPENDED", 0) + ) + else: + popen_kwargs["start_new_session"] = True + + try: + proc = subprocess.Popen(spawn_cmd, **popen_kwargs) + except OSError as exc: + if job: + job.close() + result.update( + status="refused", exit_code=EXIT_REFUSED, + reason=f"spawn failed: {exc}", + ) + return _emit(result, as_json) + + confined = True + if job is not None: + if not job.ok or not job.assign(proc.pid): + confined = False + result["backend"] = f"windows-fallback ({job.reason})" + # La racine n'execute rien tant qu'elle n'est pas dans le job. + n_resumed = resume_process(proc.pid) + result["threads_resumed"] = n_resumed + if n_resumed == 0: + result["backend"] = ( + f"{result['backend']}-resume-failed" + ) + + _write_run_record(run_id, { + "pid": proc.pid, + "host": host_id(), + "cmd": cmd, + "cwd": os.getcwd(), + "budget": budget, + "cap": cap, + "started_utc": time.strftime( + "%Y-%m-%dT%H:%M:%SZ", time.gmtime() + ), + "started_epoch": started, + "confined": confined, + }) + result["population_before"] = { + "native": native_pop, + "wsl": scan_wsl_population()[0], + } + except TimeoutError as exc: + result.update( + status="refused", exit_code=EXIT_REFUSED, reason=str(exc) + ) + return _emit(result, as_json) + + # --- Supervision : timeout / interruption / violation de budget --- + def terminate_tree() -> list[int]: + """Tue l'arbre par tous les moyens, puis verifie la postcondition.""" + if os.name == "nt": + if confined and job is not None and job.handle: + job.terminate() + time.sleep(1.0) + kill_pids(sorted(descendants_of(proc.pid), reverse=True)) + else: + try: + os.killpg(proc.pid, signal.SIGKILL) + except OSError: + pass + kill_pids(sorted(descendants_of(proc.pid), reverse=True)) + orphans = [] + for _ in range(4): # grace ~6 s, re-sonde + time.sleep(1.5) + orphans = find_orphans(proc.pid, job if confined else None) + if not orphans: + break + kill_pids(orphans) + return orphans + + status = "ok" + child_exit = None + killed = False + orphans: list[int] = [] + try: + while True: + try: + child_exit = proc.wait(timeout=2.0) + break + except subprocess.TimeoutExpired: + if timeout_s is not None and time.time() - started > timeout_s: + status = "timeout" + killed = True + orphans = terminate_tree() + break + # Budget supervise : ce run ne doit jamais heberger plus de + # lean/lake vivants que son budget declare. + if confined and os.name == "nt" and job is not None: + table = snapshot_processes() + own = sum( + 1 for p in descendants_of(proc.pid, table) + if _proc_name(p, table) in LEAN_PROC_NAMES + ) + if own > budget: + result["budget_violation"] = ( + f"{own} lean/lake descendants > budget {budget}" + ) + status = "timeout" + killed = True + orphans = terminate_tree() + break + except KeyboardInterrupt: + status = "interrupted" + killed = True + orphans = terminate_tree() + + # --- Postcondition finale : zero descendant orphelin, echec visible --- + if not killed: + for _ in range(2): + time.sleep(1.0) + orphans = find_orphans(proc.pid, job if confined else None) + if not orphans: + break + kill_pids(orphans) + time.sleep(1.0) + orphans = find_orphans(proc.pid, job if confined else None) + if orphans: + kill_pids(orphans) + time.sleep(1.5) + orphans = find_orphans(proc.pid, job if confined else None) + + if job is not None: + job.close() # kill-on-close : dernier filet, meme sur chemin lent + _remove_run_record(run_id) + + result.update({ + "pid": proc.pid, + "status": status, + "child_exit_code": child_exit, + "killed": killed, + "orphans": orphans, + "duration_s": round(time.time() - started, 2), + }) + if orphans: + result["exit_code"] = EXIT_ORPHANS + result["reason"] = f"cleanup incomplete: survivors {orphans}" + elif status == "timeout": + result["exit_code"] = EXIT_TIMEOUT + elif status == "interrupted": + result["exit_code"] = EXIT_INTERRUPTED + elif child_exit is not None and child_exit != 0: + result["exit_code"] = EXIT_CHILD + result["status"] = "child_failed" + else: + result["exit_code"] = EXIT_OK + return _emit(result, as_json) + + +def _proc_name(pid: int, _table: dict[int, int]) -> str: + """Nom executable d'un pid (best-effort, pour le filtre lean/lake).""" + try: + if os.name == "nt": + out = subprocess.run( + ["tasklist", "/FI", f"PID eq {pid}", "/FO", "CSV", "/NH"], + capture_output=True, text=True, timeout=10, + encoding="utf-8", errors="replace", + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + for line in out.stdout.splitlines(): + if line.strip(): + return line.split('","')[0].strip('"').lower() + return "" + out = subprocess.run( + ["ps", "-o", "comm=", "-p", str(pid)], + capture_output=True, text=True, timeout=10, + encoding="utf-8", errors="replace", + ) + return out.stdout.strip().lower() + except (OSError, subprocess.SubprocessError): + return "" + + +# --------------------------------------------------------------------------- +# status +# --------------------------------------------------------------------------- + +def status(as_json: bool = False) -> int: + cfg = config() + swept = sweep_stale_runs() + native_pop, native_src = scan_native_population() + wsl_pop, wsl_src = scan_wsl_population() + _registered, live = live_registered_budgets() + payload = { + "state_dir": str(state_dir()), + "config": cfg, + "population": { + "native": native_pop, "native_src": native_src, + "wsl": wsl_pop, "wsl_src": wsl_src, + }, + "live_runs": live, + "swept_stale": swept, + "headroom": max(0, cfg["cap"] - native_pop), + } + if as_json: + print(json.dumps(payload, indent=2)) + else: + print(f"state: {payload['state_dir']}") + print( + f"population: native={native_pop} ({native_src}) " + f"wsl={wsl_pop} ({wsl_src}) cap={cfg['cap']}" + ) + print(f"live runs: {len(live)} (swept {len(swept)} stale)") + return EXIT_OK + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Organe d'execution Lean confine (T1, See #15666)" + ) + sub = parser.add_subparsers(dest="action") + + p_run = sub.add_parser("run", help="Executer une commande lean/lake confinee") + p_run.add_argument("--timeout", type=float, default=None, + help="Timeout en secondes (l'arbre entier meurt)") + p_run.add_argument("--cap", type=int, default=None, + help="Cap machine-wide pour CE run uniquement") + p_run.add_argument("--budget", type=int, default=None, + help="Budget lean/lake max de ce run") + p_run.add_argument("--json", action="store_true", + help="Resultat JSON sur stdout") + p_run.add_argument("cmd", nargs=argparse.REMAINDER, + help="commande apres --") + + p_status = sub.add_parser("status", help="Population, cap, runs vivants") + p_status.add_argument("--json", action="store_true") + + args = parser.parse_args(argv) + try: + if args.action == "status": + return status(as_json=args.json) + if args.action == "run": + cmd = args.cmd + if cmd and cmd[0] == "--": + cmd = cmd[1:] + if not cmd: + parser.error("run: commande requise (run -- lake build ...)") + return run_command( + cmd, timeout_s=args.timeout, cap_override=args.cap, + budget_override=args.budget, as_json=args.json, + ) + except Exception as exc: # l'organe echoue visiblement, jamais en trace + print(f"[lean_exec] internal error: {exc}", file=sys.stderr) + return EXIT_INTERNAL + parser.print_help() + return EXIT_INTERNAL + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/lean/tests/test_lean_exec.py b/scripts/lean/tests/test_lean_exec.py new file mode 100644 index 0000000000..a6d19c9cb2 --- /dev/null +++ b/scripts/lean/tests/test_lean_exec.py @@ -0,0 +1,374 @@ +#!/usr/bin/env python3 +"""Tests de l'organe d'execution Lean confine (T1, issue #15666). + +Discriminants exiges par le cahier des charges §6 et le dispatch P0 : + +- ``test_admission_cap_machine_wide_two_worktrees`` : deux demandeurs + concurrents depuis deux repertoires distincts ne depassent JAMAIS ensemble + le cap — c'est le controle qui distingue un cap machine-wide d'un cap par + arbre. +- ``test_timeout_kills_whole_tree`` : le timeout tue TOUTE la descendance. +- ``test_planted_orphan_is_detected`` : controle par **faux negatif** du + detecteur — un enfant orphelin qui DOIT etre attrape l'est ; sans ce + controle, l'organe rendrait « propre » exactement comme sur une machine + sans Lean. +- ``test_positive_control_real_lake`` : une compilation ciblee REELLE + (``lake env lean``) passe sous le budget et publie ses metriques. + +Les tests d'admission et de confinement tournent avec des enfants Python +(sleepers) : ils ne dependent pas du toolchain Lean. Le controle positif, lui, +est skippe si ``lake`` est absent. + +Execution directe (sans pytest) : + python scripts/lean/tests/test_lean_exec.py +Ou via pytest : + pytest scripts/lean/tests/test_lean_exec.py +""" + +import json +import os +import shutil +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +import lean_exec as le # noqa: E402 + +LEAN_EXEC = str(Path(__file__).resolve().parent.parent / "lean_exec.py") +PY = sys.executable +REPO_ROOT = Path(__file__).resolve().parents[3] + +SLEEP_CMD = [PY, "-c", "import time; time.sleep(4)"] + + +def _env(state: Path, **extra) -> dict: + env = os.environ.copy() + env["LEAN_EXEC_STATE_DIR"] = str(state) + env["LEAN_EXEC_WSL"] = "off" + env.update({k: str(v) for k, v in extra.items()}) + return env + + +def _run(state: Path, args: list[str], cwd: Path | None = None, + timeout: float = 120, **extra) -> subprocess.CompletedProcess: + return subprocess.run( + [PY, LEAN_EXEC, *args], env=_env(state, **extra), + cwd=str(cwd) if cwd else None, + capture_output=True, text=True, timeout=timeout, + encoding="utf-8", errors="replace", + ) + + +def _last(state: Path) -> dict: + return json.loads( + (state / "last_run.json").read_text(encoding="utf-8") + ) + + +# --------------------------------------------------------------------------- +# Resolution d'etat + bornage de commande (pur, sans spawn) +# --------------------------------------------------------------------------- + +def test_state_dir_resolution(): + """Meme chaine que gpu.py:438, plus un override d'isolation pour les tests.""" + saved = {k: os.environ.get(k) for k in + ("LEAN_EXEC_STATE_DIR", "LOCALAPPDATA", "XDG_STATE_HOME")} + try: + for k in saved: + os.environ.pop(k, None) + os.environ["LEAN_EXEC_STATE_DIR"] = r"C:\tmp\iso" + assert le.state_dir() == Path(r"C:\tmp\iso") + os.environ.pop("LEAN_EXEC_STATE_DIR") + os.environ["LOCALAPPDATA"] = r"C:\Users\x\AppData\Local" + os.environ["XDG_STATE_HOME"] = "/xdg" + assert le.state_dir() == Path(r"C:\Users\x\AppData\Local") / ( + "CoursIA" if os.name == "nt" else "coursia") / "lean_exec" + os.environ.pop("LOCALAPPDATA") + got = le.state_dir() + assert "xdg" in got.parts, got + os.environ.pop("XDG_STATE_HOME") + assert ".local" in str(le.state_dir()) + finally: + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + +def test_bound_command_inserts_kjobs(): + """Le parallelisme est borne explicitement, jamais le defaut.""" + assert le.bound_command(["lake", "build"], 5) == [ + "lake", "build", "-Kjobs=5"] + # Un flag deja pose par l'appelant est respecte (decidable, pas ecrase). + assert le.bound_command(["lake", "build", "-Kjobs=2"], 5) == [ + "lake", "build", "-Kjobs=2"] + # Hors build : jamais de reecriture de la ligne de commande. + assert le.bound_command(["lake", "env", "lean", "X.lean"], 5) == [ + "lake", "env", "lean", "X.lean"] + assert le.bound_command(["lean", "X.lean"], 5) == ["lean", "X.lean"] + + +# --------------------------------------------------------------------------- +# Admission machine-wide — deux worktrees concurrents +# --------------------------------------------------------------------------- + +def test_admission_cap_machine_wide_two_worktrees(): + with tempfile.TemporaryDirectory() as td: + state = Path(td) / "state" + w1, w2, w3 = (Path(td) / n for n in ("w1", "w2", "w3")) + for w in (w1, w2, w3): + w.mkdir() + cap = dict(LEAN_EXEC_CAP=2, LEAN_EXEC_BUDGET=1) + procs = [ + subprocess.Popen( + [PY, LEAN_EXEC, "run", "--json", "--", *SLEEP_CMD], + env=_env(state, **cap), cwd=str(w), + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + for w in (w1, w2) + ] + time.sleep(0.5) # laisse les deux premiers s'enregistrer + third = _run(state, ["run", "--json", "--", *SLEEP_CMD], + cwd=w3, **cap) + for p in procs: + p.wait(timeout=60) + codes = [p.returncode for p in procs] + [third.returncode] + assert codes.count(0) == 2, f"attendu 2 admissions, vu {codes}" + assert third.returncode == le.EXIT_REFUSED, ( + f"le 3e demandeur devait etre refuse, vu {third.returncode}") + # Le refus se lit sur le stdout du 3e run : last_run.json est partage + # par le state dir et les 3 runs concurrents y ecrivent. + refused = json.loads(third.stdout[third.stdout.index("{"):]) + assert "cap" in (refused.get("reason") or ""), refused.get("reason") + + +def test_admission_refuses_when_population_unmeasurable(): + """Fail-closed : pas de telemetrie = pas de lancement optimiste. + + Le scan est patche plutot que rendu introuvable : sous Windows + ``tasklist.exe`` se resout via System32 meme avec un PATH vide, donc + vider PATH ne testerait pas la branche fail-closed. + """ + saved_state = os.environ.get("LEAN_EXEC_STATE_DIR") + saved_scan = le.scan_native_population + with tempfile.TemporaryDirectory() as td: + os.environ["LEAN_EXEC_STATE_DIR"] = str(Path(td) / "s") + le.scan_native_population = lambda: (-1, "unavailable: test") + try: + rc = le.run_command(SLEEP_CMD, timeout_s=10, as_json=True) + assert rc == le.EXIT_REFUSED, rc + res = json.loads( + (le.state_dir() / "last_run.json").read_text("utf-8")) + assert "unmeasurable" in res["reason"], res["reason"] + finally: + le.scan_native_population = saved_scan + if saved_state is None: + os.environ.pop("LEAN_EXEC_STATE_DIR", None) + else: + os.environ["LEAN_EXEC_STATE_DIR"] = saved_state + + +# --------------------------------------------------------------------------- +# Confinement : le timeout tue toute la descendance +# --------------------------------------------------------------------------- + +def test_timeout_kills_whole_tree(): + with tempfile.TemporaryDirectory() as td: + state = Path(td) / "state" + pidfile = Path(td) / "grandchild.pid" + root = ( + "import subprocess,sys,time;" + "p=subprocess.Popen([sys.executable,'-c'," + "'import time;time.sleep(120)']);" + f"open(r'{pidfile}','w').write(str(p.pid));" + "time.sleep(120)" + ) + rc = _run(state, ["run", "--json", "--timeout", "3", "--", + PY, "-c", root], timeout=90) + res = _last(state) + assert rc.returncode == le.EXIT_TIMEOUT, ( + f"attendu {le.EXIT_TIMEOUT}, vu {rc.returncode}") + assert res["status"] == "timeout" + assert res["orphans"] == [], f"orphelins: {res['orphans']}" + deadline = time.time() + 10 + while time.time() < deadline and not pidfile.exists(): + time.sleep(0.2) + if pidfile.exists(): + gpid = int(pidfile.read_text().strip()) + time.sleep(1.5) + assert not le.pid_alive(gpid), ( + f"le petit-fils {gpid} a survecu au timeout") + + +def test_normal_exit_is_clean(): + with tempfile.TemporaryDirectory() as td: + state = Path(td) / "state" + rc = _run(state, ["run", "--json", "--", PY, "-c", "print('ok')"], + timeout=60) + res = _last(state) + assert rc.returncode == le.EXIT_OK, rc.returncode + assert res["orphans"] == [] + assert res["child_exit_code"] == 0 + + +def test_child_failure_maps_to_exit_1(): + with tempfile.TemporaryDirectory() as td: + state = Path(td) / "state" + rc = _run(state, ["run", "--json", "--", PY, "-c", "import sys; sys.exit(3)"], + timeout=60) + res = _last(state) + assert rc.returncode == le.EXIT_CHILD, rc.returncode + assert res["child_exit_code"] == 3 + assert res["status"] == "child_failed" + + +def test_planted_orphan_is_detected(): + """Controle par faux negatif : un orphelin qui DOIT etre attrape l'est. + + L'enfant est plante HORS confinement (parent mort, enfant vivant) : le + detecteur doit le lister — sinon l'organe rendrait « propre » a tort. + """ + with tempfile.TemporaryDirectory() as td: + pidfile = Path(td) / "orphan.pid" + planter = ( + "import subprocess,sys,time;" + "p=subprocess.Popen([sys.executable,'-c'," + "'import time;time.sleep(120)']);" + f"open(r'{pidfile}','w').write(str(p.pid));" + "time.sleep(1)" + ) + parent = subprocess.Popen([PY, "-c", planter]) + parent.wait(timeout=30) + deadline = time.time() + 10 + while time.time() < deadline and not pidfile.exists(): + time.sleep(0.2) + assert pidfile.exists(), "le planteur n'a pas ecrit le pid" + orphan = int(pidfile.read_text().strip()) + try: + time.sleep(0.5) + orphans = le.find_orphans(parent.pid, None) + assert orphan in orphans, ( + f"orphelin {orphan} non detecte (vu {orphans})") + finally: + le.kill_pids([orphan]) + + +# --------------------------------------------------------------------------- +# Registre : peremption (reprise de tree_lock) +# --------------------------------------------------------------------------- + +def test_stale_run_record_is_swept_foreign_host_preserved(): + with tempfile.TemporaryDirectory() as td: + state = Path(td) / "state" + runs = state / "runs" + runs.mkdir(parents=True) + dead = {"pid": 999999999, "host": le.host_id(), "budget": 2} + foreign = {"pid": 1, "host": "autre-machine/nt", "budget": 2} + (runs / "dead.json").write_text(json.dumps(dead), encoding="utf-8") + (runs / "foreign.json").write_text(json.dumps(foreign), encoding="utf-8") + saved = os.environ.get("LEAN_EXEC_STATE_DIR") + os.environ["LEAN_EXEC_STATE_DIR"] = str(state) + try: + swept = le.sweep_stale_runs() + assert "dead" in swept + assert (runs / "foreign.json").exists(), ( + "un run d'un autre namespace de pids ne doit jamais etre " + "auto-retire (tree_lock.py:138)") + total, live = le.live_registered_budgets() + assert total == 0 and live == [] + finally: + if saved is None: + os.environ.pop("LEAN_EXEC_STATE_DIR", None) + else: + os.environ["LEAN_EXEC_STATE_DIR"] = saved + + +# --------------------------------------------------------------------------- +# Controle positif : compilation ciblee reelle sous le cap +# --------------------------------------------------------------------------- + +def _find_toolchain() -> str | None: + """Un toolchain du depot DEJA installe (elan), pour ne pas declencher de + telechargement pendant un test. Override : LEAN_EXEC_CONTROL_TOOLCHAIN.""" + override = os.environ.get("LEAN_EXEC_CONTROL_TOOLCHAIN") + if override: + return override + try: + out = subprocess.run(["elan", "toolchain", "list"], + capture_output=True, text=True, timeout=30, + encoding="utf-8", errors="replace") + installed = {ln.strip() for ln in out.stdout.splitlines() if ln.strip()} + except (OSError, subprocess.SubprocessError): + installed = set() + for candidate in REPO_ROOT.rglob("lean-toolchain"): + if ".lake" in candidate.parts: + continue + content = candidate.read_text(encoding="utf-8").strip() + if content in installed: + return content + return None + + +def test_positive_control_real_lake(): + """Une compilation ciblee REELLE passe sous le budget et publie ses metriques.""" + if shutil.which("lake") is None: + print("SKIP: lake absent du PATH") + return + toolchain = _find_toolchain() + if not toolchain: + print("SKIP: aucun lean-toolchain du depot n'est installe via elan") + return + with tempfile.TemporaryDirectory() as td: + state = Path(td) / "state" + proj = Path(td) / "control" + proj.mkdir() + (proj / "lean-toolchain").write_text(toolchain, encoding="utf-8") + (proj / "lakefile.toml").write_text( + 'name = "t1control"\nversion = "0.1.0"\n' + "defaultTargets = [\"T1Control\"]\n\n" + "[[lean_lib]]\nname = \"T1Control\"\n", + encoding="utf-8", + ) + (proj / "T1Control.lean").write_text( + "theorem t1_control : 1 + 1 = 2 := rfl\n", encoding="utf-8") + rc = _run( + state, + ["run", "--json", "--timeout", "900", "--budget", "2", "--", + "lake", "env", "lean", "T1Control.lean"], + cwd=proj, timeout=1200, + LEAN_EXEC_CAP=4, LEAN_NUM_THREADS=4, + ) + res = _last(state) + assert rc.returncode == le.EXIT_OK, ( + f"compilation reelle en echec: {rc.returncode}\n" + f"stderr={rc.stderr[-2000:]}\nstdout={rc.stdout[-2000:]}") + assert res["orphans"] == [] + print( + f"POSITIVE CONTROL OK: duration={res['duration_s']}s " + f"backend={res['backend']} cap={res['cap']} " + f"population_before={res['population_before']} " + f"cmd={res.get('cmd_effective')}" + ) + + +# --------------------------------------------------------------------------- +# Runner direct (convention des tests scripts/lean) +# --------------------------------------------------------------------------- + +if __name__ == "__main__": + tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")] + failed = 0 + for fn in tests: + started = time.time() + try: + fn() + print(f"PASS {fn.__name__} ({time.time() - started:.1f}s)") + except Exception as exc: # noqa: BLE001 + failed += 1 + print(f"FAIL {fn.__name__}: {type(exc).__name__}: {exc}") + print(f"\n{len(tests) - failed}/{len(tests)} tests passes") + sys.exit(1 if failed else 0) From 6ff48c37521366750ae70852e35a7a3d9d58e03d Mon Sep 17 00:00:00 2001 From: jsboige Date: Sun, 13 Sep 2026 03:16:19 +0200 Subject: [PATCH 2/3] =?UTF-8?q?fix(lean,#15666):=20arbitrage=20T1=20?= =?UTF-8?q?=E2=80=94=20CREATE=5FSUSPENDED=20litteral=20+=20garde=20fail-cl?= =?UTF-8?q?osed,=20skipif=20POSIX,=20pytest.skip=20reel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reserve 1 : le getattr(subprocess,"CREATE_SUSPENDED",0) retombait sur 0 (CREATE_SUSPENDED n'est exporte ni par subprocess ni par _winapi — mesure Hermes c.5649329240 sur Modules/_winapi.c v3.13.5). Constante de module CREATE_SUSPENDED = 0x00000004 (winbase.h:416), et reprise a 0 fil = echec dur EXIT_INTERNAL + status internal-error + racine tuee — plus jamais un suffixe -resume-failed sur un run vert qui certifie un confinement qui n'a pas eu lieu. Reserve 2 : test_planted_orphan_is_detected mesure une propriete Windows (PPID conserve du parent mort) que POSIX contredit (reparentage PID 1 -> descendants_of structurellement vide) — skipif(os.name != "nt") avec la raison ecrite ; rouge programme sur le runner coursia-linux evite. Reserve 3 : test_positive_control_real_lake faisait print("SKIP"); return (= faux « passed » 0.22 s) — pytest.skip(...) desormais, rend un « s » visible dans le rapport. Controles : suite 10 passed en 40.31 s (le controle reel lake TOURN E — 40 s vs 0.22 s faux vert) ; sonde comportementale reserve 1 in-process : resume_process -> 0 rend rc=127 / internal-error / windows-job-resume-failed, racine suspendue tuee avant retour. Co-Authored-By: Claude Sonnet 5 --- scripts/lean/lean_exec.py | 35 ++++++++++++++++++++++++---- scripts/lean/tests/test_lean_exec.py | 20 ++++++++++++---- 2 files changed, 47 insertions(+), 8 deletions(-) diff --git a/scripts/lean/lean_exec.py b/scripts/lean/lean_exec.py index 60047b24a8..b142a71d06 100644 --- a/scripts/lean/lean_exec.py +++ b/scripts/lean/lean_exec.py @@ -68,6 +68,16 @@ EXIT_INTERNAL = 127 EXIT_INTERRUPTED = 130 +# winbase.h:416 -- CREATE_SUSPENDED n'est exporte ni par subprocess ni par +# _winapi (mesure Hermes c.5649329240 sur Modules/_winapi.c v3.13.5 : les +# cinq constantes CREATE_* exportees ne l'incluent pas, et le cycle +# suspendu->assigne->repris n'est pas completable en Python pur via _winapi). +# Valeur ABI Win32 stable depuis Windows NT : le litteral est la forme +# robuste. JAMAIS de getattr(..., 0) sur ce drapeau -- une retombee silencieuse +# lance la racine courante et fabrique un run vert non confine (reserve 1 de +# l'arbitrage #15666, issuecomment-5649841267). +CREATE_SUSPENDED = 0x00000004 + LEAN_PROC_NAMES = {"lean", "lean.exe", "lake", "lake.exe"} _STATE_SUBDIR_WIN = ("CoursIA", "lean_exec") @@ -794,8 +804,7 @@ def run_command( # AVANT de pouvoir exécuter la moindre instruction, aucun # enfant ne peut naitre hors du job. popen_kwargs["creationflags"] = ( - getattr(subprocess, "CREATE_NO_WINDOW", 0) - | getattr(subprocess, "CREATE_SUSPENDED", 0) + getattr(subprocess, "CREATE_NO_WINDOW", 0) | CREATE_SUSPENDED ) else: popen_kwargs["start_new_session"] = True @@ -820,9 +829,27 @@ def run_command( n_resumed = resume_process(proc.pid) result["threads_resumed"] = n_resumed if n_resumed == 0: - result["backend"] = ( - f"{result['backend']}-resume-failed" + # Reserve 1 (arbitrage #15666) : un root cree suspendu et + # non repris n'a PAS ete confine -- l'echec doit invalider + # le run, pas decorer le backend d'un suffixe vert. Le root + # est encore suspendu : on tue le job avant de rendre + # l'echec (meme ordre que terminate_tree). + confined = False + if job is not None and job.handle: + job.terminate() + time.sleep(1.0) + kill_pids(sorted(descendants_of(proc.pid), reverse=True)) + result.update( + status="internal-error", + exit_code=EXIT_INTERNAL, + reason=( + "resume_process resumed 0 threads: the root was " + "spawned with CREATE_SUSPENDED and could not be " + "resumed -- confinement is NOT delivered" + ), + backend=f"{result['backend']}-resume-failed", ) + return _emit(result, as_json) _write_run_record(run_id, { "pid": proc.pid, diff --git a/scripts/lean/tests/test_lean_exec.py b/scripts/lean/tests/test_lean_exec.py index a6d19c9cb2..73df1f78ae 100644 --- a/scripts/lean/tests/test_lean_exec.py +++ b/scripts/lean/tests/test_lean_exec.py @@ -34,6 +34,8 @@ import time from pathlib import Path +import pytest + sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) import lean_exec as le # noqa: E402 @@ -226,6 +228,15 @@ def test_child_failure_maps_to_exit_1(): assert res["status"] == "child_failed" +@pytest.mark.skipif( + os.name != "nt", + reason=( + "propriete Windows : un orphelin garde le PID de son parent mort " + "comme PPID ; sous POSIX le noyau reparente a PID 1 et " + "descendants_of(parent_mort) est structurellement vide -- le test " + "mesurerait le noyau, pas find_orphans (reserve 2, arbitrage #15666)" + ), +) def test_planted_orphan_is_detected(): """Controle par faux negatif : un orphelin qui DOIT etre attrape l'est. @@ -315,13 +326,14 @@ def _find_toolchain() -> str | None: def test_positive_control_real_lake(): """Une compilation ciblee REELLE passe sous le budget et publie ses metriques.""" + # Reserve 3 (arbitrage #15666) : un print+return rend « passed » sans + # rien controler -- pire que pas de controle. pytest.skip rend un « s » + # visible dans le rapport. if shutil.which("lake") is None: - print("SKIP: lake absent du PATH") - return + pytest.skip("lake absent du PATH") toolchain = _find_toolchain() if not toolchain: - print("SKIP: aucun lean-toolchain du depot n'est installe via elan") - return + pytest.skip("aucun lean-toolchain du depot n'est installe via elan") with tempfile.TemporaryDirectory() as td: state = Path(td) / "state" proj = Path(td) / "control" From b826fb8b13e63e4359912d79a00632bc29af5b77 Mon Sep 17 00:00:00 2001 From: jsboige Date: Sun, 13 Sep 2026 14:30:25 +0200 Subject: [PATCH 3/3] fix(lean,#15666): synchroniser l'admission sur runs/*.json, pas sur un delai test_admission_cap_machine_wide_two_worktrees attendait time.sleep(0.5) entre le Popen des deux premiers runs et le lancement du 3e demandeur. Un delai fixe court apres la machine : quand w2 n'avait pas encore ecrit son enregistrement, le 3e demandeur se faisait admettre a sa place et le test rendait {w1, w3} au lieu de {w1, w2}. L'enregistrement runs/.json est ecrit SOUS le verrou d'admission, juste apres le spawn+resume (lean_exec.py:854) : sa presence prouve que le run occupe deja sa part du cap. On attend donc cet etat observable, borne a 30 s, au lieu d'une duree arbitraire. Les assertions d'admission sont inchangees (2 codes 0 parmi les concurrents, 3e refuse avec une raison de cap). Une attente passive de type LEAN_EXEC_TEST_WAIT n'aurait pas repondu : elle deplace le probleme d'un facteur d'echelle sans synchroniser sur l'etat. Mesure : 5 executions consecutives vertes du seul test d'admission, puis 10 passed sur le fichier entier (77 s). See #15666 Co-Authored-By: Claude Sonnet 5 --- scripts/lean/tests/test_lean_exec.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/scripts/lean/tests/test_lean_exec.py b/scripts/lean/tests/test_lean_exec.py index 73df1f78ae..ee1822b5b5 100644 --- a/scripts/lean/tests/test_lean_exec.py +++ b/scripts/lean/tests/test_lean_exec.py @@ -133,7 +133,23 @@ def test_admission_cap_machine_wide_two_worktrees(): ) for w in (w1, w2) ] - time.sleep(0.5) # laisse les deux premiers s'enregistrer + # Synchronisation sur l'etat OBSERVABLE, pas sur une duree : le 3e + # demandeur ne doit partir qu'une fois les DEUX premiers enregistres. + # L'enregistrement est ecrit sous le verrou d'admission, juste apres + # le spawn+resume (lean_exec.py:854), donc sa presence prouve que le + # run occupe deja sa part du cap. Un delai fixe courait apres la + # machine : mesure #15940, sleep(0.5) laissait w2 non enregistre et + # le 3e demandeur se faisait admettre a sa place -> {w1, w3}. + def _registered() -> int: + return len(list((state / "runs").glob("*.json"))) + + deadline = time.monotonic() + 30.0 + while _registered() < 2 and time.monotonic() < deadline: + time.sleep(0.05) + assert _registered() >= 2, ( + "les deux premiers runs devaient s'enregistrer sous 30 s, " + f"vu {_registered()}") + third = _run(state, ["run", "--json", "--", *SLEEP_CMD], cwd=w3, **cap) for p in procs: