Expose owned C breakpoints through shared MCP session

This commit is contained in:
Александр Петров
2026-09-17 23:03:59 +03:00
parent 0ade3b4821
commit 113d468c48
12 changed files with 78 additions and 16 deletions
+14 -1
View File
@@ -38,7 +38,7 @@ async def probe() -> None:
names = {tool.name for tool in (await client.list_tools()).tools}
if not {'start_session', 'stop_session', 'where', 'read_variable',
'press_key'} <= names:
'press_key', 'list_breakpoints'} <= names:
raise RuntimeError('Нет инструментов автономного запуска')
initial = await call('session_status')
if initial['phase'] != 'idle':
@@ -99,6 +99,12 @@ async def probe() -> None:
raise RuntimeError('DAP disconnect завершил MCP-owned MAME')
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:
@@ -115,6 +121,13 @@ async def probe() -> None:
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:
+6
View File
@@ -225,6 +225,12 @@ class ServerTests(unittest.TestCase):
mine = controller.call('break_line',
{'file': '/src/main.c', 'line': 3, 'owner': 'mcp:a'})
other = controller.call('break_function', {'name': 'main', 'owner': 'mcp:b'})
controller.breakpoint_info[mine['id']]['guard_conditions'] = ['PG3 == 0x12']
listed = controller.call('list_breakpoints', {})['breakpoints']
self.assertEqual([(item['id'], item['owner']) for item in listed],
[(mine['id'], 'mcp:a'), (other['id'], 'mcp:b')])
self.assertEqual(listed[0]['locations'][0]['link_address'], 0x8100)
self.assertEqual(listed[0]['locations'][0]['bank_guard'], 'PG3 == 0x12')
self.assertTrue(controller._has_stop_breakpoint(session.where()))
with self.assertRaisesRegex(SessionError, 'не принадлежит'):
controller.call('clear_breakpoint', {'id': other['id'], 'owner': 'mcp:a'})