# End-to-end test of a brain on a server working on a person's computer, all on this machine and # all made up for the test: an account server, a headless brain, a headless "hands" computer (it # also plays the person's screen, as a Mac showing the brain does), and a scripted fake model. # Never touches a real account, and no Docker box (the bot's own computer is a sealed folder). # Build first (cargo build --release; and in relay/), then, from app/: # python3 -I deploy/test/remote_e2e.py . # On a Mac the terminal needs 輔助使用 (the hands open Calculator through agent-desktop). import json, os, shutil, socket, subprocess, sys, tempfile, threading, time, urllib.request, urllib.error from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer APP = os.path.abspath(sys.argv[1]) BIN = os.path.join(APP, 'target/release/anybot') RELAY = os.path.join(APP, 'relay/target/release/anybot-relay') WORK = tempfile.mkdtemp(prefix='anybot-e2e-') USER, PW = 'e2euser', 'e2e-pass-12345' HANDS = 'E2E手腳' results = [] def free_port(): s = socket.socket(); s.bind(('127.0.0.1', 0)); p = s.getsockname()[1]; s.close(); return p def check(name, ok, detail=''): results.append((name, bool(ok), detail)) print(('PASS ' if ok else 'FAIL ') + name + ('' if ok else f' — {detail}'), flush=True) # ---------- the scripted model ---------- def tool_call(name, args): return {'tool': (name, json.dumps(args, ensure_ascii=False))} SCRIPTS = { 'SC_LIST': [tool_call('list_computers', {})], 'SC_OPEN': [tool_call('computer_use_open_app', {'app': 'Calculator', 'computer': HANDS})], 'SC_SHELL': [tool_call('shell', {'command': 'echo e2e-$((6*7))', 'computer': HANDS})], 'SC_REDIRECT': [tool_call('shell', {'command': 'echo moved-here', 'computer': HANDS})], 'SC_HANDOFF': [tool_call('hand_over_computer', {'instruction': 'E2E：請按「我完成了」', 'computer': HANDS})], 'SC_OWN': [tool_call('shell', {'command': 'echo on-own-$((2+3))'})], } class Model(BaseHTTPRequestHandler): def log_message(self, *a): pass def do_GET(self): body = json.dumps({'data': [{'id': 'e2e'}]}).encode() self.send_response(200); self.send_header('Content-Type', 'application/json'); self.send_header('Content-Length', str(len(body))); self.end_headers(); self.wfile.write(body) def do_POST(self): n = int(self.headers.get('Content-Length', '0') or 0) req = json.loads(self.rfile.read(n) or b'{}') msgs = req.get('messages', []) last_user = max((i for i, m in enumerate(msgs) if m.get('role') == 'user'), default=-1) text = msgs[last_user].get('content', '') if last_user >= 0 else '' if isinstance(text, list): text = ' '.join(p.get('text', '') for p in text if isinstance(p, dict)) done = sum(1 for m in msgs[last_user + 1:] if m.get('role') == 'tool') last_tool = next((m.get('content', '') for m in reversed(msgs) if m.get('role') == 'tool'), '') script = next((v for k, v in SCRIPTS.items() if k in text), []) if done < len(script): name, args = script[done]['tool'] chunk = {'choices': [{'delta': {'tool_calls': [{'index': 0, 'id': f'c{int(time.time()*1000)}', 'function': {'name': name, 'arguments': args}}]}}]} else: chunk = {'choices': [{'delta': {'content': 'E2E 完成：' + (last_tool if isinstance(last_tool, str) else '')[:300]}}]} self.send_response(200); self.send_header('Content-Type', 'text/event-stream'); self.end_headers() self.wfile.write(('data: ' + json.dumps(chunk, ensure_ascii=False) + '\n\ndata: [DONE]\n\n').encode()) # ---------- relay helpers ---------- RP = free_port(); MP = free_port() RURL = f'http://127.0.0.1:{RP}' def http(method, url, body=None, token=None, timeout=60, raw=False): data = None if body is None else json.dumps(body).encode() req = urllib.request.Request(url, method=method, data=data, headers={'Content-Type': 'application/json', **({'Authorization': 'Bearer ' + token} if token else {})}) try: with urllib.request.urlopen(req, timeout=timeout) as r: b = r.read(); return r.status, (b if raw else json.loads(b or b'null')) except urllib.error.HTTPError as e: b = e.read() try: return e.code, json.loads(b) except Exception: return e.code, b procs = {} def start(name, args, env): log = open(os.path.join(WORK, name + '.log'), 'a') e = dict(os.environ); e.update(env) procs[name] = subprocess.Popen(args, env=e, stdout=log, stderr=log, cwd=WORK) def stop(name): p = procs.pop(name, None) if p: p.terminate() try: p.wait(10) except Exception: p.kill() def template(): path = os.path.join(WORK, 'sample.json') subprocess.run([BIN, '--sample-state', path], check=True) t = json.load(open(path)) for k in ('channels', 'mcp_servers', 'skills', 'pages', 'review_allow_rules', 'review_ask_rules', 'remote_computers', 'folders'): if k in t: t[k] = [] t['providers'] = {'custom': {'api_key': '', 'base_url': f'http://127.0.0.1:{MP}/v1', 'default_model': 'e2e', 'known_models': ['e2e']}} t['remote'] = {'lan': False, 'password': '', 'devices': [], 'mac_from_phone': False, 'device_link': False, 'relay_url': '', 'relay_user': '', 'host_id': '', 'host_token': '', 'host_name': '', 'backup_key': '', 'auto_backup': False, 'last_backup': 0} b = t['bots'][0] b.update({'name': 'Master', 'provider': 'Custom', 'model': 'e2e', 'messages': [], 'memories': [], 'allow_rules': [], 'last_turn': None, 'default_computer': 'Bot 的電腦', 'tools_enabled': True, 'instructions': ''}) b.pop('threads', None) t['bots'] = [b]; t['primary_bot'] = b['id']; t['selected'] = b['id'] t['local_execution'] = False; t['direct_mode'] = False; t['auto_review'] = False return t NOBOX = {'DOCKER_HOST': 'unix:///nonexistent/docker.sock', 'PATH': '/usr/bin:/bin:/usr/sbin:/sbin'} def run(): srv = ThreadingHTTPServer(('127.0.0.1', MP), Model) threading.Thread(target=srv.serve_forever, daemon=True).start() os.makedirs(os.path.join(WORK, 'relay')) start('relay', [RELAY, '--bind', '127.0.0.1', '--port', str(RP), '--data', os.path.join(WORK, 'relay')], {}) for _ in range(50): try: if http('GET', RURL + '/api/hello')[0] == 200: break except Exception: time.sleep(0.2) # The brain: a seeded bot on the fake model, joins (and makes) the account. bdir = os.path.join(WORK, 'brain'); os.makedirs(bdir) st = template(); st['remote']['host_name'] = 'E2E大腦' json.dump(st, open(os.path.join(bdir, 'state.json'), 'w'), ensure_ascii=False) benv = {'ANYBOT_DATA_DIR': bdir, 'ANYBOT_ACCOUNT': f'{RURL}|{USER}', 'ANYBOT_ACCOUNT_PASSWORD': PW, 'ANYBOT_ACCOUNT_CREATE': '1', 'ANYBOT_HOST_NAME': 'E2E大腦', **NOBOX} start('brain', [BIN, '--serve'], benv) brain_id = None for _ in range(60): time.sleep(0.5) try: s = json.load(open(os.path.join(bdir, 'state.json'))) if s['remote'].get('host_id'): brain_id = s['remote']['host_id']; break except Exception: pass check('brain joins the account', brain_id, 'no host id') # The hands: a second AnyBot of the account, lending its hands to that brain only. _, v = http('POST', RURL + '/relay/login', {'user': USER, 'password': PW, 'device': HANDS}) _, h = http('POST', RURL + '/relay/hosts', {'name': HANDS}, token=v['token']) hdir = os.path.join(WORK, 'hands'); os.makedirs(hdir) hs = template(); hs['bots'] = [] hs['remote'].update({'relay_url': RURL, 'relay_user': USER, 'host_id': h['id'], 'host_token': h['token'], 'host_name': HANDS, 'device_link': True, 'hands_for': brain_id, 'hands_for_name': 'E2E大腦'}) json.dump(hs, open(os.path.join(hdir, 'state.json'), 'w'), ensure_ascii=False) henv = {'ANYBOT_DATA_DIR': hdir, **NOBOX} start('hands', [BIN, '--serve'], henv) tok = h['token'] sync = lambda method, rest='', body=None, q='': http(method, f'{RURL}/relay/hosts/{brain_id}/api/sync{rest}{q}', body, token=tok) def peers(): return {x['name']: x for x in http('GET', RURL + '/relay/host/peers', token=tok)[1]['hosts']} for _ in range(40): p = peers() if p.get('E2E大腦', {}).get('online') and p.get(HANDS, {}).get('online'): break time.sleep(0.5) p = peers() check('brain and hands are online', p.get('E2E大腦', {}).get('online') and p.get(HANDS, {}).get('online'), json.dumps(p, ensure_ascii=False)) # Tell the account this computer shows that brain (as the Mac's screen does). copy = {'version': 0, 'msgs': {}} def pull(full=False, client='screen'): since = 0 if full else copy['version'] code, v = sync('GET', q=f'?since={since}&client={client}') if code != 200: raise RuntimeError(f'pull {code} {v}') if client == 'screen': copy['version'] = v.get('version', 0) for d in v.get('bots', []): cur = copy['msgs'].setdefault(d['id'], []) if len(cur) < d['from']: copy['version'] = 0 # the screen asks for everything again continue del cur[d['from']:] cur.extend(d['messages']) return v # The brain looks the account's computers up every 15 s: a computer that just came on is # there within that. for _ in range(25): v = pull() comps = v.get('computers', {}).get('1', {}) names = [x['name'] for x in comps.get('names', [])] if HANDS in names: break time.sleep(1) check('computers by their own names; no server computer', names == ['Master 的電腦', HANDS], names) check('the bot works on its own computer by default', comps.get('current') == 'Master 的電腦', comps) def wait_turn(answer=None, timeout=90, during=None): """Until the turn ends; answers requests with `answer` ("once"/"reject").""" t0 = time.time(); seen = [] while time.time() - t0 < timeout: v = pull(client='poll') for a in v.get('approvals', []): key = (a.get('call_id'), a.get('kind')) if key not in [s[:2] for s in seen]: seen.append((a.get('call_id'), a.get('kind'), a)) if during: during(a) elif answer: sync('POST', '/act', {'bot': 1, 'action': 'answer', 'answer': answer}) if not v.get('live') and not v.get('approvals') and time.time() - t0 > 2.5: return seen, v time.sleep(0.5) return seen, None def last_tool(v): msgs = [m for m in v['bots'][0]['messages'] if m['role'] == 'Tool'] if v and v.get('bots') else [] return msgs[-1]['text'] if msgs else '' def full_messages(): return pull(full=True, client='check')['bots'][0]['messages'] def tool_text_after(marker): ms = full_messages() i = max((k for k, m in enumerate(ms) if m['role'] == 'User' and marker in m['text']), default=-1) return [m['text'] for m in ms[i + 1:] if m['role'] == 'Tool'] # 1. 查看電腦: the person's computer is the one they typed on. sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '我現在用哪台？SC_LIST'}) wait_turn('once') t = ' '.join(tool_text_after('SC_LIST')) check('list_computers names the hands as where the person is', HANDS in t and '使用者現在就是在這台' in t, t[:300]) # 4. Not asked for: back to the chosen computer, with a word for the bot. sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '跑一下 SC_REDIRECT'}) wait_turn('once') t = ' '.join(tool_text_after('SC_REDIRECT')) check('a step the person did not ask for goes to the chosen computer, and the bot is told', 'moved-here' in t and '[Master 的電腦]' in t and '這一步你指定了' in t, t[:400]) # 2. A step on the person's computer, named by them: asks once, runs there (agent-desktop). sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦打開計算機 SC_OPEN'}) seen, _ = wait_turn('once') t = ' '.join(tool_text_after('SC_OPEN')) check('open_app on the person computer asks first', any(s[1] == 'action' for s in seen), seen) check('open_app ran on the hands via agent-desktop', f'[{HANDS}]' in t and ('已打開' in t), t[:400]) # 3. A command there. sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦跑指令 SC_SHELL'}) wait_turn('once') t = ' '.join(tool_text_after('SC_SHELL')) check('shell runs on the hands', 'e2e-42' in t and f'[{HANDS}]' in t, t[:300]) # 5. Its own computer. sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '用你自己的 SC_OWN'}) wait_turn('once') t = ' '.join(tool_text_after('SC_OWN')) check('own computer by default', 'on-own-5' in t and '[Master 的電腦]' in t, t[:300]) # 6. 需要處理: 我完成了. sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦 SC_HANDOFF'}) got = {} phone = {} def on_handoff(a): got.update(a) # The phone (signed in as the person) sees it too, as 需要處理 with the computer. _, ses = http('POST', RURL + '/relay/login', {'user': USER, 'password': PW, 'device': 'e2e-phone'}) code, view = http('GET', f'{RURL}/h/{brain_id}/api/state?since=0', token=ses.get('token')) phone.update(next((x for x in (view or {}).get('approvals', []) if x.get('bot') == 1), {}) if code == 200 else {'code': code, 'body': str(view)[:200]}) sync('POST', '/act', {'bot': 1, 'action': 'answer', 'answer': 'once'}) wait_turn(during=on_handoff) t = ' '.join(tool_text_after('在我的電腦 SC_HANDOFF')) check('hand-off shows as 需要處理 with the instruction and computer', got.get('kind') == 'handoff' and got.get('reason') == 'E2E：請按「我完成了」' and got.get('grant') == HANDS, got) check('我完成了 lets the bot carry on', '我完成了' in t, t[:300]) check('the phone shows the hand-off with its computer', phone.get('kind') == 'handoff' and phone.get('computer') == HANDS and not phone.get('can_always'), phone) # 7. 需要處理: 跳過. sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '再一次在我的電腦 SC_HANDOFF'}) wait_turn('reject') t = ' '.join(tool_text_after('再一次在我的電腦 SC_HANDOFF')) check('跳過 tells the bot the step is skipped', '跳過' in t, t[:300]) # 8. 需要處理: answered by a message. sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '第三次在我的電腦 SC_HANDOFF'}) def on_reply(a): sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '我已經登入好了'}) seen, v = wait_turn(during=on_reply) t = ' '.join(tool_text_after('第三次在我的電腦 SC_HANDOFF')) check('a message answers a hand-off', '傳了訊息' in t, t[:300]) ms = full_messages() check('the reply is in the chat once', sum(1 for m in ms if m['role'] == 'User' and m['text'] == '我已經登入好了') == 1, [m['text'] for m in ms if m['role'] == 'User']) # 9. The screen's copy, kept from small updates, matches the brain's chat exactly. for _ in range(3): pull() mine = copy['msgs'].get(1, []) full = full_messages() sig = lambda ms: [(m['role'], m['text'], m.get('at')) for m in ms] check('screen copy equals the brain chat (no duplicates)', sig(mine) == sig(full), f'{len(mine)} vs {len(full)}') users = [m['text'] for m in full if m['role'] == 'User'] check('no user message twice', len(users) == len(set(users)), users) # 10. Choosing a computer from the screen reaches the brain. sync('POST', '/edit', {'client': 'screen', 'seq': 1, 'ops': [{'op': 'bot_set', 'id': 1, 'key': 'default_computer', 'value': HANDS}]}) time.sleep(1.0) c = pull(full=True, client='check2')['computers']['1'] check('the computer chosen on the screen is the brain\'s choice', c['current'] == HANDS, c) sync('POST', '/edit', {'client': 'screen', 'seq': 2, 'ops': [{'op': 'bot_set', 'id': 1, 'key': 'default_computer', 'value': 'Bot 的電腦'}]}) time.sleep(1.0) c = pull(full=True, client='check3')['computers']['1'] check('and back to its own', c['current'] == 'Master 的電腦', c) # 11. The hands go away: plain words, not codes. stop('hands') time.sleep(10) sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦跑指令 SC_SHELL 二'}) wait_turn('once', timeout=120) t = ' '.join(tool_text_after('SC_SHELL 二')) check('hands offline: the bot hears it in words', ('連不到' in t or '離線' in t) and 'e2e-42' not in t and '這一步你指定了' not in t and not any(code in t for code in ('503', '502', '404')), t[:300]) start('hands', [BIN, '--serve'], henv) for _ in range(40): if peers().get(HANDS, {}).get('online'): break time.sleep(0.5) sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦跑指令 SC_SHELL 三'}) wait_turn('once') t = ' '.join(tool_text_after('SC_SHELL 三')) check('hands back: works again', 'e2e-42' in t and f'[{HANDS}]' in t, t[:300]) # 12. The brain restarts: the screen picks everything up again, and work goes on. before = len(full_messages()) stop('brain') time.sleep(3) start('brain', [BIN, '--serve'], benv) for _ in range(60): try: if peers().get('E2E大腦', {}).get('online'): v = pull() break except Exception: pass time.sleep(0.5) time.sleep(2) pull() check('after a brain restart the screen has the whole chat', len(copy['msgs'].get(1, [])) >= before, f'{len(copy["msgs"].get(1, []))} < {before}') sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '在我的電腦跑指令 SC_SHELL 四'}) wait_turn('once') t = ' '.join(tool_text_after('SC_SHELL 四')) check('after a brain restart the hands still work', 'e2e-42' in t and f'[{HANDS}]' in t, t[:300]) pull(); pull() check('screen copy still equals the brain chat', sig(copy['msgs'].get(1, [])) == sig(full_messages())) # 13. 清除聊天記錄 from the screen clears the brain's chat. sync('POST', '/act', {'bot': 1, 'action': 'clear'}) time.sleep(1.5) check('clear from the screen empties the brain chat', full_messages() == []) pull(); pull() check('and the screen copy follows', copy['msgs'].get(1, []) == [], len(copy['msgs'].get(1, []))) try: run() except Exception as e: import traceback; traceback.print_exc() check('harness ran to the end', False, repr(e)) finally: for n in list(procs): stop(n) failed = [r for r in results if not r[1]] print(f'\n{len(results) - len(failed)}/{len(results)} passed; logs in {WORK}') sys.exit(1 if failed else 0)