22668923328

Committed 04 Mar 2026 12:15PM UTC coverage: 83.979%. First build

Build # 22668923328

Build Type

push

github

Committed by

arcivanov

Commit Message

Initial commit

Coverage Stats

139 of 198 branches covered (70.2%)

Branch coverage included in aggregate %.

663 of 757 new or added lines in 6 files covered. (87.58%)

663 of 757 relevant lines covered (87.58%)

3.5 hits per line

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

81.68

/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 logging
import signal

from mcp.server.fastmcp import FastMCP

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,
    ReplayServer, RrError,
)

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()


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 _format_stop_event(stop):
    if stop is None:
        return "Program stopped (no details available)"
    parts = ["Stopped: %s" % stop.reason]
    if stop.frame:
        f = stop.frame
        loc = f.function or "??"
        if f.file and f.line:
            loc += " at %s:%d" % (f.file, f.line)
        parts.append("Location: %s" % loc)
        parts.append("Address: %s" % f.address)
    if stop.breakpoint_number is not None:
        parts.append("Breakpoint: #%d" % stop.breakpoint_number)
    if stop.signal_name:
        parts.append("Signal: %s (%s)" % (stop.signal_name,
                                          stop.signal_meaning or ""))
    return "\n".join(parts)


def _format_breakpoint(bp):
    parts = ["Breakpoint #%d: %s" % (bp.number, bp.location)]
    if bp.file and bp.line:
        parts.append("  File: %s:%d" % (bp.file, bp.line))
    if bp.condition:
        parts.append("  Condition: %s" % bp.condition)
    parts.append("  Enabled: %s" % bp.enabled)
    return "\n".join(parts)


def _format_frame(frame):
    loc = frame.function or "??"
    if frame.file and frame.line:
        loc += " at %s:%d" % (frame.file, frame.line)
    return "#%d  %s  (%s)" % (frame.level, frame.address, loc)


# --- Session Lifecycle Tools ---

@mcp.tool()
def rr_record(command: list[str], working_directory: str = None,
              env: dict[str, str] = None) -> str:
    """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.
    """
    try:
        trace_dir, exit_code, stdout, stderr = rr_record_cmd(
            command, working_directory=working_directory, env=env)
        parts = ["Recording complete."]
        if trace_dir:
            parts.append("Trace directory: %s" % trace_dir)
        parts.append("Exit code: %d" % exit_code)
        if stdout.strip():
            parts.append("--- stdout ---\n%s" % stdout.strip())
        if stderr.strip():
            parts.append("--- stderr ---\n%s" % stderr.strip())
        return "\n".join(parts)
    except RrError as e:
        return "Error: %s" % e


@mcp.tool()
def rr_replay_start(trace_dir: str = None) -> str:
    """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.
    """
    global _replay_server, _gdb_session
    try:
        if _gdb_session is not None:
            return "Error: A replay session is already active. Call rr_replay_stop first."

        server = ReplayServer(trace_dir=trace_dir)
        server.start()

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

        _replay_server = server
        _gdb_session = session
        return "Replay session started on port %d. Program is paused at start." % server.port
    except (RrError, GdbSessionError) as e:
        # Clean up on failure
        if session is not None:
            try:
                session.close()
            except Exception:
                pass
        if server is not None:
            try:
                server.stop()
            except Exception:
                pass
        return "Error: %s" % e


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


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

    Args:
        trace_base_dir: Base directory for traces. Defaults to ~/.local/share/rr.
    """
    recordings = rr_list(trace_base_dir=trace_base_dir)
    if not recordings:
        return "No recordings found."
    return "Available recordings:\n" + "\n".join("  - %s" % r for r in recordings)


# --- Breakpoint Tools ---

@mcp.tool()
def rr_breakpoint_set(location: str, condition: str = None,
                      temporary: bool = False) -> str:
    """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.
    """
    try:
        session = _require_session()
        bp = session.breakpoint_set(location, condition=condition, temporary=temporary)
        return _format_breakpoint(bp)
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        breakpoint_number: The breakpoint number to remove.
    """
    try:
        session = _require_session()
        session.breakpoint_delete(breakpoint_number)
        return "Breakpoint #%d removed." % breakpoint_number
    except GdbSessionError as e:
        return "Error: %s" % e


