tegwick
5da8b2405c
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1045b-933d-7aa1-b2db-104557487a79
160 lines
7.7 KiB
Python
160 lines
7.7 KiB
Python
"""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):
|
|
# Before packet publication Git may report a missing path; after publication
|
|
# the fresh temporary assignment/freeze differs from HEAD. Both must stop
|
|
# execution before adapter construction.
|
|
with pytest.raises((subprocess.CalledProcessError, ValueError)):
|
|
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)
|