| """Public-only gates for role coverage and independent aggregate verification. |
| |
| Opaque checkpoint bytes exercise manifest checks without loading a neural model. |
| The aggregate fixtures use a scripted STOP control on public generated examples. |
| No real experiment artifacts or sealed data are opened or modified. |
| """ |
| from collections import Counter |
| from copy import deepcopy |
| import gzip |
| import json |
| from pathlib import Path |
| import random |
| import shutil |
|
|
| import pytest |
|
|
| pytest.importorskip('torch') |
| from experiments import dependency_eval as ev |
| from experiments import morphology_data as data |
| from experiments.architecture_eval import score_episode |
| from lightnest.iterative_patch import PatchDecision |
|
|
|
|
| def put(path, value): |
| path.parent.mkdir(parents=True, exist_ok=True) |
| path.write_text(json.dumps(value, sort_keys=True) + '\n') |
|
|
|
|
| @pytest.fixture |
| def experiment(tmp_path, monkeypatch): |
| real_root = ev.ROOT |
| monkeypatch.setattr(ev, 'ROOT', tmp_path) |
| old_auditor = 'work/architecture-20261017/audit_evaluation.py' |
| copied_auditor = tmp_path / old_auditor |
| copied_auditor.parent.mkdir(parents=True) |
| shutil.copyfile(real_root / old_auditor, copied_auditor) |
| monkeypatch.setattr(ev, 'SOURCE_FILES', (old_auditor,)) |
| work = tmp_path / 'work' / 'public-fixture' |
| data_dir = work / 'data' |
| put(data_dir / 'retention.json', []) |
| |
| put(data_dir / 'protocol.json', dict(sources={}, files={ |
| 'retention.json': ev.sha(data_dir / 'retention.json'), |
| 'sealed/final.json': '0' * 64, |
| })) |
| seeds, arms = (11, 13), ('control', 'alternative') |
| checkpoints, parents = {}, {} |
| for seed in seeds: |
| path = work / f'opaque-parent-{seed}.pt' |
| path.write_bytes(f'fixture-only checkpoint {seed}'.encode()) |
| checkpoints[seed] = ev.file_record(path) |
| parents[str(seed)] = ev.file_record(path) |
| training = work / 'training-protocol.json' |
| put(training, dict(status='frozen', seeds=seeds, arms={arm: {} for arm in arms}, parents=parents)) |
| training_sha = ev.sha(training) |
| specifications, reports = [], {} |
| for seed in seeds: |
| roles = [] |
| anchor = work / f'anchor-{seed}.json' |
| put(anchor, dict(seed=seed, protocol_sha256=training_sha, |
| parent=parents[str(seed)], checkpoint=checkpoints[seed])) |
| roles.append(dict(seed=seed, arm='original-parent', role='original', |
| source_report=str(anchor.relative_to(tmp_path)))) |
| for arm in arms: |
| report = work / f'report-{arm}-{seed}.json' |
| state = dict(checkpoint=checkpoints[seed], kind='relation') |
| put(report, dict(complete=True, seed=seed, arm=dict(name=arm), |
| protocol=dict(sha256=training_sha), selected=state, last=state)) |
| reports[seed, arm] = report |
| for role in ('selected', 'terminal'): |
| roles.append(dict(seed=seed, arm=arm, role=role, |
| source_report=str(report.relative_to(tmp_path)))) |
| |
| specifications.append(dict(id=f'public-model-{seed}', kind='relation', |
| path=checkpoints[seed]['path'], roles=roles)) |
| spec = work / 'model-spec.json' |
| put(spec, specifications) |
| return dict(root=tmp_path, work=work, training=training, spec=spec, |
| reports=reports, seeds=seeds, arms=arms) |
|
|
|
|
| def do_freeze(fixture): |
| return ev.freeze(fixture['work'], fixture['spec'], fixture['training']) |
|
|
|
|
| def test_complete_arm_seed_role_inventory_accepts_checkpoint_aliases(experiment): |
| result = do_freeze(experiment) |
| frozen = ev.verify_inventory(experiment['work']) |
| assert result['predictors'] == 2 |
| assert sum(len(model['roles']) for model in frozen['models']) == 10 |
| assert len(frozen['role_reports']) == 6 |
| assert not (experiment['work'] / 'data/sealed/final.json').exists() |
|
|
|
|
| def test_missing_single_terminal_role_is_rejected(experiment): |
| spec = ev.read_json(experiment['spec']) |
| spec[0]['roles'] = [role for role in spec[0]['roles'] |
| if not (role['arm'] == 'control' and role['role'] == 'terminal')] |
| put(experiment['spec'], spec) |
| with pytest.raises(ValueError, match='incomplete declared'): |
| do_freeze(experiment) |
|
|
|
|
| @pytest.mark.parametrize('corruption', ('not_complete', 'foreign_protocol', 'wrong_checkpoint')) |
| def test_forged_training_completion_or_checkpoint_is_rejected(experiment, corruption): |
| path = experiment['reports'][11, 'control'] |
| report = ev.read_json(path) |
| if corruption == 'not_complete': |
| report['complete'] = False |
| elif corruption == 'foreign_protocol': |
| report['protocol']['sha256'] = 'f' * 64 |
| else: |
| report['selected']['checkpoint']['sha256'] = 'f' * 64 |
| put(path, report) |
| with pytest.raises(ValueError, match='completed training report|committed training-report role'): |
| do_freeze(experiment) |
|
|
|
|
| def test_role_cannot_borrow_another_seeds_source_report(experiment): |
| spec = ev.read_json(experiment['spec']) |
| role = next(role for role in spec[0]['roles'] |
| if role['arm'] == 'control' and role['role'] == 'selected') |
| role['source_report'] = str(experiment['reports'][13, 'control'].relative_to(experiment['root'])) |
| put(experiment['spec'], spec) |
| with pytest.raises(ValueError, match='completed training report'): |
| do_freeze(experiment) |
|
|
|
|
| @pytest.fixture |
| def public_results(experiment): |
| do_freeze(experiment) |
| work = experiment['work'] |
| inventory_path = work / 'models-frozen.json' |
| inventory = ev.read_json(inventory_path) |
| episodes = data.make_group(random.Random(55391), data.FAMILIES[0], 0, 'public-gate-test') |
| public_path = work / 'data/public-evaluation.json' |
| put(public_path, episodes) |
| |
| |
| record = dict(path=str(public_path.relative_to(experiment['root'])), |
| sha256=ev.sha(public_path), count=len(episodes)) |
| inventory['suites'] = dict(final=record, retention=record) |
| put(inventory_path, inventory) |
| inventory_sha = ev.sha(inventory_path) |
| folder = work / 'evaluation' |
| folder.mkdir() |
| for model in inventory['models']: |
| scores = [score_episode(lambda *args, **kwargs: PatchDecision(stop=True, reason='model_stop'), |
| episode, predictor_id=model['id']) for episode in episodes] |
| summaries = {suite: dict(groups=data.summarize_scored(episodes, scores), |
| stops=dict(Counter(score['stop_reason'] for score in scores))) |
| for suite in inventory['suites']} |
| trace = folder / (model['id'] + '.jsonl.gz') |
| with gzip.open(trace, 'wt', encoding='utf-8') as output: |
| for suite in inventory['suites']: |
| for score in scores: |
| output.write(json.dumps(dict(suite=suite, scored=score)) + '\n') |
| put(folder / (model['id'] + '.json'), dict(model=model, inventory_sha256=inventory_sha, |
| traces=ev.file_record(trace), suites=summaries)) |
| return experiment |
|
|
|
|
| def test_public_scores_pass_complete_independent_counter_audit(public_results): |
| result = ev.audit(public_results['work']) |
| assert result['all_checks_passed'] |
| assert result['audited_predictors'] == 2 |
| assert result['audited_episodes'] == 32 |
| assert not (public_results['work'] / 'data/sealed/final.json').exists() |
|
|
|
|
| @pytest.mark.parametrize('field', ('edit_strict_rate', 'noop_strict_rate', 'pair_both_strict_rate', 'stops')) |
| def test_incorrect_reported_rates_and_stop_counts_are_rejected(public_results, field): |
| path = public_results['work'] / 'evaluation/public-model-11.json' |
| report = ev.read_json(path) |
| suite = report['suites']['final'] |
| if field == 'stops': |
| suite['stops']['predictor_stop'] += 1 |
| else: |
| suite['groups']['all'][field] += .125 |
| put(path, report) |
| with pytest.raises(ValueError, match='aggregate rate|STOP-reason aggregate'): |
| ev.audit(public_results['work']) |
| assert not (public_results['work'] / 'independent-audit.json').exists() |
|
|
