22788375261

Committed 07 Mar 2026 12:59AM UTC coverage: 84.4% (-1.1%) from 85.485%

Build # 22788375261

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #8 from karellen/fix_stdio_orphan_process

Fix MCP server shutdown: clean exit on SIGTERM and parent death

Coverage Stats

153 of 222 branches covered (68.92%)

Branch coverage included in aggregate %.

3 of 16 new or added lines in 1 file covered. (18.75%)

1 existing line in 1 file now uncovered.

810 of 919 relevant lines covered (88.14%)

4.38 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.27

/src/main/python/karellen_rr_mcp/server.py

#   -*- coding: utf-8 -*-
#   Copyright 2026 Karellen, Inc.
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

"""FastMCP server with tool definitions for rr reverse debugging."""

import atexit
import functools
import logging
import os
import signal
import traceback

from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.exceptions import ToolError

from karellen_rr_mcp.gdb_session import GdbSession, GdbSessionError
from karellen_rr_mcp.rr_manager import (
    record as rr_record_cmd,
    list_recordings as rr_list,
    list_processes as rr_ps_cmd,
    trace_info as rr_trace_info_cmd,
    remove_recording as rr_rm_cmd,
    ReplayServer, RrError,
)
from karellen_rr_mcp.types import (
    Breakpoint, Frame, StopEvent, Variable, ThreadInfo,
    ProcessInfo, RecordResult, EvalResult, MemoryResult,
    ReplayStatus, StringResult, IntResult, RegisterValues,
)

logger = logging.getLogger(__name__)

mcp = FastMCP("karellen-rr-mcp", instructions=(
    "rr reverse debugging server. Use rr_record to record a failing test, "
    "rr_replay_start to begin debugging, then use execution control and "
    "inspection tools to investigate. Use reverse=True to go backwards."
))

# Module-level singleton session state
_replay_server = None
_gdb_session = None


def _cleanup():
    global _replay_server, _gdb_session
    if _gdb_session is not None:
        try:
            _gdb_session.close()
        except Exception:
            pass
        _gdb_session = None
    if _replay_server is not None:
        try:
            _replay_server.stop()
        except Exception:
            pass
        _replay_server = None


atexit.register(_cleanup)


def _handle_signal(signum, frame):
    _cleanup()
    os._exit(0)


signal.signal(signal.SIGINT, _handle_signal)
signal.signal(signal.SIGTERM, _handle_signal)


def _require_session():
    if _gdb_session is None or not _gdb_session.is_connected():
        raise GdbSessionError("No active replay session. Call rr_replay_start first.")
    return _gdb_session


def _tag_errors(fn):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except GdbSessionError as e:
            raise ToolError("gdb: %s" % e) from e
        except RrError as e:
            raise ToolError("rr: %s" % e) from e
        except ToolError:
            raise
        except Exception as e:
            tb = traceback.extract_tb(e.__traceback__)
            tb_lines = ["%s:%d in %s" % (f.filename, f.lineno, f.name) for f in tb[-3:]]
            raise ToolError("internal: %s: %s\n  %s" % (
                type(e).__name__, e, "\n  ".join(tb_lines))) from e
    return wrapper


def _require_stop(stop):
    if stop is None:
        raise GdbSessionError("No stop event received")
    return stop


# --- Session Lifecycle Tools ---

@mcp.tool()
@_tag_errors
def rr_record(command: list[str], working_directory: str = None,
              env: dict[str, str] = None,
              trace_dir: str = None) -> RecordResult:
    """Record a command with rr. Returns trace directory path.

    Args:
        command: Command and arguments to record (e.g. ["./my_test"]).
        working_directory: Working directory for the recorded process.
        env: Optional extra environment variables for the recorded process.
        trace_dir: Output trace directory. If omitted, rr uses its default (~/.local/share/rr/).
    """
    trace_dir, exit_code, stdout, stderr = rr_record_cmd(
        command, working_directory=working_directory, env=env,
        trace_dir=trace_dir)
    return RecordResult(trace_dir=trace_dir, exit_code=exit_code,
                        stdout=stdout, stderr=stderr)


