Model-Observatory / verify-experiments.py
alphabell-research's picture
Fix explicit experiment entry paths for static hosting (#4)
933c910
Raw History Blame Contribute Delete
21.7 kB
"""Verify the recorded alphaBell experiments in one sequential browser session."""
import argparse
import json
import pathlib
import traceback
from urllib.parse import parse_qs, urljoin, urlparse
from playwright.sync_api import expect, sync_playwright
def pair_order(item):
value = 2166136261
for char in item['id']:
value = ((value ^ ord(char)) * 16777619) & 0xffffffff
return list(reversed(item['conditions'])) if value % 2 else item['conditions']
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--base', default='http://127.0.0.1:8917')
parser.add_argument('--chromium')
parser.add_argument('--out', default='reports-experiments')
args = parser.parse_args()
base = args.base.rstrip('/') + '/'
lab_url = urljoin(base, 'experiments/index.html')
out = pathlib.Path(args.out)
out.mkdir(parents=True, exist_ok=True)
checks, errors, failed_requests = [], [], []
report = {'checks': checks, 'browser_errors': errors, 'failed_requests': failed_requests, 'browser_closed': False}
def passed(name):
checks.append(name)
print('PASS ' + name, flush=True)
def observe(page):
page.on('pageerror', lambda error: errors.append(str(error)))
page.on('response', lambda response: failed_requests.append({'url': response.url, 'status': response.status}) if response.status >= 400 else None)
def lab_ready(page):
expect(page.locator('#caseSelect')).to_be_visible(timeout=45000)
def choose(page, experiment, case_id=None):
page.locator('[data-experiment="' + experiment + '"]').click()
if case_id:
page.locator('#caseSelect').select_option(case_id)
expect(page.locator('[data-experiment="' + experiment + '"]')).to_have_attribute('aria-current', 'page')
def outputs(page):
return page.locator('#experiment .condition .output').evaluate_all('(elements) => elements.map(element => element.textContent)')
def download_json(page, locator, name):
with page.expect_download() as pending:
locator.click()
path = out / name
pending.value.save_as(path)
return json.loads(path.read_text())
def screenshot(page, name):
# Start at the top so full-page captures do not relocate off-screen fixed links.
page.evaluate('window.scrollTo(0,0)')
page.screenshot(path=str(out / name), full_page=True)
def assert_metrics(page, condition, card):
actual = card.locator('.metrics > div').evaluate_all('(rows) => Object.fromEntries(rows.map(row => [row.querySelector("dt").textContent,row.querySelector("dd").textContent]))')
expected = page.evaluate('''object => Object.fromEntries(Object.entries(object).filter(([k,v])=>v!==undefined&&v!==null).map(([key,value]) => [key.replace(/_/g,' ').replace(/\\bms\\b/g,'(ms)').replace(/\\bseconds\\b/g,'(s)'),typeof value==='number'?(Number.isInteger(value)?value.toLocaleString('en-US'):value.toLocaleString('en-US',{maximumFractionDigits:5})):typeof value==='boolean'?(value?'Yes':'No'):typeof value==='object'?JSON.stringify(value):String(value)]))''', condition.get('metrics', {}))
assert actual == expected, {'actual': actual, 'expected': expected}
browser = None
try:
with sync_playwright() as playwright:
launch = {'headless': True, 'args': ['--disable-dev-shm-usage']}
if args.chromium:
launch['executable_path'] = args.chromium
browser = playwright.chromium.launch(**launch)
try:
context = browser.new_context(viewport={'width': 1440, 'height': 1100}, permissions=['clipboard-read', 'clipboard-write'], accept_downloads=True)
page = context.new_page()
observe(page)
response = page.request.get(urljoin(lab_url, 'data/results.json'))
assert response.ok
data = response.json()
page.goto(lab_url)
lab_ready(page)
assert len(data['experiments']['quant']['cases']) == 12
assert len(data['experiments']['reroute']['cases']) == 6
assert len(data['experiments']['surgery']['cases']) == 6
assert len(data['experiments']['memory']['cases']) == 6
passed('All four labs load their real 30 recorded cases')
choose(page, 'quant')
for item in data['experiments']['quant']['cases']:
page.locator('#caseSelect').select_option(item['id'])
ordered = pair_order(item)
assert outputs(page) == [condition['output'] or '(The model produced no answer.)' for condition in ordered]
expect(page.locator('#experiment .condition .tag')).to_have_count(0)
expect(page.locator('#experiment .condition .metrics')).to_have_count(0)
assert page.locator('#experiment .condition h4').all_text_contents() == ['Answer A', 'Answer B']
low_index = next(index for index, condition in enumerate(ordered) if condition['quant'].upper().startswith('Q4'))
page.get_by_role('button', name='Answer ' + 'AB'[low_index] + ' is 4-bit', exact=True).click()
expect(page.locator('.result-note')).to_contain_text('You spotted it.')
expect(page.locator('#experiment .condition .tag')).to_have_count(2)
for index, condition in enumerate(ordered):
card = page.locator('#experiment .condition').nth(index)
expect(card.locator('.tag')).to_have_text(condition['model'] + ' · ' + condition['quant'])
assert_metrics(page, condition, card)
passed('Quant ' + item['id'] + ': deterministic blind order, guess, actual labels and every measured metric')
expect(page.locator('.local-tally')).to_contain_text('12 / 12 correct')
first_quant = data['experiments']['quant']['cases'][0]
page.goto(lab_url + '?experiment=quant&case=' + first_quant['id'])
lab_ready(page)
assert outputs(page) == [condition['output'] or '(The model produced no answer.)' for condition in pair_order(first_quant)]
expect(page.locator('#experiment .condition .tag')).to_have_count(0)
expect(page.locator('.local-tally')).to_contain_text('12 / 12 correct')
page.get_by_role('button', name='Reveal without guessing', exact=True).click()
screenshot(page, 'quant-desktop.png')
quant_case = download_json(page, page.get_by_role('button', name='Download this case ↓', exact=True), 'quant-case.json')
assert quant_case['case']['id'] == first_quant['id']
passed('Quant local tally persists through reload, reveal resets, and case download parses')
choose(page, 'memory')
for item in data['experiments']['memory']['cases']:
page.locator('#caseSelect').select_option(item['id'])
expect(page.locator('#experiment .condition')).to_have_count(0)
expect(page.get_by_role('button', name='Reveal model answers', exact=True)).to_be_disabled()
for choice in item['choices']:
expect(page.get_by_role('button', name=choice, exact=True)).to_be_enabled()
page.get_by_role('button', name=str(item['ground_truth']), exact=True).click()
expect(page.get_by_role('button', name='Reveal model answers', exact=True)).to_be_enabled()
expect(page.locator('#experiment .condition')).to_have_count(0)
page.get_by_role('button', name='Reveal model answers', exact=True).click()
expect(page.locator('.result-note').first).to_contain_text('Expected answer: ' + str(item['ground_truth']))
assert outputs(page) == [condition['output'] or '(The model produced no answer.)' for condition in item['conditions']]
for index, condition in enumerate(item['conditions']):
assert_metrics(page, condition, page.locator('#experiment .condition').nth(index))
passed('Memory ' + item['id'] + ': answer gate, ground truth, both real model outputs and checks')
screenshot(page, 'memory-desktop.png')
memory_case = download_json(page, page.get_by_role('button', name='Download this case ↓', exact=True), 'memory-case.json')
assert memory_case['experiment'] == 'memory'
choose(page, 'surgery')
for item in data['experiments']['surgery']['cases']:
page.locator('#caseSelect').select_option(item['id'])
expect(page.locator('#experiment .condition')).to_have_count(3)
expect(page.locator('#experiment .notice')).to_contain_text('activations during execution')
expect(page.locator('#experiment .notice')).to_contain_text('does not rewrite the model weights')
assert item['intervention']['weights_modified'] is False
assert [condition['settings']['intervention_scale'] for condition in item['conditions']] == [1, .5, 0]
assert outputs(page) == [condition['output'] or '(The model produced no answer.)' for condition in item['conditions']]
for index, condition in enumerate(item['conditions']):
card = page.locator('#experiment .condition').nth(index)
expect(card.locator('h4')).to_have_text(condition['label'])
assert_metrics(page, condition, card)
passed('Surgery ' + item['id'] + ': genuine activation disclosure, 1/0.5/0 conditions and exact measured metrics')
screenshot(page, 'surgery-desktop.png')
surgery_case = download_json(page, page.get_by_role('button', name='Download this case ↓', exact=True), 'surgery-case.json')
assert surgery_case['experiment'] == 'surgery'
choose(page, 'reroute')
for item in data['experiments']['reroute']['cases']:
page.locator('#caseSelect').select_option(item['id'])
assert outputs(page) == [condition['output'] or '(The model produced no answer.)' for condition in item['conditions']]
expect(page.locator('#experiment .condition-link')).to_have_count(2)
routing = [condition for condition in item['conditions'] if condition.get('routing', {}).get('tokens')]
expect(page.locator('.routing-panel')).to_have_count(len(routing))
if routing:
max_frames = max(len(condition['routing']['tokens']) for condition in routing)
for step in sorted({0, max_frames - 1}):
page.locator('#routingStep').fill(str(step))
for index, condition in enumerate(routing):
panel = page.locator('.routing-panel').nth(index)
expect(panel.locator('.routing-label')).to_contain_text('block ' + str(condition['routing']['layer']))
tokens = condition['routing']['tokens']
expected = tokens[step] if step < len(tokens) else None
active = panel.locator('.expert-dot').evaluate_all('(dots) => dots.flatMap((dot,index) => dot.classList.contains("active") ? [index] : [])')
assert active == sorted(expected['experts'] if expected else [])
expect(panel.locator('.routing-token')).to_have_text(expected['token'] if expected else 'This answer has ended.')
page.locator('#routingStep').focus()
page.keyboard.press('Home')
page.keyboard.press('ArrowRight')
assert page.locator('#routingStep').input_value() == str(min(1, max_frames - 1))
if routing:
passed('Prompt pair ' + item['id'] + ': captured answers, correct block, exact selected experts and token stepping')
else:
passed('Prompt pair ' + item['id'] + ': captured Qwen answers and trace links without an invented expert grid')
choose(page, 'reroute', data['experiments']['reroute']['cases'][0]['id'])
screenshot(page, 'reroute-desktop.png')
reroute_case = download_json(page, page.get_by_role('button', name='Download this case ↓', exact=True), 'reroute-case.json')
assert reroute_case['experiment'] == 'reroute'
all_results = download_json(page, page.locator('#downloadResults'), 'all-results.json')
fixtures = download_json(page, page.locator('#downloadFixtures'), 'all-prompts.json')
assert all_results == data
assert fixtures['schema_version'] == 1
for experiment in fixtures['experiments'].values():
for item in experiment['cases']:
for condition in item['conditions']:
assert 'output' not in condition and isinstance(condition['prompt'], str)
passed('All four case downloads, full results and prompt-only fixtures contain parseable actual evidence')
choose(page, 'memory', data['experiments']['memory']['cases'][-1]['id'])
page.get_by_role('button', name='Share this case ↗', exact=True).click()
expect(page.locator('#toast')).to_have_text('Case link copied.')
shared = page.evaluate('navigator.clipboard.readText()')
query = parse_qs(urlparse(shared).query)
assert query['experiment'] == ['memory'] and query['case'] == [data['experiments']['memory']['cases'][-1]['id']]
page.goto(shared)
lab_ready(page)
assert page.locator('#caseSelect').input_value() == data['experiments']['memory']['cases'][-1]['id']
expect(page.locator('#experiment .condition')).to_have_count(0)
passed('Copied share URL reloads the exact case without leaking a reveal')
page.locator('[data-experiment="quant"]').focus()
page.keyboard.press('Enter')
expect(page.locator('[data-experiment="quant"]')).to_have_attribute('aria-current', 'page')
page.locator('[data-experiment="surgery"]').focus()
page.keyboard.press('Enter')
expect(page.locator('[data-experiment="surgery"]')).to_have_attribute('aria-current', 'page')
page.go_back()
expect(page.locator('[data-experiment="quant"]')).to_have_attribute('aria-current', 'page')
page.go_forward()
expect(page.locator('[data-experiment="surgery"]')).to_have_attribute('aria-current', 'page')
passed('Keyboard experiment navigation and browser back/forward restore tab and case')
page.locator('#resetGuesses').click()
saved = page.evaluate('JSON.parse(localStorage.getItem("alphabell-experiments-v1"))')
assert saved == {'quant': {}, 'memory': {}}
choose(page, 'quant')
expect(page.locator('.local-tally')).to_contain_text('0 / 0 correct')
passed('Clear local guesses resets saved tally and reveals')
context.close()
for width in [390, 320]:
context = browser.new_context(viewport={'width': width, 'height': 844}, is_mobile=True, has_touch=True)
page = context.new_page()
observe(page)
page.goto(lab_url)
lab_ready(page)
for key in ['reroute', 'quant', 'surgery', 'memory']:
choose(page, key)
if key == 'quant':
page.get_by_role('button', name='Reveal without guessing', exact=True).click()
elif key == 'memory':
first = data['experiments']['memory']['cases'][0]
page.get_by_role('button', name=str(first['ground_truth']), exact=True).click()
page.get_by_role('button', name='Reveal model answers', exact=True).click()
assert page.evaluate('document.documentElement.scrollWidth <= innerWidth'), {'width': width, 'tab': key}
if width == 390:
screenshot(page, key + '-phone-390.png')
passed(str(width) + 'px phone: all four revealed labs fit the viewport')
context.close()
context = browser.new_context(viewport={'width': 1440, 'height': 1100}, permissions=['clipboard-read', 'clipboard-write'])
page = context.new_page()
observe(page)
target = data['experiments']['reroute']['cases'][0]['conditions'][0]
trace_link = urljoin(lab_url, target['trace_url'])
page.goto(trace_link)
page.wait_for_function('typeof S!=="undefined"&&S.trace&&!S.loadingTrace', timeout=90000)
wanted = parse_qs(urlparse(trace_link).query)['capture'][0]
assert page.evaluate('S.trace.id') == wanted
assert page.locator('#prompt').input_value() == target['prompt']
assert page.evaluate('S.frames.map(frame=>frame.text).join("")') == target['output']
assert page.locator('#history .history-row').count() >= 13
page.locator('#modelSelect').select_option('qwen')
page.wait_for_function('S.model.id==="qwen"&&!S.switching&&!S.loadingTrace', timeout=90000)
assert 'capture' not in parse_qs(urlparse(page.url).query)
page.locator('#modelSelect').select_option('flint')
page.wait_for_function('S.model.id==="flint"&&!S.switching&&!S.loadingTrace', timeout=90000)
assert page.evaluate('S.trace.id') == wanted
assert parse_qs(urlparse(page.url).query)['capture'] == [wanted]
history_target = next(item for item in data['experiments']['memory']['cases'] if item['conditions'][0]['model'] == 'Recursive Flint')
page.locator('#history .history-row').filter(has_text=history_target['title']).first.click()
page.wait_for_function('S.trace.id.startsWith("memory-")&&!S.loadingTrace', timeout=90000)
assert page.locator('#prompt').input_value() == history_target['conditions'][0]['prompt']
assert parse_qs(urlparse(page.url).query)['capture'][0] == page.evaluate('S.trace.id')
passed('Main viewer opens exact trace query, restores per-model capture and history selection updates query')
page.goto(urljoin(lab_url, 'sound.html'))
expect(page.locator('#soundPlay')).to_be_enabled(timeout=45000)
page.locator('#soundModel').select_option('qwen')
page.wait_for_function('document.querySelector("#soundStatus").textContent.includes("44 measured")&&!document.querySelector("#soundPlay").disabled', timeout=45000)
expect(page.locator('#soundLayer option')).to_have_count(48)
passed('Sonification still verifies the updated Qwen source capture')
context.close()
assert not errors, errors
assert not failed_requests, failed_requests
passed('No uncaught browser errors or failed release requests in successful flows')
context = browser.new_context(viewport={'width': 390, 'height': 844})
page = context.new_page()
page.on('pageerror', lambda error: errors.append(str(error)))
fail = [True]
def route_results(route):
if fail[0]:
fail[0] = False
route.fulfill(status=503, body='Temporary test failure')
else:
route.continue_()
page.route('**/experiments/data/results.json', route_results)
page.goto(lab_url)
expect(page.get_by_role('alert')).to_contain_text('could not be opened')
expect(page.locator('#evidence')).to_be_hidden()
page.get_by_role('button', name='Try again', exact=True).click()
lab_ready(page)
expect(page.locator('#evidence')).to_be_visible()
assert not errors, errors
passed('Failed results fetch hides evidence and Retry loads the real cases')
context.close()
finally:
browser.close()
browser = None
report['browser_closed'] = True
except Exception:
report['failure'] = traceback.format_exc()
raise
finally:
report['passed'] = len(checks)
(out / 'results.json').write_text(json.dumps(report, indent=2) + '\n')
print(json.dumps({'passed': len(checks), 'browser_closed': report['browser_closed']}))
if __name__ == '__main__':
main()