Per-agent x64dbg session through debugger()/debugger_tools()

Harbor starts the x64dbg-automate MCP server (stdio) on first use and keeps
it for the life of the agent's Harbor process; the debuggee is terminated
when that process exits. Configured by debugger_command/_path/_temp in
[reverse].

harbor/cli.py

@@ -255,7 +255,7 @@ def build_parser():
     p = sub.add_parser("who", help="who has worked on a scope (notes, reviews, claims, messages)")
     p.add_argument("scope")
     p = sub.add_parser("re", help="reverse-engineering lookups from the project's image and annotations")
-    p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools"])
+    p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools", "debugger-tools"])
     p.add_argument("target", nargs="?", default="", help="an address (0x4AD850) or a name; for ghidra, the tool")
     p.add_argument("arguments", nargs="?", default="{}", help="ghidra: the tool's arguments as JSON")
     p.add_argument("--max-bytes", type=int, dest="max_bytes")
@@ -675,8 +675,13 @@ def main(argv=None):
         out(r, f"summary #{r['id']} pinned to {r['thread']}")
     elif cmd == "re":
         from .reverse import Ghidra, Reverse
-        r = None if args.action.startswith("ghidra") else Reverse(harbor.project.reverse)
-        if args.action == "ghidra-tools":
+        r = None if args.action.startswith(("ghidra", "debugger")) else Reverse(harbor.project.reverse)
+        if args.action == "debugger-tools":
+            from .reverse import Debugger
+            res = Debugger(harbor.project.reverse).tools()
+            out(res, "\n".join(f"{t['name']:<22} {(t['description'] or '').splitlines()[0][:90] if t['description'] else ''}"
+                                for t in res))
+        elif args.action == "ghidra-tools":
             res = Ghidra(harbor.project.reverse).tools()
             out(res, "\n".join(f"{t['name']:<26} {t['description'].splitlines()[0][:90]}" for t in res))
         elif args.action == "ghidra":

harbor/mcp_server.py

@@ -37,7 +37,9 @@ review queue, presence and stop flags. Protocol:
 11. To read the original machine code use re_disassemble, re_xrefs, re_bytes and re_lookup: they read the project's
     own image with the team's current names, so do not write your own disassembler or offset maths.
     For Ghidra's view (decompiler, p-code, stack frames, struct field references, types) use ghidra_tools() once,
