COS Method: formal continuation

Download original file

This is the original source text. Reading it does not run or verify the program.

1 #!/usr/bin/env python3
2 """Offline, isolated replay of the K-COS end-to-end development companion.
3 
4 Requires the exact pinned portable Lean runtime. Does not fetch dependencies,
5 modify the official monograph, or overwrite an existing evidence directory.
6 """
7 from __future__ import annotations
8 import argparse, datetime, hashlib, json, os, pathlib, re, shutil, subprocess, sys, tempfile, time
9 ROOT=pathlib.Path(__file__).resolve().parent
10 ALLOWED={'propext','Classical.choice','Quot.sound'}
11 
12 def digest(p: pathlib.Path) -> str:
13     with p.open('rb') as f: return hashlib.file_digest(f,'sha256').hexdigest()
14 
15 def main() -> int:
16     ap=argparse.ArgumentParser(description=__doc__)
17     ap.add_argument('--runtime',type=pathlib.Path,required=True)
18     ap.add_argument('--output',type=pathlib.Path,default=ROOT/'verification/local_replay')
19     ap.add_argument('--timeout',type=int,default=900)
20     args=ap.parse_args(); rt=args.runtime.resolve(); out=args.output.resolve()
21     if out.exists() and any(out.iterdir()): ap.error('Output must be new or empty; existing evidence is preserved.')
22     if not (rt/'toolchain/bin/lean').is_file(): ap.error('Portable runtime not found.')
23     out.mkdir(parents=True,exist_ok=True)
24     env=os.environ.copy()
25     for k in ('LEAN_PATH','LEAN_SRC_PATH','LEAN_SYSROOT','LAKE_HOME','LD_PRELOAD'): env.pop(k,None)
26     env['PATH']=str(rt/'toolchain/bin')+os.pathsep+env.get('PATH','')
27     env['LEAN_NUM_THREADS']='1'
28     checks=[]; report={'status':'RUNNING','started_utc':datetime.datetime.now(datetime.timezone.utc).isoformat(),'checks':checks}
29     def save(): (out/'run.json').write_text(json.dumps(report,indent=2)+'\n')
30     save()
31     with tempfile.TemporaryDirectory(prefix='kcos-e2e-') as td:
32         work=pathlib.Path(td)/'lean'
33         shutil.copytree(ROOT/'lean',work,ignore=shutil.ignore_patterns('.lake','__pycache__'))
34         def run(name,cmd,expected=0,cwd=work,required=None,forbidden=()):
35             start=time.monotonic()
36             try:
37                 p=subprocess.run(cmd,cwd=cwd,env=env,text=True,stdout=subprocess.PIPE,stderr=subprocess.STDOUT,timeout=args.timeout)
38                 text=p.stdout;code=p.returncode
39             except subprocess.TimeoutExpired as exc:
40                 text=exc.stdout or ''; text=text.decode(errors='replace') if isinstance(text,bytes) else text
41                 (out/(name+'.log')).write_text(text)
42                 checks.append({'name':name,'command':cmd,'passed':False,'timeout_seconds':args.timeout})
43                 save(); raise RuntimeError(f'{name} timed out') from exc
44             (out/(name+'.log')).write_text(text)
45             ok=code==expected and (required is None or required.lower() in text.lower()) and not any(s.lower() in text.lower() for s in forbidden)
46             checks.append({'name':name,'command':cmd,'cwd':str(cwd),'exit_code':code,'expected_exit_code':expected,'passed':ok,'seconds':time.monotonic()-start,'log':name+'.log'})
47             save();print(('PASS ' if ok else 'FAIL ')+name,flush=True)
48             if not ok: raise RuntimeError(f'{name}: unexpected result; inspect its retained log')
49             return text
50         try:
51             run('compiler',['lean','--version'],required='6a10ac8c22beadecabdbb0919c2b50214762f91d')
52             run('lake',['lake','--version'],required='4.34.0-rc2')
53             if (work/'lean-toolchain').read_bytes() != (rt/'workspace/lean-toolchain').read_bytes():
54                 raise RuntimeError('Toolchain pin mismatch')
55             target=json.loads((work/'lake-manifest.json').read_text())['packages']
56             available={x['name']:x for x in json.loads((rt/'workspace/lake-manifest.json').read_text())['packages']}
57             deps=[]
58             for d in target:
59                 c=available.get(d['name'])
60                 if c is None or (c['url'],c['rev'])!=(d['url'],d['rev']): raise RuntimeError('Dependency manifest mismatch: '+d['name'])
61                 folder=rt/'workspace/.lake/packages'/d['name']
62                 head=run('pin_'+d['name'],['git','rev-parse','HEAD'],cwd=folder).strip()
63                 status=run('clean_'+d['name'],['git','status','--porcelain','--untracked-files=no'],cwd=folder)
64                 if head!=d['rev'] or status.strip(): raise RuntimeError('Dirty or mismatched dependency: '+d['name'])
65                 deps.append({'name':d['name'],'revision':head,'tracked_clean':True})
66             report['dependencies']=deps;save()
67             production=[work/'KomorebiQCOS.lean',*sorted((work/'KomorebiQCOS').glob('*.lean'))]
68             for p in production:
69                 text=p.read_text()
70                 if re.search(r'\b(sorry|admit|sorryAx|native_decide)\b|(?m:^\s*(axiom|unsafe)\s)',text):
71                     raise RuntimeError('Disallowed production token: '+str(p.relative_to(work)))
72             inv=json.loads((ROOT/'docs/DECLARATIONS.json').read_text())
73             audit=['import KomorebiQCOS','']+['#print axioms '+x['name'] for x in inv['declarations']]
74             (work/'AxiomAudit.lean').write_text('\n'.join(audit)+'\n')
75             run('attach',['bash',str(rt/'run.sh'),'attach',str(work)])
76             run('strict_build',['lake','build'])
77             text=run('production_axioms',['lake','env','lean','-j1','-DwarningAsError=true','AxiomAudit.lean'])
78             rows=[{'name':n,'axioms':[a.strip() for a in ax.split(',') if a.strip()]} for n,ax in re.findall(r"'([^']+)' depends on axioms: \[([^\]]*)\]",text)]
79             if {x['name'] for x in rows}!={x['name'] for x in inv['declarations']} or len(rows)!=len(inv['declarations']):
80                 raise RuntimeError('Exact declaration audit inventory mismatch')
81             if any(set(x['axioms'])-ALLOWED for x in rows): raise RuntimeError('Unexpected axiom')
82             (out/'axioms.json').write_text(json.dumps(rows,indent=2)+'\n')
83             run('project_kernel_replay',['lake','env','leanchecker','--verbose','KomorebiQCOS'])
84             for stem in ('Positive','RecordedTilted'):
85                 dest=work/'.lake/build/lib/lean/Controls';dest.mkdir(parents=True,exist_ok=True)
86                 run('control_'+stem,['lake','env','lean','-j1','-DwarningAsError=true','-o',str(dest/(stem+'.olean')),'Controls/'+stem+'.lean'])
87             control_rows=inv['controls']
88             (work/'ControlAxiomAudit.lean').write_text('import Controls.Positive\nimport Controls.RecordedTilted\n'+''.join('#print axioms '+x+'\n' for x in control_rows))
89             text=run('control_axioms',['lake','env','lean','-j1','-DwarningAsError=true','ControlAxiomAudit.lean'])
90             ca=[{'name':n,'axioms':[a.strip() for a in ax.split(',') if a.strip()]} for n,ax in re.findall(r"'([^']+)' depends on axioms: \[([^\]]*)\]",text)]
91             if {x['name'] for x in ca}!=set(control_rows) or any(set(x['axioms'])-ALLOWED for x in ca):
92                 raise RuntimeError('Control axiom inventory mismatch')
93             (out/'control_axioms.json').write_text(json.dumps(ca,indent=2)+'\n')
94             reject=[('RejectFalseArithmetic','unsolved goals'),('RejectMissingBudget','mismatch'),('RejectReversedProbability','unsolved goals'),('RejectMidpointAsWidth','unsolved goals'),('RejectPlaceholder','declaration uses `sorry`')]
95             for stem,message in reject:
96                 run(stem,['lake','env','lean','-j1','-DwarningAsError=true','Controls/'+stem+'.lean'],expected=1,required=message,forbidden=['unknown identifier','object file','failed to synthesize','unexpected token'])
97             correspondence={}
98             for p in production:
99                 rel=p.relative_to(work)
100                 if p.read_bytes()!=(ROOT/'lean'/rel).read_bytes(): raise RuntimeError('Source mutated during replay')
101                 correspondence[str(rel)]=digest(p)
102             (out/'source_hashes.json').write_text(json.dumps(correspondence,indent=2)+'\n')
103             shutil.copytree(work/'.lake/build/lib/lean/KomorebiQCOS',out/'compiled_project/KomorebiQCOS')
104             for ext in ('olean','ilean','olean.private','olean.server'):
105                 p=work/'.lake/build/lib/lean'/('KomorebiQCOS.'+ext)
106                 if p.exists(): shutil.copy2(p,out/'compiled_project'/p.name)
107             report.update(status='PASS',theorems=inv['theorems'],definitions=inv['definitions'],named_instances=inv['instances'],audited_declarations=len(rows),positive_control_theorems=len(control_rows),rejected_controls=len(reject),axiom_union=sorted({a for r in rows for a in r['axioms']}),kernel_scope='Project modules only, against pinned imported environments; same Lean kernel; not an independent verifier or rebuild of all Mathlib.')
108         except Exception as e:
109             report.update(status='FAIL',error=str(e));print('FAILED: '+str(e),flush=True)
110         finally:
111             report['finished_utc']=datetime.datetime.now(datetime.timezone.utc).isoformat();save()
112     return 0 if report['status']=='PASS' else 1
113 if __name__=='__main__':raise SystemExit(main())

Back to source controls

Back to COS Method: formal continuation