"""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', []) # The nonexistent sealed file proves freeze consumes its commitment only. 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)))) # Aliases are deliberate: all reporting roles may share one physical model. 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) # Replace only this synthetic fixture's suites before recording any result. # Both use public examples, so the real read_suite and hash checks still run. 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()