finalization.py

1"""Owner-side post-cleanup selection; missing certification always fails closed."""
2import json,hashlib,os,time
3from pathlib import Path
4def sha(path):
5    h=hashlib.sha256()
6    with Path(path).open('rb') as f:
7        for block in iter(lambda:f.read(8*1024**2),b''):h.update(block)
8    return h.hexdigest()
9def digest(v):return hashlib.sha256(json.dumps(v,sort_keys=True,separators=(',',':')).encode()).hexdigest()
10def atomic(path,value):
11    path=Path(path);tmp=path.with_suffix(path.suffix+'.tmp')
12    with tmp.open('w') as f:json.dump(value,f,sort_keys=True);f.flush();os.fsync(f.fileno())
13    os.replace(tmp,path)
14    fd=os.open(path.parent,os.O_RDONLY)
15    try:os.fsync(fd)
16    finally:os.close(fd)
17def checked_ref(value,root):
18    if set(value)!={'path','sha256'}:raise ValueError('checkpoint reference shape')
19    path=Path(value['path']).resolve()
20    if path.parent!=Path(root).resolve() or str(path)!=value['path'] or sha(path)!=value['sha256']:raise ValueError('checkpoint reference integrity')
21    return value
22def finalize(cfg,result):
23    """Called only after supervise returns and its guardian has reaped workers."""
24    root=Path(cfg['output']);baseline=None;reason=None
25    try:
26        base=json.loads((root/'baseline.json').read_text())
27        if base['config_sha256']!=digest(cfg):raise ValueError('baseline identity')
28        baseline=checked_ref(base['checkpoint'],root)
29        if result.get('returncode')!=0 or result.get('deadline_expired') or result.get('supervisor_lost') or result.get('error'):raise ValueError('owned attempt failed')
30        terminal=json.loads((root/'terminal.json').read_text())
31        if terminal['status']!='complete' or terminal['baseline_checkpoint']!=baseline:raise ValueError('terminal baseline mismatch')
32        if terminal['reason'] not in ('single_schedule_complete','original_deadline_soft_stop','retention_guard'):raise ValueError('uncertified stopping reason')
33        selected=checked_ref(terminal['best_checkpoint'],root)
34        if json.loads((root/'best.json').read_text())!=selected:raise ValueError('selected pointer mismatch')
35        if sha(root/'last.pt')!=terminal['last_sha256']:raise ValueError('last checkpoint mismatch')
36        if terminal['reason']=='retention_guard' and selected!=baseline:raise ValueError('guard did not roll back')
37    except Exception as error:reason=str(error)
38    if reason is not None:
39        if baseline is not None:atomic(root/'best.json',baseline)
40        receipt={'status':'failed','config_sha256':digest(cfg),'result':result,'reason':reason,'selected_fallback':baseline,'unix':time.time(),'evaluation_allowed':False}
41        atomic(root/'failure.json',receipt);atomic(root/'completion.json',receipt)
42        return receipt
43    receipt={'status':'certified','config_sha256':digest(cfg),'result':result,'terminal_sha256':sha(root/'terminal.json'),'selected_checkpoint':selected,'baseline_checkpoint':baseline,'unix':time.time(),'evaluation_allowed':True}
44    atomic(root/'completion.json',receipt);return receipt
45def require_certified(cfg):
46    root=Path(cfg['output']);receipt=json.loads((root/'completion.json').read_text())
47    if receipt.get('status')!='certified' or receipt.get('evaluation_allowed') is not True or receipt['config_sha256']!=digest(cfg):raise ValueError('no certified completion')
48    if receipt['result'].get('returncode')!=0 or receipt['result'].get('deadline_expired') or receipt['result'].get('supervisor_lost'):raise ValueError('failed supervision')
49    if sha(root/'terminal.json')!=receipt['terminal_sha256']:raise ValueError('terminal receipt changed')
50    ref=checked_ref(receipt['selected_checkpoint'],root)
51    if json.loads((root/'best.json').read_text())!=ref:raise ValueError('selection changed')
52    return ref
53