@mcp.tool()
def rr_breakpoint_list() -> str:
    """List all breakpoints."""
    try:
        session = _require_session()
        bps = session.breakpoint_list()
        if not bps:
            return "No breakpoints set."
        return "\n\n".join(_format_breakpoint(bp) for bp in bps)
    except GdbSessionError as e:
        return "Error: %s" % e


@mcp.tool()
def rr_watchpoint_set(expression: str,
                      access_type: str = "write") -> str:
    """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".
    """
    try:
        session = _require_session()
        wp = session.watchpoint_set(expression, access_type=access_type)
        if wp:
            return _format_breakpoint(wp)
        return "Watchpoint set on %s (%s)" % (expression, access_type)
    except GdbSessionError as e:
        return "Error: %s" % e


# --- Execution Control Tools ---

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

    Args:
        reverse: If true, continue backward (reverse execution).
    """
    try:
        session = _require_session()
        stop = session.continue_execution(reverse=reverse)
        return _format_stop_event(stop)
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        count: Number of steps (forward only).
        reverse: If true, step backward.
    """
    try:
        session = _require_session()
        stop = session.step(count=count, reverse=reverse)
        return _format_stop_event(stop)
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        count: Number of steps (forward only).
        reverse: If true, step backward.
    """
    try:
        session = _require_session()
        stop = session.next(count=count, reverse=reverse)
        return _format_stop_event(stop)
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        reverse: If true, run backward to the call site.
    """
    try:
        session = _require_session()
        stop = session.finish(reverse=reverse)
        return _format_stop_event(stop)
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        event_number: The rr event number to seek to.
    """
    try:
        session = _require_session()
        stop = session.run_to_event(event_number)
        return _format_stop_event(stop)
    except GdbSessionError as e:
        return "Error: %s" % e


# --- State Inspection Tools ---

@mcp.tool()
def rr_backtrace(max_depth: int = None) -> str:
    """Get the call stack (backtrace).

    Args:
        max_depth: Maximum number of frames to return.
    """
    try:
        session = _require_session()
        frames = session.backtrace(max_depth=max_depth)
        if not frames:
            return "No stack frames."
        return "\n".join(_format_frame(f) for f in frames)
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        expression: Expression to evaluate (e.g. "x + 1", "sizeof(struct foo)").
    """
    try:
        session = _require_session()
        value = session.evaluate(expression)
        return "%s = %s" % (expression, value)
    except GdbSessionError as e:
        return "Error: %s" % e


@mcp.tool()
def rr_locals() -> str:
    """List local variables with their current values."""
    try:
        session = _require_session()
        variables = session.locals()
        if not variables:
            return "No local variables."
        lines = []
        for v in variables:
            line = "%s = %s" % (v.name, v.value)
            if v.type:
                line += "  (%s)" % v.type
            lines.append(line)
        return "\n".join(lines)
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        address: Memory address to read (e.g. "0x7ffd1234").
        count: Number of bytes to read (default 64).
    """
    try:
        session = _require_session()
        hex_bytes = session.read_memory(address, count=count)
        return "Memory at %s (%d bytes): %s" % (address, count, hex_bytes)
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        register_names: Specific register names to read. If omitted, reads all.
    """
    try:
        session = _require_session()
        regs = session.registers(register_names=register_names)
        if not regs:
            return "No register values."
        return "\n".join("  %s = %s" % (k, v) for k, v in regs.items())
    except GdbSessionError as e:
        return "Error: %s" % e


@mcp.tool()
def rr_source_lines(file: str = None, line: int = None,
                    count: int = 10) -> str:
    """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).
    """
    try:
        session = _require_session()
        src = session.source_lines(file=file, line=line, count=count)
        return src if src else "No source available."
    except GdbSessionError as e:
        return "Error: %s" % e


# --- Checkpoint Tools ---

