Digital Drosophila Archive
228 lines
14 KiB
Python
228 lines
14 KiB
Python
"""Post-completion ONLY: fresh terminal inventory, fixed science, historical seals.
|
|
|
|
No training, repair, retry or change to the registered model/experiment. Output
|
|
goes to an independently provisioned EMPTY sibling directory. The real normal
|
|
runner's final ledger report is retained unchanged, even though a subsequent
|
|
full-tree audit sees its separately authenticated completion report as extra.
|
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
from progress_normal import safe, read_json, require
|
|
|
|
ROOT = Path('/data/digital-drosophila')
|
|
MODEL = ROOT/'models/male-cns-lif/normal-conditioning-v1'
|
|
EXP = ROOT/'experiments/phase4ar-normal-conditioning'
|
|
CONTROL = ROOT/'logs/phase4ar-normal-conditioning'
|
|
OUT = ROOT/'experiments/phase4ar-normal-conditioning-final-audit'
|
|
TOOLS = Path(__file__).resolve().parent
|
|
PYTHON = ROOT/'models/male-cns-lif/.venv/bin/python'
|
|
REG_SHA = 'b63e6281b6505ca3b60acb467dd0b3fae274cf513dcbfd653649ba52e9ba9740'
|
|
START_PINS = {
|
|
'launch-attempt.json': dict(bytes=229,sha256='93274a58b83a3e508e378f3972a62c2e43a461bc4a890314f6e38530fa05719c'),
|
|
'launch.json': dict(bytes=523,sha256='94c7dba031149451a0d653dd7c083fa5c918b9363fa1caef31b988d02d72672e'),
|
|
'supervisor-start.json': dict(bytes=187,sha256='3fa980cfcf0993bb7aa29de87eff31f5bc48be215a2cfbc03b715d4b84672839'),
|
|
}
|
|
|
|
def identity(path):
|
|
p=safe(path);require(p.is_file(),'Missing regular artifact: '+str(p))
|
|
size=0;h=hashlib.sha256()
|
|
with p.open('rb') as f:
|
|
for part in iter(lambda:f.read(8<<20),b''):size+=len(part);h.update(part)
|
|
return dict(bytes=size,sha256=h.hexdigest())
|
|
|
|
def read(path):
|
|
return read_json(path)[0]
|
|
|
|
def save(path,value):
|
|
with safe(path).open('x',encoding='utf-8',newline='\n') as f:
|
|
json.dump(value,f,sort_keys=True,indent=2,ensure_ascii=False,allow_nan=False);f.write('\n')
|
|
|
|
def verify(pins):
|
|
for p,sha in pins.items():require(identity(p)['sha256']==sha,'Actual dependency changed: '+str(p))
|
|
|
|
def live_runtime_rows(ps_text,model,exp):
|
|
"""Exact argv tokens, not substring matching or a missing-PID success rule."""
|
|
runtime={str(model/n) for n in ('run_normal.py','normal_runner.py','normal_inputs.py')}
|
|
worker=str(model.parent/'conditioning-interface-v2/isolated_workers.py')
|
|
matches=[]
|
|
for line in ps_text.splitlines():
|
|
fields=line.strip().split(None,1)
|
|
if len(fields)!=2:continue
|
|
try:pid=int(fields[0]);argv=shlex.split(fields[1])
|
|
except (ValueError,TypeError):continue
|
|
if not argv:continue
|
|
owned=bool(runtime.intersection(argv))
|
|
if worker in argv and '--job' in argv:
|
|
i=argv.index('--job')+1
|
|
owned=owned or (i<len(argv) and Path(argv[i]).is_relative_to(exp/'jobs'))
|
|
# Also reject a direct independent CPU-update verifier for this run.
|
|
validator=str(model.parent/'conditioning-execution-v1/validate_update_cpu.py')
|
|
if validator in argv and '--authority' in argv:
|
|
i=argv.index('--authority')+1
|
|
owned=owned or (i<len(argv) and Path(argv[i]).is_relative_to(exp/'jobs'))
|
|
if owned:matches.append(dict(pid=pid,argv=argv))
|
|
return matches
|
|
|
|
def process_census(model,exp):
|
|
require(sys.platform.startswith('linux'),'Actual Linux process census required')
|
|
result=subprocess.run(['ps','-eo','pid=,args='],capture_output=True,text=True,timeout=15,check=True)
|
|
rows=live_runtime_rows(result.stdout,model,exp)
|
|
return dict(observed_utc=datetime.now(timezone.utc).isoformat(),matching_live_processes=rows,
|
|
actual_OS_query=True,missing_PID_alone_is_not_completion=True)
|
|
|
|
def readiness(*,exp=EXP,control=CONTROL,model=MODEL,regsha=REG_SHA,start_pins=START_PINS,census=None):
|
|
"""Small read-only guard. Refuses live/incomplete runs before any output write."""
|
|
regpath=exp/'protocol/preregistered.json'
|
|
require(identity(regpath)['sha256']==regsha,'Registration differs')
|
|
reg=read(regpath)
|
|
require(reg.get('schema')=='normal-conditioning-registration-v1' and
|
|
reg.get('scientific_scope')=='normal-acquisition-only' and reg.get('subsequent_assays_authorized') is False,'Wrong authority')
|
|
require(not (exp/'audit/stopped.json').exists(),'Stopped run requires a separate failure audit, never retry')
|
|
for name,info in start_pins.items():require(identity(control/name)==info,'Startup record changed')
|
|
observed=(process_census(model,exp) if census is None else census())
|
|
require(observed.get('actual_OS_query') is True,'Actual process census missing')
|
|
require(observed.get('matching_live_processes')==[],'Registered runtime is still live; terminal audit forbidden')
|
|
require((control/'exit.json').is_file() and (exp/'audit/execution-complete.json').is_file(),
|
|
'No complete real terminal receipts; missing process alone proves nothing')
|
|
exited=read(control/'exit.json');completion=read(exp/'audit/execution-complete.json')
|
|
expected=[str(PYTHON),'-B',str(model/'normal_runner.py'),'--registration-sha256',regsha]
|
|
require(exited.get('registration_sha256')==regsha and exited.get('argv')==expected and
|
|
type(exited.get('exit_code')) is int and exited['exit_code']==0 and exited.get('final_exit_observed') is True and
|
|
exited.get('complete') is True and exited.get('error') is None and exited.get('learning_success_claimed') is False and
|
|
exited.get('scientific_analysis_required') is True,'Parent did not record exact successful run exit')
|
|
require(completion.get('complete') is True and completion.get('registration_sha256')==regsha and
|
|
completion.get('inventory')==reg.get('normal_inventory') and completion.get('learning_success_claimed') is False,
|
|
'Actual execution completion binding differs')
|
|
return dict(ready=True,registration_sha256=regsha,process_census=observed,
|
|
exit_identity=identity(control/'exit.json'),completion_identity=identity(exp/'audit/execution-complete.json'),
|
|
scientific_success_inferred=False)
|
|
|
|
def source_identity():
|
|
commit=subprocess.check_output(['git','-C',str(ROOT),'rev-parse','HEAD']).decode().strip()
|
|
files=sorted(TOOLS.glob('*.py'));require(files,'Audit source directory empty')
|
|
result={}
|
|
for p in files:
|
|
info=identity(p)
|
|
blob=subprocess.check_output(['git','-C',str(ROOT),'show',commit+':'+p.relative_to(ROOT).as_posix()])
|
|
require(hashlib.sha256(blob).hexdigest()==info['sha256'],'Uncommitted terminal audit source: '+str(p))
|
|
result[str(p)]=info
|
|
return dict(commit=commit,files=result)
|
|
|
|
def child(name,argv,timeout):
|
|
"""Exclusive real subprocess receipts; audit failure is not a learning FAIL."""
|
|
save(OUT/(name+'-start.json'),dict(argv=argv,started_utc=datetime.now(timezone.utc).isoformat(),automatic_retry=False))
|
|
start=time.perf_counter();rc=None;timed=False;error=None
|
|
with (OUT/(name+'.stdout.log')).open('xb') as stdout,(OUT/(name+'.stderr.log')).open('xb') as stderr:
|
|
try:
|
|
rc=subprocess.run(argv,stdout=stdout,stderr=stderr,timeout=timeout,
|
|
env={**os.environ,'CUDA_VISIBLE_DEVICES':'','PYTHONDONTWRITEBYTECODE':'1'}).returncode
|
|
except subprocess.TimeoutExpired:timed=True
|
|
except OSError as exc:error=str(exc)
|
|
receipt=dict(argv=argv,exit_code=rc,timed_out=timed,error=error,actual_parent_observation=True,
|
|
complete=type(rc)is int and rc==0 and not timed,wall_seconds=time.perf_counter()-start,
|
|
stdout=identity(OUT/(name+'.stdout.log')),stderr=identity(OUT/(name+'.stderr.log')))
|
|
save(OUT/(name+'-exit.json'),receipt)
|
|
require(receipt['complete'],'Audit child failed; preserve output and do not retry')
|
|
return receipt
|
|
|
|
def analyze_child(inventory_sha,terminal_sha):
|
|
require(sys.dont_write_bytecode and Path(sys.executable)==PYTHON,'Fresh exact CPU audit interpreter required')
|
|
readiness()
|
|
require(identity(OUT/'verified-inventory.json')['sha256']==inventory_sha and
|
|
identity(OUT/'terminal-before.json')['sha256']==terminal_sha,'Actual outer inventory authority changed')
|
|
reg=read(EXP/'protocol/preregistered.json');verify(reg['source_sha256'])
|
|
inventory=json.loads((OUT/'verified-inventory.json').read_text())
|
|
terminal=json.loads((OUT/'terminal-before.json').read_text())
|
|
require(terminal['terminal_reconciliation_pass'] is True and terminal['registration_sha256']==REG_SHA,
|
|
'Terminal reconciliation missing')
|
|
sys.path.insert(0,str(MODEL))
|
|
import normal_metrics
|
|
result=normal_metrics.analyze(EXP,REG_SHA,verified_inventory=inventory,final_ledger=terminal['final_ledger'])
|
|
save(OUT/'analysis.json',result)
|
|
|
|
def seal_output():
|
|
require(not (OUT/'manifest.json').exists() and not (OUT/'checksums.sha256').exists(),'Audit already sealed')
|
|
files={}
|
|
for p in sorted(OUT.rglob('*')):
|
|
safe(p)
|
|
if p.is_file():files[p.relative_to(OUT).as_posix()]=identity(p)
|
|
else:require(p.is_dir(),'Nonregular audit output')
|
|
save(OUT/'manifest.json',dict(schema='normal-terminal-audit-seal-v1',root=str(OUT),files=files,
|
|
referenced_experiment_modified=False,candidate_generated=False))
|
|
files['manifest.json']=identity(OUT/'manifest.json')
|
|
with (OUT/'checksums.sha256').open('x',encoding='utf-8',newline='\n') as f:
|
|
for name,info in sorted(files.items()):f.write(info['sha256']+' '+name+'\n')
|
|
for name,info in files.items():require(identity(OUT/name)==info,'Audit seal reread differs')
|
|
actual=set()
|
|
for p in OUT.rglob('*'):
|
|
safe(p)
|
|
if p.is_file():actual.add(p.relative_to(OUT).as_posix())
|
|
else:require(p.is_dir(),'Nonregular seal artifact')
|
|
require(actual==set(files)|{'checksums.sha256'},'Extra or missing audit file at seal boundary')
|
|
return dict(manifest=identity(OUT/'manifest.json'),checksums=identity(OUT/'checksums.sha256'),files=len(files))
|
|
|
|
def finalize():
|
|
require(sys.dont_write_bytecode and Path(sys.executable)==PYTHON,'Exact registered -B science interpreter required')
|
|
ready=readiness() # no writes and no large reads if run is still live
|
|
require(safe(OUT).is_dir() and not any(OUT.iterdir()),'Exact independently provisioned empty audit root required')
|
|
reserve=read(EXP/'protocol/preregistered.json')['budget']['minimum_free_reserve_bytes']
|
|
require(shutil.disk_usage(OUT).free>=reserve+(4<<30),'No space for terminal audit while preserving registered disk reserve')
|
|
sources=source_identity();save(OUT/'source-identity.json',sources);save(OUT/'readiness-before.json',ready)
|
|
try:
|
|
from terminal_ledger import terminal_inventory
|
|
from normal_outcome import summarize_normal_analysis, analysis_digest
|
|
terminal=terminal_inventory(EXP,REG_SHA,MODEL,supervisor_dir=CONTROL,supervisor_start_pins=START_PINS)
|
|
inventory=terminal.pop('verified_inventory')
|
|
save(OUT/'terminal-before.json',terminal);save(OUT/'verified-inventory.json',inventory)
|
|
command=[str(PYTHON),'-B',str(TOOLS/'normal_finalize.py'),'analyze-child',
|
|
'--inventory-sha256',identity(OUT/'verified-inventory.json')['sha256'],
|
|
'--terminal-sha256',identity(OUT/'terminal-before.json')['sha256']]
|
|
observed=child('analysis',command,43200)
|
|
analysis_sha=identity(OUT/'analysis.json')['sha256'];analysis=json.loads((OUT/'analysis.json').read_text())
|
|
child('historical-integrity',[str(PYTHON),'-B',str(MODEL/'normal_prepare.py'),'integrity',
|
|
'--output',str(OUT/'historical-integrity-after.json')],1800)
|
|
historical=json.loads((OUT/'historical-integrity-after.json').read_text())
|
|
require(historical.get('complete') is True,'Historical integrity failed')
|
|
ready_after=readiness();save(OUT/'readiness-after.json',ready_after)
|
|
require(ready_after['exit_identity']==ready['exit_identity'] and ready_after['completion_identity']==ready['completion_identity'],
|
|
'Terminal control bytes changed while analyzing')
|
|
final=terminal_inventory(EXP,REG_SHA,MODEL,supervisor_dir=CONTROL,supervisor_start_pins=START_PINS)
|
|
require(final.pop('verified_inventory')==inventory,'Actual full inventory changed during science audit')
|
|
require(final==terminal,'Terminal audit changed during science analysis')
|
|
save(OUT/'terminal-after.json',final)
|
|
for p,info in sources['files'].items():require(identity(p)==info,'Terminal source changed during audit')
|
|
require(identity(OUT/'analysis.json')['sha256']==analysis_sha,'Actual analysis bytes changed')
|
|
# This outer binding contains actual byte/exit verification, never only a caller's PASS Boolean.
|
|
binding=dict(registration_sha256=REG_SHA,terminal_reconciliation_pass=True,historical_integrity_after_pass=True,
|
|
analysis_sha256=analysis_digest(analysis),analysis_file_sha256=analysis_sha,
|
|
actual_parent_exit_code=observed['exit_code'],
|
|
analyzed_source_sha256=read(EXP/'protocol/preregistered.json')['source_sha256'][str(MODEL/'normal_metrics.py')])
|
|
outcome=summarize_normal_analysis(analysis,binding)
|
|
save(OUT/'outer-evidence.json',binding);save(OUT/'final-status.json',outcome)
|
|
return dict(outcome=outcome,seal=seal_output())
|
|
except BaseException as exc:
|
|
if not (OUT/'audit-stopped.json').exists():save(OUT/'audit-stopped.json',dict(status='INCOMPLETE-AUDIT',
|
|
error_type=type(exc).__name__,error=str(exc),learning_failure_inferred=False,automatic_retry=False))
|
|
raise
|
|
|
|
if __name__=='__main__':
|
|
parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('mode',choices=('check','audit','analyze-child'))
|
|
parser.add_argument('--inventory-sha256');parser.add_argument('--terminal-sha256');args=parser.parse_args()
|
|
if args.mode=='check':print(json.dumps(readiness(),indent=2))
|
|
elif args.mode=='audit':print(json.dumps(finalize(),indent=2))
|
|
else:
|
|
require(args.inventory_sha256 and args.terminal_sha256,'Actual outer report hashes required')
|
|
analyze_child(args.inventory_sha256,args.terminal_sha256)
|