Александр Петров
75 lines
3.4 KiB
Python
75 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Живой MCP: долгий машинный out из tail-jump можно прервать Pause."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
from mcp import Client, StdioServerParameters
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
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-out-mcp-', dir='/tmp') as temp:
|
|
parameters = StdioServerParameters(
|
|
command=sys.executable,
|
|
args=[str(ROOT / 'toolchain/sdbg_mcp.py'), '--build', str(package),
|
|
'--socket', str(Path(temp) / 'session.sock'),
|
|
'--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
|
|
|
|
await call('start_session')
|
|
try:
|
|
deadline = time.monotonic() + 90
|
|
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('hello не дошёл до main')
|
|
entered = await call('step_instruction', {'count': 1})
|
|
code = await call('disassemble_logical',
|
|
{'address': hex(entered['pc']), 'length': 12})
|
|
if 'jp (hl)' not in code['text']:
|
|
raise RuntimeError('Ожидался tail-jump helper: ' + code['text'])
|
|
started_at = time.monotonic()
|
|
accepted = await call('step_out_instruction')
|
|
response_ms = round((time.monotonic() - started_at) * 1000, 1)
|
|
if not accepted['accepted'] or response_ms > 3000:
|
|
raise RuntimeError('Машинный out не ответил быстро: ' + repr(accepted))
|
|
await asyncio.sleep(.3)
|
|
if not (await call('session_status'))['running']:
|
|
raise RuntimeError('Tail-jump out неожиданно завершился')
|
|
paused = await call('pause_execution')
|
|
if (await call('session_status'))['running']:
|
|
raise RuntimeError('Pause не остановил долгий out')
|
|
print(json.dumps({'event': 'machine_out_pause_verified',
|
|
'response_ms': response_ms,
|
|
'entered_pc': entered['pc'],
|
|
'paused_pc': paused['pc']}, ensure_ascii=False),
|
|
flush=True)
|
|
finally:
|
|
await call('stop_session')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
asyncio.run(probe())
|