"""Guard isolation, immutable budgets and evidence-bearing atomic scoring.""" from pathlib import Path import shutil import subprocess import sys from types import SimpleNamespace import pytest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / 'scripts')) import language_replication as replication @pytest.fixture def packet(tmp_path): shutil.copytree(ROOT / 'reference', tmp_path / 'reference') (tmp_path / '.git').symlink_to(ROOT / '.git', target_is_directory=True) for name in (*replication.INPUTS,): target = tmp_path / replication.PACKET / name target.parent.mkdir(parents=True, exist_ok=True) shutil.copy(ROOT / replication.PACKET / name, target) prior = Path('research/evaluations/language-pilot-01/scenarios.json') (tmp_path / prior).parent.mkdir(parents=True, exist_ok=True) shutil.copy(ROOT / prior, tmp_path / prior) for relative in replication.DEPENDENCIES: (tmp_path / relative).parent.mkdir(parents=True, exist_ok=True) shutil.copy(ROOT / relative, tmp_path / relative) replication.prepare(tmp_path) return tmp_path def test_preparation_checks_equivalence_isolation_and_counterbalance(packet): frozen, assignments, scenarios = replication.check(packet) assert len(assignments) == 12 and sum(len(s['criteria']) for s in scenarios) == 36 guidance = replication.read(packet / replication.PACKET / 'guidance.json') for item in guidance['items']: for condition in ('patterns', 'prose'): assert (packet / replication.PACKET / f'pack-{condition}.md').read_text().count(item['body']) == 1 for assignment in assignments: text = (packet / replication.PACKET / 'requests' / (assignment['run'] + '.txt')).read_text() scenario = next(s for s in scenarios if s['id'] == assignment['scenario']) assert text.endswith(scenario['brief'] + '\n') assert '"criteria"' not in text and '"prohibited"' not in text assert sum(text.count(s['brief']) for s in scenarios) == 1 with pytest.raises(ValueError, match='already frozen'): replication.prepare(packet) @pytest.mark.parametrize('path', ['requests/r01.txt', 'pack-prose.md', 'scores-template.json']) def test_tampered_prompt_guidance_or_blank_sheet_is_rejected(packet, path): p = packet / replication.PACKET / path p.write_text(p.read_text() + '\nchanged\n') with pytest.raises(ValueError, match='Frozen input changed'): replication.check(packet) def test_no_calls_before_frozen_inputs_are_committed(packet): with pytest.raises(subprocess.CalledProcessError): replication.run(12, packet, lambda: pytest.fail('No adapter before commit verification')) def test_call_ceiling_retains_failure_and_never_retries_reserved_sessions(packet, monkeypatch): monkeypatch.setattr(replication, 'committed', lambda *_: None) attempts = [] class Adapter: endpoint = 'fixture:no-provider' config = SimpleNamespace(to_dict=lambda: {'max_tokens': 1400}) responses = [] rejected_content = None def generate(self, request): attempts.append(request.prompt) if len(attempts) == 2: raise RuntimeError('Retained synthetic failure') return SimpleNamespace(markdown='Synthetic fixture response; no empirical result.\n') with pytest.raises(ValueError, match='aggregate call ceiling'): replication.run(11, packet, Adapter) assert not attempts replication.run(12, packet, Adapter) assert len(attempts) == 12 assert replication.read(packet / replication.PACKET / 'raw/r02/receipt.json')['status'] == 'failed' replication.run(12, packet, Adapter) assert len(attempts) == 12 assert len(list((packet / replication.PACKET / 'masked').glob('*.md'))) == 11 @pytest.fixture def scenarios(): return replication.read(ROOT / replication.PACKET / 'scenarios.json') def reviewed_sheet(scenarios): sheet = replication.score_template(scenarios) sheet['reviewer'].update(name='Fixture reviewer', role='author-model', assignment_seen=False, prior_scores_seen=False) for answer in sheet['answers'].values(): answer.update(available=True, unsupported_claims=[]) for rating in answer['ratings'].values(): rating.update(value=0, locator='Synthetic fixture answer', reason='Fixture miss, not an actual model judgment.') return sheet def test_blank_template_keeps_unreviewed_values_null(scenarios): sheet = replication.score_template(scenarios) replication.scores_check(sheet, scenarios) assert all(r['value'] is None for a in sheet['answers'].values() for r in a['ratings'].values()) with pytest.raises(ValueError, match='Reviewer provenance'): replication.scores_check(sheet, scenarios, complete=True) @pytest.mark.parametrize('mutation,match', [ (lambda s: next(iter(s['answers'].values()))['ratings']['A1'].update(value=0.5), 'binary'), (lambda s: next(iter(s['answers'].values()))['ratings']['A1'].update(value=True), 'binary'), (lambda s: next(iter(s['answers'].values()))['ratings']['A1'].update(reason=None), 'evidence and reason'), (lambda s: next(iter(s['answers'].values())).update(available=False), 'cannot be scored zero'), (lambda s: next(iter(s['answers'].values())).update(unsupported_claims=None), 'Unsupported-claim review'), (lambda s: s['reviewer'].update(independent_human=True), 'Model review cannot'), (lambda s: s['reviewer'].update(assignment_seen=None), 'Exposure disclosure'), (lambda s: next(iter(s['answers'].values()))['ratings'].pop('A6'), 'criterion inventory'), ]) def test_invalid_or_missing_score_evidence_cannot_become_validation(scenarios, mutation, match): sheet = reviewed_sheet(scenarios) mutation(sheet) with pytest.raises(ValueError, match=match): replication.scores_check(sheet, scenarios, complete=True) def test_failed_pair_is_excluded_and_retained_answers_are_verified(packet, scenarios, monkeypatch): monkeypatch.setattr(replication, 'committed', lambda *_: None) calls = [] class Adapter: endpoint = 'fixture:no-provider' config = SimpleNamespace(to_dict=lambda: {}) responses = [] rejected_content = None def generate(self, request): calls.append(request) if len(calls) == 2: raise RuntimeError('Fixture failure') return SimpleNamespace(markdown='Fixture output.\n') replication.run(12, packet, Adapter) sheet = reviewed_sheet(scenarios) assignment = replication.read(packet / replication.PACKET / 'assignment.json')[1] unavailable = sheet['answers'][assignment['blind_id']] unavailable['available'] = False for rating in unavailable['ratings'].values(): rating['value'] = None summary = replication.summarize(sheet, packet) assert not summary['pairs'][0]['included'] assert all(t['criteria'] == 30 and t['completed_pairs'] == 5 for t in summary['totals'].values()) first = replication.read(packet / replication.PACKET / 'assignment.json')[0]['blind_id'] sheet['answers'][first]['ratings']['A1']['quote'] = 'Invented quote absent from output.' with pytest.raises(ValueError, match='Quoted score evidence'): replication.summarize(sheet, packet) sheet['answers'][first]['ratings']['A1']['quote'] = None (packet / replication.PACKET / 'raw/r01/answer.md').write_text('Modified output.\n') with pytest.raises(ValueError, match='Retained answer changed'): replication.summarize(sheet, packet)