96 lines
5.2 KiB
Python
96 lines
5.2 KiB
Python
"""One bounded CPU run; waits for verified download and binary dependencies.
|
|
|
|
No installation, retry search, cleanup, training, GPU selection or Git mutation.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import traceback
|
|
from .run import write_json
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser();p.add_argument('--root',type=Path,required=True);p.add_argument('--download-wait-seconds',type=int,default=14400)
|
|
a=p.parse_args();root=a.root.resolve();start=time.monotonic()
|
|
state={'state':'waiting','phase':'dependencies','weight_retention':'until_explicit_user_delete_request'}
|
|
def status(**kwargs):
|
|
state.update(kwargs,supervisor_elapsed_seconds=time.monotonic()-start)
|
|
write_json(root/'supervisor-status.json',state)
|
|
def execute(module,args,timeout,log):
|
|
ledger_path=root/'compute-budget-ledger.json'
|
|
ledger=json.loads(ledger_path.read_text()) if ledger_path.exists() else {'spent_seconds':0.,'calls':[]}
|
|
timeout=min(timeout,int(7200-ledger['spent_seconds']))
|
|
if timeout<1:raise TimeoutError('Cumulative two-hour compute budget exhausted')
|
|
t=time.monotonic()
|
|
try:
|
|
with (root/log).open('w') as out:
|
|
r=subprocess.run([sys.executable,'-B','-m',module,*args],stdout=out,stderr=subprocess.STDOUT,timeout=timeout)
|
|
finally:
|
|
elapsed=time.monotonic()-t
|
|
ledger['spent_seconds']+=elapsed
|
|
ledger['calls'].append({'module':module,'seconds':elapsed})
|
|
write_json(ledger_path,ledger)
|
|
if r.returncode:raise RuntimeError(f'{module} failed: exit {r.returncode}, see {log}')
|
|
status()
|
|
try:
|
|
while True:
|
|
ready=False
|
|
if all(importlib.util.find_spec(name) for name in ('torch','tokenizers')):
|
|
try:
|
|
r=subprocess.run([sys.executable,'-B','-c','import torch, tokenizers; torch.empty(1)'],capture_output=True,timeout=30)
|
|
ready=r.returncode==0
|
|
except subprocess.TimeoutExpired:pass
|
|
if ready:break
|
|
if time.monotonic()-start>3600:raise TimeoutError('Binary reference dependencies not ready within one hour')
|
|
status()
|
|
time.sleep(15)
|
|
compute_start=time.monotonic()
|
|
status(state='running',phase='reference_check')
|
|
provenance={str(p):hashlib.sha256(p.read_bytes()).hexdigest() for p in sorted(Path('functional').glob('*')) if p.is_file()}
|
|
write_json(root/'functional-code-hashes.json',provenance)
|
|
execute('functional.reference_check',['--source',str(root/'reference.py'),'--output',str(root/'reference-check.json')],300,'reference-check.log')
|
|
reference_seconds=time.monotonic()-compute_start
|
|
status(state='waiting',phase='verified_weights',reference_seconds=reference_seconds)
|
|
while True:
|
|
path=root/'manifest.json'
|
|
try:
|
|
manifest=json.loads(path.read_text())
|
|
weights=[f for f in manifest.get('files',[]) if f['name'].endswith('.safetensors')]
|
|
ready=(manifest.get('completed_unix') is not None and weights and
|
|
all((root/'weights'/f['name']).is_file() and (root/'weights'/f['name']).stat().st_size==f['bytes'] and f['sha256']==f['upstream_sha256'] for f in weights))
|
|
except (OSError,ValueError,KeyError):ready=False
|
|
if ready:break
|
|
if time.monotonic()-start>a.download_wait_seconds:raise TimeoutError('Verified weight download not ready before wait deadline')
|
|
status()
|
|
time.sleep(15)
|
|
status(state='running',phase='real_weight_preflight')
|
|
compute_start=time.monotonic()
|
|
execute('functional.preflight',['--root',str(root)],1200,'preflight.log')
|
|
preflight_seconds=time.monotonic()-compute_start
|
|
remaining=int(7200-json.loads((root/'compute-budget-ledger.json').read_text())['spent_seconds']-60)
|
|
if remaining<60:raise TimeoutError('No remaining bounded functional computation budget')
|
|
status(phase='functional_analysis',preflight_seconds=preflight_seconds,remaining_seconds=remaining)
|
|
execute('functional.run',['--root',str(root),'--seconds',str(remaining)],remaining+15,'functional-run.log')
|
|
status(phase='storage_accounting')
|
|
execute('functional.storage',['--root',str(root)],120,'storage-accounting.log')
|
|
status(phase='report')
|
|
execute('functional.report',['--root',str(root)],120,'functional-report.log')
|
|
status(state='completed',phase='done')
|
|
except Exception as exc:
|
|
status(state='stopped',error_type=type(exc).__name__,error=str(exc))
|
|
(root/'supervisor-error.txt').write_text(traceback.format_exc())
|
|
# Retain all weights/results. No automatic retries or candidate expansion.
|
|
if not (root/'functional-status.json').exists():
|
|
write_json(root/'functional-status.json',{'state':'stopped','phase':state['phase'],'error_type':type(exc).__name__,'error':str(exc)})
|
|
try:execute('functional.report',['--root',str(root)],120,'functional-report.log')
|
|
except Exception:pass
|
|
raise
|
|
|
|
|
|
if __name__=='__main__':main()
|