import subprocess
import textwrap
import re
import os
import shutil
import tempfile
from pathlib import Path
# --- Path resolution: find the grothendieck_lean Lake project ---
# Must work both interactively (CWD = repo root) and under Papermill (CWD may differ).
def find_grothendieck_lean_project():
"""Find the grothendieck_lean Lake project directory.
Searches from multiple starting points to handle both interactive use
and Papermill exécution (where CWD may differ from notebook location).
Returns an ABSOLUTE path.
"""
starts = [Path.cwd().resolve()]
nb_file = os.environ.get('NB_FILE') or globals().get('__vsc_ipynb_file__')
if nb_file:
starts.append(Path(nb_file).resolve().parent)
for start in starts:
current = start
for _ in range(10):
candidate = current / 'grothendieck_lean'
if candidate.exists() and (candidate / 'lakefile.lean').exists():
return candidate.resolve()
current = current.parent
if current == current.parent:
break
raise FileNotFoundError("grothendieck_lean/ not found -- check working directory")
def win_to_wsl(win_path: Path) -> str:
"""Convert Windows path to WSL path using drive letter."""
p = win_path.resolve()
drive_letter = p.drive
if not drive_letter or len(drive_letter) < 2:
s = str(p)
if s.startswith('/mnt/'):
return s
return s
drive = drive_letter[0].lower()
return f'/mnt/{drive}{p.as_posix()[2:]}'
WIN_LEAN_PROJECT = find_grothendieck_lean_project()
LEAN_PROJECT = win_to_wsl(WIN_LEAN_PROJECT)
# Chemins portables dans les sorties : le prefixe absolu varie par machine
# (D:\... ou /mnt/d/...) et n'a pas sa place dans un output commite.
REPO_RELATIVE_PROJECT = '<repo>/MyIA.AI.Notebooks/SymbolicAI/Lean/grothendieck_lean'
def sanitize_lean_paths(text):
variants = {str(WIN_LEAN_PROJECT), str(WIN_LEAN_PROJECT).replace('\\', '/'), LEAN_PROJECT}
for v in sorted(variants, key=len, reverse=True):
if v:
text = text.replace(v, REPO_RELATIVE_PROJECT)
return text
USE_NATIVE_LEAN = shutil.which('lake') is not None and os.name != 'nt'
def wsl(cmd, timeout=60):
"""Run a bash command inside WSL Ubuntu when available.
Captures stdout/stderr via temp files rather than capture_output=True, to
avoid the CPython ``_readerthread`` race on Windows that silently dropped
subprocess output (the committed cells 25-27 previously showed only an
``Exception in thread (_readerthread)`` trace instead of Lean output).
"""
# pipefail (#17616) : sans lui, le rc d'un pipeline bash est celui de la
# DERNIERE commande (tail) -- un `lake build ... 2>&1 | tail -20` echoue
# alors avec rc=0 et l'appelant affirmerait SUCCESS sur une build ratee.
full = ['wsl', '-d', 'Ubuntu', '--', 'bash', '-lc', 'set -o pipefail; ' + cmd]
out_f = tempfile.NamedTemporaryFile('wb', delete=False, suffix='.out')
err_f = tempfile.NamedTemporaryFile('wb', delete=False, suffix='.err')
out_path, err_path = out_f.name, err_f.name
out_f.close()
err_f.close()
try:
with open(out_path, 'wb') as o, open(err_path, 'wb') as e:
r = subprocess.run(full, stdout=o, stderr=e, timeout=timeout)
out = Path(out_path).read_text(encoding='utf-8', errors='replace')
err = Path(err_path).read_text(encoding='utf-8', errors='replace')
return r.returncode, out, err
except FileNotFoundError:
return 127, '', 'WSL executable not found'
except subprocess.TimeoutExpired:
return -1, '', f'TIMEOUT after {timeout}s'
finally:
for p in (out_path, err_path):
try:
Path(p).unlink()
except OSError:
pass
# --- Lean file reading ---
def read_lean_module(module_name):
"""Read a .lean source file from the grothendieck_lean project.
module_name: e.g. 'CategoryAndSites' -> reads Grothendieck/CategoryAndSites.lean
Returns the file content as a string.
"""
path = WIN_LEAN_PROJECT / 'Grothendieck' / f'{module_name}.lean'
if not path.exists():
return f'[FICHIER INTROUVABLE] {REPO_RELATIVE_PROJECT}/Grothendieck/{module_name}.lean'
return path.read_text(encoding='utf-8')
def display_lean_module(module_name, max_lines=None, highlight=None):
"""Display a .lean source file with line numbers.
max_lines: if set, only show the first N lines
highlight: list of line numbers to mark with '>>>' (1-indexed)
"""
content = read_lean_module(module_name)
if content.startswith('[FICHIER INTROUVABLE]'):
print(content)
return
lines = content.splitlines()
if max_lines:
lines = lines[:max_lines]
highlight = set(highlight or [])
print(f'--- Grothendieck/{module_name}.lean ---')
for i, line in enumerate(lines, 1):
marker = ' >>>' if i in highlight else ' '
print(f'{marker} {i:>3d} | {line}')
total = len(content.splitlines())
if max_lines and total > max_lines:
print(f' ... ({total - max_lines} lignes restantes sur {total} total)')
print(f'--- fin ({total} lignes) ---')
# --- Lake build ---
def run_lake_build(targets='Grothendieck', timeout=1500):
"""Run lake build against the grothendieck_lean project."""
if USE_NATIVE_LEAN:
try:
r = subprocess.run(
['lake', 'build', targets],
cwd=WIN_LEAN_PROJECT,
capture_output=True,
text=True,
timeout=timeout,
)
return r.returncode, sanitize_lean_paths(r.stdout or ''), sanitize_lean_paths(r.stderr or '')
except subprocess.TimeoutExpired:
return -1, '', f'TIMEOUT after {timeout}s'
rc, out, err = wsl(
f'source ~/.elan/env 2>/dev/null; cd {LEAN_PROJECT} && lake build {targets} 2>&1 | tail -20',
timeout=timeout,
)
return rc, sanitize_lean_paths(out or ''), sanitize_lean_paths(err or '')
# --- Lean snippet exécution ---
def run_lean(snippet, timeout_s=300):
"""Run a Lean snippet against the grothendieck_lean project using lake env lean.
The snippet is written to a temp file and executed with the project's Lake env.
Returns combined stdout+stderr.
"""
snippet = textwrap.dedent(snippet).strip() + '\n'
if USE_NATIVE_LEAN:
with tempfile.NamedTemporaryFile('w', suffix='.lean', delete=False, encoding='utf-8') as tmp:
tmp.write(snippet)
tmp_path = tmp.name
try:
r = subprocess.run(
['lake', 'env', 'lean', tmp_path],
cwd=WIN_LEAN_PROJECT,
capture_output=True,
text=True,
timeout=timeout_s,
)
return sanitize_lean_paths((r.stdout or '') + (r.stderr or ''))
except subprocess.TimeoutExpired:
return f'TIMEOUT after {timeout_s}s'
finally:
try:
Path(tmp_path).unlink()
except OSError:
pass
write_cmd = f"cat > /tmp/lean13_snippet.lean << 'LEAN_EOF'\n{snippet}LEAN_EOF"
lean_cmd = f'cd {LEAN_PROJECT} && lake env lean /tmp/lean13_snippet.lean 2>&1'
full_cmd = f'{write_cmd}\n{lean_cmd}'
rc, out, err = wsl(full_cmd, timeout=timeout_s)
if rc == -1:
return f'TIMEOUT after {timeout_s}s'
return sanitize_lean_paths((out or '') + (err or ''))
# --- Module inventory ---
GROTHENDIECK_MODULES = {
'CategoryAndSites': 'Part 1: Sieves, topologies, 3 axioms',
'SchemesTour': 'Part 2: Scheme, Spec, Gamma',
'ZariskiSite': 'Part 3: Zariski pretopology, bridge theorem',
'MathlibMap': 'Part 4: #check index Mathlib',
'Calibration': 'Part 5: 4 micro-preuves P1-P4',
'SieveLattice': 'Part 6: Pullback identities',
'SheafBasics': 'Part 7: Sheaves, sheaf condition',
'SieveOps': 'Part 8: Sieve lattice opérations',
'CoverageGen': 'Part 9: Coverage generators',
'CanonicalProps': 'Part 10: Canonical topology properties',
'SieveGenerate': 'Part 11: Sieve generation',
'DenseTopology': 'Part 12: Dense topology',
'Sheafification': 'Part 13: Sheafification functor (Mathlib bridge)',
'LeftExact': 'Part 14: Left exactness of sheafification',
'SitePoints': 'Part 15: Points of a site',
'Subcanonical': 'Part 16: Subcanonical topologies, Yoneda sheaves',
'SheafHom': 'Part 17: Sheaf hom, internal hom',
'ConstantSheaf': 'Part 18: Constant sheaf, adjunction',
'Conservative': 'Part 19: Conservative families of points',
'SheafCohomology/Basic': 'Part 20: Ext-based sheaf cohomology H^n',
'MayerVietorisSquare': 'Part 21: Mayer-Vietoris squares',
'SheafCohomology/MayerVietoris': 'Part 22: Mayer-Vietoris long exact sequence',
'SheafCohomology/Cech': 'Part 23: Cech cohomology complex',
}
# Verify project is accessible
assert (WIN_LEAN_PROJECT / 'lakefile.lean').exists(), 'grothendieck_lean/lakefile.lean not found'
mode = 'native lake env lean' if USE_NATIVE_LEAN else 'WSL lake env lean'
print(f'Setup OK : grothendieck_lean project trouve a {REPO_RELATIVE_PROJECT}')
print(f' Exécution Lean : {mode}')
print(f' {len(GROTHENDIECK_MODULES)} modules Grothendieck detectes')