-    then ghidra(tool, arguments): one shared read-only Ghidra session holding the team's annotations."""
+    then ghidra(tool, arguments): one shared read-only Ghidra session holding the team's annotations.
+    To check a value in the running game (x87 control word, `this`, struct fields), use debugger_tools() and
+    debugger(tool, arguments): your own x32dbg session; always end it with debugger('terminate_session')."""
 
 KindName = Literal["note", "question", "answer", "request", "finding", "retraction", "decision", "status", "handoff"]
 Importance = Literal["normal", "high", "urgent"]
@@ -517,6 +519,37 @@ def ghidra(tool: str, arguments: dict | None = None) -> str:
     return _gh(lambda g: g.call(tool, arguments or {}))
 
 
+_debugger = {}
+
+
+def _dbg(fn):
+    try:
+        h = harbor()
+        d = _debugger.get(h.project.name)
+        if d is None:
+            from .reverse import Debugger
+            d = _debugger[h.project.name] = Debugger(h.project.reverse)
+        return fn(d)
+    except (HarborError, projects.ProjectError) as error:
+        raise ToolError(str(error)) from None
+
+
+@mcp.tool()
+def debugger_tools() -> list:
+    """The tools of your own x64dbg session (x64dbg-automate on x32dbg): start_session, set_breakpoint, go,
+    get_all_registers, read_memory, eval_expression, ... Call them with debugger(tool, arguments)."""
+    return _dbg(lambda d: d.tools())
+
+
+@mcp.tool()
+def debugger(tool: str, arguments: dict | None = None) -> str:
+    """Run one x64dbg tool in your own debugger session, e.g. debugger('start_session', {'target_exe':
+    "R:/SteamLibrary/steamapps/common/Sid Meier's Pirates!/Pirates!.exe"}), debugger('set_breakpoint', ...),
+    debugger('go'), debugger('get_all_registers'). The session lasts until debugger('terminate_session') or
+    until your app closes. Use it to read values, not to patch the game."""
+    return _dbg(lambda d: d.call(tool, arguments or {}))
+
+
 def _report_in():
     """Tell Harbor this agent's app has its tools running: at start, then every 5 minutes while the app keeps the
     server open (updates the connection only, not the agent's last-seen time)."""

harbor/reverse.py

@@ -356,3 +356,99 @@ class Ghidra:
         if result.get("isError"):
             raise HarborError(text.strip() or f"Ghidra refused {tool}")
         return re.sub(r"^\[Context\] Operating on: [^\n]*\n+", "", text)
+
+
+class Debugger:
+    """A per-agent x64dbg session through the x64dbg-automate MCP server (stdio), started on first use and kept for
+    the life of this Harbor process, so breakpoints and the paused game survive between calls. Configured by
+    `debugger_command`, `debugger_path` (x32dbg.exe) and `debugger_temp` in the [reverse] table. The plugin and the
+    client find each other through a session file in TEMP, so both get the same TEMP."""
+
+    def __init__(self, config):
+        config = config or {}
+        self.command = config.get("debugger_command")
+        if not self.command:
+            raise HarborError("this project has no debugger_command in its [reverse] table")
+        self.env = dict(os.environ)
+        if config.get("debugger_path"):
+            self.env["X64DBG_PATH"] = str(config["debugger_path"])
+        if config.get("debugger_temp"):
+            self.env["TEMP"] = self.env["TMP"] = str(config["debugger_temp"])
+        self.proc = None
+        self.counter = 0
+        self.lock = __import__("threading").Lock()
+
+    def _start(self):
+        import atexit
+        import queue
+        import subprocess
+        import threading
+        command = self.command if isinstance(self.command, list) else [self.command]
+        self.proc = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                                     stderr=subprocess.DEVNULL, env=self.env, text=True, encoding="utf-8",
+                                     creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
+        self.answers = queue.Queue()
+
+        def read(stream, answers):
+            for line in stream:
+                if line.strip():
+                    try:
+                        answers.put(json.loads(line))
+                    except ValueError:
+                        pass
+            answers.put(None)
+        threading.Thread(target=read, args=(self.proc.stdout, self.answers), daemon=True).start()
+        self._rpc("initialize", {"protocolVersion": "2025-06-18", "capabilities": {},
+                                 "clientInfo": {"name": "harbor", "version": "2"}})
+        self._send({"jsonrpc": "2.0", "method": "notifications/initialized"})
+        atexit.register(self.close)
+
+    def _send(self, body):
+        self.proc.stdin.write(json.dumps(body) + "\n")
+        self.proc.stdin.flush()
+
+    def _rpc(self, method, params, timeout=120):
+        import queue
+        self.counter += 1
+        self._send({"jsonrpc": "2.0", "id": self.counter, "method": method, "params": params})
+        while True:
+            try:
+                answer = self.answers.get(timeout=timeout)
+            except queue.Empty:
+                raise HarborError(f"the debugger did not answer {method} within {timeout}s") from None
+            if answer is None:
+                self.proc = None
+                raise HarborError("the debugger's MCP server exited")
+            if answer.get("id") == self.counter:
+                if "error" in answer:
+                    raise HarborError(f"debugger: {answer['error'].get('message', answer['error'])}")
+                return answer.get("result")
+
+    def tools(self):
+        with self.lock:
+            if self.proc is None or self.proc.poll() is not None:
+                self._start()
+            return [{"name": t["name"], "description": t.get("description", ""),
+                     "arguments": (t.get("inputSchema") or {}).get("properties", {}),
+                     "required": (t.get("inputSchema") or {}).get("required", [])}
+                    for t in self._rpc("tools/list", {})["tools"]]
+
+    def call(self, tool, arguments=None, timeout=300):
+        with self.lock:
+            if self.proc is None or self.proc.poll() is not None:
+                self._start()
+            result = self._rpc("tools/call", {"name": tool, "arguments": arguments or {}}, timeout=timeout)
+        text = "\n".join(c.get("text", "") for c in result.get("content", []) if isinstance(c, dict))
+        if result.get("isError"):
+            raise HarborError(text.strip() or f"the debugger refused {tool}")
+        return text
+
+    def close(self):
+        """End the debuggee and the debugger with the Harbor process, so no paused game is left behind."""
+        if self.proc and self.proc.poll() is None:
+            try:
+                self._rpc("tools/call", {"name": "terminate_session", "arguments": {}}, timeout=15)
+            except HarborError:
+                pass
+            self.proc.terminate()
+        self.proc = None

tests/fake_stdio_mcp.py

@@ -0,0 +1,18 @@
+"""A fake stdio MCP server for the debugger forwarder test: one tool that echoes, one that fails."""
+import json
+import sys
+
+for line in sys.stdin:
+    msg = json.loads(line)
+    if "id" not in msg:
+        continue
+    if msg["method"] == "initialize":
+        result = {"protocolVersion": "2025-06-18"}
+    elif msg["method"] == "tools/list":
+        result = {"tools": [{"name": "echo", "description": "Echo", "inputSchema": {"properties": {"x": {}}}}]}
+    elif msg["params"]["name"] == "echo":
+        result = {"content": [{"type": "text", "text": "echo " + json.dumps(msg["params"]["arguments"])}]}
+    else:
+        result = {"isError": True, "content": [{"type": "text", "text": "no such session"}]}
+    print(json.dumps({"jsonrpc": "2.0", "method": "notifications/message", "params": {}}), flush=True)
+    print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": result}), flush=True)

tests/test_reverse.py

@@ -117,3 +117,22 @@ class GhidraForwarder(unittest.TestCase):
         with self.assertRaises(HarborError):
             from harbor.reverse import Ghidra
             Ghidra({"ghidra_url": "http://127.0.0.1:9/mcp"}).tools()
+
+
+class DebuggerForwarder(unittest.TestCase):
+    """debugger() keeps one stdio MCP server per Harbor process and skips its notifications."""
+
+    def test_call_and_error(self):
+        import sys
+        from pathlib import Path
+        from harbor.reverse import Debugger
+        d = Debugger({"debugger_command": [sys.executable, "-B", str(Path(__file__).with_name("fake_stdio_mcp.py"))]})
+        try:
+            self.assertEqual(d.tools()[0]["name"], "echo")
+            self.assertEqual(d.call("echo", {"x": 1}), 'echo {"x": 1}')
+            with self.assertRaises(HarborError):
+                d.call("terminate_session")
+        finally:
+            d.close()
+        with self.assertRaises(HarborError):
+            Debugger({})