@mcp.tool()
def rr_checkpoint_save() -> str:
    """Save a checkpoint at the current position. Returns checkpoint info."""
    try:
        session = _require_session()
        output = session.checkpoint_save()
        return output if output else "Checkpoint saved."
    except GdbSessionError as e:
        return "Error: %s" % e


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

    Args:
        checkpoint_id: The checkpoint ID to restore.
    """
    try:
        session = _require_session()
        result = session.checkpoint_restore(checkpoint_id)
        if hasattr(result, "reason"):
            return _format_stop_event(result)
        return str(result) if result else "Checkpoint %d restored." % checkpoint_id
    except GdbSessionError as e:
        return "Error: %s" % e


def main():
    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	4✔
19	import logging	4✔
20	import signal	4✔
21
22	from mcp.server.fastmcp import FastMCP	4✔
23
24	from karellen_rr_mcp.gdb_session import GdbSession, GdbSessionError	4✔
25	from karellen_rr_mcp.rr_manager import (	4✔
26	record as rr_record_cmd,
27	list_recordings as rr_list,
28	ReplayServer, RrError,
29	)
30
31	logger = logging.getLogger(__name__)	4✔
32
33	mcp = FastMCP("karellen-rr-mcp", instructions=(	4✔
34	"rr reverse debugging server. Use rr_record to record a failing test, "
35	"rr_replay_start to begin debugging, then use execution control and "
36	"inspection tools to investigate. Use reverse=True to go backwards."
37	))
38
39	# Module-level singleton session state
40	_replay_server = None	4✔
41	_gdb_session = None	4✔
42
43
44	def _cleanup():	4✔
45	global _replay_server, _gdb_session
46	if _gdb_session is not None:	4!
47	try:	4✔
48	_gdb_session.close()	4✔
NEW 49	except Exception:	×
NEW 50	pass	×
51	_gdb_session = None	4✔
52	if _replay_server is not None:	4!
53	try:	4✔
54	_replay_server.stop()	4✔
NEW 55	except Exception:	×
NEW 56	pass	×
57	_replay_server = None	4✔
58
59
60	atexit.register(_cleanup)	4✔
61
62
63	def _handle_signal(signum, frame):	4✔
NEW 64	_cleanup()	×
65
66
67	signal.signal(signal.SIGTERM, _handle_signal)	4✔
68
69
70	def _require_session():	4✔
71	if _gdb_session is None or not _gdb_session.is_connected():	4✔
72	raise GdbSessionError("No active replay session. Call rr_replay_start first.")	4✔
73	return _gdb_session	4✔
74
75
76	def _format_stop_event(stop):	4✔
77	if stop is None:	4✔
78	return "Program stopped (no details available)"	4✔
79	parts = ["Stopped: %s" % stop.reason]	4✔
80	if stop.frame:	4!
81	f = stop.frame	4✔
82	loc = f.function or "??"	4✔
83	if f.file and f.line:	4✔
84	loc += " at %s:%d" % (f.file, f.line)	4✔
85	parts.append("Location: %s" % loc)	4✔
86	parts.append("Address: %s" % f.address)	4✔
87	if stop.breakpoint_number is not None:	4✔
88	parts.append("Breakpoint: #%d" % stop.breakpoint_number)	4✔
89	if stop.signal_name:	4✔
90	parts.append("Signal: %s (%s)" % (stop.signal_name,	4✔
91	stop.signal_meaning or ""))
92	return "\n".join(parts)	4✔
93
94
95	def _format_breakpoint(bp):	4✔
96	parts = ["Breakpoint #%d: %s" % (bp.number, bp.location)]	4✔
97	if bp.file and bp.line:	4✔
98	parts.append(" File: %s:%d" % (bp.file, bp.line))	4✔
99	if bp.condition:	4✔
100	parts.append(" Condition: %s" % bp.condition)	4✔
101	parts.append(" Enabled: %s" % bp.enabled)	4✔
102	return "\n".join(parts)	4✔
103
104
105	def _format_frame(frame):	4✔
106	loc = frame.function or "??"	4✔
107	if frame.file and frame.line:	4✔
108	loc += " at %s:%d" % (frame.file, frame.line)	4✔
109	return "#%d %s (%s)" % (frame.level, frame.address, loc)	4✔
110
111
112	# --- Session Lifecycle Tools ---
113
114	@mcp.tool()	4✔
115	def rr_record(command: list[str], working_directory: str = None,	4✔
116	env: dict[str, str] = None) -> str:
117	"""Record a command with rr. Returns trace directory path.
118
119	Args:
120	command: Command and arguments to record (e.g. ["./my_test"]).
121	working_directory: Working directory for the recorded process.
122	env: Optional extra environment variables for the recorded process.
123	"""
124	try:	4✔
125	trace_dir, exit_code, stdout, stderr = rr_record_cmd(	4✔
126	command, working_directory=working_directory, env=env)
127	parts = ["Recording complete."]	4✔
128	if trace_dir:	4!
129	parts.append("Trace directory: %s" % trace_dir)	4✔
130	parts.append("Exit code: %d" % exit_code)	4✔
131	if stdout.strip():	4!
132	parts.append("--- stdout ---\n%s" % stdout.strip())	4✔
133	if stderr.strip():	4!
134	parts.append("--- stderr ---\n%s" % stderr.strip())	4✔
135	return "\n".join(parts)	4✔
136	except RrError as e:	4✔
137	return "Error: %s" % e	4✔
138
139
140	@mcp.tool()	4✔
141	def rr_replay_start(trace_dir: str = None) -> str:	4✔
142	"""Start a replay session. Launches rr gdbserver and connects GDB/MI.
143
144	Args:
145	trace_dir: Path to rr trace directory. If omitted, uses the latest trace.
146	"""
147	global _replay_server, _gdb_session
148	try:	4✔
149	if _gdb_session is not None:	4✔
150	return "Error: A replay session is already active. Call rr_replay_stop first."	4✔
151
152	server = ReplayServer(trace_dir=trace_dir)	4✔
153	server.start()	4✔
154
155	session = GdbSession()	4✔
156	session.start()	4✔
157	session.connect("localhost", server.port)	4✔
158
159	_replay_server = server	4✔
160	_gdb_session = session	4✔
161	return "Replay session started on port %d. Program is paused at start." % server.port	4✔
NEW 162	except (RrError, GdbSessionError) as e:	×
163	# Clean up on failure
NEW 164	if session is not None:	×
NEW 165	try:	×
NEW 166	session.close()	×
NEW 167	except Exception:	×
NEW 168	pass	×
NEW 169	if server is not None:	×
NEW 170	try:	×
NEW 171	server.stop()	×
NEW 172	except Exception:	×
NEW 173	pass	×
NEW 174	return "Error: %s" % e	×
175
176
177	@mcp.tool()	4✔
178	def rr_replay_stop() -> str:	4✔
179	"""Stop the current replay session and clean up."""
180	if _gdb_session is None and _replay_server is None:	4✔
181	return "No active replay session."	4✔
182	_cleanup()	4✔
183	return "Replay session stopped."	4✔
184
185
186	@mcp.tool()	4✔
187	def rr_list_recordings(trace_base_dir: str = None) -> str:	4✔
188	"""List available rr trace recordings.
189
190	Args:
191	trace_base_dir: Base directory for traces. Defaults to ~/.local/share/rr.
192	"""
193	recordings = rr_list(trace_base_dir=trace_base_dir)	4✔
194	if not recordings:	4✔
195	return "No recordings found."	4✔
196	return "Available recordings:\n" + "\n".join(" - %s" % r for r in recordings)	4✔
197
198
199	# --- Breakpoint Tools ---
200
201	@mcp.tool()	4✔
202	def rr_breakpoint_set(location: str, condition: str = None,	4✔
203	temporary: bool = False) -> str:
204	"""Set a breakpoint at a function, file:line, or address.
205
206	Args:
207	location: Breakpoint location (e.g. "main", "foo.c:42", "*0x400500").
208	condition: Optional condition expression (e.g. "i > 10").
209	temporary: If true, breakpoint is deleted after first hit.
210	"""
211	try:	4✔
212	session = _require_session()	4✔
213	bp = session.breakpoint_set(location, condition=condition, temporary=temporary)	4✔
214	return _format_breakpoint(bp)	4✔
215	except GdbSessionError as e:	4✔
216	return "Error: %s" % e	4✔
217
218
219	@mcp.tool()	4✔
220	def rr_breakpoint_remove(breakpoint_number: int) -> str:	4✔
221	"""Remove a breakpoint by its number.
222
223	Args:
224	breakpoint_number: The breakpoint number to remove.
225	"""
226	try:	4✔
227	session = _require_session()	4✔
228	session.breakpoint_delete(breakpoint_number)	4✔
229	return "Breakpoint #%d removed." % breakpoint_number	4✔
NEW 230	except GdbSessionError as e:	×
NEW 231	return "Error: %s" % e	×
232
233
234	@mcp.tool()	4✔
235	def rr_breakpoint_list() -> str:	4✔
236	"""List all breakpoints."""
237	try:	4✔
238	session = _require_session()	4✔
239	bps = session.breakpoint_list()	4✔
240	if not bps:	4✔
241	return "No breakpoints set."	4✔
242	return "\n\n".join(_format_breakpoint(bp) for bp in bps)	4✔
NEW 243	except GdbSessionError as e:	×
NEW 244	return "Error: %s" % e	×
245
246
247	@mcp.tool()	4✔
248	def rr_watchpoint_set(expression: str,	4✔
249	access_type: str = "write") -> str:
250	"""Set a hardware watchpoint on a variable or expression.
251
252	Args:
253	expression: Expression to watch (e.g. "my_var", "*0x601050").
254	access_type: One of "write", "read", or "access".
255	"""
256	try:	4✔
257	session = _require_session()	4✔
258	wp = session.watchpoint_set(expression, access_type=access_type)	4✔
259	if wp:	4!
260	return _format_breakpoint(wp)	4✔
NEW 261	return "Watchpoint set on %s (%s)" % (expression, access_type)	×
NEW 262	except GdbSessionError as e:	×
NEW 263	return "Error: %s" % e	×
264
265
266	# --- Execution Control Tools ---
267
268	@mcp.tool()	4✔
269	def rr_continue(reverse: bool = False) -> str:	4✔
270	"""Continue execution forward or backward until breakpoint/signal/end.
271
272	Args:
273	reverse: If true, continue backward (reverse execution).
274	"""
275	try:	4✔
276	session = _require_session()	4✔
277	stop = session.continue_execution(reverse=reverse)	4✔
278	return _format_stop_event(stop)	4✔
279	except GdbSessionError as e:	4✔
280	return "Error: %s" % e	4✔
281
282
283	@mcp.tool()	4✔
284	def rr_step(count: int = 1, reverse: bool = False) -> str:	4✔
285	"""Step into (source-level) forward or reverse.
286
287	Args:
288	count: Number of steps (forward only).
289	reverse: If true, step backward.
290	"""
291	try:	4✔
292	session = _require_session()	4✔
293	stop = session.step(count=count, reverse=reverse)	4✔
294	return _format_stop_event(stop)	4✔
NEW 295	except GdbSessionError as e:	×
NEW 296	return "Error: %s" % e	×
297
298
299	@mcp.tool()	4✔
300	def rr_next(count: int = 1, reverse: bool = False) -> str:	4✔
301	"""Step over (source-level) forward or reverse.
302
303	Args:
304	count: Number of steps (forward only).
305	reverse: If true, step backward.
306	"""
307	try:	4✔
308	session = _require_session()	4✔
309	stop = session.next(count=count, reverse=reverse)	4✔
310	return _format_stop_event(stop)	4✔
NEW 311	except GdbSessionError as e:	×
NEW 312	return "Error: %s" % e	×
313
314
315	@mcp.tool()	4✔
316	def rr_finish(reverse: bool = False) -> str:	4✔
317	"""Run to function return (or to call site if reverse).
318
319	Args:
320	reverse: If true, run backward to the call site.
321	"""
322	try:	4✔
323	session = _require_session()	4✔
324	stop = session.finish(reverse=reverse)	4✔
325	return _format_stop_event(stop)	4✔
NEW 326	except GdbSessionError as e:	×
NEW 327	return "Error: %s" % e	×
328
329
330	@mcp.tool()	4✔
331	def rr_run_to_event(event_number: int) -> str:	4✔
332	"""Jump to a specific rr event number.
333
334	Args:
335	event_number: The rr event number to seek to.
336	"""
337	try:	4✔
338	session = _require_session()	4✔
339	stop = session.run_to_event(event_number)	4✔
340	return _format_stop_event(stop)	4✔
NEW 341	except GdbSessionError as e:	×
NEW 342	return "Error: %s" % e	×
343
344
345	# --- State Inspection Tools ---
346
347	@mcp.tool()	4✔
348	def rr_backtrace(max_depth: int = None) -> str:	4✔
349	"""Get the call stack (backtrace).
350
351	Args:
352	max_depth: Maximum number of frames to return.
353	"""
354	try:	4✔
355	session = _require_session()	4✔
356	frames = session.backtrace(max_depth=max_depth)	4✔
357	if not frames:	4✔
358	return "No stack frames."	4✔
359	return "\n".join(_format_frame(f) for f in frames)	4✔
360	except GdbSessionError as e:	4✔
361	return "Error: %s" % e	4✔
362
363
364	@mcp.tool()	4✔
365	def rr_evaluate(expression: str) -> str:	4✔
366	"""Evaluate a C/C++ expression in the current context.
367
368	Args:
369	expression: Expression to evaluate (e.g. "x + 1", "sizeof(struct foo)").
370	"""
371	try:	4✔
372	session = _require_session()	4✔
373	value = session.evaluate(expression)	4✔
374	return "%s = %s" % (expression, value)	4✔
375	except GdbSessionError as e:	4✔
376	return "Error: %s" % e	4✔
377
378
379	@mcp.tool()	4✔
380	def rr_locals() -> str:	4✔
381	"""List local variables with their current values."""
382	try:	4✔
383	session = _require_session()	4✔
384	variables = session.locals()	4✔
385	if not variables:	4✔
386	return "No local variables."	4✔
387	lines = []	4✔
388	for v in variables:	4✔
389	line = "%s = %s" % (v.name, v.value)	4✔
390	if v.type:	4!
391	line += " (%s)" % v.type	4✔
392	lines.append(line)	4✔
393	return "\n".join(lines)	4✔
NEW 394	except GdbSessionError as e:	×
NEW 395	return "Error: %s" % e	×
396
397
398	@mcp.tool()	4✔
399	def rr_read_memory(address: str, count: int = 64) -> str:	4✔
400	"""Read raw memory bytes at an address.
401
402	Args:
403	address: Memory address to read (e.g. "0x7ffd1234").
404	count: Number of bytes to read (default 64).
405	"""
406	try:	4✔
407	session = _require_session()	4✔
408	hex_bytes = session.read_memory(address, count=count)	4✔
409	return "Memory at %s (%d bytes): %s" % (address, count, hex_bytes)	4✔
NEW 410	except GdbSessionError as e:	×
NEW 411	return "Error: %s" % e	×
412
413
414	@mcp.tool()	4✔
415	def rr_registers(register_names: list[str] = None) -> str:	4✔
416	"""Read CPU registers.
417
418	Args:
419	register_names: Specific register names to read. If omitted, reads all.
420	"""
421	try:	4✔
422	session = _require_session()	4✔
423	regs = session.registers(register_names=register_names)	4✔
424	if not regs:	4!
NEW 425	return "No register values."	×
426	return "\n".join(" %s = %s" % (k, v) for k, v in regs.items())	4✔
NEW 427	except GdbSessionError as e:	×
NEW 428	return "Error: %s" % e	×
429
430
431	@mcp.tool()	4✔
432	def rr_source_lines(file: str = None, line: int = None,	4✔
433	count: int = 10) -> str:
434	"""List source code around current position or a specific location.
435
436	Args:
437	file: Source file path. If omitted, uses current position.
438	line: Line number. If omitted, uses current position.
439	count: Number of lines to show (default 10).
440	"""
441	try:	4✔
442	session = _require_session()	4✔
443	src = session.source_lines(file=file, line=line, count=count)	4✔
444	return src if src else "No source available."	4✔
NEW 445	except GdbSessionError as e:	×
NEW 446	return "Error: %s" % e	×
447
448
449	# --- Checkpoint Tools ---
450
451	@mcp.tool()	4✔
452	def rr_checkpoint_save() -> str:	4✔
453	"""Save a checkpoint at the current position. Returns checkpoint info."""
454	try:	4✔
455	session = _require_session()	4✔
456	output = session.checkpoint_save()	4✔
457	return output if output else "Checkpoint saved."	4✔
NEW 458	except GdbSessionError as e:	×
NEW 459	return "Error: %s" % e	×
460
461
462	@mcp.tool()	4✔
463	def rr_checkpoint_restore(checkpoint_id: int) -> str:	4✔
464	"""Restore to a previously saved checkpoint.
465
466	Args:
467	checkpoint_id: The checkpoint ID to restore.
468	"""
469	try:	4✔
470	session = _require_session()	4✔
471	result = session.checkpoint_restore(checkpoint_id)	4✔
472	if hasattr(result, "reason"):	4✔
473	return _format_stop_event(result)	4✔
474	return str(result) if result else "Checkpoint %d restored." % checkpoint_id	4✔
NEW 475	except GdbSessionError as e:	×
NEW 476	return "Error: %s" % e	×
477
478
479	def main():	4✔
NEW 480	mcp.run(transport="stdio")	×

karellen / karellen-rr-mcp / 22668923328

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