Address review comments

doc/gdb-examples.md

@@ -5,12 +5,6 @@
 
 Speakeasy supports interactive debugging via the GDB Remote Serial Protocol. When you pass `--gdb`, the emulator pauses before the first instruction and waits for a GDB client to connect. You can then set breakpoints, inspect registers and memory, single-step, and continue execution — all through a standard GDB interface.
 
-GDB support is included in the standard installation:
-
-```
-pip install speakeasy-emulator
-```
-
 ### Debugging a 32-bit DLL
 
 Start speakeasy with `--gdb` on a sample DLL. It will pause before the first instruction and wait for GDB to connect. In a second terminal, connect with `gdb` (or `gdb-multiarch`) and set the architecture to `i386`:

doc/gdb.md

@@ -2,24 +2,6 @@
 
 Speakeasy supports interactive debugging of emulated binaries via the GDB Remote Serial Protocol. When enabled, the emulator pauses before the first instruction and waits for a GDB client to connect. You can then set breakpoints, inspect registers and memory, single-step, and continue execution — all through a standard GDB interface.
 
-The GDB server is implemented directly by Speakeasy and has no optional runtime dependencies.
-
----
-
-## Installation
-
-GDB support is included in a normal installation:
-
-```console
-pip install speakeasy-emulator
-```
-
-Or, when installing from source:
-
-```console
-pip install -e .
-```
-
 ---
 
 ## Quick Start

doc/install.md

@@ -8,8 +8,6 @@ Speakeasy requires Python 3.10+.
 python3 -m pip install speakeasy-emulator
 ```
 
-GDB remote debugging support is included in the standard package.
-
 Verify installation:
 
 ```console

speakeasy/engines/unicorn_eng.py

@@ -271,7 +271,13 @@ class EmuEngine:
             return hook.disable()
 
     def hook_remove(self, hid):
-        return self.emu.hook_del(hid)  # type: ignore[union-attr]
+        if hid not in self._callbacks:
+            return self.emu.hook_del(hid)  # type: ignore[union-attr]
+
+        status = _uc.uc_hook_del(self.emu._uch, uc_hook_h(hid))  # type: ignore[union-attr]
+        if status != uc.UC_ERR_OK:
+            raise uc.UcError(status)
+        self._callbacks.pop(hid, None)
 
     def close(self):
         if self.emu is None:

speakeasy/gdb.py

@@ -77,7 +77,7 @@ class RegisterInfo:
 class GdbServer:
     """An all-stop RSP server backed directly by a WindowsEmulator."""
 
-    def __init__(self, emu: Any, port: int, host: str = "127.0.0.1"):
+    def __init__(self, emu: Any, port: int, host: str):
         self.emu = emu
         self.host = host
         self.port = port
@@ -97,8 +97,28 @@ class GdbServer:
         self._exec_breakpoints: dict[tuple[int, int], int] = {}
         self._read_watchpoints: dict[tuple[int, int], int] = {}
         self._write_watchpoints: dict[tuple[int, int], int] = {}
+        self._access_watchpoints: dict[tuple[int, int], int] = {}
         self._registers = self._make_registers()
         self._hooks: list[Any] = []
+        self._code_hook: Any | None = None
+        self._read_hook: Any | None = None
+        self._write_hook: Any | None = None
+
+    def __enter__(self) -> GdbServer:
+        try:
+            self.start()
+        except BaseException:
+            self.close()
+            raise
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: Any,
+    ) -> None:
+        self.close()
 
     # ------------------------------------------------------------------
     # Transport
@@ -126,6 +146,13 @@ class GdbServer:
             except OSError:
                 pass
 
+    def notify_signal(self, signal: int) -> None:
+        if not self._closed.is_set():
+            try:
+                self._reply(f"X{signal & 0xFF:02x}".encode("ascii"))
+            except OSError:
+                pass
+
     def close(self) -> None:
         self._closed.set()
         for sock in (self._client, self._listener):
@@ -140,9 +167,14 @@ class GdbServer:
                     pass
         for hook in self._hooks:
             try:
-                hook.disable()
+                self._remove_hook(hook)
             except Exception:
                 pass
