verify-experiments.py

"""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()