runner.py

1"""Single bounded local head-to-head attempt; review required before launch."""
2import os,sys,json,hashlib,time,uuid,importlib.util,fcntl
3from pathlib import Path
4H=Path(__file__).resolve().parent;P=H.parent
5sys.path[:0]=[str(H),str(P)]
6import importlib.util
7if importlib.util.find_spec('chess') is None:
8    package=Path('/REDACTED')
9    spec=importlib.util.spec_from_file_location('chess',package/'__init__.py',submodule_search_locations=[str(package)])
10    module=importlib.util.module_from_spec(spec);sys.modules['chess']=module;spec.loader.exec_module(module)
11GUARD=P/'scaleup-v1/runtime/training_deadline.py'
12
13def sha(p):
14    h=hashlib.sha256()
15    with Path(p).open('rb') as f:
16        while b:=f.read(8*1024*1024):h.update(b)
17    return h.hexdigest()
18def read(p):return json.loads(Path(p).read_text())
19def write(p,v):
20    Path(p).write_text(json.dumps(v,indent=2) if not isinstance(v,str) else v)
21def inputs():
22    c=read(H/'execution-contract.json');s=read(H/'suite.json');p=read(H/'proposal.json')
23    if sha(H/'execution-contract.json')!=p['execution_contract_sha256'] or sha(H/'suite.json')!=p['suite_sha256']:raise ValueError('contract/suite changed')
24    for path,value in c['source_hashes'].items():
25        if sha(path)!=value:raise ValueError('source changed: '+path)
26    original=read(P/'scaleup-v1/evaluation-suite/frozen-v1/suite.json')
27    if s['schedule']!=original['schedule'] or s['openings']!=original['openings'] or len(s['schedule'])!=100:raise ValueError('frozen schedule changed')
28    if s['protocol']!=c['protocol'] or p['players']!=c['players'] or p['budget_seconds']!=14400:raise ValueError('protocol/players/budget')
29    return s,c,p
30def guardian():
31    spec=importlib.util.spec_from_file_location('head_guardian',GUARD);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m);return m
32def require_guardian():
33    if os.environ.get('CHESS_MATCH_GUARDED')!='1' or os.getpgrp()!=os.getpid():raise RuntimeError('guardian required')
34    args=Path(f'/proc/{os.getppid()}/cmdline').read_bytes().split(b'\0')
35    if str(GUARD.resolve()).encode() not in args or b'--guardian' not in args:raise RuntimeError('wrong guardian parent')
36def model_pair(c,seed,clock,factory):
37    """Close the first worker if second load/admission fails."""
38    models={}
39    try:
40        from resource_guard import check
41        for name in ('CM-7','CM-8'):
42            remaining=min(60.,clock.remaining()-5)
43            if remaining<=0:raise TimeoutError('no model load budget')
44            models[name]=factory('adapter:load',Path(c['players'][name]['path']),seed,remaining)
45            check(force=True)
46        return models
47    except BaseException:
48        for model in models.values():model.close()
49        raise
50def play(s,c,out,clock,factory=None):
51    import harness as h
52    from head_game import game,audit,summarize
53    from resource_guard import check
54    factory=factory or h.ModelWorker;rows=[];games=[];opens={o['opening_id']:o for o in s['openings']}
55    write(out/'summary.json',summarize(rows))
56    for item in s['schedule']:
57        if clock.soft_stop():break
58        models={};folder=out/item['game_id'];opening=opens[item['opening_id']]
59        try:
60            before=time.monotonic();models=model_pair(c,item['seed'],clock,factory);load_seconds=time.monotonic()-before
61            remaining=min(180.,clock.remaining()-5)
62            if remaining<=0:break
63            report=game(models,item['model_color']=='white',folder,opening['uci'],walltime=remaining,guard=check)
64            parsed,row=audit(folder,report,opening,item['model_color']=='white')
65            row.update(item,model_load_seconds=load_seconds)
66            parsed.headers.update(OpeningID=opening['opening_id'],AutoplayPlies=str(opening['autoplay_plies']),SourceOpeningName=opening['name'],CM7SHA256=c['players']['CM-7']['sha256'],CM8SHA256=c['players']['CM-8']['sha256'])
67            write(folder/'game.pgn',str(parsed)+'\n')
68            row.update(pgn_sha256=sha(folder/'game.pgn'),game_json_sha256=sha(folder/'game.json'))
69            rows.append(row);games.append(str(parsed));write(folder/'audit.json',row)
70            write(out/'results.json',rows);write(out/'summary.json',summarize(rows));write(out/'games.pgn','\n\n'.join(games)+'\n')
71            if report['error']:break
72        finally:
73            for model in models.values():model.close()
74    return rows
75def worker(request_path):
76    require_guardian();s,c,p=inputs();request=read(request_path);out=request_path.parent
77    expected=dict(p,approved=True)
78    if request!=expected or out!=(H/'runs'/p['run_id']).resolve():raise ValueError('worker request')
79    g=guardian();value=g.read_deadline(out/'deadline.json')
80    if value['identity']!=request or value['budget_seconds']!=14400 or value['checkpoint_margin_seconds']!=30:raise ValueError('worker clock')
81    clock=g.Deadline(value)
82    with (out/'worker-started.json').open('x') as f:json.dump({'pid':os.getpid(),'unix':time.time()},f);f.flush();os.fsync(f.fileno())
83    import torch,chess,transformers
84    import harness as h
85    from resource_guard import check,baseline
86    from audit_helpers import model_metrics
87    os.environ['CHESS_INFERENCE_DEVICE']='cuda:0';baseline(out)
88    write(out/'manifest.json',{'request':request,'protocol':s['protocol'],'players':c['players'],'source_hashes':c['source_hashes'],'torch':str(torch.__version__),'transformers':transformers.__version__,'python_chess':chess.__version__,'python':sys.version,'model_reload_each_game':True})
89    try:
90        models=model_pair(c,20261030,clock,h.ModelWorker)
91        try:
92            checks=[];board=chess.Board()
93            for _ in range(13):
94                for name in ('CM-7','CM-8'):
95                    time.sleep(.25);check(force=True)
96                    models[name].timeout=min(60.,clock.remaining()-5)
97                    if models[name].timeout<=0:raise TimeoutError('probe deadline')
98                    move,metrics=models[name].choose(board)
99                    if move not in board.legal_moves:raise ValueError('probe illegal move')
100                    model_metrics(dict(metrics,model_query='queried'),board)
101                    checks.append({'player':name,'move':move.uci(),'health':check(force=True)})
102            write(out/'joint-probe.json',{'both_models_resident':True,'checks':checks,'players':c['players'],'physical_gpu':1})
103        finally:
104            for model in models.values():model.close()
105        rows=play(s,c,out,clock)
106        write(out/'terminal.json',{'status':'all_scheduled_audited' if len(rows)==100 and not any(r['error'] for r in rows) else 'partial_or_error','audited_games':len(rows),'unix':time.time()})
107    except BaseException as exc:write(out/'failure.json',{'error':repr(exc),'unix':time.time()});raise
108def launch():
109    s,c,p=inputs();approval=read(H/'approval.json')
110    if approval!=dict(p,approved=True):raise ValueError('exact release required')
111    if str(uuid.UUID(p['run_id']))!=p['run_id']:raise ValueError('run UUID')
112    for ref in c['players'].values():
113        if str(Path(ref['path']).resolve())!=ref['path'] or sha(ref['path'])!=ref['sha256']:raise ValueError('checkpoint changed')
114    from resource_guard import check
115    check(force=True)
116    out=(H/'runs'/p['run_id']).resolve()
117    if p['run_directory']!=str(out):raise ValueError('run directory')
118    out.parent.mkdir(exist_ok=True);out.mkdir(exist_ok=False)
119    with (out/'owner.lock').open('x') as lock:
120        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
121        g=guardian();value=g.create_or_resume(out/'deadline.json',approval,budget=14400,margin=30);write(out/'request.json',approval)
122        env=dict(os.environ,CHESS_JOB_LOCK_FD=str(lock.fileno()),CHESS_MATCH_GUARDED='1')
123        result=g.supervise([sys.executable,str(H/'runner.py'),'--worker',str(out/'request.json')],value,env)
124        write(out/'supervision.json',result)
125        return result
126def main():
127    if len(sys.argv)==3 and sys.argv[1]=='--worker':
128        path=Path(sys.argv[2]).resolve()
129        if path.name!='request.json' or path.parent.parent!=(H/'runs').resolve():raise ValueError('worker path')
130        with (path.parent/'worker.log').open('a',buffering=1) as f:os.dup2(f.fileno(),1);os.dup2(f.fileno(),2);worker(path)
131    elif sys.argv[1:]==['--authorize-matches']:
132        result=launch()
133        if result['returncode'] or result['deadline_expired']:raise SystemExit(result['returncode'] or 124)
134    else:
135        s,c,p=inputs();print(json.dumps({'status':'plan_only','games':len(s['schedule']),'players':c['players'],'proposal_sha256':sha(H/'proposal.json')}))
136if __name__=='__main__':main()
137