+        # Prevent BinaryEmulator.set_hooks() from restoring debugger hooks if
+        # emulation is started again after this session closes.
+        for registered_hooks in getattr(self.emu, "hooks", {}).values():
+            if isinstance(registered_hooks, list):
+                registered_hooks[:] = [hook for hook in registered_hooks if hook not in self._hooks]
 
     @staticmethod
     def _extract_packet(buffer: bytearray) -> tuple[bytes, bytes, bool] | None:
@@ -233,17 +265,25 @@ class GdbServer:
                 return ResumeAction(detach=True)
             action = self._handle_command(payload)
             if action is not None:
+                if not action.detach and not action.kill:
+                    # Mark the resume transition so Ctrl-C arriving before
+                    # begin_run() becomes a pending interrupt, not a queued '?'.
+                    with self._state_lock:
+                        self._running = True
                 return action
         return ResumeAction(detach=True)
 
-    def begin_run(self, action: ResumeAction) -> None:
+    def begin_run(self, action: ResumeAction) -> bool:
         with self._state_lock:
             self._running = True
-            self._stop_pending = False
+            if self._stop_pending:
+                return False
             self._stop_reason = StopReason()
             pc = self.emu.get_pc()
             at_breakpoint = any(begin <= pc < begin + size for begin, size in self._exec_breakpoints)
             self._resume_from_breakpoint = pc if at_breakpoint else None
+        self._refresh_hooks()
+        return True
 
     def finish_run(self, action: ResumeAction) -> StopReason | None:
         with self._state_lock:
@@ -272,13 +312,38 @@ class GdbServer:
             self._commands.put(b"?")
 
     def _install_hooks(self) -> None:
-        self._hooks.extend(
-            (
-                self.emu.add_code_hook(self._on_code),
-                self.emu.add_mem_read_hook(self._on_read),
-                self.emu.add_mem_write_hook(self._on_write),
-            )
+        self._code_hook = self.emu.add_code_hook(self._on_code)
+        self._read_hook = self.emu.add_mem_read_hook(self._on_read)
+        self._write_hook = self.emu.add_mem_write_hook(self._on_write)
+        self._hooks.extend((self._code_hook, self._read_hook, self._write_hook))
+        self._refresh_hooks()
+
+    @staticmethod
+    def _remove_hook(hook: Any) -> None:
+        if hook.added and hook.native_hook:
+            hook.emu_eng.hook_remove(hook.handle)
+        hook.handle = 0
+        hook.added = False
+        hook.enabled = False
+
+    @classmethod
+    def _set_hook_enabled(cls, hook: Any | None, enabled: bool) -> None:
+        if hook is None:
+            return
+        if enabled and not hook.added:
+            hook.add()
+        elif not enabled and hook.added:
+            # Hook.disable() still enters Python for every native event. Remove
+            # the Unicorn hook entirely to avoid that callback overhead.
+            cls._remove_hook(hook)
+
+    def _refresh_hooks(self) -> None:
+        self._set_hook_enabled(
+            self._code_hook,
+            bool(self._exec_breakpoints) or self._resume_from_breakpoint is not None,
         )
+        self._set_hook_enabled(self._read_hook, bool(self._read_watchpoints or self._access_watchpoints))
+        self._set_hook_enabled(self._write_hook, bool(self._write_watchpoints or self._access_watchpoints))
 
     def _on_code(self, _emu: Any, address: int, _size: int) -> bool:
         if self._resume_from_breakpoint == address:
@@ -292,12 +357,16 @@ class GdbServer:
         return True
 
     def _on_read(self, _emu: Any, _access: int, address: int, size: int, _value: int) -> bool:
-        if self._overlaps_watchpoint(self._read_watchpoints, address, size):
+        if self._overlaps_watchpoint(self._access_watchpoints, address, size):
+            self._request_stop(StopReason(kind="awatch", address=address))
+        elif self._overlaps_watchpoint(self._read_watchpoints, address, size):
             self._request_stop(StopReason(kind="rwatch", address=address))
         return True
 
     def _on_write(self, _emu: Any, _access: int, address: int, size: int, _value: int) -> bool:
-        if self._overlaps_watchpoint(self._write_watchpoints, address, size):
+        if self._overlaps_watchpoint(self._access_watchpoints, address, size):
+            self._request_stop(StopReason(kind="awatch", address=address))
+        elif self._overlaps_watchpoint(self._write_watchpoints, address, size):
             self._request_stop(StopReason(kind="watch", address=address))
         return True
 
@@ -411,12 +480,7 @@ class GdbServer:
             elif bp_type == 3:
                 target = self._read_watchpoints
             elif bp_type == 4:
-                target = self._read_watchpoints
-                other = self._write_watchpoints
-                if set_breakpoint:
-                    other[key] = bp_type
-                else:
-                    other.pop(key, None)
+                target = self._access_watchpoints
             else:
                 self._reply(b"")
                 return
@@ -424,6 +488,7 @@ class GdbServer:
                 target[key] = bp_type
             else:
                 target.pop(key, None)
+            self._refresh_hooks()
             self._reply(b"OK")
         except ValueError:
             self._reply(b"E01")
@@ -470,7 +535,7 @@ class GdbServer:
                 ("gs", ux.UC_X86_REG_GS),
             )
         )
