#!/usr/bin/env python3 """MCP сам запускает hello, затем DAP подключается к тому же MAME.""" from __future__ import annotations import asyncio import json import os from pathlib import Path import subprocess import sys import tempfile import time from mcp import Client, StdioServerParameters ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(Path(__file__).resolve().parent)) from run_vscode_dap_probe import send, wait_response # noqa: E402 async def probe() -> None: home = Path(os.environ['MAME_HOME']).resolve() binary = Path(os.environ.get('MAME_BIN', home / 'sprinter')).resolve() package = ROOT / 'tests/hello/.sprinter-cc-hello' with tempfile.TemporaryDirectory(prefix='sprinter-managed-mcp-', dir='/tmp') as temp: socket = str(Path(temp) / 'session.sock') parameters = StdioServerParameters( command=sys.executable, args=[str(ROOT / 'toolchain/sdbg_mcp.py'), '--build', str(package), '--socket', socket, '--mame-home', str(home), '--mame-bin', str(binary)]) async with Client(parameters) as client: async def call(name: str, arguments: dict | None = None) -> dict: result = await client.call_tool(name, arguments or {}) if result.is_error or result.structured_content is None: raise RuntimeError(name + ': ' + repr(result.content)) return result.structured_content names = {tool.name for tool in (await client.list_tools()).tools} if not {'start_session', 'stop_session', 'where', 'read_variable', 'press_key', 'list_breakpoints', 'disassemble_logical'} <= names: raise RuntimeError('Нет инструментов автономного запуска') initial = await call('session_status') if initial['phase'] != 'idle': raise RuntimeError('Ожидался idle: ' + repr(initial)) started_at = time.monotonic() accepted = await call('start_session') start_ms = round((time.monotonic() - started_at) * 1000, 1) if not accepted['accepted'] or start_ms > 3000: raise RuntimeError('MCP start_session не ответил быстро: ' + repr(accepted)) duplicate = await client.call_tool('start_session', {}) if not duplicate.is_error or 'уже запущена' not in str(duplicate.content): raise RuntimeError('Повторный start_session не был отклонён') mame_pid = None try: deadline = time.monotonic() + 85 while time.monotonic() < deadline: status = await call('session_status') if status['phase'] == 'ready': break if status['phase'] in ('failed', 'stopped'): raise RuntimeError('Launcher: ' + repr(status)) await asyncio.sleep(.2) else: raise TimeoutError('MCP session не дошла до main') mame_pid = status['mame_pid'] location = await call('where') if location.get('function', {}).get('name') != 'main': raise RuntimeError('MCP не остановился в main: ' + repr(location)) bytes_before = await call('read_memory', {'address': hex(location['pc']), 'length': 32}) disassembly = await call('disassemble_logical', {'address': hex(location['pc']), 'length': 32}) bytes_after = await call('read_memory', {'address': hex(location['pc']), 'length': 32}) if disassembly['space'] != 'logical_z80' or \ disassembly['address'] != location['pc'] or \ not disassembly['text'].upper().startswith( f"{location['pc']:04X}:") or \ bytes_before['hex'] != bytes_after['hex']: raise RuntimeError('Нет дизассемблирования main: ' + repr(disassembly)) await call('read_variable', {'name': 'errno'}) shot = await call('screenshot') if Path(shot['path']).read_bytes()[:8] != b'\x89PNG\r\n\x1a\n': raise RuntimeError('MCP не создал PNG') dap = subprocess.Popen([sys.executable, str(ROOT / 'toolchain/sdbg_dap.py')], cwd=ROOT, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) buffer = b'' try: send(dap, 1, 'initialize') _, buffer, _ = wait_response(dap, buffer, 'initialize', 10) send(dap, 2, 'attach', {'socket': socket}) _, buffer, _ = wait_response(dap, buffer, 'attach', 10) send(dap, 3, 'configurationDone') _, buffer, _ = wait_response(dap, buffer, 'configurationDone', 10) send(dap, 4, 'stackTrace') frame, buffer, _ = wait_response(dap, buffer, 'stackTrace', 10) if frame['body']['stackFrames'][0]['name'] != 'main': raise RuntimeError('DAP attach не увидел main') send(dap, 5, 'disconnect') wait_response(dap, buffer, 'disconnect', 10) finally: if dap.poll() is None: dap.terminate() dap.communicate(timeout=8) after_dap = await call('session_status') if after_dap['phase'] != 'ready' or after_dap['session_id'] != status['session_id']: raise RuntimeError('DAP disconnect завершил MCP-owned MAME') lines = disassembly['text'].splitlines() addresses = [int(line.split(':', 1)[0], 16) for line in lines] calls = [index for index, line in enumerate(lines) if ' call ' in ' ' + line.lower() + ' '] if len(calls) < 2 or calls[0] != 0 or calls[1] <= 1: raise RuntimeError('Нет двух call в main: ' + disassembly['text']) async def wait_machine_stop() -> dict: deadline = time.monotonic() + 10 while time.monotonic() < deadline: if not (await call('session_status'))['running']: return await call('where') await asyncio.sleep(.05) paused = await call('pause_execution') raise TimeoutError('Машинный шаг не завершился; Pause PC=' + hex(paused['pc'])) await call('step_over_instruction', {'count': 1}) after_over = await wait_machine_stop() if after_over['pc'] != addresses[1]: raise RuntimeError('over call не остановился после вызова: ' + repr(after_over)) await call('step_over_instruction', {'count': calls[1] - 1}) at_call = await wait_machine_stop() if at_call['pc'] != addresses[calls[1]]: raise RuntimeError('over count не дошёл до второго call: ' + repr(at_call)) entered = await call('step_instruction', {'count': 1}) if entered['pc'] == at_call['pc']: raise RuntimeError('Машинный step не вошёл в возвращаемый вызов') await call('step_out_instruction') returned = await wait_machine_stop() if returned['pc'] != addresses[calls[1] + 1]: raise RuntimeError('Машинный out не вернулся после call: ' + repr(returned)) stepped = await call('step_instruction', {'count': 3}) if stepped['pc'] == returned['pc'] or \ (await call('session_status'))['running']: raise RuntimeError('Три машинных шага не остановились: ' + repr(stepped)) source = str(ROOT / 'tests/hello/hello.c') before = await call('set_line_breakpoint', {'file': source, 'line': 62}) points = await call('list_breakpoints') active = [item for item in points['breakpoints'] if item['id'] == before['id']] if len(active) != 1 or not active[0]['owner'].startswith('mcp:') or \ not any(loc.get('line') == 62 for loc in active[0]['locations']): raise RuntimeError('Личная точка не отражена в списке: ' + repr(points)) await call('continue_execution') deadline = time.monotonic() + 10 while time.monotonic() < deadline: at_getchar = await call('session_status') if not at_getchar['running']: break await asyncio.sleep(.1) else: raise TimeoutError('MCP не остановился перед getchar') location = await call('where') if not any(item['line'] == 62 for item in location['sources']): raise RuntimeError('MCP остановился не перед getchar: ' + repr(location)) await call('clear_breakpoint', {'identifier': before['id']}) after = await call('set_line_breakpoint', {'file': source, 'line': 63}) await call('continue_execution') await asyncio.sleep(.4) running = await call('session_status') if not running['running']: raise RuntimeError('CPU не ожидает getchar перед вводом') while_running = await call('list_breakpoints') if not any(item['id'] == after['id'] for item in while_running['breakpoints']): raise RuntimeError('Не прочитана точка при running CPU') key = await call('press_key', {'key': 'x', 'frames': 3}) deadline = time.monotonic() + 10 while time.monotonic() < deadline: after_key = await call('session_status') if not after_key['running']: break await asyncio.sleep(.1) else: raise TimeoutError('MCP press_key не завершил getchar') location = await call('where') if not any(item['line'] == 63 for item in location['sources']): raise RuntimeError('После клавиши ожидалась строка 63: ' + repr(location)) await call('clear_breakpoint', {'identifier': after['id']}) print(json.dumps({'event': 'mcp_launch_dap_attach_verified', 'tools': len(names), 'start_ms': start_ms, 'session_id': status['session_id'], 'socket': socket, 'pc': location['pc'], 'disassembly_head': disassembly['text'][:90], 'key': key['key'], 'after_getchar_line': 63}, ensure_ascii=False), flush=True) finally: await call('stop_session') deadline = time.monotonic() + 12 while time.monotonic() < deadline: ended = await call('session_status') if ended['phase'] == 'stopped': break await asyncio.sleep(.1) else: raise RuntimeError('stop_session не завершил launcher') if Path(socket).exists(): raise RuntimeError('После stop_session остался Unix socket') if mame_pid is not None: deadline = time.monotonic() + 8 while time.monotonic() < deadline: try: os.kill(mame_pid, 0) except ProcessLookupError: break await asyncio.sleep(.1) else: raise RuntimeError('После stop_session остался процесс MAME') if __name__ == '__main__': asyncio.run(probe())