From 89633a8c12b72df32499cba7c6eca9e0c0b51577 Mon Sep 17 00:00:00 2001 From: jsboige Date: Tue, 15 Sep 2026 14:14:48 +0200 Subject: [PATCH 1/3] feat(lean,#15666): T5a validation charge bornee -- harnais de sondes pendant un build lake reel sous lean_exec Sondes pendant la charge (population via la mesure de l'organe, DriveFS sous garde-fou timeout, spawn p95 borne a la baseline, RAM) + postconditions last_run.json (status ok, orphelins 0, cmd reconciliee). --touch N genere la charge par mtime-bump (contenu inchange). 17 contrats, aucun lake requis. Mesure reelle (po-2026) : budget defaut 2 -> l'organe tue l'arbre a 96,8 s (budget_violation 4 > 2), orphelins 0, machine vivable (DriveFS p95 234 ms, spawn p95 78 ms) -- le scenario incident arrete net, chiffre. Co-Authored-By: Claude Sonnet 5 --- scripts/lean/validate_bounded_load.py | 455 ++++++++++++++++++++ scripts/tests/test_validate_bounded_load.py | 262 +++++++++++ 2 files changed, 717 insertions(+) create mode 100644 scripts/lean/validate_bounded_load.py create mode 100644 scripts/tests/test_validate_bounded_load.py diff --git a/scripts/lean/validate_bounded_load.py b/scripts/lean/validate_bounded_load.py new file mode 100644 index 0000000000..02bb7184b3 --- /dev/null +++ b/scripts/lean/validate_bounded_load.py @@ -0,0 +1,455 @@ +"""Validation charge bornee de l'organe lean_exec (T5a, See #15666). + +Incident fondateur (12 septembre 2026) : ~30 processus ``lean.exe`` a ~95 % CPU +ont etouffe une machine worker -- DriveFS tombe, puis les services de fond, +puis reboot du cluster. L'organe ``scripts/lean/lean_exec.py`` (T1-T3) confine +chaque invocation : admission machine-wide sous cap, Job Object kill-on-close, +budget supervise, postcondition zero orphelin. Cette tranche valide l'organe +SUR CHARGE REELLE : une compilation lake effective, echantillonnee pendant +qu'elle tourne. + +Le harnais mesure PENDANT la compilation (l'organe garantit le cap ; le +harnais verifie que la machine reste vivable) : + + - **population native lean/lake** : echantillonnee avec la MEME fonction de + mesure que l'organe (``lean_exec.scan_native_population``) -- le max + observe doit rester <= cap. C'est la contradiction directe du regime de + l'incident (30 lean sous cap 8). + - **sonde DriveFS** : ``listdir`` chronometre sur le lecteur monte (defaut + ``G:\\Mon Drive``). Un TIMEOUT de sonde = ECHEC : c'est le premier + symptome mesurable de l'asphyxie de septembre. Lecteur absent = sonde + ``unavailable``, notee honnetement, sans faux pass. La latence (hors + timeout) est rapportee, pas seuillee -- le regime mortel est le hang, pas + la milliseconde. + - **latence de spawn** : lancer un python trivial et attendre sa mort -- + proxy de la sante ordonnanceur du systeme, ce qui meurt en premier dans + l'incident. p95 pendant la charge <= max(rapport x p95 baseline, plancher). + - **RAM libre** : rapportee sans seuil invente (l'organe plafonne deja la + memoire du run via Job Object). + +Post-conditions lues dans l'enregistrement de l'organe (``last_run.json``, +ecrit a CHAQUE sortie de ``run``, y compris refuse) : status ``ok``, +orphelins vides, sortie 0. + +Generation de la charge : le harnais ne fabrique pas de charge fictive. Il +lance ``lean_exec.py run -- lake build`` dans le lake fourni ; pour qu'un +build cache-chaud produise du travail reel, ``--touch N`` augmente le mtime +de N modules propres du lake (contenu inchange, arbre git propre) -- la +recompilation en cascade qui suit est la charge. + +Codes de sortie : 0 = PASS, 1 = FAIL (chaque critere dur violé est nomme). +""" + +from __future__ import annotations + +import argparse +import ctypes +import json +import os +import subprocess +import tempfile +import sys +import threading +import time +from pathlib import Path + +# L'organe est un sibling du present script ; l'importer plutot que dupliquer +# sa mesure de population (memes regles de comptage, meme source tasklist). +sys.path.insert(0, str(Path(__file__).resolve().parent)) +import lean_exec # noqa: E402 + +DRIVEFS_OK = "ok" +DRIVEFS_TIMEOUT = "timeout" +DRIVEFS_ERROR = "error" +DRIVEFS_UNAVAILABLE = "unavailable" + +DEFAULT_DRIVEFS_ROOT = r"G:\Mon Drive" + + +# --------------------------------------------------------------------------- +# Sondes +# --------------------------------------------------------------------------- + +def probe_spawn_ms() -> float: + """Latence pour spawner un processus trivial et attendre sa mort.""" + t0 = time.monotonic() + subprocess.run( + [sys.executable, "-c", "pass"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + return (time.monotonic() - t0) * 1000.0 + + +def probe_drivefs(root: str | None, timeout_s: float) -> tuple[str, float | None]: + """Listdir chronometre sous garde-fou temps. + + Le listdir d'un DriveFS mort peut se bloquer des minutes : la sonde court + dans un fil daemon et est abandonnee au bout de ``timeout_s`` -- c'est le + comportement a detecter, pas un cas d'erreur a contourner. + """ + if not root or not os.path.isdir(root): + return DRIVEFS_UNAVAILABLE, None + box: dict = {} + + def _list() -> None: + try: + t0 = time.monotonic() + os.listdir(root) + box["ms"] = (time.monotonic() - t0) * 1000.0 + except OSError as exc: + box["error"] = str(exc) + + thread = threading.Thread(target=_list, daemon=True) + thread.start() + thread.join(timeout_s) + if thread.is_alive(): + return DRIVEFS_TIMEOUT, None + if "error" in box: + return DRIVEFS_ERROR, None + return DRIVEFS_OK, box.get("ms") + + +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), + ] + + +def ram_free_mb() -> float | None: + """RAM physiquement disponible, None si non mesurable.""" + try: + if os.name == "nt": + stat = _MemoryStatusEx() + stat.dwLength = ctypes.sizeof(_MemoryStatusEx) + if ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(stat)): + return round(stat.ullAvailPhys / (1024 * 1024), 1) + return None + with open("/proc/meminfo", encoding="ascii") as fh: + for line in fh: + if line.startswith("MemAvailable:"): + return round(int(line.split()[1]) / 1024, 1) + return None + except OSError: + return None + + +def take_sample(phase: str, drivefs_root: str | None, + probe_timeout_s: float) -> dict: + pop, pop_src = lean_exec.scan_native_population() + state, ms = probe_drivefs(drivefs_root, probe_timeout_s) + return { + "t": round(time.time(), 3), + "phase": phase, + "population": pop, + "population_src": pop_src, + "spawn_ms": round(probe_spawn_ms(), 1), + "drivefs": state, + "drivefs_ms": round(ms, 1) if ms is not None else None, + "ram_free_mb": ram_free_mb(), + } + + +# --------------------------------------------------------------------------- +# Echantillonnage pendant la charge +# --------------------------------------------------------------------------- + +class Sampler: + """Preleve des echantillons tant que le processus de charge vit.""" + + def __init__(self, drivefs_root: str | None, probe_timeout_s: float, + sample_fn=None) -> None: + self._sample = sample_fn or ( + lambda phase: take_sample(phase, drivefs_root, probe_timeout_s) + ) + + def collect(self, proc: subprocess.Popen, interval: float, + max_wall: float) -> tuple[list[dict], bool]: + """Echantillonne jusqu'a la mort de ``proc`` ou depassement du mur. + + Renvoie (echantillons, mur_depasse). Le premier echantillon est pris + immediatement : un build court ne doit pas echapper a la mesure. + """ + samples: list[dict] = [] + start = time.monotonic() + while proc.poll() is None: + samples.append(self._sample("load")) + if time.monotonic() - start > max_wall: + return samples, True + time.sleep(interval) + return samples, False + + +# --------------------------------------------------------------------------- +# Verdict +# --------------------------------------------------------------------------- + +def _p95(values: list[float]) -> float | None: + if not values: + return None + ordered = sorted(values) + idx = min(len(ordered) - 1, max(0, int(0.95 * len(ordered) + 0.5))) + return ordered[idx] + + +def evaluate(baseline: list[dict], load: list[dict], + organ_record: dict | None, cap: int, + spawn_ratio: float, spawn_floor_ms: float) -> dict: + """Verdict : chaque critere dur est cite avec sa mesure, PASS ou FAIL.""" + checks: list[dict] = [] + + def hard(name: str, ok: bool, detail: str) -> None: + checks.append({"criterion": name, "pass": bool(ok), "detail": detail}) + + status = (organ_record or {}).get("status") + orphans = (organ_record or {}).get("orphans") or [] + hard( + "organ_postcondition", + organ_record is not None and status == "ok" and not orphans, + f"organ status={status!r} orphelins={len(orphans)}", + ) + + pops = [s["population"] for s in load] + max_pop = max(pops) if pops else None + hard( + "population_cap", + bool(pops) and max_pop is not None and max_pop <= cap, + (f"max population observee {max_pop} <= cap {cap} " + f"({len(pops)} echantillons)") + if pops else + f"0 echantillon pendant la charge -- critere non verifiable", + ) + + timeouts = [s for s in load if s["drivefs"] == DRIVEFS_TIMEOUT] + unavailable = [s for s in load if s["drivefs"] == DRIVEFS_UNAVAILABLE] + hard( + "drivefs_alive", + not timeouts, + (f"{len(timeouts)} sonde(s) DriveFS en timeout") + if timeouts else + (f"aucun timeout ({len(unavailable)} sonde(s) unavailable)" + if unavailable else "aucune sonde DriveFS en timeout"), + ) + + base_spawn = _p95([s["spawn_ms"] for s in baseline]) + load_spawn = _p95([s["spawn_ms"] for s in load]) + bound = spawn_floor_ms + if base_spawn is not None: + bound = max(spawn_ratio * base_spawn, spawn_floor_ms) + hard( + "spawn_p95", + load_spawn is not None and load_spawn <= bound, + f"spawn p95 charge {load_spawn} ms <= borne {round(bound, 1)} ms " + f"(baseline p95 {base_spawn} ms, rapport max {spawn_ratio}, " + f"plancher {spawn_floor_ms} ms)", + ) + + ram_values = [s["ram_free_mb"] for s in load if s["ram_free_mb"] is not None] + drivefs_ms = [s["drivefs_ms"] for s in load if s["drivefs_ms"] is not None] + return { + "pass": all(c["pass"] for c in checks), + "checks": checks, + "report": { + "echantillons_baseline": len(baseline), + "echantillons_charge": len(load), + "population_max": max_pop, + "spawn_p95_baseline_ms": base_spawn, + "spawn_p95_charge_ms": load_spawn, + "drivefs_p95_ms": _p95(drivefs_ms) if drivefs_ms else None, + "ram_libre_min_mb": min(ram_values) if ram_values else None, + }, + } + + +# --------------------------------------------------------------------------- +# Charge +# --------------------------------------------------------------------------- + +def touch_own_modules(lake_dir: Path, n: int) -> list[str]: + """Mtime-bump des n modules propres les plus anciens du lake. + + Contenu strictement inchange (seul le mtime bouge) : l'arbre git reste + propre, mais lake considere les oleans perimes et recompile -- c'est la + charge. Les modules sous ``.lake/`` (paquets vendores) sont exclus. + """ + modules = [ + p for p in lake_dir.rglob("*.lean") + if ".lake" not in p.parts and p.name != "lakefile.lean" + ] + modules.sort(key=lambda p: p.stat().st_mtime) + now = time.time() + touched = [] + for path in modules[:n]: + os.utime(path, (now, now)) + touched.append(path.relative_to(lake_dir).as_posix()) + return touched + + +def read_organ_record(expected_cmd: list[str]) -> dict | None: + """Lit ``last_run.json`` ; None si absent ou manifestement stale. + + Un enregistrement d'un run precedent (cmd differente) n'est pas une + preuve pour CE run : le harnais le rejete plutot que de valider sur du + stale. + """ + path = lean_exec.state_dir() / "last_run.json" + try: + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + return None + if record.get("cmd") != expected_cmd: + return None + return record + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Validation charge bornee de l'organe lean_exec (T5a, #15666)") + parser.add_argument("--lake-dir", required=True, + help="repertoire du lake a compiler (reel)") + parser.add_argument("--cmd", nargs=argparse.REMAINDER, + default=["lake", "build"], + help="commande de charge (defaut: lake build) ; " + "REMAINDER : dernier argument du harnais, " + "tout ce qui suit est la commande") + parser.add_argument("--cap", type=int, default=None, + help="cap machine-wide transmis a l'organe") + parser.add_argument("--budget", type=int, default=None, + help="budget lean/lake transmis a l'organe") + parser.add_argument("--timeout", type=float, default=None, + help="timeout (s) transmis a l'organe") + parser.add_argument("--touch", type=int, default=0, metavar="N", + help="mtime-bump de N modules propres pour generer " + "la charge (contenu inchange)") + parser.add_argument("--baselines", type=int, default=3) + parser.add_argument("--interval", type=float, default=2.0) + parser.add_argument("--probe-timeout", type=float, default=10.0) + parser.add_argument("--drivefs-root", default=DEFAULT_DRIVEFS_ROOT) + parser.add_argument("--spawn-ratio", type=float, default=5.0) + parser.add_argument("--spawn-floor-ms", type=float, default=5000.0) + parser.add_argument("--max-wall", type=float, default=None, + help="mur du harnais en s (defaut: timeout organe " + "+ 120, sinon 1200)") + parser.add_argument("--json", action="store_true", + help="verdict complet en JSON sur stdout") + args = parser.parse_args(argv) + + load_cmd = list(args.cmd) + if load_cmd and load_cmd[0] == "--": + load_cmd = load_cmd[1:] + if not load_cmd: + parser.error("commande de charge vide") + + lake_dir = Path(args.lake_dir).resolve() + if not ((lake_dir / "lakefile.lean").exists() + or (lake_dir / "lakefile.toml").exists()): + parser.error(f"pas de lakefile.lean/.toml sous {lake_dir}") + + max_wall = args.max_wall + if max_wall is None: + max_wall = (args.timeout + 120.0) if args.timeout else 1200.0 + + touched: list[str] = [] + if args.touch > 0: + touched = touch_own_modules(lake_dir, args.touch) + + cap = args.cap if args.cap is not None else lean_exec.config()["cap"] + + baseline = [ + take_sample("baseline", args.drivefs_root, args.probe_timeout) + for _ in range(max(1, args.baselines)) + ] + + organ_cli = [ + sys.executable, + str(Path(__file__).resolve().parent / "lean_exec.py"), + "run", + ] + if args.cap is not None: + organ_cli += ["--cap", str(args.cap)] + if args.budget is not None: + organ_cli += ["--budget", str(args.budget)] + if args.timeout is not None: + organ_cli += ["--timeout", str(args.timeout)] + organ_cli += ["--", *load_cmd] + + started = time.monotonic() + # Sortie de l'organe vers FICHIER, jamais vers un pipe non lu : un build + # lake emet bien plus que le buffer de pipe, et personne ne draine tant + # que le harnais echantillonne -- un PIPE ici est un deadlock garanti. + log_fd, log_path = tempfile.mkstemp( + prefix="t5a_organ_log_", suffix=".txt" + ) + log_file = os.fdopen(log_fd, "w", encoding="utf-8") + proc = subprocess.Popen( + organ_cli, + cwd=str(lake_dir), + stdout=log_file, + stderr=subprocess.STDOUT, + creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0), + ) + sampler = Sampler(args.drivefs_root, args.probe_timeout) + load, wall_exceeded = sampler.collect(proc, args.interval, max_wall) + if wall_exceeded: + proc.kill() + proc.wait() + log_file.close() + with open(log_path, encoding="utf-8", errors="replace") as fh: + output_tail = fh.read().splitlines()[-15:] + + record = read_organ_record(load_cmd) + verdict = evaluate(baseline, load, record, cap, + args.spawn_ratio, args.spawn_floor_ms) + if wall_exceeded: + verdict["checks"].append({ + "criterion": "harness_max_wall", + "pass": False, + "detail": f"mur du harnais {max_wall} s depasse -- charge tuee", + }) + verdict["pass"] = False + verdict["wall_s"] = round(time.monotonic() - started, 1) + verdict["cap"] = cap + verdict["lake_dir"] = str(lake_dir) + verdict["touched_modules"] = touched + verdict["organ_exit_code"] = proc.returncode + verdict["organ_output_tail"] = output_tail + verdict["organ_log"] = log_path + verdict["organ_record"] = record + + if args.json: + print(json.dumps(verdict, indent=2)) + else: + rep = verdict["report"] + print(f"[T5a] charge : {' '.join(load_cmd)} dans {lake_dir} " + f"({len(touched)} modules touches, wall {verdict['wall_s']} s)") + print(f"[T5a] organ : exit={proc.returncode} " + f"record={'lu' if record else 'ABSENT/stale'}") + for check in verdict["checks"]: + mark = "PASS" if check["pass"] else "FAIL" + print(f"[T5a] {mark} {check['criterion']}: {check['detail']}") + print(f"[T5a] drivefs p95 {rep['drivefs_p95_ms']} ms ; " + f"spawn p95 {rep['spawn_p95_charge_ms']} ms " + f"(baseline {rep['spawn_p95_baseline_ms']}) ; " + f"ram libre min {rep['ram_libre_min_mb']} MB") + print(f"[T5a] VERDICT : {'PASS' if verdict['pass'] else 'FAIL'} " + f"({sum(1 for c in verdict['checks'] if c['pass'])}/" + f"{len(verdict['checks'])} criteres durs)") + + return 0 if verdict["pass"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/tests/test_validate_bounded_load.py b/scripts/tests/test_validate_bounded_load.py new file mode 100644 index 0000000000..dd7cfc7da0 --- /dev/null +++ b/scripts/tests/test_validate_bounded_load.py @@ -0,0 +1,262 @@ +"""Tests du harnais de validation charge bornee (T5a, #15666). + +Proprietes sous test, dans l'ordre de ce qu'elles protegent : + +1. **Un verdict propre cite ses mesures** -- chaque critere dur PASS/FAIL + porte le chiffre qui le fonde (population max vs cap, p95 vs borne), + jamais un aveu nu. +2. **Le regime de l'incident echoue** -- population au-dessus du cap, sonde + DriveFS en timeout, spawn etouffe, organe refuse ou orphelins : chacun + de ces visages de l'asphyxie de septembre doit casser un critere NOMME. +3. **Pas de faux pass** -- sonde DriveFS indisponible n'est pas un succes + de sonde ; zero echantillon pendant la charge n'est pas une population + conforme ; un enregistrement d'organe absent ne valide rien. +4. **Le harnais ne se bloque pas** -- la sonde DriveFS abandonnee au bout + de son timeout rend la main ; l'echantillonneur s'arrete a la mort de + la charge et respecte son propre mur. + +Aucun lake ni lean requis : le verdict est une fonction pure sur des +echantillons fabriques, l'echantillonneur se teste sur un enfant python. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +import textwrap +import time +from pathlib import Path + +LEAN_DIR = Path(__file__).resolve().parents[1] / "lean" +sys.path.insert(0, str(LEAN_DIR)) + +import validate_bounded_load as vbl # noqa: E402 + + +def _sample(phase="load", population=0, spawn_ms=100.0, + drivefs="ok", drivefs_ms=40.0, ram=8000.0) -> dict: + return { + "t": 0.0, "phase": phase, "population": population, + "population_src": "test", "spawn_ms": spawn_ms, + "drivefs": drivefs, "drivefs_ms": drivefs_ms, + "ram_free_mb": ram, + } + + +def _organ_ok() -> dict: + return {"status": "ok", "orphans": [], "exit_code": 0, + "cmd": ["lake", "build"], "duration_s": 60.0} + + +def _evaluate(baseline, load, organ=_organ_ok(), cap=8): + return vbl.evaluate(baseline, load, organ, cap, + spawn_ratio=5.0, spawn_floor_ms=5000.0) + + +def _check(verdict, name): + return next(c for c in verdict["checks"] if c["criterion"] == name) + + +# --- 1. verdict propre ------------------------------------------------------ + +def test_verdict_propre_quatre_criteres_cites(): + baseline = [_sample("baseline") for _ in range(3)] + load = [_sample(population=5) for _ in range(20)] + verdict = _evaluate(baseline, load) + assert verdict["pass"] is True + assert len(verdict["checks"]) == 4 + assert all(c["pass"] for c in verdict["checks"]) + pop = _check(verdict, "population_cap") + assert "5 <= cap 8" in pop["detail"] and "20 echantillons" in pop["detail"] + spawn = _check(verdict, "spawn_p95") + assert "5000.0 ms" in spawn["detail"] # plancher, baseline basse + + +def test_p95_connu(): + assert vbl._p95([1, 2, 3, 4, 5, 6, 7, 8, 9, 100]) == 100 + assert vbl._p95([5]) == 5 + assert vbl._p95([]) is None + + +# --- 2. le regime de l'incident echoue -------------------------------------- + +def test_population_hors_cap_echoue_et_cite_les_nombres(): + load = [_sample(population=5), _sample(population=12), _sample()] + verdict = _evaluate([_sample("baseline")], load, cap=8) + assert verdict["pass"] is False + pop = _check(verdict, "population_cap") + assert pop["pass"] is False + assert "12" in pop["detail"] and "cap 8" in pop["detail"] + + +def test_sonde_drivefs_en_timeout_echoue(): + load = [_sample(drivefs="ok"), _sample(drivefs="timeout", drivefs_ms=None), + _sample()] + verdict = _evaluate([_sample("baseline")], load) + drivefs = _check(verdict, "drivefs_alive") + assert drivefs["pass"] is False + assert "1 sonde" in drivefs["detail"] + + +def test_spawn_etouffe_au_dela_de_la_borne_echoue(): + baseline = [_sample("baseline", spawn_ms=100.0) for _ in range(4)] + load = [_sample(spawn_ms=40000.0) for _ in range(10)] + verdict = _evaluate(baseline, load) + spawn = _check(verdict, "spawn_p95") + assert spawn["pass"] is False + assert "40000.0 ms" in spawn["detail"] + + +def test_spawn_eleve_mais_sous_plancher_passe(): + # Baseline lente (2000 ms) : borne = max(5 x 2000, 5000) = 10000. + # La charge a 8000 ms est degradée mais SOUS la borne -> PASS : la borne + # est relative a la machine, pas un absolu invente. + baseline = [_sample("baseline", spawn_ms=2000.0) for _ in range(4)] + load = [_sample(spawn_ms=8000.0) for _ in range(10)] + verdict = _evaluate(baseline, load) + assert _check(verdict, "spawn_p95")["pass"] is True + + +def test_organ_refuse_ou_orphelins_echoue(): + refused = _evaluate([_sample("baseline")], [_sample()], + organ={"status": "refused", "orphans": []}) + assert _check(refused, "organ_postcondition")["pass"] is False + assert "refused" in _check(refused, "organ_postcondition")["detail"] + + orphans = _evaluate([_sample("baseline")], [_sample()], + organ={"status": "ok", "orphans": [4242]}) + detail = _check(orphans, "organ_postcondition")["detail"] + assert _check(orphans, "organ_postcondition")["pass"] is False + assert "1" in detail # un orphelin, compte + + +def test_enregistrement_organ_absent_ne_valide_rien(): + verdict = _evaluate([_sample("baseline")], [_sample()], organ=None) + assert _check(verdict, "organ_postcondition")["pass"] is False + + +# --- 3. pas de faux pass ---------------------------------------------------- + +def test_sonde_indisponible_nest_pas_un_succes_mais_pas_un_echec(): + load = [_sample(drivefs="unavailable", drivefs_ms=None) for _ in range(5)] + verdict = _evaluate([_sample("baseline", drivefs="unavailable", + drivefs_ms=None)], load) + drivefs = _check(verdict, "drivefs_alive") + assert drivefs["pass"] is True + assert "5 sonde(s) unavailable" in drivefs["detail"] # note honnete + + +def test_zero_echantillon_de_charge_est_non_verifiable(): + verdict = _evaluate([_sample("baseline")], []) + pop = _check(verdict, "population_cap") + assert pop["pass"] is False + assert "non verifiable" in pop["detail"] + + +# --- 4. le harnais ne se bloque pas ----------------------------------------- + +def test_sonde_drivefs_abandonnee_rend_la_main(monkeypatch, tmp_path): + def _hanging_listdir(_root): + time.sleep(5.0) + + monkeypatch.setattr(os, "listdir", _hanging_listdir) + t0 = time.monotonic() + state, ms = vbl.probe_drivefs(str(tmp_path), timeout_s=0.3) + elapsed = time.monotonic() - t0 + assert state == vbl.DRIVEFS_TIMEOUT + assert ms is None + assert elapsed < 3.0 # rendu la main au timeout, pas au bout des 5 s + + +def test_sonde_drivefs_hors_racine_indisponible(): + state, ms = vbl.probe_drivefs(None, timeout_s=1.0) + assert state == vbl.DRIVEFS_UNAVAILABLE + assert ms is None + + +def test_sampler_echantillonne_pendant_la_vie_arrete_a_la_mort(): + calls = [] + + def fake_sample(phase): + calls.append(phase) + return _sample(phase) + + sampler = vbl.Sampler(None, 1.0, sample_fn=fake_sample) + proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(1.2)"], + stdout=subprocess.DEVNULL, + ) + try: + samples, wall_exceeded = sampler.collect(proc, interval=0.15, + max_wall=30.0) + finally: + if proc.poll() is None: + proc.kill() + assert wall_exceeded is False + assert len(samples) >= 3 + assert proc.poll() == 0 # collect a rendu la main APRES la mort + + +def test_sampler_renvoie_le_mur_depasse(): + sampler = vbl.Sampler(None, 1.0, sample_fn=_sample) + proc = subprocess.Popen( + [sys.executable, "-c", "import time; time.sleep(5)"], + stdout=subprocess.DEVNULL, + ) + try: + _, wall_exceeded = sampler.collect(proc, interval=0.1, max_wall=0.01) + assert wall_exceeded is True + finally: + proc.kill() + proc.wait() + + +# --- charge : touch de modules ---------------------------------------------- + +def test_touch_own_modules_exclut_lake_et_bump_seulement_les_n(tmp_path): + lake = tmp_path / "mylake" + (lake / "Beta").mkdir(parents=True) + (lake / ".lake" / "packages" / "x").mkdir(parents=True) + files = { + "lakefile.lean": lake / "lakefile.lean", + "Alpha.lean": lake / "Alpha.lean", + "Beta/B.lean": lake / "Beta" / "B.lean", + "vendored": lake / ".lake" / "packages" / "x" / "X.lean", + } + old = 1000000000 + for path in files.values(): + path.write_text("-- stub\n", encoding="utf-8") + os.utime(path, (old, old)) + + touched = vbl.touch_own_modules(lake, 2) + + assert len(touched) == 2 + assert "vendored" not in str(touched) + assert "lakefile.lean" not in touched + for key in ("Alpha.lean", "Beta/B.lean"): + assert key in touched + assert files[key].stat().st_mtime > old + assert files["vendored"].stat().st_mtime == old # .lake intouché + + +def test_cli_exige_un_lakefile(tmp_path): + import pytest + with pytest.raises(SystemExit): + vbl.main(["--lake-dir", str(tmp_path)]) + + +def test_read_organ_record_rejette_le_stale(tmp_path, monkeypatch): + import json + state = tmp_path / "state" + monkeypatch.setattr( + vbl.lean_exec, "state_dir", lambda: state, raising=True + ) + state.mkdir() + (state / "last_run.json").write_text( + json.dumps({"cmd": ["lake", "build"], "status": "ok"}), encoding="utf-8" + ) + # cmd differente : enregistrement d'un AUTRE run -> None, pas de + # validation sur du stale. + assert vbl.read_organ_record(["lake", "exe", "cache", "get"]) is None + assert vbl.read_organ_record(["lake", "build"]) is not None From a3bc4f9620a045a9e6c9b6c3c54f81946d84b370 Mon Sep 17 00:00:00 2001 From: jsboige Date: Wed, 16 Sep 2026 20:53:09 +0200 Subject: [PATCH 2/3] Fix(lean,#15666): T5a harness fails closed on unmeasurable population and stale organ record Two fail-open paths found in exact-head review of #16299: 1. scan_native_population() returns -1 when the scan fails; evaluate() kept those samples, so an all-invalid run produced max_pop = -1 <= cap and marked population_cap PASS with zero valid measurements. Invalid samples are now filtered and the criterion FAILs when none remain, citing valid/total counts. 2. read_organ_record() reconciled only cmd: an organ crash before emit leaves the previous run's last_run.json (same lake build command) accepted as evidence for the current run. The harness now captures known run_ids (last_run.json + live run registry) before launching and rejects any record whose run_id is missing or preexisting. Co-Authored-By: Claude Sonnet 5 --- scripts/lean/validate_bounded_load.py | 74 ++++++++++++++++++--- scripts/tests/test_validate_bounded_load.py | 74 +++++++++++++++++++++ 2 files changed, 137 insertions(+), 11 deletions(-) diff --git a/scripts/lean/validate_bounded_load.py b/scripts/lean/validate_bounded_load.py index 02bb7184b3..8b12261916 100644 --- a/scripts/lean/validate_bounded_load.py +++ b/scripts/lean/validate_bounded_load.py @@ -29,7 +29,10 @@ Post-conditions lues dans l'enregistrement de l'organe (``last_run.json``, ecrit a CHAQUE sortie de ``run``, y compris refuse) : status ``ok``, -orphelins vides, sortie 0. +orphelins vides, sortie 0. L'enregistrement est lie a CETTE invocation : +son ``run_id`` doit etre frais (inconnu avant le lancement du harnais) -- +un crash de l'organe avant emission laisse le ``last_run.json`` du run +precedent, qui porte un run_id deja connu et est rejete. Generation de la charge : le harnais ne fabrique pas de charge fictive. Il lance ``lean_exec.py run -- lake build`` dans le lake fourni ; pour qu'un @@ -218,15 +221,26 @@ def hard(name: str, ok: bool, detail: str) -> None: f"organ status={status!r} orphelins={len(orphans)}", ) - pops = [s["population"] for s in load] + pops_all = [s["population"] for s in load] + # -1 = scan_native_population en echec : une mesure INVALIDE ne doit + # jamais valider le critere (un run tout -1 produirait max=-1 <= cap). + pops = [p for p in pops_all if p is not None and p >= 0] max_pop = max(pops) if pops else None + if pops: + pop_detail = (f"max population observee {max_pop} <= cap {cap} " + f"({len(pops)} mesures valides / {len(pops_all)} " + f"echantillons)") + elif pops_all: + pop_detail = (f"0 mesure de population valide sur {len(pops_all)} " + f"echantillons (scan en echec) -- critere non " + f"verifiable") + else: + pop_detail = ("0 echantillon pendant la charge -- critere non " + "verifiable") hard( "population_cap", bool(pops) and max_pop is not None and max_pop <= cap, - (f"max population observee {max_pop} <= cap {cap} " - f"({len(pops)} echantillons)") - if pops else - f"0 echantillon pendant la charge -- critere non verifiable", + pop_detail, ) timeouts = [s for s in load if s["drivefs"] == DRIVEFS_TIMEOUT] @@ -294,12 +308,41 @@ def touch_own_modules(lake_dir: Path, n: int) -> list[str]: return touched -def read_organ_record(expected_cmd: list[str]) -> dict | None: +def known_run_ids() -> set[str]: + """Run_ids deja presents AVANT le lancement de la charge. + + Sources : le ``last_run.json`` preexistant (run le plus recent, son + registre est nettoye en fin de run propre) et le registre des runs + vivants (``runs/*.json``). L'organe genere un run_id frais (uuid) par + invocation : apres le run, un ``last_run.json`` qui porte un de ces + run_ids n'a pas ete reecrit par CE run. + """ + ids: set[str] = set() + state = lean_exec.state_dir() + try: + prev = json.loads((state / "last_run.json").read_text( + encoding="utf-8")) + if isinstance(prev, dict) and prev.get("run_id"): + ids.add(prev["run_id"]) + except (OSError, ValueError): + pass + try: + ids.update(p.stem for p in (state / "runs").glob("*.json")) + except OSError: + pass + return ids + + +def read_organ_record(expected_cmd: list[str], + preexisting_run_ids: set[str] | None = None) -> dict | None: """Lit ``last_run.json`` ; None si absent ou manifestement stale. - Un enregistrement d'un run precedent (cmd differente) n'est pas une - preuve pour CE run : le harnais le rejete plutot que de valider sur du - stale. + Un enregistrement d'un run precedent n'est pas une preuve pour CE run. + Deux defenses cumulees : la cmd doit etre celle attendue, et (quand le + harnais fournit les run_ids connus avant lancement) le ``run_id`` doit + etre frais -- un organe crashe avant emission laisse le fichier du run + precedent, qui porte un run_id deja connu ; un fichier sans run_id + (format anterieur) est traite en stale potentiel, pas en preuve. """ path = lean_exec.state_dir() / "last_run.json" try: @@ -308,6 +351,10 @@ def read_organ_record(expected_cmd: list[str]) -> dict | None: return None if record.get("cmd") != expected_cmd: return None + if preexisting_run_ids is not None: + run_id = record.get("run_id") + if not run_id or run_id in preexisting_run_ids: + return None return record @@ -386,6 +433,11 @@ def main(argv: list[str] | None = None) -> int: organ_cli += ["--timeout", str(args.timeout)] organ_cli += ["--", *load_cmd] + # Run_ids connus AVANT lancement : le record final doit en porter un + # frais -- sinon c'est le fichier d'un run precedent (crash avant + # emission), pas une preuve pour CE run. + pre_run_ids = known_run_ids() + started = time.monotonic() # Sortie de l'organe vers FICHIER, jamais vers un pipe non lu : un build # lake emet bien plus que le buffer de pipe, et personne ne draine tant @@ -410,7 +462,7 @@ def main(argv: list[str] | None = None) -> int: with open(log_path, encoding="utf-8", errors="replace") as fh: output_tail = fh.read().splitlines()[-15:] - record = read_organ_record(load_cmd) + record = read_organ_record(load_cmd, pre_run_ids) verdict = evaluate(baseline, load, record, cap, args.spawn_ratio, args.spawn_floor_ms) if wall_exceeded: diff --git a/scripts/tests/test_validate_bounded_load.py b/scripts/tests/test_validate_bounded_load.py index dd7cfc7da0..9e417b3e56 100644 --- a/scripts/tests/test_validate_bounded_load.py +++ b/scripts/tests/test_validate_bounded_load.py @@ -260,3 +260,77 @@ def test_read_organ_record_rejette_le_stale(tmp_path, monkeypatch): # validation sur du stale. assert vbl.read_organ_record(["lake", "exe", "cache", "get"]) is None assert vbl.read_organ_record(["lake", "build"]) is not None + + +def test_population_non_mesurable_ne_valide_rien(): + # scan_native_population en echec retourne -1 : un run tout -1 produirait + # max=-1 <= cap -- FAUX PASS. Le critere doit echouer avec 0 mesure + # valide, meme s'il y a des echantillons. + load = [_sample(population=-1) for _ in range(5)] + verdict = _evaluate([_sample("baseline")], load, organ=None, cap=8) + pop = _check(verdict, "population_cap") + assert pop["pass"] is False + assert "0 mesure de population valide" in pop["detail"] + assert "5 echantillons" in pop["detail"] + assert verdict["pass"] is False + + +def test_population_mixte_juge_sur_les_mesures_valides(): + # Quelques scans en echec parmi des mesures valides : le verdict porte + # sur les valides et cite les deux comptes. + load = ([_sample(population=-1), _sample(population=6)] + + [_sample(population=3) for _ in range(6)]) + verdict = _evaluate([_sample("baseline")], load) + pop = _check(verdict, "population_cap") + assert pop["pass"] is True + assert "6 <= cap 8" in pop["detail"] + assert "6 mesures valides / 8 echantillons" in pop["detail"] + + +def test_read_organ_record_rejette_le_stale_meme_commande(tmp_path, monkeypatch): + # Crash de l'organe AVANT emission : last_run.json porte encore le run + # PRECEDENT -- meme cmd, donc la reconciliation par cmd seule passe. Le + # run_id deja connu avant lancement doit le demasquer. + import json + state = tmp_path / "state" + monkeypatch.setattr( + vbl.lean_exec, "state_dir", lambda: state, raising=True + ) + state.mkdir() + + def _write(record): + (state / "last_run.json").write_text( + json.dumps(record), encoding="utf-8" + ) + + _write({"run_id": "runprecedent", "cmd": ["lake", "build"], + "status": "ok", "orphans": []}) + pre_ids = vbl.known_run_ids() + assert pre_ids == {"runprecedent"} + # Le fichier n'a pas ete reecrit (crash) : run_id preexistant -> stale. + assert vbl.read_organ_record(["lake", "build"], pre_ids) is None + # Format sans run_id : stale potentiel, jamais une preuve. + _write({"cmd": ["lake", "build"], "status": "ok", "orphans": []}) + assert vbl.read_organ_record(["lake", "build"], pre_ids) is None + # Run_id frais : CE run a atteint l'emission -> accepte. + _write({"run_id": "runfrais", "cmd": ["lake", "build"], + "status": "ok", "orphans": []}) + assert vbl.read_organ_record(["lake", "build"], pre_ids) is not None + + +def test_known_run_ids_lit_aussi_le_registre_des_runs(tmp_path, monkeypatch): + # Les runs vivants (registre runs/*.json) comptent aussi comme connus : + # un run concurrent ne doit pas pouvoir etre valide comme preuve. + import json + state = tmp_path / "state" + monkeypatch.setattr( + vbl.lean_exec, "state_dir", lambda: state, raising=True + ) + state.mkdir() + (state / "runs").mkdir() + (state / "runs" / "runvivant.json").write_text("{}", encoding="utf-8") + (state / "last_run.json").write_text( + json.dumps({"run_id": "runprecedent", "cmd": ["lake", "build"]}), + encoding="utf-8", + ) + assert vbl.known_run_ids() == {"runprecedent", "runvivant"} From c6a3d4d13ab2a01f4a3bb95d72c488402377dd4c Mon Sep 17 00:00:00 2001 From: jsboige Date: Wed, 16 Sep 2026 21:04:27 +0200 Subject: [PATCH 3/3] Fix(lean,#15666): correct valid-count assertion in mixed-population contract 7 valid measurements (1x6 + 6x3) among 8 samples, not 6. Co-Authored-By: Claude Sonnet 5 --- scripts/tests/test_validate_bounded_load.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/tests/test_validate_bounded_load.py b/scripts/tests/test_validate_bounded_load.py index 9e417b3e56..0a718eaff4 100644 --- a/scripts/tests/test_validate_bounded_load.py +++ b/scripts/tests/test_validate_bounded_load.py @@ -284,7 +284,7 @@ def test_population_mixte_juge_sur_les_mesures_valides(): pop = _check(verdict, "population_cap") assert pop["pass"] is True assert "6 <= cap 8" in pop["detail"] - assert "6 mesures valides / 8 echantillons" in pop["detail"] + assert "7 mesures valides / 8 echantillons" in pop["detail"] def test_read_organ_record_rejette_le_stale_meme_commande(tmp_path, monkeypatch):