-        definitions.extend((f"st{i}", getattr(ux, f"UC_X86_REG_ST{i}"), 10, "float") for i in range(8))
+        definitions.extend((f"st{i}", getattr(ux, f"UC_X86_REG_FP{i}"), 10, "float") for i in range(8))
         # Unicorn does not expose all x87 environment fields independently.
         # Keep their standard RSP slots so clients retain the canonical layout.
         definitions.extend(
@@ -489,8 +554,11 @@ class GdbServer:
         register = self._registers[index]
         if register.unicorn_id is None:
             return bytes(register.size)
-        value = int(self._uc.reg_read(register.unicorn_id))
-        return value.to_bytes(register.size, "little")
+        value = self._uc.reg_read(register.unicorn_id)
+        if register.name.startswith("st"):
+            mantissa, exponent = value
+            return int(mantissa).to_bytes(8, "little") + int(exponent).to_bytes(2, "little")
+        return int(value).to_bytes(register.size, "little")
 
     def _read_registers(self) -> None:
         try:
@@ -504,11 +572,10 @@ class GdbServer:
             data = bytes.fromhex(encoded.decode("ascii"))
             offset = 0
             for register in self._registers:
-                value = int.from_bytes(data[offset : offset + register.size], "little")
-                if len(data[offset : offset + register.size]) != register.size:
+                register_data = data[offset : offset + register.size]
+                if len(register_data) != register.size:
                     raise ValueError
-                if register.unicorn_id is not None:
-                    self._uc.reg_write(register.unicorn_id, value)
+                self._write_register_value(register, register_data)
                 offset += register.size
             if offset != len(data):
                 raise ValueError
@@ -516,6 +583,16 @@ class GdbServer:
         except Exception:
             self._reply(b"E01")
 
+    def _write_register_value(self, register: RegisterInfo, value: bytes) -> None:
+        if register.unicorn_id is None:
+            return
+        if register.name.startswith("st"):
+            mantissa = int.from_bytes(value[:8], "little")
+            exponent = int.from_bytes(value[8:], "little")
+            self._uc.reg_write(register.unicorn_id, (mantissa, exponent))
+        else:
+            self._uc.reg_write(register.unicorn_id, int.from_bytes(value, "little"))
+
     def _read_register(self, encoded_index: bytes) -> None:
         try:
             index = int(encoded_index, 16)
@@ -531,8 +608,7 @@ class GdbServer:
             value = bytes.fromhex(value_text.decode("ascii"))
             if len(value) != register.size:
                 raise ValueError
-            if register.unicorn_id is not None:
-                self._uc.reg_write(register.unicorn_id, int.from_bytes(value, "little"))
+            self._write_register_value(register, value)
             self._reply(b"OK")
         except (ValueError, IndexError, UnicodeDecodeError):
             self._reply(b"E01")
@@ -576,8 +652,13 @@ class GdbServer:
     # XML and process metadata
 
     def _handle_xfer(self, payload: bytes) -> bytes | None:
+        # qXfer packets have the form:
+        # qXfer:<object>:<operation>:<annex>:<offset>,<length>
+        qxfer_prefix = b"qXfer:"
         try:
-            object_name, operation, annex, position = payload[6:].split(b":", 3)
+            if not payload.startswith(qxfer_prefix):
+                raise ValueError
+            object_name, operation, annex, position = payload.removeprefix(qxfer_prefix).split(b":", 3)
             if operation != b"read":
                 return None
             offset_text, length_text = position.split(b",", 1)

speakeasy/windows/kernel.py

@@ -27,7 +27,7 @@ class WinKernelEmulator(WindowsEmulator, IoManager):
     Class used to emulate Windows drivers
     """
 
-    def __init__(self, config, debug=False, exit_event=None, gdb_port=None, gdb_host="127.0.0.1"):
+    def __init__(self, config, debug=False, exit_event=None, gdb_port=None, gdb_host=None):
         super().__init__(config, debug=debug, exit_event=exit_event, gdb_port=gdb_port, gdb_host=gdb_host)
 
         self.disasm_eng: object | None = None

speakeasy/windows/win32.py

@@ -31,7 +31,7 @@ class Win32Emulator(WindowsEmulator):
     User Mode Windows Emulator Class
     """
 
-    def __init__(self, config, argv=None, debug=False, exit_event=None, gdb_port=None, gdb_host="127.0.0.1"):
+    def __init__(self, config, argv=None, debug=False, exit_event=None, gdb_port=None, gdb_host=None):
         super().__init__(config, debug=debug, exit_event=exit_event, gdb_port=gdb_port, gdb_host=gdb_host)
 
         self.last_error = 0

speakeasy/windows/winemu.py

@@ -17,6 +17,7 @@ import speakeasy.winenv.defs.nt.ddk as ddk
 import speakeasy.winenv.defs.windows.windows as windef
 from speakeasy.binemu import BinaryEmulator
 from speakeasy.errors import WindowsEmuError
+from speakeasy.gdb import GdbServer, ResumeAction, StopReason
 from speakeasy.profiler import MemAccess, Run
 from speakeasy.profiler_events import TracePosition
 from speakeasy.report import ErrorInfo, RegionInfo
@@ -81,12 +82,12 @@ class WindowsEmulator(BinaryEmulator):
         """Initialize configured processes. Subclasses must implement."""
         ...
 
-    def __init__(self, config, exit_event=None, debug=False, gdb_port=None, gdb_host="127.0.0.1"):
+    def __init__(self, config, exit_event=None, debug=False, gdb_port=None, gdb_host=None):
         super().__init__(config)
 
         self.debug: bool = debug
         self.gdb_port: int | None = gdb_port
-        self.gdb_host: str = gdb_host
+        self.gdb_host: str | None = gdb_host
         self.arch: int = 0
         self.modules: list[Any] = []
         self._setup_done: bool = False
@@ -546,8 +547,10 @@ class WindowsEmulator(BinaryEmulator):
         self.mem_write(base, bytes(data))
 
     def resume(self, addr, count=-1):
-        """
-        Resume emulation at the specified address.
+        """Resume emulation directly at an address.
+
+        This low-level API bypasses the GDB command loop; callers that enable
+        GDB should drive execution through :meth:`start` instead.
         """
         timeout = 0 if self.gdb_port is not None else self.config.timeout
         self.emu_eng.start(addr, timeout=timeout, count=count)  # type: ignore[union-attr]
@@ -569,23 +572,35 @@ class WindowsEmulator(BinaryEmulator):
         # so the first stop reports a meaningful PC/SP/etc.
         self._prepare_run_context(run)
 
-        debugger = None
-        debug_action = None
-        detached_resume_addr = None
+        completed = True
         if self.gdb_port is not None:
-            from speakeasy.gdb import GdbServer
+            if self.gdb_host is None:
+                raise WindowsEmuError("A GDB bind host is required when gdb_port is set")
+            with GdbServer(self, self.gdb_port, self.gdb_host) as debugger:
+                debug_action = debugger.command_loop()
+                if debug_action.kill:
+                    completed = True
+                elif debug_action.detach:
+                    debugger.close()
+                    completed = self._execute_runs()
+                else:
+                    completed = self._execute_runs(debugger, debug_action)
+        else:
+            completed = self._execute_runs()
 
-            debugger = GdbServer(self, self.gdb_port, self.gdb_host)
-            debugger.start()
-            debug_action = debugger.command_loop()
-            if debug_action.kill:
-                debugger.close()
-                self.on_emu_complete()
-                return
-            if debug_action.detach:
-                debugger.close()
-                debugger = None
+        if completed:
+            self.on_emu_complete()
 
+    def _execute_runs(
+        self,
+        debugger: GdbServer | None = None,
+        debug_action: ResumeAction | None = None,
+    ) -> bool:
+        """Execute prepared runs, optionally under control of an active GDB session."""
+        if debugger is not None:
+            assert debug_action is not None
+        detached_resume_addr = None
+        terminal_signal = 0
         timeout = 0 if debugger is not None else self.config.timeout
 
         if self.profiler:
@@ -600,17 +615,15 @@ class WindowsEmulator(BinaryEmulator):
                     resume_addr = detached_resume_addr or self.curr_run.start_addr  # type: ignore[union-attr]
                     detached_resume_addr = None
                 instruction_count = 1 if debugger is not None and debug_action.step else self.config.max_instructions
-                if debugger is not None:
-                    debugger.begin_run(debug_action)
-                self.emu_eng.start(resume_addr, timeout=timeout, count=instruction_count)  # type: ignore[union-attr]
+                should_execute = debugger is None or debugger.begin_run(debug_action)
+                if should_execute:
+                    self.emu_eng.start(resume_addr, timeout=timeout, count=instruction_count)  # type: ignore[union-attr]
                 if debugger is not None:
                     stop_reason = debugger.finish_run(debug_action)
                     if stop_reason is not None:
                         debug_action = debugger.command_loop(stop_reason)
                         if debug_action.kill:
-                            debugger.close()
-                            self.on_emu_complete()
-                            return
+                            return True
                         if debug_action.detach:
                             detached_resume_addr = self.get_pc()
                             debugger.close()
@@ -622,10 +635,12 @@ class WindowsEmulator(BinaryEmulator):
                         logger.error("* Timeout of %d sec(s) reached.", timeout)
             except KeyboardInterrupt:
                 logger.error("* User exited.")
-                return
+                if debugger is not None:
+                    debugger.notify_signal(2)
+                return False
             except Exception as e:
                 if self.exit_event and self.exit_event.is_set():
-                    return
+                    return False
                 stack_trace = traceback.format_exc()
 
                 try:
@@ -635,6 +650,21 @@ class WindowsEmulator(BinaryEmulator):
 
                 error = self.get_error_info(str(e), self.get_pc(), traceback=stack_trace)
                 self.curr_run.error = error  # type: ignore[union-attr]
+                terminal_signal = 11
+
+                if debugger is not None:
+                    # Ensure a pending Ctrl-C cannot be lost, then report the
+                    # target fault while its register state is still available.
+                    debugger.finish_run(debug_action)
+                    debug_action = debugger.command_loop(
+                        StopReason(signal=terminal_signal, kind="exception", address=self.get_pc())
+                    )
+                    if debug_action.kill:
+                        return True
+                    if debug_action.detach:
+                        debugger.close()
+                        debugger = None
+                        timeout = self.config.timeout
 
                 run = self.on_run_complete()
                 if not run:
@@ -646,9 +676,11 @@ class WindowsEmulator(BinaryEmulator):
             break
 
         if debugger is not None:
-            debugger.notify_exit(0)
-            debugger.close()
-        self.on_emu_complete()
+            if terminal_signal:
+                debugger.notify_signal(terminal_signal)
+            else:
+                debugger.notify_exit(0)
+        return True
 
     def get_current_run(self):
         """

tests/test_gdb.py

@@ -5,12 +5,40 @@ import subprocess
 import sys
 import textwrap
 import time
+from dataclasses import dataclass
 
 import pytest
 
 TESTS_DIR = os.path.dirname(__file__)
 
 
+@dataclass(frozen=True)
+class X86Registers:
+    eax: int
+    ecx: int
+    edx: int
+    ebx: int
+    esp: int
+    ebp: int
+    esi: int
+    edi: int
+    eip: int
+    eflags: int
+    cs: int
+    ss: int
+    ds: int
+    es: int
+    fs: int
+    gs: int
+
+    @classmethod
+    def from_rsp(cls, encoded: str):
+        # The i386 core feature begins with these sixteen 32-bit registers in
+        # the same order as the dataclass fields above.
+        core_layout = struct.Struct("<16I")
+        return cls(*core_layout.unpack(bytes.fromhex(encoded)[: core_layout.size]))
+
+
 class GdbRspClient:
     def __init__(self, port: int, timeout: float = 10.0):
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -68,6 +96,9 @@ class GdbRspClient:
     def read_registers(self) -> str:
         return self.send("g")
 
+    def read_x86_registers(self) -> X86Registers:
+        return X86Registers.from_rsp(self.read_registers())
+
     def read_memory(self, addr: int, size: int) -> str:
         return self.send(f"m{addr:x},{size:x}")
 
@@ -138,6 +169,29 @@ _PAUSE_SERVER_SCRIPT = textwrap.dedent("""\
 """)
 
 
+_FAULT_SERVER_SCRIPT = textwrap.dedent("""\
+    import json
+    import sys
+
+    from speakeasy import Speakeasy
+
+    port = int(sys.argv[1])
+    config_path = sys.argv[2]
+    with open(config_path) as f:
+        cfg = json.load(f)
+
+    se = Speakeasy(config=cfg, gdb_port=port)
+    address = se.load_shellcode(data=b"\\x90\\xc3", arch="x86")
+
+    def fail_start(*args, **kwargs):
+        raise RuntimeError("forced execution failure")
+
+    se.emu.emu_eng.start = fail_start
+    se.run_shellcode(address)
+    se.shutdown()
+""")
+
+
 _EXIT_SERVER_SCRIPT = textwrap.dedent("""\
     import json
     import sys
@@ -213,9 +267,7 @@ def gdb_emulator():
 
 @pytest.fixture
 def gdb_x64_emulator():
-    # Exercise the explicit non-loopback override while clients still connect
-    # through loopback in the isolated test process.
-    port, proc = _start_module_server("dll_test_x64.dll.xz", host="0.0.0.0")
+    port, proc = _start_module_server("dll_test_x64.dll.xz")
     yield port
     _stop_server(proc)
 
@@ -231,11 +283,7 @@ def test_gdb_connect_and_read_registers(gdb_emulator):
         assert len(regs) > 0
         assert all(char in "0123456789abcdefxx" for char in regs.lower())
 
-        # x86 GDB register order: eax(0), ecx(8), edx(16), ebx(24),
-        # esp(32), ebp(40), esi(48), edi(56), eip(64)
-        # Each register is 4 bytes = 8 hex chars.
-        eip_hex = regs[64:72].lower()
-        assert eip_hex not in ("00000000", "xxxxxxxx")
+        assert client.read_x86_registers().eip != 0
 
         client.continue_()
     finally:
@@ -248,11 +296,8 @@ def test_gdb_read_memory(gdb_emulator):
     try:
         client.query_halt_reason()
 
-        regs_hex = client.read_registers()
-        sp_hex = regs_hex[32:40]  # esp is at offset 32 (register #4)
-        sp = struct.unpack("<I", bytes.fromhex(sp_hex))[0]
-
-        mem = client.read_memory(sp, 4)
+        registers = client.read_x86_registers()
+        mem = client.read_memory(registers.esp, 4)
         assert len(mem) == 8
         assert all(char in "0123456789abcdef" for char in mem.lower())
 
@@ -305,6 +350,20 @@ def gdb_pause_emulator():
     _stop_server(proc)
 
 
+@pytest.fixture
+def gdb_fault_emulator():
+    port = _find_free_port()
+    config_path = os.path.join(TESTS_DIR, "test.json")
+    proc = subprocess.Popen(
+        [sys.executable, "-c", _FAULT_SERVER_SCRIPT, str(port), config_path],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    _wait_for_port(port, proc)
+    yield port, proc
+    _stop_server(proc)
+
+
 @pytest.fixture
 def gdb_exit_emulator():
     port = _find_free_port()
@@ -326,8 +385,7 @@ def test_gdb_breakpoint_stop_keeps_session_alive(gdb_emulator):
     try:
         client.query_halt_reason()
         client.step()
-        registers = client.read_registers()
-        pc = struct.unpack("<I", bytes.fromhex(registers[64:72]))[0]
+        pc = client.read_x86_registers().eip
         code = bytes.fromhex(client.read_memory(pc, 16))
         instruction = next(iter(capstone.Cs(capstone.CS_ARCH_X86, capstone.CS_MODE_32).disasm(code, pc)))
         breakpoint = pc + instruction.size
@@ -337,11 +395,9 @@ def test_gdb_breakpoint_stop_keeps_session_alive(gdb_emulator):
         assert stop.startswith("T05")
         assert "swbreak:;" in stop
 
-        # The old udbserver backend closed its socket immediately after the
-        # stop reply. A register query proves the native session remains live.
-        stopped_registers = client.read_registers()
-        stopped_pc = struct.unpack("<I", bytes.fromhex(stopped_registers[64:72]))[0]
-        assert stopped_pc == breakpoint
+        # A register query after the stop reply verifies that the native
+        # debugger session remains live.
+        assert client.read_x86_registers().eip == breakpoint
 
         assert client.query(f"z0,{breakpoint:x},1") == "OK"
         assert client.continue_().startswith(("T", "W"))
@@ -385,6 +441,19 @@ def test_gdb_detach(gdb_pause_emulator):
         client.close()
 
 
+def test_gdb_reports_target_fault_and_nonzero_termination(gdb_fault_emulator):
+    port, proc = gdb_fault_emulator
+    client = GdbRspClient(port)
+    try:
+        client.query_halt_reason()
+        assert client.continue_().startswith("T0b")
+        assert client.read_x86_registers().eip != 0
+        assert client.continue_() == "X0b"
+        assert proc.wait(timeout=10) == 0
+    finally:
+        client.close()
+
+
 def test_gdb_reports_clean_exit(gdb_exit_emulator):
     port, proc = gdb_exit_emulator
     client = GdbRspClient(port)
@@ -402,14 +471,12 @@ def test_gdb_single_step(gdb_emulator):
     try:
         client.query_halt_reason()
 
-        registers = client.read_registers()
-        previous_eip = struct.unpack("<I", bytes.fromhex(registers[64:72]))[0]
+        previous_eip = client.read_x86_registers().eip
 
         for _ in range(3):
             response = client.step()
             assert response.startswith(("S", "T"))
-            registers = client.read_registers()
-            eip = struct.unpack("<I", bytes.fromhex(registers[64:72]))[0]
+            eip = client.read_x86_registers().eip
             assert eip != previous_eip
             previous_eip = eip
 

tests/test_gdb_compat.py

@@ -1,7 +1,10 @@
 from dataclasses import dataclass
+from types import SimpleNamespace
 from xml.etree import ElementTree
 
-from speakeasy.gdb import GdbServer, _rsp_escape, _rsp_packet
+from unicorn import UC_ARCH_X86, UC_MODE_64, Uc
+
+from speakeasy.gdb import GdbServer, ResumeAction, _rsp_escape, _rsp_packet
 from speakeasy.winenv import arch
 
 
@@ -15,6 +18,20 @@ class DummyThread:
     teb: DummyTeb
 
 
+class DummyHook:
+    def __init__(self):
+        self.enabled = True
+        self.added = True
+        self.native_hook = True
+        self.handle = 1
+        self.emu_eng = SimpleNamespace(hook_remove=lambda handle: None)
+
+    def add(self):
+        self.enabled = True
+        self.added = True
+        self.handle = 1
+
+
 class DummyModule:
     def __init__(self, path, base):
         self.emu_path = path
@@ -32,12 +49,21 @@ class DummyEmulator:
     def get_arch(self):
         return arch.ARCH_AMD64
 
+    def add_code_hook(self, callback):
+        return DummyHook()
+
+    def add_mem_read_hook(self, callback):
+        return DummyHook()
+
+    def add_mem_write_hook(self, callback):
+        return DummyHook()
+
 
 def make_server():
-    return GdbServer(DummyEmulator(), 0)
+    return GdbServer(DummyEmulator(), 0, "127.0.0.1")
 
 
-def test_server_bind_host_defaults_to_loopback_and_can_be_overridden():
+def test_server_uses_provided_bind_host():
     assert make_server().host == "127.0.0.1"
     assert GdbServer(DummyEmulator(), 0, "0.0.0.0").host == "0.0.0.0"
 
@@ -112,6 +138,80 @@ def test_packet_extraction_validates_checksum_across_fragmented_input():
     assert GdbServer._extract_packet(bytearray(oversized))[2] is False
 
 
+def test_interrupt_during_resume_transition_is_preserved():
+    server = make_server()
+    server.emu.emu_eng = SimpleNamespace(stop=lambda: None)
+    server._running = True
+
+    server._interrupt()
+
+    assert not server.begin_run(ResumeAction())
+    reason = server.finish_run(ResumeAction())
+    assert reason is not None
+    assert reason.signal == 2
+
+
+def test_instruction_and_memory_hooks_are_enabled_only_when_needed():
+    server = make_server()
+    server._reply = lambda payload: None
+    server._install_hooks()
+
+    assert not server._code_hook.enabled
+    assert not server._read_hook.enabled
+    assert not server._write_hook.enabled
+
+    server._change_breakpoint(b"Z0,1000,1")
+    assert server._code_hook.enabled
+    server._change_breakpoint(b"z0,1000,1")
+    assert not server._code_hook.enabled
+
+    server._change_breakpoint(b"Z4,2000,4")
+    assert server._read_hook.enabled
+    assert server._write_hook.enabled
+    server._change_breakpoint(b"z4,2000,4")
+    assert not server._read_hook.enabled
+    assert not server._write_hook.enabled
+
+
+def test_access_watchpoint_does_not_clobber_write_watchpoint():
+    server = make_server()
+    replies = []
+    server._reply = replies.append
+
+    server._change_breakpoint(b"Z4,1000,4")
+    server._change_breakpoint(b"Z2,1000,4")
+    server._change_breakpoint(b"z4,1000,4")
+
+    assert (0x1000, 4) not in server._access_watchpoints
+    assert (0x1000, 4) in server._write_watchpoints
+    assert replies == [b"OK", b"OK", b"OK"]
+
+
+def test_access_watchpoint_reports_awatch_stop_reason():
+    server = make_server()
+    reasons = []
+    server._access_watchpoints[(0x1000, 4)] = 4
+    server._request_stop = reasons.append
+
+    server._on_read(None, 0, 0x1001, 1, 0)
+
+    assert len(reasons) == 1
+    assert reasons[0].kind == "awatch"
+    assert b"awatch:1001;" in server._stop_reply(reasons[0])
+
+
+def test_x87_registers_use_full_80_bit_wire_format():
+    emulator = DummyEmulator()
+    emulator.emu_eng = SimpleNamespace(emu=Uc(UC_ARCH_X86, UC_MODE_64))
+    server = GdbServer(emulator, 0, "127.0.0.1")
+    st0_index = next(index for index, register in enumerate(server._registers) if register.name == "st0")
+    wire_value = (3).to_bytes(8, "little") + (0x7FFF).to_bytes(2, "little")
+
+    server._write_register_value(server._registers[st0_index], wire_value)
+
+    assert server._read_register_value(st0_index) == wire_value
+
+
 def test_xfer_escaping_respects_advertised_packet_size():
     server = make_server()
     server._executable_path = lambda: "$#}*" * 5000