#!/usr/bin/env bash
# Experiment A847-LEAN-CAPACITY-32549b50e95c / Branch A.
# This file embeds the complete standard-library runner. No global installation.
# --check-evidence validates delivered bindings and does not rerun any proof.
# --run needs the pinned checkout, authenticated official toolchain archive,
# and an explicit budget representing the remaining normal assigned allowance.
set -euo pipefail
python3 -I - "$@" <<'IMD_LEAN_CAPACITY_PY'
SOURCES = {'runner.py': 'import os,sys,re,json,time,hashlib,subprocess,pathlib,signal,concurrent.futures,tarfile,shutil\nP=pathlib.Path\nREPO=P(os.environ.get(\'CAPACITY_REPO\',P(__file__).resolve().parents[3])).resolve()\nW=P(os.environ.get(\'CAPACITY_WORK\',REPO/\'test/scratch/lean-capacity-a\')).resolve(); E=W/\'evidence\'; E.mkdir(exist_ok=True)\nB=W/\'build\'; B.mkdir(exist_ok=True)\nTC=W/\'lean-4.35.0-rc2-linux\'; LEAN=TC/\'bin/lean\'\nCOMMIT=\'048bb245b60abce79f1a2d0c82336ff8508ec752\'; TREE=\'c6111a7d32daf2906275ee1ac265bf6865ebd9a5\'\nFLAGS=[\'-j1\',\'-DElab.async=false\']; ALLOWED=[\'propext\',\'Quot.sound\',\'Classical.choice\']\n# The bootstrap derives this deadline from the remaining assigned allowance and reserve.\n# Direct helper invocations have no implicit proof-time allowance.\nDEADLINE=float(os.environ.get(\'CAPACITY_DEADLINE\',\'0\'))\nENV=dict(os.environ,LEAN_PATH=str(B)); commands=[]\ndef sha(b):return hashlib.sha256(b).hexdigest()\ndef dump(p,x):\n p.parent.mkdir(parents=True,exist_ok=True);tmp=p.with_name(p.name+\'.tmp\');tmp.write_text(json.dumps(x,indent=2)+\'\\n\');os.replace(tmp,p)\ndef run(cmd,log,timeout=None,env=None):\n t=time.monotonic(); log=E/log; log.parent.mkdir(parents=True,exist_ok=True)\n remaining=DEADLINE-time.time(); limit=min(timeout or remaining,remaining)\n if limit<=0: return {\'status\':\'NOT_RUN\',\'reason\':\'safe budget exhausted\'}\n with log.open(\'wb\') as f:\n  interrupted=False;loader_error=None\n  try:p=subprocess.Popen([str(x) for x in cmd],stdout=f,stderr=subprocess.STDOUT,env=env or ENV,start_new_session=True,cwd=REPO)\n  except OSError as exc:loader_error=str(exc);f.write((loader_error+\'\\n\').encode());rc=None\n  else:\n   try:rc=p.wait(timeout=limit)\n   except subprocess.TimeoutExpired:\n    interrupted=True; os.killpg(p.pid,signal.SIGTERM)\n    try:rc=p.wait(timeout=3)\n    except subprocess.TimeoutExpired:os.killpg(p.pid,signal.SIGKILL);rc=p.wait()\n d={\'command\':[str(x) for x in cmd],\'exit\':rc,\'wall_seconds\':round(time.monotonic()-t,6),\'log\':str(log.relative_to(E)),\'log_sha256\':sha(log.read_bytes()),\'status\':\'BLOCKED\' if loader_error else \'INCONCLUSIVE\' if interrupted or rc in (-9,137,124) else \'PASS\' if rc==0 else \'FAIL\'}\n if loader_error:d[\'reason\']=\'loader failure: \'+loader_error\n commands.append(d); dump(E/\'commands.json\',commands); return d\n\ndef uncomment(s):\n # Nested Lean block comments and strings; preserve positions and newlines.\n out=list(s);i=0;depth=0;string=False\n while i<len(s):\n  if depth:\n   if s.startswith(\'/-\',i):out[i:i+2]=\'  \';depth+=1;i+=2\n   elif s.startswith(\'-/\',i):out[i:i+2]=\'  \';depth-=1;i+=2\n   else:\n    if s[i]!=\'\\n\':out[i]=\' \'\n    i+=1\n  elif string:\n   if s[i]==\'\\\\\':out[i:i+2]=\'  \';i+=2\n   elif s[i]==\'"\':out[i]=\' \';string=False;i+=1\n   else:\n    if s[i]!=\'\\n\':out[i]=\' \'\n    i+=1\n  elif s.startswith(\'/-\',i):out[i:i+2]=\'  \';depth=1;i+=2\n  elif s.startswith(\'--\',i):\n   j=s.find(\'\\n\',i);j=len(s) if j<0 else j;out[i:j]=\' \'*(j-i);i=j\n  elif s[i]==\'"\':out[i]=\' \';string=True;i+=1\n  else:i+=1\n if depth or string:raise ValueError(\'unclosed comment/string\')\n return \'\'.join(out)\n\nfiles={}; declarations={}; deps={}; manifest={}\ndef authenticate():\n assert subprocess.check_output([\'git\',\'rev-parse\',COMMIT+\'^{tree}\'],cwd=REPO,text=True).strip()==TREE\n records=subprocess.check_output([\'git\',\'ls-tree\',\'-r\',\'-z\',COMMIT,\'--\',\'src\',\'lean-toolchain\',\'lakefile.lean\',\'lake-manifest.json\'],cwd=REPO).split(b\'\\0\')\n # Compare actual bytes to Git objects through cat-file batch (not git diff/index).\n entries=[]\n for rec in records:\n  if rec:\n   meta,path=rec.split(b\'\\t\'); mode,kind,oid=meta.split();assert kind==b\'blob\' and mode==b\'100644\'\n   entries.append((path.decode(),oid.decode()))\n proc=subprocess.Popen([\'git\',\'cat-file\',\'--batch\'],cwd=REPO,stdin=subprocess.PIPE,stdout=subprocess.PIPE)\n for path,oid in entries:\n  proc.stdin.write((oid+\'\\n\').encode());proc.stdin.flush();header=proc.stdout.readline().split();n=int(header[2]);blob=proc.stdout.read(n);assert proc.stdout.read(1)==b\'\\n\'\n  assert not (REPO/path).is_symlink(),(path,\'source symlink\');actual=(REPO/path).read_bytes();assert actual==blob,(path,\'differs from Git blob\')\n  manifest[path]={\'git_blob\':oid,\'sha256\':sha(actual),\'bytes\':len(actual)}\n  if path.endswith(\'.lean\') and path.startswith(\'src/\'):\n   m=path[4:-5].replace(\'/\',\'.\');s=actual.decode();files[m]=s; clean=uncomment(s)\n   dd=[]\n   for match in re.finditer(r\'^\\s*(?:(?:public|meta|private)\\s+)*import\\s+(?:all\\s+)?([^\\n]+)\',clean,re.M):\n    for dep in match[1].split():\n     assert re.fullmatch(r\'[A-Za-z_][A-Za-z_0-9.]*\',dep),(m,dep)\n     if dep.split(\'.\')[0] not in (\'Init\',\'Std\',\'Lean\',\'Lake\'):dd.append(dep)\n   deps[m]=dd\n   ns=re.findall(r\'^namespace\\s+([A-Za-z0-9_.]+)\',clean,re.M)\n   # Backlog declarations use a single fully qualified namespace in each file.\n   matches=list(re.finditer(r\'^theorem\\s+([A-Za-z_0-9]+)\\s*:\',clean,re.M))\n   for mat in matches:\n    end=clean.find(\':=\',mat.end());assert end>=0\n    typ=s[mat.end():end].strip(); endbody=re.search(r\'^(?:theorem |#print|end\\b)\',clean[end:],re.M);stop=end+endbody.start() if endbody else len(s)\n    if len(ns)==1:\n     name=ns[0]+\'.\'+mat[1]\n     assert name not in declarations,name\n     declarations[name]={\'name\':name,\'module\':m,\'path\':path,\'line\':clean[:mat.start()].count(\'\\n\')+1,\'type\':typ,\'body\':s[end+2:stop]}\n proc.stdin.close();assert proc.wait()==0\n assert (REPO/\'lean-toolchain\').read_text().strip()==\'leanprover/lean4:v4.35.0-rc2\'\n dump(E/\'source-manifest.json\',{\'commit\':COMMIT,\'tree\':TREE,\'comparison\':\'consumed filesystem bytes equal Git cat-file blobs\',\'files\':manifest})\n\ndef closure(mod):\n visited=set();order=[]\n def visit(m,active):\n  if m in visited:return\n  assert m not in active,(\'import cycle\',m)\n  if m not in deps:raise ValueError(\'missing project dependency \'+m)\n  for d in deps[m]:visit(d,active|{m})\n  visited.add(m);order.append(m)\n visit(mod,set());return order\n\ndef unit(name):\n d=declarations[name];base=name[:-len(\'_checked\')]; roots=[name]\n if \'Chunk\' in name:\n  # Trace actual alias intermediate if present, then collect actual referenced slices.\n  body=d[\'body\'];aliases=re.findall(r\'\\b(root\\w+_projected_slices_pilot_checked)\\b\',body)\n  for alias in aliases:\n   q=name.rsplit(\'.\',1)[0]+\'.\'+alias;assert q in declarations;roots.append(q);body+=\'\\n\'+declarations[q][\'body\']\n  slice_names=re.findall(r\'\\b(root\\w+_(?:projected_)?slice\\d+_checked)\\b\',body)\n  for s in dict.fromkeys(slice_names):\n   q=name.rsplit(\'.\',1)[0]+\'.\'+s;assert q in declarations;roots.append(q)\n  rows=re.findall(r\'\\b(root\\w+_rows_exact)\\b\',body)\n  assert len(set(rows))==1,(name,rows)\n  roots.append(name.rsplit(\'.\',1)[0]+\'.\'+rows[0]);assert roots[-1] in declarations\n  assert slice_names and \'checkNarrowing_flatten\' in body,name\n  # Binding equality lists the same rows, order, scope, and domains as composition.\n  binding=declarations[roots[-1]][\'type\'];rowlist=re.search(r\'checkNarrowing_flatten\\s+(\\w+)\\s+(\\[[^\\]]+\\])\\s+(\\w+)\\s+(\\w+)\',body)\n  assert rowlist,(name,\'composition parse\')\n  scope,rows,dom0,dom1=rowlist.groups();assert re.sub(r\'\\s+\',\'\',rows)+\'.flatten\' in re.sub(r\'\\s+\',\'\',binding),(name,\'binding row order\')\n  for s in dict.fromkeys(slice_names):\n   st=declarations[name.rsplit(\'.\',1)[0]+\'.\'+s][\'type\'];assert scope in st and dom0 in st and dom1 in st,(name,s,\'scope/domains\')\n  rowterms=[x.strip() for x in rows[1:-1].split(\',\')]\n  sliceterms=list(dict.fromkeys(slice_names))\n  assert len(rowterms)==len(sliceterms),(name,\'slice count\')\n  for rowterm,slicename in zip(rowterms,sliceterms):\n   st=declarations[name.rsplit(\'.\',1)[0]+\'.\'+slicename][\'type\']\n   assert re.sub(r\'\\s+\',\'\',f\'⟨{scope},{rowterm}⟩\') in re.sub(r\'\\s+\',\'\',st),(name,\'slice row order\',slicename)\n  assert scope in d[\'type\'] and dom0 in d[\'type\'] and dom1 in d[\'type\']\n  parent_table=re.search(r\'checkNarrowing\\s+⟨\\s*\'+re.escape(scope)+r\'\\s*,\\s*([^⟩]+)⟩\',d[\'type\'])\n  assert parent_table and binding.split(\'=\')[0].strip()==parent_table[1].strip(),(name,\'parent row binding\')\n return {\'id\':name,\'module\':d[\'module\'],\'path\':d[\'path\'],\'line\':d[\'line\'],\'roots\':[declarations[n] for n in dict.fromkeys(roots)],\'closure\':closure(d[\'module\']),\'stages\':{k:{\'status\':\'NOT_RUN\'} for k in [\'compile\',\'type\',\'axioms\',\'export\',\'lean_replay\',\'nanoda\',\'con_ron\']}}\n\nbuilt={};failed={}\ndef compile_unit(u):\n for m in u[\'closure\']:\n  if m in built:continue\n  if any(d in failed for d in deps[m]):failed[m]={\'status\':\'BLOCKED\',\'reason\':\'failed dependency\'};continue\n  raw=files[m];clean=uncomment(raw)\n  bad=re.findall(r\'\\b(?:sorry|admit|native_decide|axiom|implemented_by|extern|ofReduceBool|ofReduceNat|trustCompiler|skipKernelTC)\\b\',clean)\n  if bad:failed[m]={\'status\':\'BLOCKED\',\'reason\':\'kernel bypass token(s): \'+str(sorted(set(bad)))};continue\n  dest=B.joinpath(*m.split(\'.\')).with_suffix(\'.olean\');dest.parent.mkdir(parents=True,exist_ok=True)\n  result=run([LEAN,\'-R\',REPO/\'src\',\'-o\',dest,*FLAGS,REPO/(\'src/\'+m.replace(\'.\',\'/\')+\'.lean\')], \'compile/\'+m+\'.log\')\n  print(\'MODULE\',m,result[\'status\'],result.get(\'wall_seconds\'),flush=True)\n  if result[\'status\']==\'PASS\':\n   built[m]={\'source_sha256\':sha(raw.encode()),\'toolchain_lean_sha256\':sha(LEAN.read_bytes()),\'toolchain_archive_sha256\':\'b67d0e0ba959fc74e186f1b44aebca32d6ecb6a28f8e9a029fc4e9f73398a049\',\'flags\':FLAGS,\'imports\':{d:built[d][\'output_sha256\'] for d in deps[m]},\'output_sha256\':sha(dest.read_bytes()),\'all_outputs\':{str(q.relative_to(B)):sha(q.read_bytes()) for q in dest.parent.glob(dest.stem+\'.*\') if q.is_file()},\'command\':result}\n   dump(E/\'cache-provenance.json\',built)\n  else:failed[m]=result\n  if time.time()>=DEADLINE:break\n if not all(m in built for m in u[\'closure\']):\n  u[\'stages\'][\'compile\']={\'status\':\'INCONCLUSIVE\' if time.time()>=DEADLINE else \'BLOCKED\',\'failures\':{m:failed.get(m,{\'status\':\'NOT_RUN\'}) for m in u[\'closure\'] if m not in built}};return False\n u[\'stages\'][\'compile\']={\'status\':\'PASS\',\'modules\':u[\'closure\'],\'cache\':\'cache-provenance.json\'};return True\n\ndef audit(u):\n short=u[\'id\'].rsplit(\'.\',1)[1];p=E/\'audits\'/f\'{short}.lean\';p.parent.mkdir(exist_ok=True)\n s=f\'import {u["module"]}\\nimport Lean\\nset_option autoImplicit false\\nset_option maxRecDepth 100000\\nset_option maxHeartbeats 10000000\\nnamespace {u["id"].rsplit(".",1)[0]}\\nopen SpectreFinite.FiniteTable\\nopen HigherFastChecking\\n\'\n for i,r in enumerate(u[\'roots\']):\n  n=r[\'name\'];s+=f\'\\nexample : {r["type"]} := @{n}\\n#print axioms {n}\\nopen Lean Elab Command in\\nrun_cmd do\\n  let n := ``{n}\\n  let env ← getEnv\\n  let some ci := env.find? n | throwError "missing root"\\n  unless ci matches .thmInfo _ do throwError "root is not a theorem"\\n  let axs ← liftCoreM (Lean.collectAxioms n)\\n  let allowed : List Name := [``propext, ``Classical.choice, ``Quot.sound]\\n  unless axs.toList.all (fun a => allowed.contains a) do throwError "disallowed axioms: {{axs.toList}}"\\n  logInfo m!"ROOT_OK {n} AXIOMS {{axs.toList}}"\\n\'\n s+=\'\\nend \'+u[\'id\'].rsplit(\'.\',1)[0]+\'\\n\';p.write_text(s)\n result=run([LEAN,*FLAGS,p],f\'audits/{short}.log\')\n text=(E/result[\'log\']).read_text() if \'log\' in result else \'\'\n roots={}\n for r in u[\'roots\']:\n  n=r[\'name\'];mat=re.search(r\'ROOT_OK \'+re.escape(n)+r\' AXIOMS \\[([^\\]]*)\\]\',text)\n  if mat:roots[n]={\'exact_type\':\'PASS\',\'axioms\':sorted(x.strip() for x in mat[1].split(\',\') if x.strip())}\n good=result[\'status\']==\'PASS\' and len(roots)==len(u[\'roots\']) and not re.search(r\'\\berror:|declaration uses .sorry.\',text)\n u[\'root_results\']=roots\n for stage in [\'type\',\'axioms\']:u[\'stages\'][stage]={**result,\'status\':\'PASS\' if good else result[\'status\'] if result[\'status\']!=\'PASS\' else \'FAIL\',\'audit_source\':str(p.relative_to(E))}\n return good\n\nif __name__==\'__main__\':\n authenticate()\n name=\'SpectreFinite.HexSearch.RadiusFive.root8Node0Step1Chunk8_2_checked\';u=unit(name);dump(E/\'seed-unit.json\',u)\n print(\'SEED CLOSURE\',len(u[\'closure\']),flush=True)\n good=compile_unit(u) and audit(u)\n dump(E/\'seed-unit.json\',u)\n print(json.dumps(u[\'stages\']),flush=True)\n sys.exit(0 if good else 1)\n', 'continue.py': "import runner as r\nimport re,time,json,os\nP=r.P;W=r.W;E=r.E\n\ndef queue():\n N='SpectreFinite.HexSearch.RadiusFive.'\n names=[N+f'root{x}Node0Step1Chunk{x}_{j}_checked' for x in [0,1,4,5,6,7,8] for j in range(6)]\n names += [N+f'root0Node0Step2Chunk{k}_{j}_checked' for k in [5,6,7,8] for j in range(6)]\n names += [N+'root0Node0Step3Chunk5_5_checked']+[N+f'root0Node0Step3Chunk{k}_{j}_checked' for k in [6,7,8] for j in range(6)]\n small='SpectreFinite.HexSearch.RadiusThree.root2Node0Step4_checked';stress=N+'root7Node110Step1_checked'\n names += [small,stress];assert len(set(names))==87\n units={n:r.unit(n) for n in names};assert len({x['name'] for u in units.values() for x in u['roots']})==555\n prefix=[N+f'root8Node0Step1Chunk8_{j}_checked' for j in [2,0,1,3,4,5]]+[small,stress]\n scheduled=set();ordered=[];metrics={}\n def score(u):\n  added=set(u['closure'])-scheduled\n  return (len(added),sum(len(r.files[m].encode()) for m in added),u['path'],u['line'],u['id'])\n while units:\n  n=prefix[len(ordered)] if len(ordered)<len(prefix) else min(units,key=lambda x:score(units[x]))\n  u=units.pop(n);u['schedule_score']=score(u);scheduled.update(u['closure']);ordered.append(u)\n for i,u in enumerate(ordered):u['global_queue_index']=i;u['branch']='A' if u['id']==small or re.match(N.replace('.','\\\\.')+r'root(?:0|8)Node0',u['id']) else 'B'\n selected=[u for u in ordered if u['branch']=='A'];assert len(selected)==56\n r.dump(E/'canonical-queue.json',{'method':'global canonical order, then filter ownership; scheduled union includes all earlier global closures','global_units':ordered,'branch_A_ids':[u['id'] for u in selected]})\n return selected\n\ndef load_cache():\n data=json.loads((E/'cache-provenance.json').read_text());leanhash=r.sha(r.LEAN.read_bytes())\n for m,c in data.items():\n  target=r.B.joinpath(*m.split('.')).with_suffix('.olean')\n  assert c['source_sha256']==r.sha(r.files[m].encode()) and c['flags']==r.FLAGS and c['toolchain_lean_sha256']==leanhash\n  assert target.is_file() and r.sha(target.read_bytes())==c['output_sha256']\n  if 'all_outputs' in c:\n   assert all((r.B/path).is_file() and r.sha((r.B/path).read_bytes())==h for path,h in c['all_outputs'].items())\n  assert all(data[d]['output_sha256']==h for d,h in c['imports'].items())\n r.built.update(data);r.commands[:]=json.loads((E/'commands.json').read_text())\n\nif __name__=='__main__':\n r.authenticate();qs=queue();load_cache();seed=json.loads((E/'seed-unit.json').read_text());assert seed['stages']['type']['status']=='PASS' and seed['stages']['axioms']['status']=='PASS'\n qs[0].update({k:v for k,v in seed.items() if k in ['stages','root_results']});r.dump(E/'units.json',qs)\n print('QUEUE READY',len(qs),'CACHE',len(r.built),flush=True)\n # Checker tiers and controls are attached below after the complete seed exists.\n import tiers\n tiers.setup()\n # Seed received all checker tiers in seed_tiers.py before entering this queue.\n assert qs[0]['stages']['export']['status']!='NOT_RUN'\n index=1\n while index<len(qs):\n  if time.time()+120>=r.DEADLINE:break\n  cohort=[];module=qs[index]['module']\n  while index<len(qs) and qs[index]['module']==module and len(cohort)<3:\n   u=qs[index];index+=1\n   print('UNIT',u['id'],flush=True)\n   if r.compile_unit(u) and r.audit(u):cohort.append(u)\n   r.dump(E/'units.json',qs)\n  if cohort:tiers.check(cohort)\n  r.dump(E/'units.json',qs)\n r.dump(E/'stop.json',{'reason':'queue exhaustion' if all(u['stages']['compile']['status']!='NOT_RUN' for u in qs) else 'insufficient safe assigned budget','proof_deadline_unix':r.DEADLINE,'stopped_unix':time.time(),'remaining_ids':[u['id'] for u in qs if not all(u['stages'][x]['status']=='PASS' for x in ['compile','type','axioms'])]})\n", 'tiers.py': 'import runner as r\nimport os,json,time,re,subprocess,signal,pathlib,gzip\nP=r.P; E=r.E\nr.ENV[\'PATH\']=str(r.TC/\'bin\')+os.pathsep+os.environ.get(\'PATH\',\'/usr/bin:/bin\')\nprimitives=[]; exports=[]; controls={}; unavailable={}\n\ndef setup():\n global primitives,exports,controls\n bundled=r.TC/\'src/lean/lake/Lake/CLI/Check.lean\';text=r.uncomment(bundled.read_text())\n (E/\'reference\').mkdir(exist_ok=True);(E/\'reference/Check.lean\').write_bytes(bundled.read_bytes())\n m=re.search(r\'^def primitiveTargets[^\\n]*\\n(.*?)^def \',text,re.M|re.S);assert m\n m=re.fullmatch(r\'\\s*return\\s*#\\[([^\\[\\]]*)\\]\\s*\',m[1]);assert m\n for item in m[1].split(\',\'):\n  if not item.strip():continue\n  n=re.fullmatch(r\'\\s*``([A-Za-z_][A-Za-z0-9_.\\\']*)\\s*\',item);assert n;primitives.append(n[1])\n assert primitives and len(set(primitives))==len(primitives)\n r.dump(E/\'checker-policy.json\',{\'source\':\'leanprover/lean4@11acb17ec6b07a8f9e9173e6845197929540936b/src/lake/Lake/CLI/Check.lean\',\'source_sha256\':r.sha(bundled.read_bytes()),\'primitiveTargets\':primitives,\'permitted_axioms\':r.ALLOWED,\'quot_builtins\':[\'Quot\',\'Quot.mk\',\'Quot.lift\',\'Quot.ind\'],\'strict_axiom_policy\':\'Always enforced by per-root Lean.collectAxioms before any tier PASS; nanoda config false does not waive this.\'})\n if (E/\'exports.json\').exists():exports[:]=json.loads((E/\'exports.json\').read_text())\n if (E/\'controls.json\').exists():controls.update(json.loads((E/\'controls.json\').read_text()))\n else:run_controls()\n\ndef targets(roots):return list(dict.fromkeys([\'Quot\',\'Quot.mk\',\'Quot.lift\',\'Quot.ind\',*roots,*r.ALLOWED,*primitives]))\n\ndef export(module,roots,label,timeout=180):\n dest=E/\'exports\'/f\'{label}.jsonl\';log=E/\'export-logs\'/f\'{label}.stderr.log\';dest.parent.mkdir(exist_ok=True);log.parent.mkdir(exist_ok=True)\n cmd=[str(r.TC/\'bin/leanexport\'),module,\'--\',*targets(roots)];start=time.monotonic();limit=min(timeout,r.DEADLINE-time.time());interrupted=False\n if limit<=0:return {\'status\':\'NOT_RUN\',\'reason\':\'safe budget exhausted\'}\n with dest.open(\'wb\') as stdout,log.open(\'wb\') as stderr:\n  p=subprocess.Popen(cmd,env=r.ENV,cwd=r.REPO,stdout=stdout,stderr=stderr,start_new_session=True)\n  try:rc=p.wait(timeout=limit)\n  except subprocess.TimeoutExpired:\n   interrupted=True;os.killpg(p.pid,signal.SIGTERM)\n   try:rc=p.wait(timeout=3)\n   except subprocess.TimeoutExpired:os.killpg(p.pid,signal.SIGKILL);rc=p.wait()\n d={\'command\':cmd,\'exit\':rc,\'wall_seconds\':round(time.monotonic()-start,6),\'log\':str(log.relative_to(E)),\'log_sha256\':r.sha(log.read_bytes()),\'export\':str(dest.relative_to(E)),\'export_bytes\':dest.stat().st_size,\'export_sha256\':r.sha(dest.read_bytes()),\'roots\':roots,\'targets\':targets(roots),\'status\':\'INCONCLUSIVE\' if interrupted or rc in (-9,137,124) else \'PASS\' if rc==0 else \'FAIL\'}\n if rc==0:\n  try:d[\'structure\']=inspect(dest,roots)\n  except Exception as exc:d[\'status\']=\'FAIL\';d[\'structure_error\']=str(exc)\n r.commands.append(d);r.dump(E/\'commands.json\',r.commands);exports.append(d);r.dump(E/\'exports.json\',exports)\n print(\'EXPORT\',label,d[\'status\'],d[\'export_bytes\'],d[\'wall_seconds\'],flush=True)\n return d\n\ndef inspect(p,roots):\n names={0:\'\'};declared=set();axioms=[];records=0;keys={}\n with p.open() as f:\n  for i,line in enumerate(f):\n   x=json.loads(line);records+=1\n   if i==0:assert \'meta\'in x\n   keys.update({k:keys.get(k,0)+1 for k in x})\n   if \'in\' in x and (\'str\' in x or \'num\' in x):\n    n=x.get(\'str\',x.get(\'num\'));name=\'.\'.join(v for v in [names[n.get(\'pre\',0)],str(n.get(\'str\',n.get(\'num\',\'\')))] if v);names[x[\'in\']]=name\n   for k in [\'thm\',\'def\',\'axiom\',\'opaque\']:\n    if k in x:\n     declared.add(names[x[k][\'name\']])\n     if k==\'axiom\':axioms.append(names[x[k][\'name\']])\n   # Inductive blocks contain their own names and constructor declarations.\n   if \'ind\' in x:\n    z=x[\'ind\']\n    if isinstance(z,dict) and \'name\'in z:declared.add(names[z[\'name\']])\n assert all(n in declared for n in roots),(set(roots)-declared)\n return {\'requested_roots_present\':True,\'axiom_declarations\':sorted(axioms),\'records\':records,\'record_kinds\':keys,\'note\':\'Structural root-presence check only; replay validates proof terms.\'}\n\ndef checker(tier,path,label,timeout=90,strict=True):\n if tier in unavailable:return {\'status\':\'BLOCKED\',\'reason\':unavailable[tier]}\n if tier==\'lean_replay\':cmd=[r.TC/\'bin/leanchecker\',\'--silent\',\'--from-export\',path]\n elif tier==\'con_ron\':cmd=[r.TC/\'bin/con-ron\',\'--jobs=2\',path]\n elif tier==\'nanoda\':\n  cfg=E/\'configs\'/f\'{label}.nanoda.json\'\n  r.dump(cfg,{\'use_stdin\':False,\'export_file_path\':str(path),\'permitted_axioms\':r.ALLOWED,\'unpermitted_axiom_hard_error\':False,\'num_threads\':4,\'nat_extension\':True,\'string_extension\':True})\n  cmd=[r.TC/\'bin/nanoda_bin\',cfg]\n result=r.run(cmd,f\'checkers/{label}.{tier}.log\',timeout=timeout)\n result.update({\'binary_sha256\':r.sha(P(cmd[0]).read_bytes()) if P(cmd[0]).is_file() else None,\'export_sha256\':r.sha(path.read_bytes()),\'export\':str(path.relative_to(E)),\'strict_axiom_audit_required\':strict})\n if tier==\'nanoda\':result.update({\'config\':str(cfg.relative_to(E)),\'config_sha256\':r.sha(cfg.read_bytes())})\n raw=(E/result[\'log\']).read_text(errors=\'replace\') if \'log\'in result else \'\'\n if result[\'status\']==\'FAIL\' and re.search(r\'error while loading shared libraries|could not execute external process|No such file or directory\',raw):\n  result[\'status\']=\'BLOCKED\';result[\'reason\']=\'loader failure, not a proof rejection\';unavailable[tier]=result[\'reason\']\n print(\'CHECKER\',label,tier,result[\'status\'],result.get(\'exit\'),result.get(\'wall_seconds\'),flush=True)\n return result\n\ndef run_controls():\n c=E/\'controls\';c.mkdir(exist_ok=True)\n src=c/\'CapacityControl.lean\';src.write_text(\'import Lean\\nset_option autoImplicit false\\nnamespace CapacityNegativeControl\\ntheorem prerequisite : True := True.intro\\ntheorem good : True := prerequisite\\naxiom forbidden : False\\ntheorem usesForbidden : False := forbidden\\nend CapacityNegativeControl\\n\')\n comp=r.run([r.LEAN,\'-R\',c,\'-o\',r.B/\'CapacityControl.olean\',*r.FLAGS,src],\'controls/compile.log\',timeout=45)\n controls[\'fixture_compile\']=comp;assert comp[\'status\']==\'PASS\'\n false=c/\'FalseType.lean\';false.write_text(\'import CapacityControl\\nset_option autoImplicit false\\nexample : False := CapacityNegativeControl.good\\n\')\n controls[\'false_expected_type\']=r.run([r.LEAN,*r.FLAGS,false],\'controls/false-type.log\',timeout=30)\n bad=c/\'DisallowedAxiom.lean\';bad.write_text(\'import CapacityControl\\nimport Lean\\nset_option autoImplicit false\\nopen Lean Elab Command in\\nrun_cmd do\\n  let axs ← liftCoreM (Lean.collectAxioms ``CapacityNegativeControl.usesForbidden)\\n  let allowed : List Name := [``propext, ``Classical.choice, ``Quot.sound]\\n  unless axs.toList.all (fun a => allowed.contains a) do\\n    throwError "CONTROL disallowed axioms: {axs.toList}"\\n\')\n controls[\'disallowed_axiom_audit\']=r.run([r.LEAN,*r.FLAGS,bad],\'controls/disallowed-axiom.log\',timeout=30)\n good=export(\'CapacityControl\',[\'CapacityNegativeControl.good\'],\'control-valid\',timeout=45);controls[\'valid_export\']=good\n # Isolated export mutations follow the registry\'s ill_typed_export fixture.\n source=E/good[\'export\'];records=[json.loads(l) for l in source.read_text().splitlines()];names={0:\'\'}\n for x in records:\n  if \'in\'in x and (\'str\'in x or \'num\'in x):\n   n=x.get(\'str\',x.get(\'num\'));names[x[\'in\']]=\'.\'.join(v for v in [names[n.get(\'pre\',0)],str(n.get(\'str\',n.get(\'num\',\'\')))] if v)\n nameid=next(k for k,v in names.items() if v==\'CapacityNegativeControl.good\')\n wrong=json.loads(json.dumps(records))\n thm=next(x[\'thm\'] for x in wrong if \'thm\'in x and x[\'thm\'][\'name\']==nameid);thm[\'value\']=thm[\'type\']\n ill=c/\'false-proof.jsonl\';ill.write_text(\'\'.join(json.dumps(x,separators=(\',\',\':\'))+\'\\n\' for x in wrong))\n # Remove the theorem\'s declaration, leaving a syntactically valid but incomplete root set.\n incomplete=c/\'missing-root.jsonl\';incomplete.write_text(\'\'.join(json.dumps(x,separators=(\',\',\':\'))+\'\\n\' for x in records if not (\'thm\'in x and x[\'thm\'][\'name\']==nameid)))\n try:inspect(incomplete,[\'CapacityNegativeControl.good\']);outcome={\'status\':\'FAIL\',\'reason\':\'incomplete root unexpectedly accepted\'}\n except AssertionError as exc:outcome={\'status\':\'PASS\',\'reason\':\'structural gate rejected omitted requested theorem\',\'actual_rejection\':str(exc)}\n controls[\'incomplete_export_root_gate\']=outcome\n dependency=c/\'missing-dependency.jsonl\'\n helperid=next(k for k,v in names.items() if v==\'CapacityNegativeControl.prerequisite\')\n dependency.write_text(\'\'.join(json.dumps(x,separators=(\',\',\':\'))+\'\\n\' for x in records if not (\'thm\'in x and x[\'thm\'][\'name\']==helperid)))\n badexport=export(\'CapacityControl\',[\'CapacityNegativeControl.usesForbidden\'],\'control-forbidden\',timeout=45)\n controls[\'forbidden_export\']=badexport\n for label,path in [(\'valid\',source),(\'false-proof\',ill),(\'incomplete\',incomplete),(\'missing-dependency\',dependency),(\'disallowed-axiom\',E/badexport[\'export\'])]:\n  controls[label]={tier:checker(tier,path,\'control-\'+label,timeout=30,strict=False) for tier in [\'lean_replay\',\'nanoda\',\'con_ron\']}\n controls[\'semantics\']=\'FAIL for the false-type/axiom/ill-typed fixtures means actual rejection, an expected negative-control outcome. Incomplete root gate is structural; checker acceptance of an export with a requested root omitted is not proof validation.\'\n r.dump(E/\'controls.json\',controls)\n\ndef check(units,first=False):\n if isinstance(units,dict):units=[units]\n assert all(all(u[\'stages\'][x][\'status\']==\'PASS\' for x in [\'compile\',\'type\',\'axioms\']) and len(u.get(\'root_results\',{}))==len(u[\'roots\']) for u in units),\'complete strict root audit required\'\n label=\'cohort-\'+units[0][\'id\'].rsplit(\'.\',1)[1]\n roots=list(dict.fromkeys(x[\'name\'] for u in units for x in u[\'roots\']))\n d=export(units[0][\'module\'],roots,label)\n for u in units:u[\'stages\'][\'export\']=d\n if d[\'status\']!=\'PASS\':\n  for u in units:\n   for tier in [\'lean_replay\',\'nanoda\',\'con_ron\']:u[\'stages\'][tier]={\'status\':\'BLOCKED\',\'reason\':\'export did not complete\'}\n  return\n # Reject every nonstandard exported axiom regardless of checker soft policy.\n assert set(d[\'structure\'][\'axiom_declarations\'])<=set(r.ALLOWED),d[\'structure\'][\'axiom_declarations\']\n for tier in [\'lean_replay\',\'nanoda\',\'con_ron\']:\n  res=checker(tier,E/d[\'export\'],label,timeout=120 if first else 60)\n  for u in units:u[\'stages\'][tier]=res\n', 'seed_tiers.py': "import json,runner as r,tiers\nr.commands[:]=json.loads((r.E/'commands.json').read_text())\nu=json.loads((r.E/'seed-unit.json').read_text());assert all(u['stages'][x]['status']=='PASS' for x in ['compile','type','axioms']);tiers.setup();tiers.check(u,first=True);r.dump(r.E/'seed-unit.json',u)\n", 'entry.py': "# A fresh run discovers the finite branch queue, then checks the complete seed\n# before creating any optional checker fixtures or exports.\nimport importlib,json,sys\nimport runner as r\nr.authenticate()\nunits=importlib.import_module('continue').queue();u=units[0]\nr.dump(r.E/'units.json',units);r.dump(r.E/'seed-unit.json',u)\ngood=r.compile_unit(u) and r.audit(u)\nr.dump(r.E/'seed-unit.json',u);r.dump(r.E/'units.json',units)\nprint('SEED',u['id'],'PASS' if good else 'NOT PASS',flush=True)\nsys.exit(0 if good else 1)\n", 'package.py': "import pathlib,json,hashlib,tarfile,gzip,io,time,subprocess,os\nP=pathlib.Path\nREPO=P(os.environ.get('CAPACITY_REPO',P(__file__).resolve().parents[3])).resolve();W=P(os.environ.get('CAPACITY_WORK',REPO/'test/scratch/lean-capacity-a')).resolve();E=W/'evidence';OUT=REPO/'artifacts/lean-capacity/a'\nSHA=lambda b:hashlib.sha256(b).hexdigest()\nSTAGES=['compile','type','axioms','export','lean_replay','nanoda','con_ron']\ndef dump(p,x):p.write_text(json.dumps(x,indent=2)+'\\n')\ndef packaging():\n OUT.mkdir(parents=True,exist_ok=True)\n units=json.loads((E/'units.json').read_text()) if (E/'units.json').exists() else [json.loads((E/'seed-unit.json').read_text())]\n if (E/'seed-unit.json').exists():\n  seed=json.loads((E/'seed-unit.json').read_text())\n  if units[0]['id']==seed['id']:units[0].update({k:seed[k] for k in ['stages','root_results'] if k in seed})\n for u in units:\n  u['expected_type']=u['roots'][0]['type'];u['root_names']=[root['name'] for root in u['roots']]\n source=json.loads((E/'source-manifest.json').read_text());toolchain=json.loads((E/'toolchain.json').read_text());controls=json.loads((E/'controls.json').read_text()) if (E/'controls.json').exists() else {'semantics':'Control suite not run; no outcomes claimed.'}\n complete=lambda u:all(u['stages'][s]['status']=='PASS' for s in ['compile','type','axioms'])\n totals={s:sum(complete(u) and u['stages'][s]['status']=='PASS' for u in units) for s in STAGES}\n # Every counted root must have its individual exact-type and strict axiom evidence.\n for u in units:\n  if complete(u):\n   assert len(u['roots'])==len(u['root_results'])\n   for x in u['roots']:\n    rr=u['root_results'][x['name']];assert rr['exact_type']=='PASS' and set(rr['axioms'])<={'propext','Quot.sound','Classical.choice'}\n stop=json.loads((E/'stop.json').read_text()) if (E/'stop.json').exists() else {'reason':'checkpoint or interrupted runner; see raw stage outcomes'}\n # Hash every output sidecar, including private/server OLeans and IR. Initial\n # command receipts bind the primary OLean; this final manifest binds the full\n # locally built object set without rerunning any proof.\n cache=json.loads((E/'cache-provenance.json').read_text()) if (E/'cache-provenance.json').exists() else {}\n outputs={}\n for module,provenance in cache.items():\n  base=W/'build'/module.replace('.','/');primary=base.with_suffix('.olean')\n  assert SHA(primary.read_bytes())==provenance['output_sha256']\n  outputs[module]={'build_command':provenance['command'],'source_sha256':provenance['source_sha256'],'toolchain_archive_sha256':toolchain['archive_sha256'],'flags':provenance['flags'],'primary_olean_sha256':provenance['output_sha256'],'output_hashes':{p.relative_to(W/'build').as_posix():SHA(p.read_bytes()) for p in base.parent.glob(base.name+'.*') if p.is_file()}}\n dump(E/'cache-complete-outputs.json',{'collection':'final reconciliation of locally built outputs; no proof rerun','modules':outputs})\n # Copy reference data and actual runner sources; no installed tools or build objects.\n ref=E/'reference';ref.mkdir(exist_ok=True)\n for n in ['release.json','verify_submission.py','comparator-declaration-closure.md','Check.lean']:\n  if (W/n).exists():(ref/n).write_bytes((W/n).read_bytes())\n code=E/'runner';code.mkdir(exist_ok=True)\n for n in ['runner.py','continue.py','tiers.py','seed_tiers.py','test_runner.py','package.py','entry.py','bootstrap.py','test_process.py','check_source_final.py']:\n  if (W/n).exists():(code/n).write_bytes((W/n).read_bytes())\n for n in ['seed-run.log','seed-tiers.log','continue-run.log',*sorted(p.name for p in W.glob('*.driver.log'))]:\n  if (W/n).exists():(E/n).write_bytes((W/n).read_bytes())\n # Phase times count each process once, even when one audit proves both type and axiom gates.\n commands=json.loads((E/'commands.json').read_text());timings={};seen_logs=set()\n for command in commands:\n  log=command.get('log','')\n  if log in seen_logs:continue\n  seen_logs.add(log)\n  phase=('controls' if log.startswith('controls/') or log.startswith('checkers/control-') or 'control-' in log else 'compile' if log.startswith('compile/') else 'type_and_axiom_audit' if log.startswith('audits/') else 'export' if 'export' in command else 'lean_replay' if log.endswith('.lean_replay.log') else 'nanoda' if log.endswith('.nanoda.log') else 'con_ron' if log.endswith('.con_ron.log') else 'other')\n  value=timings.setdefault(phase,{'processes':0,'summed_wall_seconds':0.0});value['processes']+=1;value['summed_wall_seconds']+=command.get('wall_seconds',0.0)\n for value in timings.values():value['summed_wall_seconds']=round(value['summed_wall_seconds'],6)\n dump(E/'phase-wall-times.json',{'semantics':'Sum of recorded process wall times, counted once; excludes acquisition, source indexing and packaging. Not an end-to-end throughput guarantee.','phases':timings})\n # The source-bound branch receipt is inside the archive, without its own archive digest.\n receipt={'schema':'imd-lean-capacity/v1','experiment':'A847-LEAN-CAPACITY-32549b50e95c','branch':'A','source':{'repository':'https://github.com/jconorgrogan/palomar-chair-verification','commit':source['commit'],'tree':source['tree'],'manifest':'source-manifest.json','comparison':source['comparison']},'toolchain':toolchain,'flags':['-j1','-DElab.async=false'],'allowed_axioms':['propext','Quot.sound','Classical.choice'],'assigned_unit_count':56,'completed_unit_count':sum(complete(u) for u in units),'unique_unit_and_support_roots':len({root['name'] for u in units for root in u['roots']}),'final_source_check':'final-source-check.json' if (E/'final-source-check.json').exists() else None,'input_replay':{'sha256':'b222cca4765faeebd619a23efd48b0bc2cc2520e37333a88e5d02cd89155341f','reference_data_only':True,'authentication_record':'reference/input-replay.json' if (E/'reference/input-replay.json').exists() else None,'status':'PASS' if (E/'reference/input-replay.json').exists() else 'NOT_RUN'},'units':units,'tier_totals':totals,'totals_semantics':'Distinct complete units per tier, never summed across tiers. Supports are not units. Type and axiom gates apply to every tier. Lean replay uses the same kernel. Structural presence is not proof validation.','commands':'commands.json','phase_timings':'phase-wall-times.json','controls':'controls.json','control_interpretation':controls.get('semantics'),'cache_provenance':['cache-provenance.json','cache-complete-outputs.json'],'cache_note':'Initial command receipts bind primary OLeans. Final reconciliation also binds every locally produced private/server OLean and IR sidecar; delivered replay records and checks full output sets before reuse.','queue':'canonical-queue.json','stop':stop,'remaining_ids':[u['id'] for u in units if not complete(u)],'remaining_by_tier':{s:[u['id'] for u in units if not(complete(u) and u['stages'][s]['status']=='PASS')] for s in STAGES},'delivery':{'policy':'earliest exports retained; no proof truncation; deduplicate identical export bytes','missing_export_delivery':[]},'scope':'Replication of helper proof units only; no final theorem certification, comparator/registry invocation, full Solution/Mathlib build, or independent throughput guarantee.'}\n dump(E/'branch-receipt.json',receipt)\n # Authenticate bytes for every archive member (the manifest itself is bound by outer archive SHA).\n members={};seen={};aliases={}\n for p in sorted(E.rglob('*')):\n  if not p.is_file() or p.name=='ARCHIVE-MANIFEST.json':continue\n  rel=p.relative_to(E).as_posix();assert p.suffix not in ['.olean','.ilean','.so','.zst']\n  digest=SHA(p.read_bytes())\n  if rel.startswith('exports/') and digest in seen:aliases[rel]=seen[digest];continue\n  if rel.startswith('exports/'):seen[digest]=rel\n  members[rel]={'bytes':p.stat().st_size,'sha256':digest}\n manifest={'schema':'imd-lean-capacity/archive/v1','members':members,'identical_export_aliases':aliases}\n dump(E/'ARCHIVE-MANIFEST.json',manifest)\n archive=OUT/'evidence.tar.gz'\n with archive.open('wb') as raw,gzip.GzipFile(filename='',fileobj=raw,mode='wb',compresslevel=9,mtime=0) as gz,tarfile.open(fileobj=gz,mode='w|',format=tarfile.PAX_FORMAT) as tar:\n  for name in [*members,'ARCHIVE-MANIFEST.json']:\n   data=(E/name).read_bytes();info=tarfile.TarInfo(name);info.size=len(data);info.mode=0o644;info.mtime=0;info.uid=info.gid=0;info.uname=info.gname='';tar.addfile(info,io.BytesIO(data))\n assert archive.stat().st_size<=60*1024*1024,('archive exceeds 60MiB',archive.stat().st_size)\n receipt['delivery'].update({'archive':'evidence.tar.gz','archive_bytes':archive.stat().st_size,'archive_sha256':SHA(archive.read_bytes()),'manifest':'ARCHIVE-MANIFEST.json','replay_sha256':SHA((OUT/'replay.sh').read_bytes()) if (OUT/'replay.sh').exists() else None})\n dump(OUT/'receipt.json',receipt)\n counted=totals['axioms'];remaining=len(receipt['remaining_ids'])\n report=f'''Experiment A847-LEAN-CAPACITY-32549b50e95c, Branch A\\n\\n{counted} of 56 assigned complete units passed compilation, their exact published types, and strict transitive axiom checks. {remaining} units remain without those three results.\\n\\n| Stage | Distinct complete units passing |\\n|---|---:|\\n'''+''.join(f'| {s} | {totals[s]} |\\n' for s in STAGES)+f'''\\nThe seed was `SpectreFinite.HexSearch.RadiusFive.root8Node0Step1Chunk8_2_checked`: parent, five published slices and row binding. It obtained actual compile/type/axiom results before optional checker work. The queue is generated from actual declarations, validates scope, slice order, domain names and parent-row binding, and includes both projected-slice intermediates. Global canonical order is generated against earlier scheduled closures before filtering Branch A. Branch B units are never selected for checking.\\n\\nSources are bound to commit `{source['commit']}` and tree `{source['tree']}`. Filesystem bytes were compared directly with Git blobs using `git cat-file`, independently of the index. The official release asset was authenticated against GitHub's release digest and the supplied hash; the Lean version and source commit match. `toolchain.json` records binary SHA-256 values. No original mathematics or build pins were changed.\\n\\nCompilation targets only required project import closures, including public imports and import-all syntax, with `-j1 -DElab.async=false`. Every unit root has an `autoImplicit=false` exact-type example and a theorem-kind/transitive-axiom audit. Only `propext`, `Quot.sound` and `Classical.choice` are permitted. Cache entries are locally compiled and bind source, authenticated toolchain, flags, dependency outputs and output hashes. Initial command receipts bind primary OLeans; final reconciliation also hashes every private/server OLean and IR sidecar. The delivered replay records and checks full output sets before reuse.\\n\\nExports use bundled `leanexport MODULE -- ROOTS...`, including complete unit roots, the permitted axioms, Quot builtins and primitive targets parsed from matching `Check.lean` with comments removed. The raw exports, hashes, roots, commands, logs, exits, wall times and NanoDa configs are retained. Lean replay uses `leanchecker --silent --from-export`; NanoDa uses the matching config, and con-ron uses `--jobs=2`. NanoDa's soft axiom configuration does not waive the strict per-root audit. Lean replay is the same kernel.\\n\\nControls are isolated fixtures, never proof units. The false expected type and added axiom were actually rejected by their Lean audits. All three replay tools rejected the ill-typed-proof and missing-dependency exports. All three accepted the omitted-root export; the structural root-presence gate rejected it. Lean replay accepted the added-axiom export, while NanoDa and con-ron rejected it. The separate strict axiom audit rejected it regardless of replay outcomes. These control failures reject deliberately invalid inputs; they do not assert that a published theorem is false. Exact results and raw logs are in `controls.json` and `controls/` / `checkers/`. Parser component tests are in `parser-tests.log`. No control outcome is inferred from a missing tool.\\n\\nStop reason: {stop['reason']}. Exact remaining IDs and per-tier coverage are in `receipt.json`; supports are never counted as units and tier totals are never added together. Final reconciliation reads saved evidence without rerunning proofs.\\n\\nThe archive contains manifests, receipts, runner sources, reference data, controls, raw logs/configs and deduplicated complete exports; it excludes installed toolchain and cache binaries. Archive: {archive.stat().st_size:,} bytes, SHA-256 `{receipt['delivery']['archive_sha256']}`.\\n\\nRun `bash artifacts/lean-capacity/a/replay.sh --check-evidence` to check packaging and evidence bindings without rerunning proofs. `--help` describes a fresh mathematical run with an authenticated official toolchain archive and an explicit remaining assigned budget. The source checkout and toolchain are prerequisites for a fresh run; the archive itself is portable evidence, not a bundled Lean installation.\\n\\nThis is replication of helper-unit checks, not new mathematics, final-theorem certification, a comparator/registry run or a throughput guarantee.\\n'''\n if 'false_expected_type' not in controls:\n  a=report.index('Controls are isolated fixtures');b=report.index('Stop reason:',a);report=report[:a]+'The optional control suite did not run; no control outcome is claimed.\\n\\n'+report[b:]\n (OUT/'report.md').write_text(report)\n print(json.dumps({'tier_totals':totals,'archive_bytes':archive.stat().st_size,'stop':stop},indent=2))\nif __name__=='__main__':packaging()\n", 'test_runner.py': 'import unittest,importlib.util,pathlib\np=pathlib.Path(__file__).with_name(\'runner.py\');s=importlib.util.spec_from_file_location(\'capacity_runner\',p);r=importlib.util.module_from_spec(s);s.loader.exec_module(r)\nclass ParserTests(unittest.TestCase):\n def test_nested_comments_and_strings(self):\n  src=\'public import A\\n/- import Bogus /- nested -/ -/\\nimport all B\\n-- import C\\n"import D"\\n\'\n  clean=r.uncomment(src)\n  self.assertEqual(len(clean),len(src));self.assertEqual(clean.count(\'\\n\'),src.count(\'\\n\'))\n  self.assertNotIn(\'Bogus\',clean);self.assertNotIn(\'import C\',clean);self.assertNotIn(\'import D\',clean)\n  imports=[]\n  for m in r.re.finditer(r\'^\\s*(?:(?:public|meta|private)\\s+)*import\\s+(?:all\\s+)?([^\\n]+)\',clean,r.re.M):imports.extend(m[1].split())\n  self.assertEqual(imports,[\'A\',\'B\'])\n def test_unclosed_comment_is_rejected(self):\n  with self.assertRaises(ValueError):r.uncomment(\'/- x\')\n def test_comment_markers_in_string(self):\n  self.assertEqual(r.uncomment(\'"/-"\\nimport A\').strip(),\'import A\')\nif __name__==\'__main__\':unittest.main()\n', 'test_process.py': 'import runner as r,pathlib,sys,time,json\noriginal=r.E;r.E=r.W/\'process-test-evidence\';r.E.mkdir(exist_ok=True);r.DEADLINE=time.time()+30\ncases=[]\na=r.run([sys.executable,\'-c\',\'raise SystemExit(7)\'],\'exit7.log\',timeout=2);assert a[\'status\']==\'FAIL\' and a[\'exit\']==7;cases.append(a)\na=r.run([sys.executable,\'-c\',\'import time;time.sleep(2)\'],\'timeout.log\',timeout=.05);assert a[\'status\']==\'INCONCLUSIVE\' and a[\'exit\']!=0;cases.append(a)\na=r.run([r.W/\'intentionally-missing-control-executable\'],\'loader.log\',timeout=2);assert a[\'status\']==\'BLOCKED\' and a[\'exit\'] is None;cases.append(a)\na=r.run([sys.executable,\'-c\',\'print("CONTROL success")\'],\'success.log\',timeout=2);assert a[\'status\']==\'PASS\' and a[\'exit\']==0;cases.append(a)\nfor d in cases:d[\'raw_log\']=(r.E/d[\'log\']).read_text()\n(original/\'process-tests.json\').write_text(json.dumps({\'status\':\'PASS\',\'cases\':cases},indent=2)+\'\\n\');print(\'Process exit/timeout/loader/success checks PASS\')\n'}
import argparse,sys,pathlib,json,hashlib,tarfile,subprocess,tempfile,os,time,datetime,io,shutil,re
P=pathlib.Path
ARCHIVE_SHA='b67d0e0ba959fc74e186f1b44aebca32d6ecb6a28f8e9a029fc4e9f73398a049'
RELEASE_URL='https://github.com/leanprover/lean4/releases/download/v4.35.0-rc2/lean-4.35.0-rc2-linux.tar.zst'
META_URL='https://api.github.com/repos/leanprover/lean4/releases/tags/v4.35.0-rc2'
def hashfile(p):
 h=hashlib.sha256()
 with open(p,'rb') as f:
  for b in iter(lambda:f.read(1024*1024),b''):h.update(b)
 return h.hexdigest()
