LazyBoy2/tests/live_team.py

"""Opt-in paid-model team test. Existing credentials, disposable local workspace."""
import json,os,signal,socket,subprocess,tempfile,time
from pathlib import Path
BINARY=Path(__file__).resolve().parents[1]/'target/debug/lazyboy'
def main():
 with tempfile.TemporaryDirectory(prefix='gb-live-team-',dir='/tmp') as root:
  root=Path(root);data=root/'data';workspace=root/'work';workspace.mkdir();(workspace/'items.txt').write_text('banana\napple\nbanana\npear\n')
  env={**os.environ,'LAZYBOY_DATA_DIR':str(data),'LAZYBOY_MAX_ROUNDS_TOTAL':'16','LAZYBOY_BROWSER_HEADED':'0'}
  with (root/'daemon.log').open('w') as log:
   p=subprocess.Popen([str(BINARY),'serve'],cwd=workspace,env=env,stdout=log,stderr=log)
   def rpc(op,agent=None,**kw):
    with socket.socket(socket.AF_UNIX) as s:
     s.settimeout(10);s.connect(str(data/'service.sock'));s.sendall((json.dumps({'op':op,'agent':agent,**kw})+'\n').encode());v=json.loads(s.makefile().readline());assert 'error' not in v,v;return v
   try:
    for _ in range(100):
     if (data/'service.sock').exists():break
     assert p.poll() is None,(root/'daemon.log').read_text();time.sleep(.1)
    for name in ('main','analyst'):rpc('create',name=name,cwd=str(workspace))
    prompt='請用 delegate_task 把這份工作交給既有的 analyst agent，讓他在背景完成，你先簡短回覆已交辦即可：讀目前工作區 items.txt，去重並依字母排序寫成 clean.txt，每行一筆，再用 read_file 實際讀回確認。請讓 analyst 完成後回報產物路徑與驗證內容。不要上網，不要新增其他工作。'
    rpc('chat','main',message=prompt);after=0;tasks=[];replied=False;pinged=False
    deadline=time.monotonic()+300
    while time.monotonic()<deadline:
     for e in rpc('events','main',after=after)['events']:
      after=e['id'];kind=e['kind'];payload=e['payload']
      if kind in ('reply','task_queued','task_ended'):print(kind,json.dumps(payload,ensure_ascii=False),flush=True)
      if kind=='reply' and tasks and all(t['state']=='terminal' for t in tasks):replied=True
     tasks=rpc('tasks','main')
     if tasks and not pinged:rpc('chat','main',message='謝謝，我還可以在這裡繼續聊天吧？不用改動背景工作。');pinged=True
     roots=[t for t in tasks if t['parent_id'] is None]
     if roots and all(t['state']=='terminal' for t in roots):
      if not all(t['verdict'] in ('done','answer') for t in roots):
       import sqlite3
       db=sqlite3.connect(data/'team.sqlite3')
       for (raw,) in db.execute('SELECT data FROM tasks'):
        saved=json.loads(raw)
        for m in saved['session']['messages']:
         if m.get('tool_calls'):print('calls:',[c['function']['name'] for c in m['tool_calls']],flush=True)
         if m['role']=='tool' and 'error' in m.get('content',''):print('tool feedback:',m['content'],flush=True)
       raise AssertionError(roots)
      if replied:break
     time.sleep(.5)
    else:raise AssertionError(('timed out',tasks))
    assert (workspace/'clean.txt').read_text().splitlines()==['apple','banana','pear']
    assert tasks and all(t['requests']<=t['limit'] for t in tasks)
    import sqlite3
    db=sqlite3.connect(data/'team.sqlite3');saved=[json.loads(row[0]) for row in db.execute('SELECT data FROM tasks')]
    assert any(any(call['function']['name']=='external_read_file' and 'clean.txt' in call['function']['arguments'] for message in t['session']['messages'] for call in message.get('tool_calls',[])) for t in saved)
    print('PASS live model: existing-agent delegation, concurrent chat, verified artifact and returned report',flush=True)
    print('Root task model requests:',[t['requests'] for t in tasks if t['parent_id'] is None],flush=True)
   finally:
    if p.poll() is None:p.send_signal(signal.SIGINT);p.wait(timeout=10)
if __name__=='__main__':main()