Spaces:
Running
Running
File size: 5,855 Bytes
d0e78ed 61a7188 d0e78ed 61a7188 d0e78ed 61a7188 d0e78ed | 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 | """Command line entry points share the browser's exact scientific runtime."""
from __future__ import annotations
import argparse
from dataclasses import asdict
import json
import os
from pathlib import Path
import sys
from .study import ADMISSIONS, EXECUTION_MODES, POLICIES, PROBE_SAFETY_MODES, RECOVERY_STRATEGIES, RunConfig, run, run_study
from .fixtures import catalog
from .backends import CANDIDATE_BACKENDS
from .receipts import check_envelope, pack, replay
def _write(path, value):
path = Path(path)
path.parent.mkdir(parents=True, exist_ok=True)
temp = path.with_name(path.name + '.tmp')
temp.write_text(json.dumps(value, indent=2, allow_nan=False) + '\n', encoding='utf-8')
os.replace(temp, path)
def main(argv=None):
parser = argparse.ArgumentParser(prog='crosswire', description='Finite behavioral interoperability laboratory')
commands = parser.add_subparsers(dest='command', required=True)
one = commands.add_parser('run')
defaults = asdict(RunConfig())
for name in ('family', 'implementation', 'split', 'policy', 'admission', 'execution_mode', 'probe_safety', 'recovery_strategy'):
choices = {'family': [f['id'] for f in catalog()], 'implementation': ['table', 'procedural'],
'split': ['development', 'evaluation'], 'policy': POLICIES, 'admission': ADMISSIONS,
'execution_mode': EXECUTION_MODES, 'probe_safety': PROBE_SAFETY_MODES,
'recovery_strategy': RECOVERY_STRATEGIES}[name]
flags = ['--' + name]
if '_' in name:
flags.append('--' + name.replace('_', '-'))
one.add_argument(*flags, default=defaults[name], choices=choices)
for name in ('seed', 'budget', 'horizon'):
one.add_argument('--' + name, default=defaults[name], type=int)
one.add_argument('--instance-seed', '--instance_seed', type=int, default=defaults['instance_seed'],
help='Structural seed for generated_protocol; --seed changes opaque labels only')
one.add_argument('--candidate-backend', '--candidate_backend', choices=CANDIDATE_BACKENDS,
default=defaults['candidate_backend'],
help='Candidate representation: exhaustive flat or lazy factored (same finite model class)')
one.add_argument('--anchors', action='store_true')
one.add_argument('--temporal', action='store_true')
one.add_argument('--escalate', action='store_true')
one.add_argument('--risk-tolerance', '--risk_tolerance', type=float, default=defaults['risk_tolerance'],
help='Allowed fraction of uniformly weighted surviving models (not calibrated target risk)')
one.add_argument('--fault-schedule', '--fault_schedule', type=json.loads, default=defaults['fault_schedule'],
help='Fault schedule JSON dict (e.g. {"step": 2, "type": "sensor_drift"})')
one.add_argument('--out', type=Path)
study = commands.add_parser('study')
study.add_argument('--config', type=Path, required=True)
study.add_argument('--out', type=Path, required=True)
for name in ('verify', 'replay', 'check'):
commands.add_parser(name).add_argument('receipt', type=Path)
server = commands.add_parser('serve')
server.add_argument('--port', type=int, default=8781)
server.add_argument('--open', action='store_true')
commands.add_parser('catalog')
args = parser.parse_args(argv)
try:
if args.command == 'serve':
from .server import serve
serve(args.port, args.open)
return 0
if args.command == 'catalog':
print(json.dumps(catalog(), indent=2))
return 0
if args.command == 'run':
cfg = {name: getattr(args, name, defaults[name]) for name in defaults}
result = run(RunConfig.from_dict(cfg))
receipt = pack(result)
if args.out:
_write(args.out, receipt)
print(json.dumps({'path': str(args.out.resolve()), 'decision': result['decision'],
'contract_satisfied': result['audit']['contract_satisfied'],
'scientific_hash': result['scientific_hash']}))
else:
print(json.dumps(receipt, indent=2))
elif args.command == 'study':
spec = json.loads(args.config.read_text(encoding='utf-8'))
result = run_study(spec)
_write(args.out, pack(result))
print(json.dumps({'path': str(args.out.resolve()), 'runs': len(result['rows']),
'summary': result['summary'], 'paired': result['paired']}))
else:
receipt = json.loads(args.receipt.read_text(encoding='utf-8'))
if args.command == 'check':
from .certificates import check_result
payload = check_envelope(receipt)
if payload['schema'] == 'crosswire-study-v1':
checks = [check_result(row) for row in payload['rows']]
checked = {'valid': bool(checks) and all(c.get('valid') is True for c in checks),
'checked_runs': len(checks), 'checks': checks,
'scope': 'Public model evidence only; does not authenticate private audits or target execution'}
else:
checked = check_result(payload)
else:
checked = replay(receipt)
print(json.dumps(checked))
return 0 if checked['valid'] else 1
return 0
except (ValueError, TypeError, OSError, RuntimeError, ImportError) as exc:
print(f'{type(exc).__name__}: {exc}', file=sys.stderr)
return 1
|