diff --git a/scripts/qualify-voice-live.py b/scripts/qualify-voice-live.py new file mode 100644 index 0000000..6bbd2dc --- /dev/null +++ b/scripts/qualify-voice-live.py @@ -0,0 +1,107 @@ +#!/usr/bin/env python3 +"""Explicit live fake-microphone qualification. Never physical microphone evidence. + +Reads ignored protected deployment credentials; emits only safe qualification facts. +Requires local synthetic PCM fixture and deployed Hope. No deployment changes. +""" +import argparse +import json +import re +import time +from pathlib import Path +from playwright.sync_api import sync_playwright, expect + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--url', default='https://192.168.86.2:9178') + parser.add_argument('--credentials', default='.runtime/deployment.json') + parser.add_argument('--wav', default='.runtime/voice-qualification/microphone.wav') + parser.add_argument('--timeout', type=int, default=180) + args = parser.parse_args() + values = json.loads(Path(args.credentials).read_text()) + fixture = Path(args.wav).resolve(strict=True) + receipt = {'mode': 'synthetic_browser_microphone_not_physical', 'connected': False, + 'user_transcript': False, 'assistant_transcript': False, + 'audio_track': False, 'audio_bytes': 0, 'errors': 0} + with sync_playwright() as playwright: + browser = playwright.chromium.launch(headless=True, args=[ + '--use-fake-ui-for-media-stream', '--use-fake-device-for-media-stream', + '--use-file-for-fake-audio-capture=' + str(fixture), + '--autoplay-policy=no-user-gesture-required']) + context = browser.new_context(ignore_https_errors=True, permissions=['microphone'], + viewport={'width': 1440, 'height': 1000}) + page = context.new_page() + page.add_init_script("window.__qualificationVoiceID=null;window.addEventListener('hope:voice-state',event=>{if(event.detail?.conversation_id)window.__qualificationVoiceID=event.detail.conversation_id;});") + page.on('pageerror', lambda error: receipt.update(errors=receipt['errors'] + 1)) + # Read-only receiver stats: exposes no tokens, credentials or private payloads. + page.add_init_script('''window.__voiceStats={bytes:0,packets:0}; + const original=RTCPeerConnection.prototype.getStats; + const connections=[]; const Native=window.RTCPeerConnection; + window.RTCPeerConnection=new Proxy(Native,{construct(target,args){ + const connection=Reflect.construct(target,args); connections.push(connection); return connection;}}); + setInterval(async()=>{for(const connection of connections){try{ + const report=await original.call(connection); + report.forEach(item=>{if(item.type==='inbound-rtp'&&(item.kind==='audio'||item.mediaType==='audio')){ + window.__voiceStats.bytes=Math.max(window.__voiceStats.bytes,item.bytesReceived||0); + window.__voiceStats.packets=Math.max(window.__voiceStats.packets,item.packetsReceived||0);}}); + }catch{}}},500);''') + try: + page.goto(args.url.rstrip('/') + '/', wait_until='networkidle') + expect(page.locator('#loginSubmit')).to_be_enabled(timeout=30000) + page.locator('#loginUsername').fill('admin') + page.locator('#loginPassword').fill(values['admin_password']) + page.locator('#loginSubmit').click() + expect(page.locator('#primaryInput')).to_be_enabled(timeout=30000) + # Prefer restored prominent voice entry, tolerate current compatibility ID. + for selector in ['#talkToHope', '#talkVoice', '#openVoice']: + if page.locator(selector).count() and page.locator(selector).is_visible(): + page.locator(selector).click() + break + expect(page.locator('#voiceModeBtn')).to_be_visible(timeout=10000) + page.locator('#voiceModeBtn').click() + expect(page.locator('#startBtn')).to_be_enabled(timeout=10000) + page.locator('#startBtn').click() + deadline = time.monotonic() + args.timeout + while time.monotonic() < deadline: + status = page.locator('#status').inner_text() + body = page.locator('#messages').inner_text() if page.locator('#messages').count() else '' + receipt['connected'] |= 'Microphone on' in status or 'Secure room' in status + # Synthetic question and expected answer facts, not transcript dumps. + user_text = ' '.join(page.locator('#messages .message-row.user').all_inner_texts()).lower() + receipt['user_transcript'] |= bool(re.search(r'(?:two|2)\s*(?:plus|\+)\s*(?:two|2)', user_text)) + assistant_text = ' '.join(page.locator('#messages .message-row.agent').all_inner_texts()).lower() + receipt['assistant_transcript'] |= 'four' in assistant_text or '4' in assistant_text + receipt['audio_track'] |= page.locator('audio[id^="audio-"]').count() > 0 + stats = page.evaluate('window.__voiceStats') + receipt['audio_bytes'] = stats['bytes'] + if receipt['assistant_transcript'] and receipt['audio_bytes'] > 0: + break + page.wait_for_timeout(500) + receipt['audio_playing'] = page.evaluate('''Array.from(document.querySelectorAll('audio[id^="audio-"]')).some(el=>!el.paused&&el.srcObject&&el.srcObject.getAudioTracks().some(track=>track.readyState==='live'))''') + output = Path('.runtime/voice-qualification/live-receipt.json') + output.write_text(json.dumps(receipt, indent=2) + '\n') + print(json.dumps(receipt, sort_keys=True)) + assert receipt['connected'], 'synthetic_voice_connection_failed' + assert receipt['user_transcript'], 'synthetic_voice_stt_not_observed' + assert receipt['assistant_transcript'], 'synthetic_voice_answer_not_observed' + assert receipt['audio_track'] and receipt['audio_bytes'] > 0, 'remote_tts_audio_not_observed' + finally: + if page.locator('#stopBtn').count() and page.locator('#stopBtn').is_visible() and page.locator('#stopBtn').is_enabled(): + page.locator('#stopBtn').click() + # Delete only the ID captured for this newly created qualification + # voice conversation. Never enumerate or remove other account chats. + try: + cleanup = page.evaluate('''async()=>{let removed=false;const cid=window.__qualificationVoiceID; + if(cid){try{const result=await HopeAuth.request('/api/conversations/'+encodeURIComponent(cid),{method:'DELETE'});removed=result.ok===true;}catch{}} + try{await HopeAuth.request('/api/auth/logout',{method:'POST'});}catch{} + return {created_conversation_seen:!!cid,removed};}''') + print(json.dumps({'qualification_cleanup': cleanup}, sort_keys=True)) + except Exception: + print('qualification_cleanup_unavailable') + context.close() + browser.close() + + +if __name__ == '__main__': + main()