@mcp.tool()
@_tag_errors
def rr_replay_start(trace_dir: str = None, pid: int = None) -> ReplayStatus:
    """Start a replay session. Launches rr gdbserver and connects GDB/MI.

    Args:
        trace_dir: Path to rr trace directory. If omitted, uses the latest trace.
        pid: PID of a specific subprocess to replay. Use rr_ps to list available processes.
    """
    global _replay_server, _gdb_session
    if _gdb_session is not None:
        raise ToolError("gdb: A replay session is already active. Call rr_replay_stop first.")

    server = ReplayServer(trace_dir=trace_dir, pid=pid)
    session = None
    try:
        server.start()

        session = GdbSession()
        session.start()
        session.connect("localhost", server.port)

        _replay_server = server
        _gdb_session = session
        return ReplayStatus(port=server.port, message="Program is paused at start.")
    except (RrError, GdbSessionError):
        if session is not None:
            try:
                session.close()
            except Exception:
                pass
        try:
            server.stop()
        except Exception:
            pass
        raise


@mcp.tool()
@_tag_errors
def rr_replay_stop() -> StringResult:
    """Stop the current replay session and clean up."""
    if _gdb_session is None and _replay_server is None:
        return StringResult(result="No active replay session.")
    _cleanup()
    return StringResult(result="Replay session stopped.")


@mcp.tool()
@_tag_errors
def rr_list_recordings(trace_base_dir: str = None) -> list[str]:
    """List available rr trace recordings.

    Args:
        trace_base_dir: Base directory for traces. Defaults to ~/.local/share/rr.
    """
    return rr_list(trace_base_dir=trace_base_dir)


@mcp.tool()
@_tag_errors
def rr_ps(trace_dir: str) -> list[ProcessInfo]:
    """List processes in an rr trace recording.

    Args:
        trace_dir: Path to rr trace directory.
    """
    return rr_ps_cmd(trace_dir)


@mcp.tool()
@_tag_errors
def rr_traceinfo(trace_dir: str) -> StringResult:
    """Get trace metadata (header info in JSON format).

    Args:
        trace_dir: Path to rr trace directory.
    """
    return StringResult(result=rr_trace_info_cmd(trace_dir))


@mcp.tool()
@_tag_errors
def rr_rm(trace_dir: str) -> StringResult:
    """Remove an rr trace recording.

    Args:
        trace_dir: Path to rr trace directory to remove.
    """
    rr_rm_cmd(trace_dir)
    return StringResult(result="Trace removed: %s" % trace_dir)


@mcp.tool()
@_tag_errors
def rr_when() -> StringResult:
    """Get the current rr event number. Useful for knowing your position in the trace."""
    session = _require_session()
    output = session.rr_when()
    return StringResult(result=output if output else "Unable to determine current event.")


# --- Breakpoint Tools ---

@mcp.tool()
@_tag_errors
def rr_breakpoint_set(location: str, condition: str = None,
                      temporary: bool = False) -> Breakpoint:
    """Set a breakpoint at a function, file:line, or address.

    Args:
        location: Breakpoint location (e.g. "main", "foo.c:42", "*0x400500").
        condition: Optional condition expression (e.g. "i > 10").
        temporary: If true, breakpoint is deleted after first hit.
    """
    session = _require_session()
    return session.breakpoint_set(location, condition=condition, temporary=temporary)


@mcp.tool()
@_tag_errors
def rr_breakpoint_remove(breakpoint_number: int) -> IntResult:
    """Remove a breakpoint by its number.

    Args:
        breakpoint_number: The breakpoint number to remove.
    """
    session = _require_session()
    session.breakpoint_delete(breakpoint_number)
    return IntResult(result=breakpoint_number)


@mcp.tool()
@_tag_errors
def rr_breakpoint_list() -> list[Breakpoint]:
    """List all breakpoints."""
    session = _require_session()
    return session.breakpoint_list()


@mcp.tool()
@_tag_errors
def rr_watchpoint_set(expression: str,
                      access_type: str = "write") -> Breakpoint:
    """Set a hardware watchpoint on a variable or expression.

    Args:
        expression: Expression to watch (e.g. "my_var", "*0x601050").
        access_type: One of "write", "read", or "access".
    """
    session = _require_session()
    wp = session.watchpoint_set(expression, access_type=access_type)
    if wp is None:
        raise GdbSessionError("Failed to parse watchpoint response")
    return wp


