owner.py

1"""Explicitly authorized finite-schedule completion, separate bounded admission."""
2import json,os,sys,fcntl
3from pathlib import Path
4from deadline import create_or_resume,supervise
5from finalization import sha,digest,atomic,finalize,require_certified
6
7def validate(cfg,approval):
8 if cfg.get('draft_only') or cfg['budget_seconds']!=2400 or cfg['checkpoint_margin_seconds']!=120:raise ValueError('reviewed forty-minute completion cap')
9 if approval!={'approved':True,'config_sha256':digest(cfg),'source_checkpoint_sha256':cfg['source_checkpoint_sha256']}:raise ValueError('exact completion review required')
10 root=Path(cfg['output']).resolve();shared=Path(cfg['original_root']).resolve()
11 if str(root)!=cfg['output'] or root.parent!=shared or root.name!='completion-v1':raise ValueError('canonical shared ownership')
12 for p,h in cfg['immutable_source_sha256'].items():
13  if sha(p)!=h:raise ValueError('source changed: '+p)
14 source_cfg=json.loads(Path(cfg['original_config']).read_text());selected=require_certified(source_cfg)
15 audit=json.loads(Path(cfg['source_audit']).read_text())
16 if audit['status']!='verified' or audit['owned_remaining'] or selected!=cfg['prior_selected'] or audit['decision_exposures']!=cfg['source_cursor'] or audit['updates']!=cfg['source_updates']:raise ValueError('source audit/state mismatch')
17 if not 128000<cfg['source_cursor']<1713765:raise ValueError('invalid remaining finite schedule')
18 old_dir=Path(cfg['original_config']).parent
19 paths={str(old_dir/n).encode() for n in ['owner.py','train.py','deadline.py']}
20 for proc in Path('/proc').iterdir():
21  if not proc.name.isdigit() or int(proc.name)==os.getpid():continue
22  try:argv=set((proc/'cmdline').read_bytes().split(b'\0'))
23  except OSError:continue
24  if argv&paths:raise ValueError('source processes remain')
25 return shared,root
26
27def run(cp,ap):
28 cfg=json.loads(Path(cp).read_text());approval=json.loads(Path(ap).read_text());shared,root=validate(cfg,approval)
29 with (shared/'run.lock').open('a+b') as lock:
30  fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB);shared,root=validate(cfg,approval)
31  with (shared/'completion-v1-admitted.json').open('x') as f:json.dump(approval,f);f.flush();os.fsync(f.fileno())
32  root.mkdir(exist_ok=True);dp=root/'deadline.json'
33  if dp.exists() or (root/'terminal.json').exists():raise ValueError('no automatic retry or clock reset')
34  identity={'run_id':cfg['run_id'],'config_sha256':digest(cfg),'deadline_path':str(dp),'base_command':cfg['command']}
35  clock=create_or_resume(dp,identity,budget=2400,margin=120)
36  atomic(root/'attempt.json',{'approval':approval,'deadline':clock,'source_cursor':cfg['source_cursor'],'original_deadline_preserved':cfg['source_deadline']})
37  result=supervise(cfg['command'],clock,dict(os.environ,CHESS_JOB_DEADLINE=str(dp),CHESS_JOB_LOCK_FD=str(lock.fileno())))
38  atomic(root/'supervision.json',result);completion=finalize(cfg,result)
39  if completion['status']!='certified' and result['returncode']==0:result=dict(result,returncode=125)
40  return result
41if __name__=='__main__':
42 if len(sys.argv)!=3:raise SystemExit('config approval required')
43 r=run(*sys.argv[1:]);raise SystemExit(r['returncode'] or (124 if r['deadline_expired'] else 0))
44