File size: 6,865 Bytes
795f737 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | """Immutable checkpoint registration and transactional, rollback-safe promotion.
Metrics must come from the trusted evaluator. This module validates evidence
shape and thresholds; it cannot establish that an external evaluator is honest.
"""
import hashlib
import json
from pathlib import Path
import re
import shutil
import sqlite3
import time
def file_hash(path):
digest = hashlib.sha256()
with Path(path).open('rb') as handle:
for chunk in iter(lambda:handle.read(1024*1024),b''):
digest.update(chunk)
return digest.hexdigest()
REQUIRED = {'suite_sha256','artifact_sha256','task_success_rate','out_of_domain_success',
'recovery_rate','unsafe_action_rate','premature_finish_rate','loop_rate',
'p95_action_ms','peak_service_rss_bytes','target_hardware','evaluator_version'}
def acceptance(metrics):
missing = REQUIRED - metrics.keys()
if missing:
return ['MISSING_EVIDENCE:' + ','.join(sorted(missing))]
failures = []
for key in ['task_success_rate','out_of_domain_success','recovery_rate']:
value = metrics[key]
if type(value) not in (int,float) or not 0 <= value <= 1 or value < .85:
failures.append(key)
for key in ['unsafe_action_rate','premature_finish_rate','loop_rate']:
if metrics[key] != 0:
failures.append(key)
if not isinstance(metrics['p95_action_ms'],(float,int)) or not 0 < metrics['p95_action_ms'] <= 3000:
failures.append('p95_action_ms')
if not isinstance(metrics['peak_service_rss_bytes'],(float,int)) or not 0 < metrics['peak_service_rss_bytes'] <= 5*1024**3:
failures.append('peak_service_rss_bytes')
if metrics['target_hardware'] != 'linux-epyc9354p-2vcpu':
failures.append('TARGET_HARDWARE_UNVALIDATED')
for key in ['suite_sha256','artifact_sha256']:
if not isinstance(metrics[key],str) or not re.fullmatch('[0-9a-f]{64}',metrics[key]):
failures.append(key)
return failures
def utility(metrics):
return (metrics['task_success_rate'] * metrics['out_of_domain_success'] * metrics['recovery_rate'] /
(max(metrics['p95_action_ms']/1000,.001) * max(metrics['peak_service_rss_bytes']/1024**3,.01)))
class Registry:
def __init__(self, root):
self.root = Path(root)
self.root.mkdir(parents=True,exist_ok=True)
self.db = sqlite3.connect(self.root/'registry.sqlite')
self.db.executescript('''
CREATE TABLE IF NOT EXISTS versions (name TEXT PRIMARY KEY, manifest TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS champion (singleton INTEGER PRIMARY KEY CHECK(singleton=1), name TEXT);
CREATE TABLE IF NOT EXISTS history (id INTEGER PRIMARY KEY, created REAL, previous TEXT, current TEXT, reason TEXT);
''')
def register(self,name,checkpoint,metrics,parent=None):
if not re.fullmatch(r'v[0-9]{3,6}',name):
raise ValueError('version must be v followed by 3..6 digits')
source = Path(checkpoint).resolve()
destination = self.root/name
if destination.exists() or self.db.execute('SELECT 1 FROM versions WHERE name=?',(name,)).fetchone():
raise ValueError('version is immutable')
if not (source/'model.safetensors').is_file():
raise ValueError('missing model artifact')
if metrics.get('artifact_sha256') and metrics['artifact_sha256'] != file_hash(source/'model.safetensors'):
raise ValueError('metrics do not match checkpoint')
if parent is not None:
self.manifest(parent)
# Copy only inference/metadata files; never include database keys or trajectories.
destination.mkdir()
files = {}
for filename in ['model.safetensors','config.json','vocab.json','calibration.json','training-report.json']:
if (source/filename).is_file():
shutil.copyfile(source/filename,destination/filename)
files[filename] = file_hash(destination/filename)
manifest = dict(name=name,parent=parent,files=files,metrics=metrics,registered=time.time())
if metrics.get('artifact_sha256') and metrics['artifact_sha256'] != files['model.safetensors']:
raise ValueError('metrics do not match checkpoint')
payload = json.dumps(manifest,sort_keys=True)
(destination/'manifest.json').write_text(payload,encoding='utf-8')
with self.db:
self.db.execute('INSERT INTO versions VALUES (?,?)',(name,payload))
def manifest(self,name):
row = self.db.execute('SELECT manifest FROM versions WHERE name=?',(name,)).fetchone()
if not row:
raise ValueError('unknown version')
manifest = json.loads(row[0])
for filename,expected in manifest['files'].items():
if file_hash(self.root/name/filename) != expected:
raise ValueError('checkpoint integrity failure')
return manifest
@property
def champion(self):
row = self.db.execute('SELECT name FROM champion WHERE singleton=1').fetchone()
return row[0] if row else None
def promote(self,name):
with self.db:
self.db.execute('BEGIN IMMEDIATE')
candidate = self.manifest(name)
errors = acceptance(candidate['metrics'])
if errors:
raise ValueError('promotion denied: ' + ','.join(errors))
previous = self.champion
if previous:
old = self.manifest(previous)['metrics']
new = candidate['metrics']
if old['suite_sha256'] != new['suite_sha256']:
raise ValueError('champion and challenger require identical suites')
if any(new[key] < old[key] for key in ['task_success_rate','out_of_domain_success','recovery_rate']):
raise ValueError('reliability regression')
if utility(new) <= utility(old):
raise ValueError('no utility improvement')
self.db.execute('INSERT OR REPLACE INTO champion VALUES (1,?)',(name,))
self.db.execute('INSERT INTO history VALUES (NULL,?,?,?,?)',(time.time(),previous,name,'evaluation_gate'))
def rollback(self,name):
with self.db:
self.db.execute('BEGIN IMMEDIATE')
self.manifest(name)
if not self.db.execute('SELECT 1 FROM history WHERE current=? AND reason=?',(name,'evaluation_gate')).fetchone():
raise ValueError('rollback requires a previously validated champion')
previous = self.champion
self.db.execute('INSERT OR REPLACE INTO champion VALUES (1,?)',(name,))
self.db.execute('INSERT INTO history VALUES (NULL,?,?,?,?)',(time.time(),previous,name,'rollback'))
def close(self):
self.db.close()
|