# --- Execution Control Tools ---

@mcp.tool()
@_tag_errors
def rr_continue(reverse: bool = False) -> StopEvent:
    """Continue execution forward or backward until breakpoint/signal/end.

    Args:
        reverse: If true, continue backward (reverse execution).
    """
    session = _require_session()
    return _require_stop(session.continue_execution(reverse=reverse))


@mcp.tool()
@_tag_errors
def rr_step(count: int = 1, reverse: bool = False) -> StopEvent:
    """Step into (source-level) forward or reverse.

    Args:
        count: Number of steps (forward only).
        reverse: If true, step backward.
    """
    session = _require_session()
    return _require_stop(session.step(count=count, reverse=reverse))


@mcp.tool()
@_tag_errors
def rr_next(count: int = 1, reverse: bool = False) -> StopEvent:
    """Step over (source-level) forward or reverse.

    Args:
        count: Number of steps (forward only).
        reverse: If true, step backward.
    """
    session = _require_session()
    return _require_stop(session.next(count=count, reverse=reverse))


@mcp.tool()
@_tag_errors
def rr_finish(reverse: bool = False) -> StopEvent:
    """Run to function return (or to call site if reverse).

    Args:
        reverse: If true, run backward to the call site.
    """
    session = _require_session()
    return _require_stop(session.finish(reverse=reverse))


@mcp.tool()
@_tag_errors
def rr_run_to_event(event_number: int) -> StopEvent:
    """Jump to a specific rr event number.

    Args:
        event_number: The rr event number to seek to.
    """
    session = _require_session()
    return _require_stop(session.run_to_event(event_number))


# --- Thread and Frame Tools ---

@mcp.tool()
@_tag_errors
def rr_thread_list() -> list[ThreadInfo]:
    """List all threads in the replayed process with their current state and location."""
    session = _require_session()
    return session.thread_info()


@mcp.tool()
@_tag_errors
def rr_thread_select(thread_id: str) -> IntResult:
    """Switch to a different thread.

    Args:
        thread_id: Thread ID to switch to (from rr_thread_list).
    """
    session = _require_session()
    session.thread_select(thread_id)
    return IntResult(result=int(thread_id))


@mcp.tool()
@_tag_errors
def rr_select_frame(frame_level: int) -> IntResult:
    """Select a stack frame for inspection. After selecting, rr_locals and rr_evaluate
    operate in the selected frame's context.

    Args:
        frame_level: Frame number from rr_backtrace (0 = innermost/current).
    """
    session = _require_session()
    session.select_frame(frame_level)
    return IntResult(result=frame_level)


# --- State Inspection Tools ---

@mcp.tool()
@_tag_errors
def rr_backtrace(max_depth: int = None) -> list[Frame]:
    """Get the call stack (backtrace).

    Args:
        max_depth: Maximum number of frames to return.
    """
    session = _require_session()
    return session.backtrace(max_depth=max_depth)


@mcp.tool()
@_tag_errors
def rr_evaluate(expression: str) -> EvalResult:
    """Evaluate a C/C++ expression in the current context.

    Args:
        expression: Expression to evaluate (e.g. "x + 1", "sizeof(struct foo)").
    """
    session = _require_session()
    value = session.evaluate(expression)
    return EvalResult(expression=expression, value=value)


@mcp.tool()
@_tag_errors
def rr_locals() -> list[Variable]:
    """List local variables with their current values."""
    session = _require_session()
    return session.locals()


@mcp.tool()
@_tag_errors
def rr_read_memory(address: str, count: int = 64) -> MemoryResult:
    """Read raw memory bytes at an address.

    Args:
        address: Memory address to read (e.g. "0x7ffd1234").
        count: Number of bytes to read (default 64).
    """
    session = _require_session()
    hex_bytes = session.read_memory(address, count=count)
    return MemoryResult(address=address, count=count, contents=hex_bytes)


@mcp.tool()
@_tag_errors
def rr_registers(register_names: list[str] = None) -> RegisterValues:
    """Read CPU registers.

    Args:
        register_names: Specific register names to read. If omitted, reads all.
    """
    session = _require_session()
    return RegisterValues(registers=session.registers(register_names=register_names))


