Spaces:
Running
Running
Download verify-experiments.py from alphabell/Model-Observatory: direct link, hf CLI and curl.
- Browser
- Download file 21.7 kB
-
https://huggingface.co/spaces/alphabell/Model-Observatory/resolve/main/verify-experiments.py
- Command line
-
hf download hf://spaces/alphabell/Model-Observatory/verify-experiments.py
-
curl -L -o verify-experiments.py https://huggingface.co/spaces/alphabell/Model-Observatory/resolve/main/verify-experiments.py
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() | |