55 lines
3.8 KiB
Python
55 lines
3.8 KiB
Python
"""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()
|