@mcp.tool()
@_tag_errors
def rr_source_lines(file: str = None, line: int = None,
                    count: int = 10) -> StringResult:
    """List source code around current position or a specific location.

    Args:
        file: Source file path. If omitted, uses current position.
        line: Line number. If omitted, uses current position.
        count: Number of lines to show (default 10).
    """
    session = _require_session()
    src = session.source_lines(file=file, line=line, count=count)
    return StringResult(result=src if src else "No source available.")


# --- Checkpoint Tools ---

@mcp.tool()
@_tag_errors
def rr_checkpoint_save() -> StringResult:
    """Save a checkpoint at the current position. Returns checkpoint info."""
    session = _require_session()
    output = session.checkpoint_save()
    return StringResult(result=output if output else "Checkpoint saved.")


@mcp.tool()
@_tag_errors
def rr_checkpoint_restore(checkpoint_id: int) -> StopEvent:
    """Restore to a previously saved checkpoint.

    Args:
        checkpoint_id: The checkpoint ID to restore.
    """
    session = _require_session()
    result = session.checkpoint_restore(checkpoint_id)
    if isinstance(result, StopEvent):
        return result
    raise GdbSessionError("No stop event after checkpoint restore")


def _watch_parent():
    """Background thread: exit when parent process dies.

    anyio.run() overrides SIGTERM handling, so os.kill(SIGTERM) would be
    swallowed. Instead, call _cleanup() directly and use os._exit(0) to
    force a clean exit that bypasses the blocked event loop.
    """
    import threading
    import time

    ppid = os.getppid()

    def _monitor():
        while True:
            time.sleep(2)
            if os.getppid() != ppid:
                _cleanup()
                os._exit(0)

    t = threading.Thread(target=_monitor, daemon=True)
    t.start()


def main():
    _watch_parent()
    mcp.run(transport="stdio")