def check_evidence(out):
 receipt=json.loads((out/'receipt.json').read_text());archive=out/'evidence.tar.gz'
 assert receipt['schema']=='imd-lean-capacity/v1' and receipt['branch']=='A'
 assert archive.stat().st_size==receipt['delivery']['archive_bytes']<=60*1024*1024
 assert hashfile(archive)==receipt['delivery']['archive_sha256']
 assert hashfile(out/'replay.sh')==receipt['delivery']['replay_sha256']
 with tarfile.open(archive,'r:gz') as tar:
  items=tar.getmembers();names=[x.name for x in items];assert len(names)==len(set(names))
  for x in items:
   q=P(x.name);assert x.isfile() and not q.is_absolute() and '..' not in q.parts
  manifest=json.load(tar.extractfile('ARCHIVE-MANIFEST.json'));expected=manifest['members']
  assert set(names)==set(expected)|{'ARCHIVE-MANIFEST.json'}
  for name,binding in expected.items():
   f=tar.extractfile(name);h=hashlib.sha256();n=0
   for b in iter(lambda:f.read(1024*1024),b''):h.update(b);n+=len(b)
   assert n==binding['bytes'] and h.hexdigest()==binding['sha256'],name
  units=receipt['units'];assert len(units)==receipt['assigned_unit_count']==56
  ids=[u['id'] for u in units];assert len(set(ids))==56 and ids[0].endswith('root8Node0Step1Chunk8_2_checked')
  totals={s:0 for s in ['compile','type','axioms','export','lean_replay','nanoda','con_ron']}
  def getbytes(name):
   name=manifest.get('identical_export_aliases',{}).get(name,name)
   return tar.extractfile(name).read()
  for u in units:
   complete=all(u['stages'][s]['status']=='PASS' for s in ['compile','type','axioms'])
   if complete:
    assert {x['name'] for x in u['roots']}==set(u['root_results'])
    for root in u['roots']:
     result=u['root_results'][root['name']]
     assert result['exact_type']=='PASS' and set(result['axioms'])<={'propext','Quot.sound','Classical.choice'}
    for s in ['type','axioms']:
     d=u['stages'][s];assert d['exit']==0 and hashlib.sha256(getbytes(d['log'])).hexdigest()==d['log_sha256']
     audit=getbytes(d['audit_source']).decode();assert 'set_option autoImplicit false' in audit
     assert all(root['name'] in audit and root['type'] in audit for root in u['roots'])
     raw=getbytes(d['log']).decode()
     for root in u['roots']:
      marker=re.search(r'ROOT_OK '+re.escape(root['name'])+r' AXIOMS \[([^\]]*)\]',raw)
      assert marker,root['name']
      assert sorted(x.strip() for x in marker[1].split(',') if x.strip())==u['root_results'][root['name']]['axioms']
   for s,d in u['stages'].items():
    if complete and d['status']=='PASS':
     totals[s]+=1
     if s in ['export','lean_replay','nanoda','con_ron']:
      data=getbytes(d['export']);assert hashlib.sha256(data).hexdigest()==d['export_sha256']
      assert d['exit']==0 and d['command']
      if s=='export':assert {x['name'] for x in u['roots']}<=set(d['roots'])
      if s=='nanoda':assert hashlib.sha256(getbytes(d['config'])).hexdigest()==d['config_sha256']
  assert totals==receipt['tier_totals'],(totals,receipt['tier_totals'])
  if 'controls.json' in names:
   controls=json.load(tar.extractfile('controls.json'))
   assert controls['false_expected_type']['status']=='FAIL' and controls['false_expected_type']['exit']!=0
   assert controls['disallowed_axiom_audit']['status']=='FAIL' and controls['disallowed_axiom_audit']['exit']!=0
   assert controls['incomplete_export_root_gate']['status']=='PASS'
   for fixture in ['false-proof','missing-dependency']:
    for tier in ['lean_replay','nanoda','con_ron']:
     d=controls[fixture][tier];assert d['status']=='FAIL' and d['exit']!=0
 print(json.dumps({'evidence_integrity':'PASS','proofs_rerun':False,'tier_totals':totals,'archive_bytes':archive.stat().st_size},indent=2))

