horizon-airdef/tools/launcher_mcp_e2e.py

#!/usr/bin/env python3
"""End-to-end check of the launcher's MCP playing the game through its game_* tools.

A scratch launcher home gets one "installed" version made of this checkout's builds
(horizon-client and horizon-ctl) and the prepared resources in .local/horizon-client. An MCP
client then talks to `horizon-launcher mcp` alone: every CLI command is listed as a tool of the same
name (versions, settings, mods_search...), a destructive one needs confirm, the game tools are
listed, a call before the game runs says to start it, play starts it under control, and
game_wait_ready, game_state, game_ui and game_screenshot play it until game_quit. Then No AI is turned on with
`horizon-launcher no-ai on`: `horizon-launcher mcp` refuses to start, and a session already running
refuses its next call.

    python3 tools/launcher_mcp_e2e.py --bin-dir target/fast --launcher launcher/target/debug/horizon-launcher

On macOS the game renders offscreen (it runs in a locked session or over SSH); on Linux it needs
a display (vscreen on the desktop). Games started here have no audio.
"""
import argparse
import base64
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

try:
    from . import lab
except ImportError:
    import lab

ROOT = Path(__file__).resolve().parents[1]
TAG = 'v2099.1.1'


class Failed(Exception):
    pass


class Mcp:
    def __init__(self, launcher, env):
        self.proc = subprocess.Popen([str(launcher), 'mcp'], env=env, stdin=subprocess.PIPE,
                                     stdout=subprocess.PIPE, text=True)
        self.next_id = 0

    def rpc(self, method, params):
        self.next_id += 1
        self.proc.stdin.write(json.dumps({'jsonrpc': '2.0', 'id': self.next_id, 'method': method,
                                          'params': params}) + '\n')
        self.proc.stdin.flush()
        reply = json.loads(self.proc.stdout.readline())
        if 'error' in reply:
            raise Failed(f'{method}: {reply["error"]}')
        return reply['result']

    def call(self, tool, **args):
        result = self.rpc('tools/call', {'name': tool, 'arguments': args})
        texts = [c['text'] for c in result['content'] if c['type'] == 'text']
        images = [c for c in result['content'] if c['type'] == 'image']
        print(f'[launcher-mcp] {tool} {args} -> {(texts[-1] if texts else "")[:300]}')
        return result, texts, images

    def close(self):
        self.proc.stdin.close()
        self.proc.wait(timeout=60)


def scenario(m, out):
    init = m.rpc('initialize', {'protocolVersion': '2025-06-18', 'capabilities': {},
                                'clientInfo': {'name': 'launcher_mcp_e2e', 'version': '1'}})
    assert init['serverInfo']['name'] == 'horizon-launcher', init
    tools = {t['name'] for t in m.rpc('tools/list', {})['tools']}
    for name in ['play', 'server', 'versions', 'settings', 'mods_search', 'modpack_list', 'servers_list',
                 'data_status', 'account_list', 'game_args', 'game_state', 'game_ui', 'game_click',
                 'game_screenshot']:
        assert name in tools, f'{name} missing from {sorted(tools)}'
    for name in ['game_launch', 'mcp', 'no_ai', 'agent_config']:
        assert name not in tools, f'{name} is a tool'

    # Generated from the CLI: the same command, its JSON output.
    result, texts, _ = m.call('versions', installed=True)
    assert json.loads(texts[-1])['installed'] == [TAG], texts
    result, texts, _ = m.call('settings', key='opening-animation', value='off')
    assert json.loads(texts[-1])['opening_animation'] is False, texts
    result, texts, _ = m.call('game_args')
    assert 'game_args' in json.loads(texts[-1]), texts
    result, texts, _ = m.call('remove', tag=TAG)
    assert result['isError'] and 'confirm' in texts[-1], texts

    result, texts, _ = m.call('game_state')
    assert result['isError'] and 'with play' in texts[-1], texts

    result, texts, _ = m.call('play', world='Agent', creative=True,
                              offscreen=sys.platform == 'darwin')
    assert not result['isError'], texts
    result, texts, _ = m.call('game_wait_ready', seconds=300)
    assert not result['isError'], texts
    result, texts, _ = m.call('game_state')
    state = json.loads(texts[-1])
    assert state['screen'] == 'playing' and state['gamemode'] == 'creative', state
    m.call('game_pause')
    result, texts, _ = m.call('game_ui')
    ids = {w['id'] for w in json.loads(texts[-1])['widgets']}
    assert 'back-to-game' in ids, ids
    result, texts, _ = m.call('game_click', id='back-to-game')
    assert json.loads(texts[-1])['screen'] == 'playing', texts
    result, _, images = m.call('game_screenshot')
    assert images, 'no image'
    (out / 'launcher-mcp.png').write_bytes(base64.b64decode(images[0]['data']))
    m.call('game_quit')
    return {'tools': len(tools)}


REFUSAL = "AI features are off: turn off No AI in the launcher's settings"


def no_ai_scenario(launcher, env, home):
    run = lambda *a: subprocess.run([str(launcher), *a], env=env, input='', capture_output=True, text=True,
                                    timeout=60)
    assert json.loads(run('no-ai', '--json').stdout) == {'no_ai': False}
    config = json.loads(run('agent-config', '--json').stdout)
    assert config['server']['args'] == ['mcp'] and len(config['clients']) == 4, config
    assert 'agent-config' in run('--help').stdout
    test = json.loads(run('agent-config', '--test', '--json').stdout)
    assert test['answered'] and test['tools'] > 100, test
    m = Mcp(launcher, env)
    m.rpc('initialize', {'protocolVersion': '2025-06-18', 'capabilities': {},
                         'clientInfo': {'name': 'launcher_mcp_e2e', 'version': '1'}})
    r = run('no-ai', 'on')
    assert r.returncode == 0, r
    assert 'noAi:true' in (home / 'horizon-extras.txt').read_text()
    try:
        m.call('links')
        raise AssertionError('links ran under No AI')
    except Failed as e:
        assert REFUSAL in str(e), e
    m.close()
    for command in ('mcp', 'agent-config'):
        r = run(command)
        print(f'[launcher-mcp] {command} under No AI -> exit {r.returncode}: {r.stderr.strip()}')
        assert r.returncode == 1 and r.stderr.strip() == f'error: {REFUSAL}', r
    assert 'agent-config' not in run('--help').stdout and ' mcp ' not in run('--help').stdout
    # What the launcher needs keeps working.
    assert run('versions', '--installed', '--json').returncode == 0
    assert run('no-ai', 'off').returncode == 0
    assert json.loads(run('no-ai', '--json').stdout) == {'no_ai': False}
    return {'no_ai': 'refused'}


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('--bin-dir', required=True, help='horizon-client and horizon-ctl built from this checkout')
    parser.add_argument('--launcher', required=True, help='a horizon-launcher binary')
    parser.add_argument('--out', default=str(ROOT / '.local/launcher-mcp-e2e'), help='the screenshot goes here')
    args = parser.parse_args()
    bins = Path(args.bin_dir).resolve()
    resources = (lab.RESOURCES).resolve()
    if not (resources / 'horizon-resources.json').exists():
        sys.exit(f'no prepared resources in {resources}')
    out = Path(args.out)
    out.mkdir(parents=True, exist_ok=True)
    work = Path(tempfile.mkdtemp(prefix='hzlm-'))
    version = work / 'launcher/versions' / TAG
    (version / '.local/horizon-client').mkdir(parents=True)
    for name in ['horizon-client', 'horizon-ctl']:
        os.symlink(bins / name, version / name)
    os.symlink(resources, version / '.local/horizon-client' / lab.VERSION)
    (version / '.installed').write_text('')
    env = lab.game_env(HORIZON_LAUNCHER_HOME=work / 'launcher', HORIZON_HOME=work / 'home', HORIZON_TELEMETRY_URL='off')
    m = Mcp(Path(args.launcher).resolve(), env)
    try:
        summary = scenario(m, out)
        summary.update(no_ai_scenario(Path(args.launcher).resolve(), env, work / 'home'))
        print(f'[launcher-mcp] PASS: {summary}')
        ok = True
    except (Failed, AssertionError, KeyError, json.JSONDecodeError) as e:
        print(f'[launcher-mcp] FAIL: {e!r}')
        ok = False
        try:
            m.call('game_quit')
        except Exception:
            pass
    finally:
        m.close()
        shutil.rmtree(work, ignore_errors=True)
    sys.exit(0 if ok else 1)


if __name__ == '__main__':
    main()