1	# -- coding: utf-8 --
2	# Copyright 2026 Karellen, Inc.
3	#
4	# Licensed under the Apache License, Version 2.0 (the "License");
5	# you may not use this file except in compliance with the License.
6	# You may obtain a copy of the License at
7	#
8	# http://www.apache.org/licenses/LICENSE-2.0
9	#
10	# Unless required by applicable law or agreed to in writing, software
11	# distributed under the License is distributed on an "AS IS" BASIS,
12	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	# See the License for the specific language governing permissions and
14	# limitations under the License.
15
16	"""FastMCP server with tool definitions for rr reverse debugging."""
17
18	import atexit	5✔
19	import functools	5✔
20	import logging	5✔
21	import os	5✔
22	import signal	5✔
23	import traceback	5✔
24
25	from mcp.server.fastmcp import FastMCP	5✔
26	from mcp.server.fastmcp.exceptions import ToolError	5✔
27
28	from karellen_rr_mcp.gdb_session import GdbSession, GdbSessionError	5✔
29	from karellen_rr_mcp.rr_manager import (	5✔
30	record as rr_record_cmd,
31	list_recordings as rr_list,
32	list_processes as rr_ps_cmd,
33	trace_info as rr_trace_info_cmd,
34	remove_recording as rr_rm_cmd,
35	ReplayServer, RrError,
36	)
37	from karellen_rr_mcp.types import (	5✔
38	Breakpoint, Frame, StopEvent, Variable, ThreadInfo,
39	ProcessInfo, RecordResult, EvalResult, MemoryResult,
40	ReplayStatus, StringResult, IntResult, RegisterValues,
41	)
42
43	logger = logging.getLogger(__name__)	5✔
44
45	mcp = FastMCP("karellen-rr-mcp", instructions=(	5✔
46	"rr reverse debugging server. Use rr_record to record a failing test, "
47	"rr_replay_start to begin debugging, then use execution control and "
48	"inspection tools to investigate. Use reverse=True to go backwards."
49	))
50
51	# Module-level singleton session state
52	_replay_server = None	5✔
53	_gdb_session = None	5✔
54
55
56	def _cleanup():	5✔
57	global _replay_server, _gdb_session
58	if _gdb_session is not None:	5!
59	try:	5✔
60	_gdb_session.close()	5✔
61	except Exception:	×
62	pass	×
63	_gdb_session = None	5✔
64	if _replay_server is not None:	5!
65	try:	5✔
66	_replay_server.stop()	5✔
67	except Exception:	×
68	pass	×
69	_replay_server = None	5✔
70
71
72	atexit.register(_cleanup)	5✔
73
74
75	def _handle_signal(signum, frame):	5✔
76	_cleanup()	×
NEW 77	os._exit(0)	×
78
79
80	signal.signal(signal.SIGINT, _handle_signal)	5✔
81	signal.signal(signal.SIGTERM, _handle_signal)	5✔
82
83
84	def _require_session():	5✔
85	if _gdb_session is None or not _gdb_session.is_connected():	5✔
86	raise GdbSessionError("No active replay session. Call rr_replay_start first.")	5✔
87	return _gdb_session	5✔
88
89
90	def _tag_errors(fn):	5✔
91	@functools.wraps(fn)	5✔
92	def wrapper(args, *kwargs):	5✔
93	try:	5✔
94	return fn(args, *kwargs)	5✔
95	except GdbSessionError as e:	5✔
96	raise ToolError("gdb: %s" % e) from e	5✔
97	except RrError as e:	5✔
98	raise ToolError("rr: %s" % e) from e	5✔
99	except ToolError:	5✔
100	raise	5✔
101	except Exception as e:	5✔
102	tb = traceback.extract_tb(e.__traceback__)	5✔
103	tb_lines = ["%s:%d in %s" % (f.filename, f.lineno, f.name) for f in tb[-3:]]	5✔
104	raise ToolError("internal: %s: %s\n %s" % (	5✔
105	type(e).__name__, e, "\n ".join(tb_lines))) from e
106	return wrapper	5✔
107
108
109	def _require_stop(stop):	5✔
110	if stop is None:	5✔
111	raise GdbSessionError("No stop event received")	5✔
112	return stop	5✔
113
114
115	# --- Session Lifecycle Tools ---
116
117	@mcp.tool()	5✔
118	@_tag_errors	5✔
119	def rr_record(command: list[str], working_directory: str = None,	5✔
120	env: dict[str, str] = None,
121	trace_dir: str = None) -> RecordResult:
122	"""Record a command with rr. Returns trace directory path.
123
124	Args:
125	command: Command and arguments to record (e.g. ["./my_test"]).
126	working_directory: Working directory for the recorded process.
127	env: Optional extra environment variables for the recorded process.
128	trace_dir: Output trace directory. If omitted, rr uses its default (~/.local/share/rr/).
129	"""
130	trace_dir, exit_code, stdout, stderr = rr_record_cmd(	5✔
131	command, working_directory=working_directory, env=env,
132	trace_dir=trace_dir)
133	return RecordResult(trace_dir=trace_dir, exit_code=exit_code,	5✔
134	stdout=stdout, stderr=stderr)
135
136
137	@mcp.tool()	5✔
138	@_tag_errors	5✔
139	def rr_replay_start(trace_dir: str = None, pid: int = None) -> ReplayStatus:	5✔
140	"""Start a replay session. Launches rr gdbserver and connects GDB/MI.
141
142	Args:
143	trace_dir: Path to rr trace directory. If omitted, uses the latest trace.
144	pid: PID of a specific subprocess to replay. Use rr_ps to list available processes.
145	"""
146	global _replay_server, _gdb_session
147	if _gdb_session is not None:	5✔
148	raise ToolError("gdb: A replay session is already active. Call rr_replay_stop first.")	5✔
149
150	server = ReplayServer(trace_dir=trace_dir, pid=pid)	5✔
151	session = None	5✔
152	try:	5✔
153	server.start()	5✔
154
155	session = GdbSession()	5✔
156	session.start()	5✔
157	session.connect("localhost", server.port)	5✔
158
159	_replay_server = server	5✔
160	_gdb_session = session	5✔
161	return ReplayStatus(port=server.port, message="Program is paused at start.")	5✔
162	except (RrError, GdbSessionError):	5✔
163	if session is not None:	5!
164	try:	5✔
165	session.close()	5✔
166	except Exception:	×
167	pass	×
168	try:	5✔
169	server.stop()	5✔
170	except Exception:	×
171	pass	×
172	raise	5✔
173
174
175	@mcp.tool()	5✔
176	@_tag_errors	5✔
177	def rr_replay_stop() -> StringResult:	5✔
178	"""Stop the current replay session and clean up."""
179	if _gdb_session is None and _replay_server is None:	5✔
180	return StringResult(result="No active replay session.")	5✔
181	_cleanup()	5✔
182	return StringResult(result="Replay session stopped.")	5✔
183
184
185	@mcp.tool()	5✔
186	@_tag_errors	5✔
187	def rr_list_recordings(trace_base_dir: str = None) -> list[str]:	5✔
188	"""List available rr trace recordings.
189
190	Args:
191	trace_base_dir: Base directory for traces. Defaults to ~/.local/share/rr.
192	"""
193	return rr_list(trace_base_dir=trace_base_dir)	5✔
194
195
196	@mcp.tool()	5✔
197	@_tag_errors	5✔
198	def rr_ps(trace_dir: str) -> list[ProcessInfo]:	5✔
199	"""List processes in an rr trace recording.
200
201	Args:
202	trace_dir: Path to rr trace directory.
203	"""
204	return rr_ps_cmd(trace_dir)	5✔
205
206
207	@mcp.tool()	5✔
208	@_tag_errors	5✔
209	def rr_traceinfo(trace_dir: str) -> StringResult:	5✔
210	"""Get trace metadata (header info in JSON format).
211
212	Args:
213	trace_dir: Path to rr trace directory.
214	"""
215	return StringResult(result=rr_trace_info_cmd(trace_dir))	×
216
217
218	@mcp.tool()	5✔
219	@_tag_errors	5✔
220	def rr_rm(trace_dir: str) -> StringResult:	5✔
221	"""Remove an rr trace recording.
222
223	Args:
224	trace_dir: Path to rr trace directory to remove.
225	"""
226	rr_rm_cmd(trace_dir)	×
227	return StringResult(result="Trace removed: %s" % trace_dir)	×
228
229
230	@mcp.tool()	5✔
231	@_tag_errors	5✔
232	def rr_when() -> StringResult:	5✔
233	"""Get the current rr event number. Useful for knowing your position in the trace."""
234	session = _require_session()	×
235	output = session.rr_when()	×
236	return StringResult(result=output if output else "Unable to determine current event.")	×
237
238
239	# --- Breakpoint Tools ---
240
241	@mcp.tool()	5✔
242	@_tag_errors	5✔
243	def rr_breakpoint_set(location: str, condition: str = None,	5✔
244	temporary: bool = False) -> Breakpoint:
245	"""Set a breakpoint at a function, file:line, or address.
246
247	Args:
248	location: Breakpoint location (e.g. "main", "foo.c:42", "*0x400500").
249	condition: Optional condition expression (e.g. "i > 10").
250	temporary: If true, breakpoint is deleted after first hit.
251	"""
252	session = _require_session()	5✔
253	return session.breakpoint_set(location, condition=condition, temporary=temporary)	5✔
254
255
256	@mcp.tool()	5✔
257	@_tag_errors	5✔
258	def rr_breakpoint_remove(breakpoint_number: int) -> IntResult:	5✔
259	"""Remove a breakpoint by its number.
260
261	Args:
262	breakpoint_number: The breakpoint number to remove.
263	"""
264	session = _require_session()	5✔
265	session.breakpoint_delete(breakpoint_number)	5✔
266	return IntResult(result=breakpoint_number)	5✔
267
268
269	@mcp.tool()	5✔
270	@_tag_errors	5✔
271	def rr_breakpoint_list() -> list[Breakpoint]:	5✔
272	"""List all breakpoints."""
273	session = _require_session()	5✔
274	return session.breakpoint_list()	5✔
275
276
277	@mcp.tool()	5✔
278	@_tag_errors	5✔
279	def rr_watchpoint_set(expression: str,	5✔
280	access_type: str = "write") -> Breakpoint:
281	"""Set a hardware watchpoint on a variable or expression.
282
283	Args:
284	expression: Expression to watch (e.g. "my_var", "*0x601050").
285	access_type: One of "write", "read", or "access".
286	"""
287	session = _require_session()	5✔
288	wp = session.watchpoint_set(expression, access_type=access_type)	5✔
289	if wp is None:	5✔
290	raise GdbSessionError("Failed to parse watchpoint response")	5✔
291	return wp	5✔
292
293
294	# --- Execution Control Tools ---
295
296	@mcp.tool()	5✔
297	@_tag_errors	5✔
298	def rr_continue(reverse: bool = False) -> StopEvent:	5✔
299	"""Continue execution forward or backward until breakpoint/signal/end.
300
301	Args:
302	reverse: If true, continue backward (reverse execution).
303	"""
304	session = _require_session()	5✔
305	return _require_stop(session.continue_execution(reverse=reverse))	5✔
306
307
308	@mcp.tool()	5✔
309	@_tag_errors	5✔
310	def rr_step(count: int = 1, reverse: bool = False) -> StopEvent:	5✔
311	"""Step into (source-level) forward or reverse.
312
313	Args:
314	count: Number of steps (forward only).
315	reverse: If true, step backward.
316	"""
317	session = _require_session()	5✔
318	return _require_stop(session.step(count=count, reverse=reverse))	5✔
319
320
321	@mcp.tool()	5✔
322	@_tag_errors	5✔
323	def rr_next(count: int = 1, reverse: bool = False) -> StopEvent:	5✔
324	"""Step over (source-level) forward or reverse.
325
326	Args:
327	count: Number of steps (forward only).
328	reverse: If true, step backward.
329	"""
330	session = _require_session()	5✔
331	return _require_stop(session.next(count=count, reverse=reverse))	5✔
332
333
334	@mcp.tool()	5✔
335	@_tag_errors	5✔
336	def rr_finish(reverse: bool = False) -> StopEvent:	5✔
337	"""Run to function return (or to call site if reverse).
338
339	Args:
340	reverse: If true, run backward to the call site.
341	"""
342	session = _require_session()	5✔
343	return _require_stop(session.finish(reverse=reverse))	5✔
344
345
346	@mcp.tool()	5✔
347	@_tag_errors	5✔
348	def rr_run_to_event(event_number: int) -> StopEvent:	5✔
349	"""Jump to a specific rr event number.
350
351	Args:
352	event_number: The rr event number to seek to.
353	"""
354	session = _require_session()	5✔
355	return _require_stop(session.run_to_event(event_number))	5✔
356
357
358	# --- Thread and Frame Tools ---
359
360	@mcp.tool()	5✔
361	@_tag_errors	5✔
362	def rr_thread_list() -> list[ThreadInfo]:	5✔
363	"""List all threads in the replayed process with their current state and location."""
364	session = _require_session()	5✔
365	return session.thread_info()	5✔
366
367
368	@mcp.tool()	5✔
369	@_tag_errors	5✔
370	def rr_thread_select(thread_id: str) -> IntResult:	5✔
371	"""Switch to a different thread.
372
373	Args:
374	thread_id: Thread ID to switch to (from rr_thread_list).
375	"""
376	session = _require_session()	5✔
377	session.thread_select(thread_id)	5✔
378	return IntResult(result=int(thread_id))	5✔
379
380
381	@mcp.tool()	5✔
382	@_tag_errors	5✔
383	def rr_select_frame(frame_level: int) -> IntResult:	5✔
384	"""Select a stack frame for inspection. After selecting, rr_locals and rr_evaluate
385	operate in the selected frame's context.
386
387	Args:
388	frame_level: Frame number from rr_backtrace (0 = innermost/current).
389	"""
390	session = _require_session()	5✔
391	session.select_frame(frame_level)	5✔
392	return IntResult(result=frame_level)	5✔
393
394
395	# --- State Inspection Tools ---
396
397	@mcp.tool()	5✔
398	@_tag_errors	5✔
399	def rr_backtrace(max_depth: int = None) -> list[Frame]:	5✔
400	"""Get the call stack (backtrace).
401
402	Args:
403	max_depth: Maximum number of frames to return.
404	"""
405	session = _require_session()	5✔
406	return session.backtrace(max_depth=max_depth)	5✔
407
408
409	@mcp.tool()	5✔
410	@_tag_errors	5✔
411	def rr_evaluate(expression: str) -> EvalResult:	5✔
412	"""Evaluate a C/C++ expression in the current context.
413
414	Args:
415	expression: Expression to evaluate (e.g. "x + 1", "sizeof(struct foo)").
416	"""
417	session = _require_session()	5✔
418	value = session.evaluate(expression)	5✔
419	return EvalResult(expression=expression, value=value)	5✔
420
421
422	@mcp.tool()	5✔
423	@_tag_errors	5✔
424	def rr_locals() -> list[Variable]:	5✔
425	"""List local variables with their current values."""
426	session = _require_session()	5✔
427	return session.locals()	5✔
428
429
430	@mcp.tool()	5✔
431	@_tag_errors	5✔
432	def rr_read_memory(address: str, count: int = 64) -> MemoryResult:	5✔
433	"""Read raw memory bytes at an address.
434
435	Args:
436	address: Memory address to read (e.g. "0x7ffd1234").
437	count: Number of bytes to read (default 64).
438	"""
439	session = _require_session()	5✔
440	hex_bytes = session.read_memory(address, count=count)	5✔
441	return MemoryResult(address=address, count=count, contents=hex_bytes)	5✔
442
443
444	@mcp.tool()	5✔
445	@_tag_errors	5✔
446	def rr_registers(register_names: list[str] = None) -> RegisterValues:	5✔
447	"""Read CPU registers.
448
449	Args:
450	register_names: Specific register names to read. If omitted, reads all.
451	"""
452	session = _require_session()	5✔
453	return RegisterValues(registers=session.registers(register_names=register_names))	5✔
454
455
456	@mcp.tool()	5✔
457	@_tag_errors	5✔
458	def rr_source_lines(file: str = None, line: int = None,	5✔
459	count: int = 10) -> StringResult:
460	"""List source code around current position or a specific location.
461
462	Args:
463	file: Source file path. If omitted, uses current position.
464	line: Line number. If omitted, uses current position.
465	count: Number of lines to show (default 10).
466	"""
467	session = _require_session()	5✔
468	src = session.source_lines(file=file, line=line, count=count)	5✔
469	return StringResult(result=src if src else "No source available.")	5✔
470
471
472	# --- Checkpoint Tools ---
473
474	@mcp.tool()	5✔
475	@_tag_errors	5✔
476	def rr_checkpoint_save() -> StringResult:	5✔
477	"""Save a checkpoint at the current position. Returns checkpoint info."""
478	session = _require_session()	5✔
479	output = session.checkpoint_save()	5✔
480	return StringResult(result=output if output else "Checkpoint saved.")	5✔
481
482
483	@mcp.tool()	5✔
484	@_tag_errors	5✔
485	def rr_checkpoint_restore(checkpoint_id: int) -> StopEvent:	5✔
486	"""Restore to a previously saved checkpoint.
487
488	Args:
489	checkpoint_id: The checkpoint ID to restore.
490	"""
491	session = _require_session()	5✔
492	result = session.checkpoint_restore(checkpoint_id)	5✔
493	if isinstance(result, StopEvent):	5✔
494	return result	5✔
495	raise GdbSessionError("No stop event after checkpoint restore")	5✔
496
497
498	def _watch_parent():	5✔
499	"""Background thread: exit when parent process dies.
500
501	anyio.run() overrides SIGTERM handling, so os.kill(SIGTERM) would be
502	swallowed. Instead, call _cleanup() directly and use os._exit(0) to
503	force a clean exit that bypasses the blocked event loop.
504	"""
NEW 505	import threading	×
NEW 506	import time	×
507
NEW 508	ppid = os.getppid()	×
509
NEW 510	def _monitor():	×
NEW 511	while True:	×
NEW 512	time.sleep(2)	×
NEW 513	if os.getppid() != ppid:	×
NEW 514	_cleanup()	×
NEW 515	os._exit(0)	×
516
NEW 517	t = threading.Thread(target=_monitor, daemon=True)	×
NEW 518	t.start()	×
519
520
521	def main():	5✔
NEW 522	_watch_parent()	×
UNCOV 523	mcp.run(transport="stdio")	×

karellen / karellen-rr-mcp / 22788375261

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous