LazyBoy2/tests/recovery_flow.py

"""Recovery choices keep the same task, goal history, plan and browser profile."""
import json,tempfile,threading,sys
from pathlib import Path
from http.server import ThreadingHTTPServer
from runtime_flow import Provider,Run,saved
from cli_flow import response,tool,paired

def main():
 with tempfile.TemporaryDirectory(prefix='gb-recovery-') as root,ThreadingHTTPServer(('127.0.0.1',0),Provider) as server:
  server.errors=[];threading.Thread(target=server.serve_forever,daemon=True).start();runs=[]
  try:
   replies=[response(tool('report_blocked',{'reason':'登入無法完成，兩次嘗試都停在驗證頁。','options':['由我在目前瀏覽器處理登入','先繼續寫文案']})),response(tool('external_write_file',{'path':'draft.txt','content':'draft without login'})),response(tool('report_done',{'message':'文案已備妥，登入仍未完成。'}))]
   def callback(req):
    if len(replies)==2:
     paired(req['messages'])
     assert any(m['role']=='user' and '先繼續寫文案' in m.get('content','') for m in req['messages']),req
     assert any(m['role']=='user' and m.get('content')=='complete the original task' for m in req['messages']),req
    return replies.pop(0)
   server.callback=callback
   run=Run(root,server);runs.append(run);run.wait_line('需要你的回覆');run.finish();before=saved(root);assert before['pending_question']['kind']=='recovery'
   run=Run(root,server,['run','--session',before['id'],'b']);runs.append(run);run.finish();after=saved(root)
   assert before['id']==after['id'] and not after['pending_question'];assert (Path(root)/'draft.txt').exists()
   print('PASS blocker -> letter choice -> alternate work in same session',flush=True)
   server.callback=lambda _:response(tool('report_blocked',{'reason':'Still blocked','options':['Try a different route']}))
   run=Run(root,server);runs.append(run);run.wait_line('需要你的回覆');run.finish();pending=saved(root)
   assert pending['pending_question']['options'][-1]=='停止這份工作'
   requests=len(server.requests)
   run=Run(root,server,['run','--session',pending['id'],'2']);runs.append(run);run.finish(code=130)
   assert saved(root)['last_verdict']=='cancelled' and not saved(root)['pending_question']
   assert len(server.requests)==requests
   print('PASS explicit stop choice cancels without another model call',flush=True)
   if '--browser' in sys.argv:
    replies=[response(tool('browser_navigate',{'url':f'http://127.0.0.1:{server.server_port}/fixture'})),response(tool('browser_handoff',{'reason':'請在原頁面處理測試登入','options':['先寫文案']}))]
    server.callback=lambda _:replies.pop(0)
    run=Run(root,server);runs.append(run);run.wait_line('需要你的回覆');run.finish()
    previous=saved(root);assert previous['pending_question']['kind']=='handoff'
    replies=[response(tool('browser_read_page',{})),response(tool('report_done',{'message':'original browser restored and inspected'}))]
    def restored(req):
     if len(replies)==2:
      assert any(m['role']=='user' and '我已經登入了' in m.get('content','') for m in req['messages']),req
      assert not any('"login_verified":true' in m.get('content','') for m in req['messages']),req
     else:
      assert any(m['role']=='tool' and 'browser checkpoint evidence' in m.get('content','') for m in req['messages']),req
     return replies.pop(0)
    server.callback=restored
    run=Run(root,server,['run','--session',previous['id'],'我已經登入了']);runs.append(run);run.finish()
    assert saved(root)['id']==previous['id']
    print('PASS parked handoff restores original browser for inspection; human reply is not proof of login',flush=True)
   assert not server.errors,server.errors
  finally:
   for run in runs:
    if run.process.poll() is None:run.process.terminate()
   server.shutdown()
if __name__=='__main__':main()