| """Frozen, one-pass evaluation of the learned-slot dependency experiment. |
| |
| The coordinator freezes all predictors before this module opens sealed examples. |
| The old action runner and scorer remain unchanged. A separate audit replays saved |
| actions and checks the supported execution subset with the previous second VM. |
| No execution feedback is passed into a predictor or used to select its actions. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| from collections import Counter, defaultdict |
| from datetime import datetime, timezone |
| import gzip |
| import hashlib |
| import importlib.util |
| import io |
| import json |
| from pathlib import Path |
| import resource |
| import sys |
| import time |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| if str(ROOT) not in sys.path: |
| sys.path.insert(0, str(ROOT)) |
|
|
| from experiments.architecture_eval import score_episode |
| from experiments.morphology_data import summarize_scored |
|
|
| SOURCE_FILES = ( |
| 'experiments/dependency_eval.py', 'experiments/dependency_train.py', |
| 'experiments/architecture_eval.py', 'experiments/morphology_data.py', |
| 'lightnest/relation_patch.py', 'lightnest/slot_patch.py', |
| 'lightnest/iterative_patch.py', 'lightnest/rv32.py', |
| 'work/architecture-20261017/audit_evaluation.py', |
| ) |
|
|
|
|
| def sha(path): |
| digest = hashlib.sha256() |
| with Path(path).open('rb') as source: |
| for block in iter(lambda: source.read(1024 * 1024), b''): |
| digest.update(block) |
| return digest.hexdigest() |
|
|
|
|
| def read_json(path): |
| return json.loads(Path(path).read_text()) |
|
|
|
|
| def write_json(path, value): |
| path = Path(path) |
| path.parent.mkdir(parents=True, exist_ok=True) |
| with path.open('x', encoding='utf-8') as output: |
| json.dump(value, output, ensure_ascii=False, sort_keys=True, |
| indent=2, allow_nan=False) |
| output.write('\n') |
|
|
|
|
| def inside(path): |
| path = (ROOT / path).resolve() |
| if not path.is_relative_to(ROOT) or not path.is_file(): |
| raise ValueError('expected an existing file inside the repository: ' + str(path)) |
| return path |
|
|
|
|
| def file_record(path): |
| path = inside(path) |
| return dict(path=str(path.relative_to(ROOT)), sha256=sha(path), bytes=path.stat().st_size) |
|
|
|
|
| def verify_file(record): |
| path = inside(record['path']) |
| if path.stat().st_size != record['bytes'] or sha(path) != record['sha256']: |
| raise ValueError('frozen artifact changed: ' + record['path']) |
| return path |
|
|
|
|
| def freeze(work, specifications, training_protocol): |
| """Read public commitments and predictors, never sealed case contents.""" |
| work = Path(work) |
| target = work / 'models-frozen.json' |
| if target.exists(): |
| raise ValueError('refuse to replace frozen model inventory') |
| training_record = file_record(training_protocol) |
| training = read_json(verify_file(training_record)) |
| if training.get('status') != 'frozen': |
| raise ValueError('training protocol is not frozen') |
| expected_roles = {(seed, arm, role) for seed in training['seeds'] |
| for arm in training['arms'] for role in ('selected', 'terminal')} |
| expected_roles.update((seed, 'original-parent', 'original') for seed in training['seeds']) |
| observed_roles, report_records = set(), {} |
| data_protocol = file_record(work / 'data/protocol.json') |
| protocol = read_json(verify_file(data_protocol)) |
| for relative, expected in protocol['sources'].items(): |
| if sha(inside(relative)) != expected: |
| raise ValueError('committed data/scoring source changed: ' + relative) |
| for relative, expected in protocol['files'].items(): |
| if relative.startswith('sealed/'): |
| continue |
| if sha(work / 'data' / relative) != expected: |
| raise ValueError('committed public data changed: ' + relative) |
| models = [] |
| for supplied in read_json(specifications): |
| item = dict(supplied) |
| item.update(file_record(item['path'])) |
| if not item.get('id') or not item.get('roles'): |
| raise ValueError('predictor requires an identifier and declared reporting roles') |
| if item['kind'] not in ('relation', 'slot'): |
| raise ValueError('undeclared predictor family') |
| if 'message_steps' in item or 'memory_mode' in item: |
| raise ValueError('formal inference settings must come from the trained checkpoint') |
| for role in item['roles']: |
| identity = (role['seed'], role['arm'], role['role']) |
| if identity not in expected_roles or identity in observed_roles: |
| raise ValueError('undeclared or duplicate arm/seed/reporting role') |
| observed_roles.add(identity) |
| report_record = file_record(role['source_report']) |
| report_records[report_record['path']] = report_record |
| report = read_json(verify_file(report_record)) |
| if role['role'] == 'original': |
| if (report['seed'] != role['seed'] or |
| report['protocol_sha256'] != training_record['sha256'] or |
| report['parent'] != training['parents'][str(role['seed'])]): |
| raise ValueError('original-parent anchor does not match the protocol') |
| committed, kind = report['checkpoint'], 'relation' |
| else: |
| if (report.get('complete') is not True or report['seed'] != role['seed'] or |
| report['arm']['name'] != role['arm'] or |
| report['protocol']['sha256'] != training_record['sha256']): |
| raise ValueError('missing completed training report for a declared role') |
| selected_role = report['last' if role['role'] == 'terminal' else 'selected'] |
| committed, kind = selected_role['checkpoint'], selected_role['kind'] |
| if (inside(committed['path']) != inside(item['path']) or |
| committed['sha256'] != item['sha256'] or kind != item['kind']): |
| raise ValueError('predictor differs from its committed training-report role') |
| models.append(item) |
| if not models or len({m['id'] for m in models}) != len(models): |
| raise ValueError('empty or duplicate model inventory') |
| if any('/' in m['id'] or '..' in m['id'] for m in models): |
| raise ValueError('model identifiers must be file-safe') |
| if observed_roles != expected_roles: |
| raise ValueError('incomplete declared arm/seed/selected/terminal/original coverage') |
| result = dict( |
| schema='lightnest-dependency-evaluation-v1', |
| created_utc=datetime.now(timezone.utc).isoformat(), |
| training_protocol=training_record, data_protocol=data_protocol, |
| specification=file_record(specifications), models=models, |
| role_reports=list(report_records.values()), |
| sources=[file_record(relative) for relative in SOURCE_FILES], |
| suites={ |
| 'final': dict(path=str((work / 'data/sealed/final.json').relative_to(ROOT)), |
| sha256=protocol['files']['sealed/final.json'], count=256), |
| 'retention': dict(path=str((work / 'data/retention.json').relative_to(ROOT)), |
| sha256=protocol['files']['retention.json'], count=128), |
| }, |
| bounds=dict(max_steps_including_stop=8, max_total_changed_bytes=8, |
| max_output_bytes=1024, confidence_threshold=0.), |
| policy='Every declared seed, arm, selected role and terminal role is reported. ' |
| 'Identical checkpoint roles share inference. No final-based selection, ' |
| 'training, threshold adjustment, or default replacement.', |
| access='The data steward generated and validated final cases. Coordinator ' |
| 'opens them only after this inventory exists. This is a recorded ' |
| 'access protocol in a shared workspace, not physical isolation.', |
| ) |
| write_json(target, result) |
| return dict(inventory=file_record(target), predictors=len(models)) |
|
|
|
|
| def verify_inventory(work): |
| frozen = read_json(Path(work) / 'models-frozen.json') |
| for record in [frozen['training_protocol'], frozen['data_protocol'], |
| frozen['specification'], *frozen['role_reports'], |
| *frozen['sources'], *frozen['models']]: |
| verify_file(record) |
| return frozen |
|
|
|
|
| def read_suite(spec): |
| path = inside(spec['path']) |
| if sha(path) != spec['sha256']: |
| raise ValueError('evaluation data changed after commitment') |
| rows = read_json(path) |
| if len(rows) != spec['count'] or len({row['id'] for row in rows}) != len(rows): |
| raise ValueError('evaluation count or identifiers differ') |
| return rows |
|
|
|
|
| def run_model(work, model_id, threads=2): |
| from experiments.dependency_train import build_predictor, configure_threads |
| configure_threads(threads) |
| work = Path(work) |
| frozen = verify_inventory(work) |
| frozen_sha = sha(work / 'models-frozen.json') |
| matches = [m for m in frozen['models'] if m['id'] == model_id] |
| if len(matches) != 1: |
| raise ValueError('model is absent from the frozen inventory') |
| spec = matches[0] |
| folder = work / 'evaluation' |
| folder.mkdir(exist_ok=True) |
| trace_path = folder / (model_id + '.jsonl.gz') |
| summary_path = folder / (model_id + '.json') |
| if trace_path.exists() or summary_path.exists(): |
| raise ValueError('refuse to repeat or overwrite a formal predictor evaluation') |
| predictor = build_predictor(spec) |
| wall_start, cpu_start = time.perf_counter(), time.process_time() |
| summaries = {} |
| with trace_path.open('xb') as raw: |
| with gzip.GzipFile(filename='', fileobj=raw, mode='wb', mtime=0) as compressed: |
| with io.TextIOWrapper(compressed, encoding='utf-8') as output: |
| for suite, data_spec in frozen['suites'].items(): |
| episodes = read_suite(data_spec) |
| scores = [] |
| for episode in episodes: |
| score = score_episode(predictor, episode, predictor_id=model_id, |
| max_steps=8) |
| output.write(json.dumps(dict(suite=suite, scored=score), |
| ensure_ascii=False, separators=(',', ':'), allow_nan=False) + '\n') |
| scores.append(score) |
| summaries[suite] = dict(groups=summarize_scored(episodes, scores), |
| stops=dict(Counter(score['stop_reason'] for score in scores))) |
| result = dict(model=spec, inventory_sha256=frozen_sha, |
| completed_utc=datetime.now(timezone.utc).isoformat(), |
| inference_and_scoring_wall_seconds=time.perf_counter() - wall_start, |
| inference_and_scoring_cpu_seconds=time.process_time() - cpu_start, |
| threads=threads, max_rss_kib=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss, |
| traces=file_record(trace_path), suites=summaries) |
| verify_inventory(work) |
| if sha(work / 'models-frozen.json') != frozen_sha: |
| raise ValueError('model inventory changed during evaluation') |
| write_json(summary_path, result) |
| return dict(model=model_id, suites={name: value['groups']['all'] |
| for name, value in summaries.items()}) |
|
|
|
|
| def independent_counts(episodes, scores): |
| """Recompute key pair and edit counters without the scoring aggregator.""" |
| groups = defaultdict(list) |
| for episode, scored in zip(episodes, scores): |
| for key in ('all', 'tier/' + episode.get('tier', 'main'), |
| 'category/' + episode['category']): |
| groups[key].append((episode, scored)) |
| result = {} |
| for key, rows in groups.items(): |
| pairs, invariant = defaultdict(list), defaultdict(list) |
| out = Counter(total=len(rows)) |
| for episode, score in rows: |
| kind = 'edit' if episode['edits'] else 'noop' |
| out[kind + '_total'] += 1 |
| out[kind + '_strict'] += int(score['strict_success']) |
| for field in ('strict_success', 'exact_final', 'semantic_final', 'explicit_stop', |
| 'false_edit_on_correct', 'proposed_edit_steps'): |
| out[field] += int(score[field]) |
| if episode.get('pair_id') is not None: |
| pairs[episode['pair_id']].append((episode, score)) |
| if episode.get('invariance_pair_id') is not None: |
| invariant[episode['invariance_pair_id']].append((episode, score)) |
| for pair in pairs.values(): |
| if len(pair) != 2 or {e['variant'] for e, _ in pair} != {0, 1}: |
| raise ValueError('independent return-pair completeness check failed') |
| out['pair_total'] += 1 |
| out['pair_both_strict'] += int(all(s['strict_success'] for _, s in pair)) |
| for pair in invariant.values(): |
| if len(pair) != 2 or {e['nuisance_variant'] for e, _ in pair} != {0, 1}: |
| raise ValueError('independent invariance-pair completeness check failed') |
| kind = 'edit' if pair[0][0]['edits'] else 'noop' |
| out['invariance_total'] += 1 |
| out['invariance_' + kind + '_total'] += 1 |
| both = int(all(s['strict_success'] for _, s in pair)) |
| out['invariance_both_strict'] += both |
| out['invariance_' + kind + '_both_strict'] += both |
| deltas = [] |
| for episode, score in pair: |
| before, after = bytes.fromhex(episode['binary']), bytes.fromhex(score['final_hex']) |
| deltas.append((len(after), [(i, value) for i, value in enumerate(after) |
| if i >= len(before) or before[i] != value])) |
| out['invariance_predicted_delta_agreement'] += int(deltas[0] == deltas[1]) |
| result[key] = dict(out) |
| return result |
|
|
|
|
| def audit(work): |
| """Post-evaluation audit: no predictor load, forward pass, or optimization.""" |
| work = Path(work) |
| frozen = verify_inventory(work) |
| frozen_sha = sha(work / 'models-frozen.json') |
| target = work / 'independent-audit.json' |
| if target.exists(): |
| raise ValueError('refuse to overwrite a completed independent audit') |
| spec = importlib.util.spec_from_file_location('dependency_second_vm', |
| ROOT / 'work/architecture-20261017/audit_evaluation.py') |
| verifier = importlib.util.module_from_spec(spec) |
| spec.loader.exec_module(verifier) |
| semantics = verifier.SemanticAudit() |
| suites = {name: read_suite(item) for name, item in frozen['suites'].items()} |
| records, checks = [], [] |
| total = 0 |
| for model in frozen['models']: |
| path = work / 'evaluation' / (model['id'] + '.json') |
| report_record = file_record(path) |
| reported = read_json(path) |
| if reported['model'] != model or reported['inventory_sha256'] != frozen_sha: |
| raise ValueError('result does not match frozen predictor inventory') |
| trace = verify_file(reported['traces']) |
| with gzip.open(trace, 'rt', encoding='utf-8') as source: |
| rows = [json.loads(line) for line in source] |
| if len(rows) != sum(len(data) for data in suites.values()): |
| raise ValueError('missing or duplicate saved episodes') |
| offset = 0 |
| for suite, episodes in suites.items(): |
| saved = rows[offset:offset + len(episodes)] |
| offset += len(episodes) |
| if any(row['suite'] != suite for row in saved): |
| raise ValueError('saved suite ordering changed') |
| scores = [row['scored'] for row in saved] |
| for episode, scored in zip(episodes, scores): |
| verifier.audit_scored_row(episode, scored, model['id'], semantics) |
| independent = independent_counts(episodes, scores) |
| committed_groups = reported['suites'][suite]['groups'] |
| if set(independent) != set(committed_groups): |
| raise ValueError('reported aggregate groups differ') |
| for group, counters in independent.items(): |
| for field, value in counters.items(): |
| if committed_groups[group].get(field, 0) != value: |
| raise ValueError('independent counter differs: ' + '/'.join((model['id'], suite, group, field))) |
| for numerator, denominator, rate in ( |
| ('edit_strict', 'edit_total', 'edit_strict_rate'), |
| ('noop_strict', 'noop_total', 'noop_strict_rate'), |
| ('pair_both_strict', 'pair_total', 'pair_both_strict_rate')): |
| count = counters.get(denominator, 0) |
| value = counters.get(numerator, 0) / count if count else None |
| if committed_groups[group].get(rate) != value: |
| raise ValueError('independent aggregate rate differs') |
| if reported['suites'][suite]['stops'] != dict(Counter(s['stop_reason'] for s in scores)): |
| raise ValueError('independent STOP-reason aggregate differs') |
| checks.append(dict(model=model['id'], suite=suite, episodes=len(episodes), |
| groups=len(independent), passed=True)) |
| total += len(episodes) |
| records.extend([report_record, reported['traces']]) |
| verify_inventory(work) |
| for record in records: |
| verify_file(record) |
| if sha(work / 'models-frozen.json') != frozen_sha: |
| raise ValueError('model inventory changed during audit') |
| result = dict(all_checks_passed=True, completed_utc=datetime.now(timezone.utc).isoformat(), |
| inventory_sha256=frozen_sha, |
| audited_predictors=len(frozen['models']), audited_episodes=total, |
| independent_executor_calls=semantics.executions, |
| execution_cache_hits=semantics.cache_hits, checks=checks, artifacts=records, |
| scope='Read-only byte-trace replay, independently recomputed scores and pair counters, ' |
| 'and the preserved second VM. No training, model inference, or checkpoint selection.') |
| write_json(target, result) |
| return {key: result[key] for key in ('all_checks_passed', 'audited_predictors', |
| 'audited_episodes', 'independent_executor_calls')} |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser() |
| parser.add_argument('action', choices=('freeze', 'run', 'audit')) |
| parser.add_argument('--work', type=Path, default=ROOT / 'work/morphology-20261017') |
| parser.add_argument('--spec', type=Path) |
| parser.add_argument('--training-protocol', type=Path) |
| parser.add_argument('--model') |
| parser.add_argument('--threads', type=int, default=2) |
| args = parser.parse_args() |
| if args.action == 'freeze': |
| if args.spec is None or args.training_protocol is None: |
| parser.error('freeze requires --spec and --training-protocol') |
| result = freeze(args.work.resolve(), args.spec, args.training_protocol) |
| elif args.action == 'run': |
| if args.model is None: |
| parser.error('run requires --model') |
| result = run_model(args.work.resolve(), args.model, args.threads) |
| else: |
| result = audit(args.work.resolve()) |
| print(json.dumps(result, ensure_ascii=False, sort_keys=True, allow_nan=False)) |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