def fresh_run(args):
 started=time.time();repo=args.repo.resolve();out=args.out_dir.resolve()
 assert args.budget_seconds and args.reserve_seconds+60<args.budget_seconds<=6840,'supply remaining assigned budget exceeding reserve + 60 seconds'
 assert args.reserve_seconds>=120,'reserve at least 120 seconds for reconciliation and packaging'
 # Task scope: all generated work stays under test/scratch, all delivery under Branch A.
 assert out==repo/'artifacts/lean-capacity/a','output directory must be the assigned Branch A directory'
 scratch=repo/'test/scratch';scratch.mkdir(parents=True,exist_ok=True)
 work=P(tempfile.mkdtemp(prefix='lean-capacity-a-replay-',dir=scratch))
 deadline=started+args.budget_seconds-args.reserve_seconds
 def cmd(c):
  remaining=deadline-time.time();assert remaining>0,'remaining assigned proof budget exhausted'
  subprocess.run([str(x) for x in c],check=True,cwd=repo,timeout=remaining)
 if args.toolchain_archive:
  archive=args.toolchain_archive.resolve();assert archive.is_file()
 elif args.download_toolchain:
  archive=work/'lean.tar.zst';cmd(['curl','-fL','--retry','2',RELEASE_URL,'-o',archive]);cmd(['curl','-fsSL',META_URL,'-o',work/'release.json'])
  release=json.loads((work/'release.json').read_text());asset=next(x for x in release['assets'] if x['name']=='lean-4.35.0-rc2-linux.tar.zst');assert asset['digest']=='sha256:'+ARCHIVE_SHA
 else:raise ValueError('fresh verification needs --toolchain-archive PATH or --download-toolchain; no unverified LEAN_HOME accepted')
 assert hashfile(archive)==ARCHIVE_SHA,'official toolchain archive hash mismatch'
 cmd(['tar','--zstd','-xf',archive,'-C',work]);tc=work/'lean-4.35.0-rc2-linux'
 version=subprocess.check_output([tc/'bin/lean','--version'],text=True).strip();assert 'version 4.35.0-rc2,'in version and 'commit 11acb17ec6b07a8f9e9173e6845197929540936b'in version
 evidence=work/'evidence';evidence.mkdir()
 toolchain={'archive_url':RELEASE_URL,'release_metadata_url':META_URL,'archive_sha256':ARCHIVE_SHA,'official_release_digest':'sha256:'+ARCHIVE_SHA,'version':version,'binary_sha256':{n:hashfile(tc/'bin'/n) for n in ['lean','leanexport','leanchecker','nanoda_bin','con-ron']}}
 (evidence/'toolchain.json').write_text(json.dumps(toolchain,indent=2)+'\n')
 for n,code in SOURCES.items():(work/n).write_text(code)
 # The bootstrap passes explicit task-local locations; it never reads global toolchain settings.
 env=dict(os.environ,CAPACITY_REPO=str(repo),CAPACITY_WORK=str(work),CAPACITY_DEADLINE=str(deadline))
 def phase(name):
  with (work/(name+'.driver.log')).open('wb') as log:
   p=subprocess.Popen([sys.executable,str(work/(name+'.py'))],cwd=repo,env=env,stdout=log,stderr=subprocess.STDOUT,start_new_session=True)
   try:return p.wait(timeout=max(1,deadline-time.time()))
   except subprocess.TimeoutExpired:
    import signal
    os.killpg(p.pid,signal.SIGTERM)
    try:p.wait(timeout=3)
    except subprocess.TimeoutExpired:os.killpg(p.pid,signal.SIGKILL);p.wait()
    return 124
 rc=phase('entry')
 if rc==0:rc=phase('seed_tiers')
 if rc==0:
  rc=phase('test_runner')
  if (work/'test_runner.driver.log').exists():shutil.copyfile(work/'test_runner.driver.log',evidence/'parser-tests.log')
 if rc==0:rc=phase('test_process')
 if rc==0:rc=phase('continue')
 if rc!=0 and not (evidence/'stop.json').exists():
  (evidence/'stop.json').write_text(json.dumps({'reason':'runner phase stopped or rejected; see raw logs','phase_exit':rc,'proof_deadline_unix':deadline,'stopped_unix':time.time()},indent=2)+'\n')
 # Reconciliation consumes saved evidence only, even on interrupted proof work.
 # Remaining proof IDs are populated by the queue metadata/receipt; no proof reruns here.
 result=subprocess.run([sys.executable,str(work/'package.py')],cwd=repo,env=env,timeout=args.reserve_seconds)
 if result.returncode:raise RuntimeError('packaging failed; raw task-local evidence remains in '+str(work))
 print('Fresh run completed; mathematical stage exit:',rc)
 return 0 if rc==0 else 2

def main():
 parser=argparse.ArgumentParser(description='Branch A complete Lean-unit runner and evidence integrity checker. Fresh runs use only the pinned source closure and official authenticated bundled tools; no comparator or full library build.')
 group=parser.add_mutually_exclusive_group(required=True);group.add_argument('--check-evidence',action='store_true');group.add_argument('--run',action='store_true')
 parser.add_argument('--repo',type=P,default=P.cwd());parser.add_argument('--out-dir',type=P,default=P('artifacts/lean-capacity/a'))
 parser.add_argument('--budget-seconds',type=int,help='Actual remaining normal assigned allowance, including reserve; no fixed 1200-second cutoff')
 parser.add_argument('--reserve-seconds',type=int,default=480);parser.add_argument('--toolchain-archive',type=P);parser.add_argument('--download-toolchain',action='store_true')
 args=parser.parse_args()
 if args.check_evidence:check_evidence(args.out_dir.resolve());return 0
 return fresh_run(args)
if __name__=='__main__':
 try:sys.exit(main())
 except (AssertionError,ValueError,RuntimeError,subprocess.SubprocessError,OSError,KeyError) as exc:print('replay error:',str(exc),file=sys.stderr);sys.exit(3)

IMD_LEAN_CAPACITY_PY