53 lines
3.7 KiB
Python
53 lines
3.7 KiB
Python
"""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()
|