AnyBot/deploy/test/remote_e2e.py

# 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})],
    '另一台：': [tool_call('hand_over_computer', {'instruction': 'E2E：在另一台', 'computer': 'E2E另一台'})],
    '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)
    # A second computer showing the same brain (as a desktop AnyBot left on the server does):
    # with two, 「我的電腦」 has to be the one the person typed on, not a guess.
    _, h2 = http('POST', RURL + '/relay/hosts', {'name': 'E2E另一台'}, token=v['token'])
    h2dir = os.path.join(WORK, 'hands2'); os.makedirs(h2dir)
    hs2 = template(); hs2['bots'] = []
    hs2['remote'].update({'relay_url': RURL, 'relay_user': USER, 'host_id': h2['id'], 'host_token': h2['token'], 'host_name': 'E2E另一台', 'device_link': True, 'hands_for': brain_id, 'hands_for_name': 'E2E大腦', 'brain_id': brain_id, 'brain_name': 'E2E大腦'})
    json.dump(hs2, open(os.path.join(h2dir, 'state.json'), 'w'), ensure_ascii=False)
    start('hands2', [BIN, '--serve'], {'ANYBOT_DATA_DIR': h2dir, **NOBOX})
    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') and p.get('E2E另一台', {}).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 and 'E2E另一台' in names:
            break
        time.sleep(1)
    check('computers by their own names; no server computer', sorted(names) == sorted(['Master 的電腦', HANDS, 'E2E另一台']), 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'])
    # 8b. A turn started on one computer; while it waits, the person types on another one:
    # 「我的電腦」 in that message is the computer it was typed on (the turn's own idea was fixed
    # when it started).
    tok2 = h2['token']
    sync2 = lambda method, rest='', body=None, q='': http(method, f'{RURL}/relay/hosts/{brain_id}/api/sync{rest}{q}', body, token=tok2)
    sync2('POST', '/act', {'bot': 1, 'action': 'send', 'text': '另一台：在這台 E2E另一台 請我接手'})
    def steer_from_hands(a):
        if a.get('kind') == 'handoff':
            sync('POST', '/act', {'bot': 1, 'action': 'send', 'text': '算了，在我的電腦跑指令 SC_SHELL 插話'})
        else:
            sync('POST', '/act', {'bot': 1, 'action': 'answer', 'answer': 'once'})
    wait_turn(during=steer_from_hands)
    t = ' '.join(tool_text_after('SC_SHELL 插話'))
    check('a message typed on another computer mid-turn: 「我的電腦」 is that one', 'e2e-42' in t and f'[{HANDS}]' in t and '這一步你指定了' not in t, t[:300])
    # 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 first 「我的電腦」 is the computer the person typed on', 'e2e-42' in t and f'[{HANDS}]' in t and '這一步你指定了' not 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)