tegwick
1e2741a959
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1045b-933d-7aa1-b2db-104557487a79
308 lines
17 KiB
Python
308 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Freeze, verify and execute the bounded fresh presentation replication."""
|
|
import argparse
|
|
from collections import Counter
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import secrets
|
|
import subprocess
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
from reference_consume import ROOT, published
|
|
from reference_preview import require
|
|
|
|
PACKET = Path('research/evaluations/language-replication-02')
|
|
INPUTS = ('protocol.md', 'instructions.txt', 'scenarios.json', 'guidance.json')
|
|
DEPENDENCIES = ('scripts/language_replication.py', 'scripts/reference_consume.py',
|
|
'scripts/reference_preview.py', 'scripts/classification_coverage.py', 'scripts/research_bench.py')
|
|
|
|
|
|
def digest(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def read(path):
|
|
return json.loads(path.read_text())
|
|
|
|
|
|
def write_new(path, value):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
text = value if isinstance(value, str) else json.dumps(value, indent=2, allow_nan=False) + '\n'
|
|
with path.open('x') as stream:
|
|
stream.write(text)
|
|
|
|
|
|
def scenarios_check(scenarios):
|
|
require(len(scenarios) == 6 and len({s['id'] for s in scenarios}) == 6, 'Six unique scenarios required')
|
|
for scenario in scenarios:
|
|
require(scenario['brief'].strip() and scenario['title'].strip() and scenario['prohibited'], 'Incomplete scenario')
|
|
require([c['id'] for c in scenario['criteria']] == [f'A{i}' for i in range(1, 7)], 'Six atomic criteria required')
|
|
require(all(c['requirement'].strip() for c in scenario['criteria']), 'Empty criterion')
|
|
|
|
|
|
def packs(guidance):
|
|
result = {}
|
|
for condition in ('patterns', 'prose'):
|
|
parts = [guidance['intro']]
|
|
for number, item in enumerate(guidance['items'], 1):
|
|
heading = item['title'] if condition == 'patterns' else f'Decision {number}'
|
|
parts.append(f'## {heading}\n\n{item["body"]}')
|
|
parts.append(guidance['closing'])
|
|
result[condition] = '\n\n'.join(parts) + '\n'
|
|
return result
|
|
|
|
|
|
def prompt(packet, assignment, scenario):
|
|
return ((packet / 'instructions.txt').read_text() + '\nREFERENCE\n' +
|
|
(packet / f'pack-{assignment["condition"]}.md').read_text() +
|
|
'\nSCENARIO\n' + scenario['brief'] + '\n')
|
|
|
|
|
|
def score_template(scenarios):
|
|
return {'schema_version': 1, 'reviewer': {'name': None, 'role': None, 'independent_human': False,
|
|
'assignment_seen': None, 'prior_scores_seen': None},
|
|
'answers': {s['id'] + '-' + label: {
|
|
'available': None,
|
|
'ratings': {c['id']: {'value': None, 'quote': None, 'locator': None, 'reason': None}
|
|
for c in s['criteria']},
|
|
'unsupported_claims': None, 'ambiguity': [], 'suspected_condition_clues': []}
|
|
for s in scenarios for label in ('U', 'V')}}
|
|
|
|
|
|
def prepare(root=ROOT):
|
|
packet = root / PACKET
|
|
require(not (packet / 'freeze.json').exists(), 'Packet already frozen; do not overwrite')
|
|
index, pin, manifest = published(root, '0.1')
|
|
scenarios = read(packet / 'scenarios.json')
|
|
scenarios_check(scenarios)
|
|
old = read(root / 'research/evaluations/language-pilot-01/scenarios.json')
|
|
require(not ({s['brief'] for s in scenarios} & {s['brief'] for s in old}), 'Consumed brief reused')
|
|
guidance = read(packet / 'guidance.json')
|
|
core = {c['id']: c for c in index['candidates'] if c['subset'] == 'core'}
|
|
require(len(guidance['items']) == 8 and {i['candidate'] for i in guidance['items']} == set(core), 'Guidance must cover pinned core')
|
|
for item in guidance['items']:
|
|
require(item['title'] == core[item['candidate']]['title'] and item['body'].strip(), 'Guidance identity/body mismatch')
|
|
pack_text = packs(guidance)
|
|
for condition, text in pack_text.items():
|
|
write_new(packet / f'pack-{condition}.md', text)
|
|
equivalence = {'method': 'Only headings vary; all eight ordered bodies, introduction and closing are byte-identical.',
|
|
'blocks': [{'candidate': i['candidate'], 'body_sha256': digest(i['body'].encode()),
|
|
'source_path': core[i['candidate']]['path'], 'source_sha256': core[i['candidate']]['sha256']}
|
|
for i in guidance['items']],
|
|
'word_counts': {k: len(v.split()) for k, v in pack_text.items()}}
|
|
write_new(packet / 'pack-equivalence.json', equivalence)
|
|
assignments = []
|
|
for number, scenario in enumerate(scenarios):
|
|
conditions = ['patterns', 'prose'] if number % 2 == 0 else ['prose', 'patterns']
|
|
labels = ['U', 'V'] if secrets.randbelow(2) else ['V', 'U']
|
|
for condition, label in zip(conditions, labels):
|
|
assignments.append({'run': f'r{len(assignments)+1:02d}', 'scenario': scenario['id'],
|
|
'condition': condition, 'blind_id': scenario['id'] + '-' + label})
|
|
write_new(packet / 'assignment.json', assignments)
|
|
generated = ['pack-patterns.md', 'pack-prose.md', 'pack-equivalence.json', 'assignment.json', 'scores-template.json']
|
|
write_new(packet / 'scores-template.json', score_template(scenarios))
|
|
by_id = {s['id']: s for s in scenarios}
|
|
for assignment in assignments:
|
|
path = 'requests/' + assignment['run'] + '.txt'
|
|
write_new(packet / path, prompt(packet, assignment, by_id[assignment['scenario']]))
|
|
generated.append(path)
|
|
paths = [PACKET / name for name in (*INPUTS, *generated)] + [Path(p) for p in DEPENDENCIES]
|
|
write_new(packet / 'freeze.json', {
|
|
'schema_version': 1, 'packet': 'language-replication-02',
|
|
'prepared_at': datetime.now(timezone.utc).isoformat(),
|
|
'reference': dict(pin, version='0.1', corpus_revision=manifest['corpus_revision']),
|
|
'files': {str(p): digest((root / p).read_bytes()) for p in paths},
|
|
'sessions': 12, 'max_output_tokens_per_call': 1400,
|
|
'limits': ['Author-created fresh variants, not held-out M3 retrieval or independent validation.',
|
|
'No inference at preparation; execution needs explicit user authorization.']})
|
|
check(root)
|
|
|
|
|
|
def check(root=ROOT):
|
|
packet = root / PACKET
|
|
frozen = read(packet / 'freeze.json')
|
|
require(frozen['schema_version'] == 1 and frozen['sessions'] == 12 and frozen['max_output_tokens_per_call'] == 1400, 'Invalid freeze contract')
|
|
require(set(DEPENDENCIES) <= set(frozen['files']), 'Missing frozen dependencies')
|
|
for relative, expected in frozen['files'].items():
|
|
path = root / relative
|
|
require(path.resolve().is_relative_to(root.resolve()), 'Unsafe frozen path')
|
|
require(digest(path.read_bytes()) == expected, 'Frozen input changed: ' + relative)
|
|
_, pin, manifest = published(root, frozen['reference']['version'])
|
|
require(frozen['reference'] == dict(pin, version=manifest['version'], corpus_revision=manifest['corpus_revision']), 'Reference pin drift')
|
|
scenarios = read(packet / 'scenarios.json')
|
|
scenarios_check(scenarios)
|
|
guidance = read(packet / 'guidance.json')
|
|
for condition, text in packs(guidance).items():
|
|
require((packet / f'pack-{condition}.md').read_text() == text, 'Presentation body/heading mismatch')
|
|
assignments = read(packet / 'assignment.json')
|
|
require([a['run'] for a in assignments] == [f'r{i:02d}' for i in range(1, 13)], 'Invalid run identities')
|
|
by_id = {s['id']: s for s in scenarios}
|
|
for number, scenario in enumerate(scenarios):
|
|
pair = assignments[2*number:2*number+2]
|
|
require({a['scenario'] for a in pair} == {scenario['id']}, 'Invalid pairing')
|
|
require({a['blind_id'] for a in pair} == {scenario['id'] + '-U', scenario['id'] + '-V'}, 'Invalid masking labels')
|
|
require([a['condition'] for a in pair] == (['patterns', 'prose'] if number % 2 == 0 else ['prose', 'patterns']), 'Counterbalance changed')
|
|
for assignment in assignments:
|
|
path = packet / 'requests' / (assignment['run'] + '.txt')
|
|
require(path.read_text() == prompt(packet, assignment, by_id[assignment['scenario']]), 'Prompt changed or scoring material supplied')
|
|
require(str(path.relative_to(root)) in frozen['files'], 'Unfrozen prompt')
|
|
require(read(packet / 'scores-template.json') == score_template(scenarios), 'Blank score template changed')
|
|
return frozen, assignments, scenarios
|
|
|
|
|
|
def committed(root, frozen):
|
|
paths = [*frozen['files'], str(PACKET / 'freeze.json')]
|
|
for path in paths:
|
|
raw = subprocess.check_output(['git', 'show', 'HEAD:' + path], cwd=root, stderr=subprocess.DEVNULL)
|
|
require(raw == (root / path).read_bytes(), 'Commit frozen packet before inference')
|
|
|
|
|
|
def run(max_calls, root=ROOT, adapter_factory=None):
|
|
frozen, assignments, _ = check(root)
|
|
require(type(max_calls) is int and 0 < max_calls <= 12, 'Explicit ceiling of at most 12 calls required')
|
|
committed(root, frozen)
|
|
packet = root / PACKET
|
|
pending = []
|
|
for assignment in assignments:
|
|
directory = packet / 'raw' / assignment['run']
|
|
if directory.exists():
|
|
started = read(directory / 'started.json')
|
|
require(started['attempts_reserved'] == 1 and started['prompt_sha256'] == digest((packet / 'requests' / (assignment['run'] + '.txt')).read_bytes()), 'Invalid prior reservation')
|
|
else:
|
|
pending.append(assignment)
|
|
used = len(assignments) - len(pending)
|
|
require(used + len(pending) <= max_calls, 'Requested sessions exceed aggregate call ceiling')
|
|
if adapter_factory is None:
|
|
from research_bench import LLMConnectAdapter
|
|
adapter_factory = lambda: LLMConnectAdapter('http://127.0.0.1:18081', 'custodian-triage-balanced', 1400, diagnostics=True)
|
|
for assignment in pending:
|
|
directory = packet / 'raw' / assignment['run']
|
|
directory.mkdir(parents=True, exist_ok=False)
|
|
text = (packet / 'requests' / (assignment['run'] + '.txt')).read_text()
|
|
started = dict(assignment, attempts_reserved=1, prompt_sha256=digest(text.encode()),
|
|
freeze_sha256=digest((packet / 'freeze.json').read_bytes()), max_output_tokens=1400)
|
|
write_new(directory / 'started.json', started) # Reserve before any external call.
|
|
adapter = adapter_factory()
|
|
receipt = dict(started, endpoint=adapter.endpoint, config=adapter.config.to_dict())
|
|
then = time.monotonic()
|
|
try:
|
|
result = adapter.generate(SimpleNamespace(prompt=text, stage_id='decision-review'))
|
|
write_new(directory / 'answer.md', result.markdown)
|
|
write_new(packet / 'masked' / (assignment['blind_id'] + '.md'), result.markdown)
|
|
receipt.update(status='completed', output_sha256=digest(result.markdown.encode()), output_words=len(result.markdown.split()))
|
|
except Exception as exc:
|
|
receipt.update(status='failed', error_type=type(exc).__name__)
|
|
if adapter.rejected_content:
|
|
write_new(directory / 'rejected-response.md', adapter.rejected_content)
|
|
receipt.update(elapsed_seconds=round(time.monotonic() - then, 3), responses=adapter.responses)
|
|
write_new(directory / 'receipt.json', receipt)
|
|
print(json.dumps({'run': assignment['run'], 'status': receipt['status']}), flush=True)
|
|
|
|
|
|
def scores_check(sheet, scenarios, complete=False):
|
|
expected = score_template(scenarios)
|
|
require(sheet.get('schema_version') == 1 and set(sheet['answers']) == set(expected['answers']), 'Score answer inventory mismatch')
|
|
reviewer = sheet['reviewer']
|
|
require(type(reviewer['independent_human']) is bool, 'Invalid review independence declaration')
|
|
require(not reviewer['independent_human'] or reviewer['role'] == 'human', 'Model review cannot be independent human validation')
|
|
if complete:
|
|
require(isinstance(reviewer['name'], str) and reviewer['name'].strip() and reviewer['role'] in {'author-model', 'independent-model', 'human'}, 'Reviewer provenance required')
|
|
require(type(reviewer['assignment_seen']) is bool and type(reviewer['prior_scores_seen']) is bool, 'Exposure disclosure required')
|
|
for ident, answer in sheet['answers'].items():
|
|
require(answer['available'] is None or type(answer['available']) is bool, 'Invalid answer availability')
|
|
require(set(answer['ratings']) == set(expected['answers'][ident]['ratings']), 'Atomic criterion inventory mismatch')
|
|
if complete:
|
|
require(type(answer['available']) is bool, 'Answer availability required')
|
|
for rating in answer['ratings'].values():
|
|
value = rating['value']
|
|
require(value is None or type(value) is int and value in (0, 1), 'Atomic scores are binary or unreviewed null')
|
|
require(all(rating[k] is None or isinstance(rating[k], str) and rating[k].strip()
|
|
for k in ('quote', 'locator', 'reason')), 'Rating evidence must be text')
|
|
require(answer['available'] is not False or value is None, 'Unavailable answer cannot be scored zero')
|
|
if value is not None:
|
|
require(rating['reason'] and (rating['quote'] or rating['locator']), 'Rating evidence and reason required')
|
|
if complete and answer['available']:
|
|
require(value is not None, 'Available answer has unreviewed criterion')
|
|
claims = answer['unsupported_claims']
|
|
require(claims is None or isinstance(claims, list) and all(c['quote'] and c['reason'] for c in claims), 'Unsupported claims need evidence')
|
|
if complete and answer['available']:
|
|
require(claims is not None, 'Unsupported-claim review required')
|
|
require(all(isinstance(answer[k], list) and all(isinstance(v, str) for v in answer[k])
|
|
for k in ('ambiguity', 'suspected_condition_clues')), 'Review caveats must be lists')
|
|
return True
|
|
|
|
|
|
def summarize(sheet, root=ROOT):
|
|
frozen, assignments, scenarios = check(root)
|
|
scores_check(sheet, scenarios, complete=True)
|
|
totals = {c: Counter(completed_pairs=0, fulfilled=0, unsafe=0, criteria=0) for c in ('patterns', 'prose')}
|
|
pairs = []
|
|
for scenario in scenarios:
|
|
pair = [a for a in assignments if a['scenario'] == scenario['id']]
|
|
valid = True
|
|
for assignment in pair:
|
|
directory = root / PACKET / 'raw' / assignment['run']
|
|
receipt_path = directory / 'receipt.json'
|
|
rating = sheet['answers'][assignment['blind_id']]
|
|
if not receipt_path.exists():
|
|
valid = False
|
|
require(not rating['available'], 'Unexecuted answer cannot be marked available')
|
|
continue
|
|
receipt = read(receipt_path)
|
|
require(receipt['freeze_sha256'] == digest((root / PACKET / 'freeze.json').read_bytes()), 'Receipt freeze mismatch')
|
|
available = receipt['status'] == 'completed'
|
|
require(rating['available'] == available, 'Scoring availability differs from retained result')
|
|
if available:
|
|
raw = (directory / 'answer.md').read_bytes()
|
|
require(digest(raw) == receipt['output_sha256'], 'Retained answer changed')
|
|
require((root / PACKET / 'masked' / (assignment['blind_id'] + '.md')).read_bytes() == raw, 'Masked answer changed')
|
|
for item in [*rating['ratings'].values(), *rating['unsupported_claims']]:
|
|
if item.get('quote'):
|
|
require(item['quote'] in raw.decode(), 'Quoted score evidence not in retained answer')
|
|
else:
|
|
valid = False
|
|
results = {}
|
|
if valid:
|
|
for assignment in pair:
|
|
rating = sheet['answers'][assignment['blind_id']]
|
|
results[assignment['condition']] = {'fulfilled': sum(r['value'] for r in rating['ratings'].values()),
|
|
'unsafe': len(rating['unsupported_claims'])}
|
|
totals[assignment['condition']].update(completed_pairs=1, criteria=6, **results[assignment['condition']])
|
|
pairs.append({'scenario': scenario['id'], 'included': valid, 'results': results})
|
|
return {'schema_version': 1, 'reviewer': sheet['reviewer'], 'totals': {k: dict(v) for k, v in totals.items()},
|
|
'pairs': pairs, 'limits': ['Descriptive author-created replication, not proof of superiority/equivalence.',
|
|
'Reviewer role is declared provenance; no automatic independent-human or M1 claim.']}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('command', choices=['prepare', 'check', 'run', 'scores-check', 'summarize'])
|
|
parser.add_argument('--max-calls', type=int)
|
|
parser.add_argument('--scores', type=Path)
|
|
parser.add_argument('--complete', action='store_true')
|
|
args = parser.parse_args()
|
|
try:
|
|
if args.command == 'prepare':
|
|
prepare()
|
|
elif args.command == 'run':
|
|
run(args.max_calls)
|
|
elif args.command in {'scores-check', 'summarize'}:
|
|
require(args.scores, 'Score record required')
|
|
_, _, scenarios = check()
|
|
sheet = read(args.scores)
|
|
if args.command == 'summarize':
|
|
print(json.dumps(summarize(sheet), indent=2))
|
|
return
|
|
scores_check(sheet, scenarios, args.complete)
|
|
else:
|
|
check()
|
|
print('Replication packet/scoring contract verified; no empirical gate inferred.')
|
|
except (ValueError, TypeError, KeyError, OSError, subprocess.CalledProcessError) as exc:
|
|
parser.exit(1, f'Replication failed: {exc}\n')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|