owner.py

1"""Eight-hour canonical run admission; explicit resume never creates a budget."""
2import json,hashlib,os,sys,fcntl
3from pathlib import Path
4from deadline import create_or_resume,supervise,Deadline
5def digest(v):return hashlib.sha256(json.dumps(v,sort_keys=True,separators=(',',':')).encode()).hexdigest()
6def validate(cfg,approval):
7    if cfg['budget_seconds']!=28800 or cfg['checkpoint_margin_seconds']!=180:raise ValueError('eight-hour immutable cap')
8    root=Path(cfg['output']).resolve()
9    if str(root)!=cfg['output']:raise ValueError('canonical output required')
10    resume=approval.get('resume')
11    if approval!={'approved':True,'config_sha256':digest(cfg),'resume':resume}:raise ValueError('exact config review required')
12    if resume is not None:
13        if set(resume)!={'path','sha256'} or str(Path(resume['path']).resolve())!=resume['path']:raise ValueError('canonical bound resume checkpoint')
14    if not cfg.get('sealed_dataset_manifest_sha256') or not cfg.get('schedule_sha256') or not cfg.get('runtime_sha256'):raise ValueError('sealed data/schedule/runtime required')
15    return root,resume
16def run(config_path,approval_path):
17    cfg=json.loads(Path(config_path).read_text());approval=json.loads(Path(approval_path).read_text());root,resume=validate(cfg,approval)
18    root.mkdir(parents=True,exist_ok=True);deadline_path=root/'deadline.json'
19    with (root/'run.lock').open('a+b') as lock:
20        fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
21        if (root/'terminal.json').exists():raise ValueError('terminal run cannot resume')
22        if resume is not None and not deadline_path.exists():raise ValueError('missing original deadline; fresh resume budget forbidden')
23        if resume is None and deadline_path.exists():raise ValueError('existing run requires explicit reviewed resume')
24        if resume is not None and not Path(resume['path']).is_file():raise ValueError('missing resume checkpoint')
25        identity={'run_id':cfg['run_id'],'config_sha256':digest(cfg),'deadline_path':str(deadline_path),'base_command':cfg['command']}
26        value=create_or_resume(deadline_path,identity,budget=28800,margin=180)
27        if Deadline(value).soft_stop():raise TimeoutError('original deadline exhausted')
28        admission=digest(approval);receipt=root/('attempt-'+admission+'.json')
29        # The guarded job must verify sealed/runtime/checkpoint hashes before GPU
30        # initialization; that expensive startup is inside the same clock.
31        command=cfg['command']+([] if resume is None else ['--resume',resume['path'],'--resume-sha256',resume['sha256']])
32        with receipt.open('x') as f:
33            json.dump({'approval':approval,'command':command,'original_deadline':value},f);f.flush();os.fsync(f.fileno())
34        env=dict(os.environ,CHESS_JOB_DEADLINE=str(deadline_path),CHESS_JOB_LOCK_FD=str(lock.fileno()))
35        result=supervise(command,value,env)
36        path=root/('attempt-'+admission+'-result.json');path.write_text(json.dumps(result))
37        return result
38if __name__=='__main__':
39    if len(sys.argv)!=3:raise SystemExit('exact config and review approval paths required')
40    result=run(*sys.argv[1:]);raise SystemExit(result['returncode'] or (124 if result['deadline_expired'] else 0))
41