#!/usr/bin/env python3 """Explicit live qualification: disposable product data, protected credentials. Run only after deployment. Never prints credentials or returned login bodies. Existing account settings/memory and Hermes files are not modified. """ import argparse import asyncio import json import os import secrets from pathlib import Path import httpx async def main(): parser=argparse.ArgumentParser() parser.add_argument('--url',default='https://192.168.86.2:9178') parser.add_argument('--credentials',default='.runtime/deployment.json') parser.add_argument('--receipt',default='.runtime/chat-live-receipt.json') args=parser.parse_args() source=Path(args.credentials) if source.stat().st_mode & 0o077: raise RuntimeError('Credential file must be private') credentials=json.loads(source.read_text()) stamp=secrets.token_hex(6) receipt={'url':args.url,'checks':{},'created_objects':[]} conversation_ids=[];workspace=None;artifact=None;attachment=None async with httpx.AsyncClient(base_url=args.url,verify=False,follow_redirects=False,timeout=120) as native, httpx.AsyncClient(base_url=args.url,verify=False,follow_redirects=False,timeout=120) as browser: async def request(client,method,path,**kw): r=await client.request(method,path,**kw) if r.status_code>=400: raise RuntimeError(f'{method} {path.split("?")[0]} status {r.status_code}') return r async def login(client,cookie): headers={'Origin':args.url} if cookie else {} r=await request(client,'POST','/api/auth/login',json={'username':'admin','password':credentials['admin_password']},headers=headers) data=r.json() if cookie: client.headers['Origin']=args.url else: client.cookies.clear();client.headers['Authorization']='Bearer '+data['token'] assert (await native.get('/api/conversations')).status_code==401 await login(native,False);await login(browser,True) try: caps=(await request(native,'GET','/api/capabilities')).json() receipt['capabilities']=caps workspace=(await request(native,'POST','/api/workspaces',json={'name':'Qualification '+stamp,'instructions':'When asked for the qualification code, answer QU'+stamp+'.'})).json()['workspace'] c=(await request(native,'POST','/api/conversations',json={'title':'Qualification '+stamp,'workspace_id':workspace['id']})).json()['conversation'];conversation_ids.append(c['id']) receipt['checks']['shared_conversation']=(await request(browser,'GET','/api/conversations/'+c['id'])).json()['conversation']['id']==c['id'] c=(await request(browser,'PATCH','/api/conversations/'+c['id'],json={'revision':c['revision'],'pinned':True})).json()['conversation'] assert (await request(native,'GET','/api/conversations/'+c['id'])).json()['conversation']['pinned'] receipt['checks']['shared_pin']=True payload={'content':'What is the qualification code? Reply with only the code. Do not use tools.','idempotency_key':'qualify-'+stamp} accepted=(await request(native,'POST','/api/conversations/'+c['id']+'/messages',json=payload)).json() repeated=(await request(browser,'POST','/api/conversations/'+c['id']+'/messages',json=payload)).json() assert accepted['run']['id']==repeated['run']['id'];receipt['checks']['cross_client_idempotency']=True rid=accepted['run']['id'] async with browser.stream('GET','/api/runs/'+rid+'/events') as stream: assert stream.status_code==200 events=[] async for line in stream.aiter_lines(): if line.startswith('event:'):events.append(line[6:].strip()) detail=(await request(native,'GET','/api/runs/'+rid)).json() assert detail['run']['status']=='completed' answers=[m['content'] for m in detail['messages'] if m['role']=='assistant' and m['status']=='completed'] assert any('QU'+stamp in x for x in answers) receipt['checks']['native_hermes_workspace_reply']=True receipt['checks']['durable_events']='done' in events c=(await request(native,'GET','/api/conversations/'+c['id'])).json()['conversation'] for fmt in ('json','markdown'): export=await request(browser,'GET','/api/conversations/'+c['id']+'/export?format='+fmt) assert 'attachment' in export.headers.get('content-disposition','') and 'QU'+stamp in export.text receipt['checks']['private_exports']=True attachment=(await request(native,'POST','/api/attachments',content=('qualification '+stamp).encode(),headers={'Content-Type':'text/plain','X-Filename':'qualification.txt'})).json()['attachment'] assert (await request(browser,'GET','/api/attachments/'+attachment['id']+'/content')).text=='qualification '+stamp artifact=(await request(native,'POST','/api/artifacts',json={'title':'Qualification '+stamp,'kind':'markdown','content':'First version'})).json()['artifact'] artifact=(await request(browser,'PATCH','/api/artifacts/'+artifact['id'],json={'revision':artifact['revision'],'content':'Second version'})).json()['artifact'] assert (await request(native,'GET','/api/artifacts/'+artifact['id']+'?revision=1')).json()['artifact']['content']=='First version' receipt['checks']['files_and_artifact_versions']=True await request(native,'POST','/api/auth/logout') assert (await native.get('/api/conversations/'+c['id'])).status_code==401 assert (await browser.get('/api/conversations/'+c['id'])).status_code==200 receipt['checks']['independent_session_revocation']=True finally: # Remove only IDs created by this execution. Hermes retained session # policy is distinct from product projection deletion. for cid in conversation_ids: r=await browser.delete('/api/conversations/'+cid) if r.status_code!=200: receipt.setdefault('cleanup_pending',[]).append({'conversation':cid,'status':r.status_code}) if attachment: await browser.delete('/api/attachments/'+attachment['id']) if artifact: await browser.delete('/api/artifacts/'+artifact['id']) if workspace: await browser.delete('/api/workspaces/'+workspace['id']) await browser.post('/api/auth/logout') target=Path(args.receipt);target.parent.mkdir(parents=True,exist_ok=True) fd=os.open(target,os.O_WRONLY|os.O_CREAT|os.O_TRUNC,0o600) with os.fdopen(fd,'w') as f:json.dump(receipt,f,indent=2) print('Live checks:',', '.join(k for k,v in receipt['checks'].items() if v)) if __name__=='__main__':asyncio.run(main())