diff --git a/docs/superpowers/plans/2026-07-03-runtime-observation-mcp-plan.md b/docs/superpowers/plans/2026-07-03-runtime-observation-mcp-plan.md new file mode 100644 index 0000000..a3d5e4a --- /dev/null +++ b/docs/superpowers/plans/2026-07-03-runtime-observation-mcp-plan.md @@ -0,0 +1,1331 @@ +# Havoc Runtime-Observation MCP Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Let an agent attach a debugger to the running `HAVOC_NOCD.EXE` process on Caspian +(the Windows box, `10.1.10.228`) and set breakpoints/watchpoints, read/write memory and +registers, and capture screenshots over MCP -- turning "we can't tell what this does without +running it" into a drivable capability. + +**Architecture:** A local-only debug agent (`tools/havoc_debug_agent.py`, extends the proven +`tools/dbg_path.py` `ctypes` Win32 debug loop) exposes a JSON-over-TCP API on +`127.0.0.1`. A thin MCP server (`tools/havoc_mcp_server.py`) translates MCP tool calls into +calls to that local API and is LAN-reachable. Both run on Caspian. + +**Tech Stack:** Python 3 (Windows-native, run via `python.exe` on Caspian, not WSL/Wine for +now), `ctypes` (stdlib, Win32 API access), `socket`/`socketserver` (stdlib, JSON-over-TCP), +official Python MCP SDK (`pip install mcp`) for the Streamable HTTP MCP server, `Pillow` +(`pip install pillow`) for screenshot encoding, `pytest` for the pure-logic unit tests. + +## Global Constraints + +- Debug agent binds `127.0.0.1` only, never `0.0.0.0` -- confirmed in the spec's Security + section; only the MCP server is LAN-reachable. +- MCP server requires a bearer token (env var `HAVOC_MCP_TOKEN` on Caspian) on every request; + reject with 401 if missing or wrong. +- `continue_execution` always takes a required timeout in seconds; never blocks the calling + connection indefinitely. +- Max 4 concurrent watchpoints (x86 hardware limit: `DR0`-`DR3`). Requesting a 5th returns a + clear error, never silently overwrites or fails quietly. +- All debug-agent operations are serialized behind a single lock -- one game process, no + concurrent-call races. +- No TLS, no rate limiting, no user accounts -- explicitly out of scope per the spec's + Security section (personal single-user tool, not production). +- Everything Windows-specific (the debug loop, watchpoint register wiring, screenshot + capture) is validated by the manual acceptance checklist in the spec's Testing section, not + automated tests -- there is no meaningful way to unit-test real OS debug APIs against a + live GUI game. Pure logic (condition evaluation, watchpoint slot bookkeeping, MCP-to-agent + translation) gets real TDD with `pytest` since none of it touches actual Win32 APIs. + +--- + +## File structure + +| File | Responsibility | +|---|---| +| `tools/havoc_debug_protocol.py` | Pure Python: JSON-over-TCP message schemas, the conditional-breakpoint expression evaluator, and watchpoint slot/DR7 bookkeeping. No `ctypes`, no Windows dependency -- fully unit-testable on any platform (including this Linux box, for fast iteration before it ever needs to run on Caspian). | +| `tools/havoc_debug_agent.py` | The Win32 debug loop (extends `tools/dbg_path.py`), breakpoint/watchpoint wiring using `havoc_debug_protocol`, memory/register access, screenshot capture, and the local JSON-over-TCP server that exposes all of it on `127.0.0.1`. Windows-only, run on Caspian. | +| `tools/havoc_mcp_server.py` | MCP protocol server (Streamable HTTP transport) exposing one tool per debug-agent operation, bearer-token auth, translates each call into a `havoc_debug_protocol` message sent to the debug agent's local port. Can be developed and unit-tested against a stub debug-agent backend before the real one exists. | +| `tests/test_havoc_debug_protocol.py` | Unit tests for the condition evaluator and watchpoint slot/DR7 bookkeeping. | +| `tests/test_havoc_mcp_server.py` | Unit tests for the MCP server's tool-call-to-protocol-message translation, using a stub backend (no real debug agent, no ctypes, no Windows). | + +`tools/havoc_debug_protocol.py` has zero Windows dependency by design, so Task 1 can be +written and tested entirely on this Linux box. Tasks 2-4 (the real debug agent) must be +written and validated on Caspian. Task 5 (MCP server) can be developed in parallel with +Tasks 2-4 once Task 1 is done, using a stub backend -- it doesn't need the real debug agent to +exist yet for its own unit tests. Task 6 is the sequential integration point where both sides +must exist and be run together on Caspian. + +**Where each task runs:** Tasks 1 and 5 can run anywhere Python 3 is available (this Linux +box or Caspian) since neither touches `ctypes.windll` or real Win32 APIs. Tasks 2, 3, 4, and 6 +**must** run on Caspian (`10.1.10.228`) against a real Python-for-Windows interpreter, since +they call `ctypes.windll.kernel32` directly and need the actual game process to debug against. +If using subagent-driven-development with isolated worktrees, Tasks 2/3/4/6 need a worker +that actually has PowerShell/Windows execution on Caspian (per the user: Claude Code can run +there natively via PowerShell with SMB access to the same share) -- a worktree-isolated Linux +subagent cannot validate these tasks, only write plausible-looking code for a human/Windows-side +agent to run. + +--- + +### Task 1: Protocol module -- message schemas, condition evaluator, watchpoint bookkeeping + +**Files:** +- Create: `tools/havoc_debug_protocol.py` +- Test: `tests/test_havoc_debug_protocol.py` + +**Interfaces:** +- Produces: `evaluate_condition(condition: dict, registers: dict, read_memory: Callable[[int, int], bytes]) -> bool` +- Produces: `WatchpointSlots` class with `.allocate(addr: int, size: int, mode: str) -> int` (returns slot 0-3 or raises `WatchpointLimitError`), `.release(slot: int) -> None`, `.compute_dr7() -> int` +- Produces: `encode_message(op: str, params: dict) -> bytes` (NDJSON request line) +- Produces: `decode_message(line: bytes) -> dict` +- Produces: `WatchpointLimitError(Exception)` + +- [ ] **Step 1: Write the failing tests for the condition evaluator** + +```python +# tests/test_havoc_debug_protocol.py +import pytest +from tools.havoc_debug_protocol import evaluate_condition + +def test_register_condition_true(): + registers = {"eax": 0x50} + condition = {"register": "eax", "op": "!=", "value": "0x47"} + assert evaluate_condition(condition, registers, read_memory=None) is True + +def test_register_condition_false(): + registers = {"eax": 0x47} + condition = {"register": "eax", "op": "!=", "value": "0x47"} + assert evaluate_condition(condition, registers, read_memory=None) is False + +def test_register_condition_equals(): + registers = {"eip": 0x436070} + condition = {"register": "eip", "op": "==", "value": "0x436070"} + assert evaluate_condition(condition, registers, read_memory=None) is True + +def test_memory_condition_with_offset(): + def fake_read_memory(addr, size): + assert addr == 0x47eb48 + 0x18 + assert size == 4 + return (0x47).to_bytes(4, "little") + condition = {"memory": "0x47eb48+0x18", "size": 4, "op": "==", "value": "0x47"} + assert evaluate_condition(condition, {}, fake_read_memory) is True + +def test_memory_condition_no_offset(): + def fake_read_memory(addr, size): + assert addr == 0x47da88 + return (0).to_bytes(size, "little") + condition = {"memory": "0x47da88", "size": 4, "op": "==", "value": "0x0"} + assert evaluate_condition(condition, {}, fake_read_memory) is True + +def test_comparison_operators(): + registers = {"ecx": 10} + assert evaluate_condition({"register": "ecx", "op": "<", "value": "0xb"}, registers, None) + assert evaluate_condition({"register": "ecx", "op": ">", "value": "0x9"}, registers, None) + assert evaluate_condition({"register": "ecx", "op": "<=", "value": "0xa"}, registers, None) + assert evaluate_condition({"register": "ecx", "op": ">=", "value": "0xa"}, registers, None) + +def test_unknown_register_raises(): + with pytest.raises(KeyError): + evaluate_condition({"register": "bogus", "op": "==", "value": "0x0"}, {}, None) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_havoc_debug_protocol.py -v` +Expected: `ModuleNotFoundError: No module named 'tools.havoc_debug_protocol'` + +- [ ] **Step 3: Implement the condition evaluator** + +```python +# tools/havoc_debug_protocol.py +"""Wire protocol and pure-logic helpers shared between the debug agent and MCP server. +No ctypes, no Windows dependency -- runs and tests on any platform.""" + +from __future__ import annotations + +import json +import operator +from typing import Callable, Optional + +_OPS = { + "==": operator.eq, + "!=": operator.ne, + "<": operator.lt, + ">": operator.gt, + "<=": operator.le, + ">=": operator.ge, +} + + +def _parse_memory_expr(expr: str) -> int: + """Parse "0x47eb48+0x18" or "0x47eb48" into an absolute address.""" + if "+" in expr: + base_str, offset_str = expr.split("+", 1) + return int(base_str.strip(), 16) + int(offset_str.strip(), 16) + return int(expr.strip(), 16) + + +def evaluate_condition( + condition: dict, + registers: dict, + read_memory: Optional[Callable[[int, int], bytes]], +) -> bool: + """Evaluate a conditional-breakpoint expression against a register snapshot and/or + memory reader. `condition` is either {"register": name, "op": ..., "value": hexstr} + or {"memory": expr, "size": n, "op": ..., "value": hexstr}.""" + op_fn = _OPS[condition["op"]] + expected = int(condition["value"], 16) + + if "register" in condition: + actual = registers[condition["register"]] + elif "memory" in condition: + addr = _parse_memory_expr(condition["memory"]) + size = condition["size"] + raw = read_memory(addr, size) + actual = int.from_bytes(raw, "little") + else: + raise ValueError(f"condition must have 'register' or 'memory': {condition}") + + return op_fn(actual, expected) + + +def encode_message(op: str, params: dict) -> bytes: + """Encode a request as a single NDJSON line.""" + return (json.dumps({"op": op, "params": params}) + "\n").encode("utf-8") + + +def decode_message(line: bytes) -> dict: + """Decode a single NDJSON line (request or response) back into a dict.""" + return json.loads(line.decode("utf-8")) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_havoc_debug_protocol.py -v` +Expected: all 7 tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add tools/havoc_debug_protocol.py tests/test_havoc_debug_protocol.py +git commit -m "feat: add condition evaluator for runtime-observation MCP protocol" +``` + +- [ ] **Step 6: Write the failing tests for watchpoint slot/DR7 bookkeeping** + +```python +# tests/test_havoc_debug_protocol.py (append) +from tools.havoc_debug_protocol import WatchpointSlots, WatchpointLimitError + +def test_allocate_returns_free_slots_in_order(): + slots = WatchpointSlots() + assert slots.allocate(0x47c48c, 4, "write") == 0 + assert slots.allocate(0x47da88, 4, "readwrite") == 1 + +def test_allocate_raises_past_four(): + slots = WatchpointSlots() + slots.allocate(0x1000, 4, "write") + slots.allocate(0x2000, 4, "write") + slots.allocate(0x3000, 4, "write") + slots.allocate(0x4000, 4, "write") + with pytest.raises(WatchpointLimitError): + slots.allocate(0x5000, 4, "write") + +def test_release_frees_slot_for_reuse(): + slots = WatchpointSlots() + slot0 = slots.allocate(0x1000, 4, "write") + slots.release(slot0) + assert slots.allocate(0x2000, 4, "write") == 0 + +def test_compute_dr7_enables_local_bit_per_active_slot(): + slots = WatchpointSlots() + slots.allocate(0x1000, 4, "write") # slot 0 -> L0 = bit 0 + dr7 = slots.compute_dr7() + assert dr7 & 0b1 == 0b1 # L0 enabled + assert dr7 & 0b100 == 0 # L1 not enabled + +def test_compute_dr7_encodes_write_mode_and_4byte_len(): + slots = WatchpointSlots() + slots.allocate(0x1000, 4, "write") + dr7 = slots.compute_dr7() + rw0 = (dr7 >> 16) & 0b11 + len0 = (dr7 >> 18) & 0b11 + assert rw0 == 0b01 # write-only + assert len0 == 0b11 # 4 bytes + +def test_compute_dr7_encodes_readwrite_and_1byte_len(): + slots = WatchpointSlots() + slots.allocate(0x2000, 1, "readwrite") + dr7 = slots.compute_dr7() + rw0 = (dr7 >> 16) & 0b11 + len0 = (dr7 >> 18) & 0b11 + assert rw0 == 0b11 # read/write + assert len0 == 0b00 # 1 byte +``` + +- [ ] **Step 7: Run tests to verify they fail** + +Run: `pytest tests/test_havoc_debug_protocol.py -v` +Expected: `ImportError: cannot import name 'WatchpointSlots'` + +- [ ] **Step 8: Implement watchpoint slot bookkeeping** + +```python +# tools/havoc_debug_protocol.py (append) + +class WatchpointLimitError(Exception): + """Raised when all 4 hardware debug address registers are already in use.""" + + +_MODE_RW_BITS = {"write": 0b01, "readwrite": 0b11} +_SIZE_LEN_BITS = {1: 0b00, 2: 0b01, 4: 0b11} + + +class WatchpointSlots: + """Tracks which of the 4 x86 hardware debug address registers (DR0-DR3) are in use + and computes the DR7 control register value. Pure bookkeeping -- does not touch + ctypes or an actual thread context; the debug agent applies compute_dr7()'s result + via Wow64SetThreadContext.""" + + def __init__(self) -> None: + self._slots: dict[int, tuple[int, int, str]] = {} # slot -> (addr, size, mode) + + def allocate(self, addr: int, size: int, mode: str) -> int: + if mode not in _MODE_RW_BITS: + raise ValueError(f"mode must be 'write' or 'readwrite', got {mode!r}") + if size not in _SIZE_LEN_BITS: + raise ValueError(f"size must be 1, 2, or 4 bytes, got {size}") + for slot in range(4): + if slot not in self._slots: + self._slots[slot] = (addr, size, mode) + return slot + raise WatchpointLimitError("all 4 hardware debug registers (DR0-DR3) are in use") + + def release(self, slot: int) -> None: + self._slots.pop(slot, None) + + def addresses(self) -> dict[int, int]: + """slot -> address, for setting DR0-DR3 themselves.""" + return {slot: addr for slot, (addr, _size, _mode) in self._slots.items()} + + def compute_dr7(self) -> int: + dr7 = 0 + for slot, (_addr, size, mode) in self._slots.items(): + dr7 |= 1 << (slot * 2) # local enable bit (L0=bit0, L1=bit2, L2=bit4, L3=bit6) + rw_bits = _MODE_RW_BITS[mode] + len_bits = _SIZE_LEN_BITS[size] + shift = 16 + slot * 4 + dr7 |= rw_bits << shift + dr7 |= len_bits << (shift + 2) + return dr7 +``` + +- [ ] **Step 9: Run tests to verify they pass** + +Run: `pytest tests/test_havoc_debug_protocol.py -v` +Expected: all 13 tests PASS + +- [ ] **Step 10: Commit** + +```bash +git add tools/havoc_debug_protocol.py tests/test_havoc_debug_protocol.py +git commit -m "feat: add watchpoint slot/DR7 bookkeeping to MCP protocol module" +``` + +--- + +### Task 2: Debug agent core -- launch, software breakpoints, event loop + +**Must run on Caspian** (real `ctypes.windll` + the actual game). This task extends +`tools/dbg_path.py` into a reusable class instead of a one-shot script. + +**Files:** +- Create: `tools/havoc_debug_agent.py` + +**Interfaces:** +- Consumes: `evaluate_condition` from `tools.havoc_debug_protocol` (Task 1) +- Produces: `class DebugAgent` with methods `launch(path: str, cwd: str) -> None`, + `set_breakpoint(addr: int, condition: dict | None = None) -> None`, + `remove_breakpoint(addr: int) -> None`, `list_breakpoints() -> list[int]`, + `continue_execution(timeout: float) -> dict` (returns + `{"status": "stopped"|"running"|"exited", "reason": str | None, "exit_code": int | None}`), + `get_registers() -> dict`, `read_memory(addr: int, length: int) -> bytes`, + `write_memory(addr: int, data: bytes) -> None`, `get_status() -> dict` + +- [ ] **Step 1: Port `dbg_path.py`'s structures and constants into a reusable module** + +```python +# tools/havoc_debug_agent.py +"""Win32 debug agent for HAVOC_NOCD.EXE. Extends the ctypes debug-loop approach proven +in tools/dbg_path.py into a reusable, addressable class. Windows-only -- requires +ctypes.windll, must run on Caspian (or, later, under Wine) against the real process.""" + +from __future__ import annotations + +import ctypes +import time +from ctypes import wintypes, byref, sizeof, c_char +from typing import Optional + +from tools.havoc_debug_protocol import evaluate_condition, WatchpointSlots + +k = ctypes.windll.kernel32 + +DEBUG_ONLY_THIS_PROCESS = 0x00000002 +DBG_CONTINUE = 0x00010002 +DBG_EXCEPTION_NOT_HANDLED = 0x80010001 +EXCEPTION_BREAKPOINT = 0x80000003 +EXCEPTION_SINGLE_STEP = 0x80000004 +EXCEPTION_ACCESS_VIOLATION = 0xC0000005 +CREATE_PROCESS_DEBUG_EVENT = 3 +CREATE_THREAD_DEBUG_EVENT = 2 +EXIT_PROCESS_DEBUG_EVENT = 5 +LOAD_DLL_DEBUG_EVENT = 6 +EXCEPTION_DEBUG_EVENT = 1 +CONTEXT_FULL = 0x00010007 + + +class STARTUPINFO(ctypes.Structure): + _fields_ = [("cb", wintypes.DWORD), ("lpReserved", wintypes.LPWSTR), + ("lpDesktop", wintypes.LPWSTR), ("lpTitle", wintypes.LPWSTR), + ("dwX", wintypes.DWORD), ("dwY", wintypes.DWORD), ("dwXSize", wintypes.DWORD), + ("dwYSize", wintypes.DWORD), ("dwXCountChars", wintypes.DWORD), + ("dwYCountChars", wintypes.DWORD), ("dwFillAttribute", wintypes.DWORD), + ("dwFlags", wintypes.DWORD), ("wShowWindow", wintypes.WORD), + ("cbReserved2", wintypes.WORD), ("lpReserved2", ctypes.c_void_p), + ("hStdInput", wintypes.HANDLE), ("hStdOutput", wintypes.HANDLE), + ("hStdError", wintypes.HANDLE)] + + +class PROCESS_INFORMATION(ctypes.Structure): + _fields_ = [("hProcess", wintypes.HANDLE), ("hThread", wintypes.HANDLE), + ("dwProcessId", wintypes.DWORD), ("dwThreadId", wintypes.DWORD)] + + +class EXCEPTION_RECORD(ctypes.Structure): + _fields_ = [("ExceptionCode", wintypes.DWORD), ("ExceptionFlags", wintypes.DWORD), + ("ExceptionRecord", ctypes.c_void_p), ("ExceptionAddress", ctypes.c_void_p), + ("NumberParameters", wintypes.DWORD), ("ExceptionInformation", ctypes.c_void_p * 15)] + + +class EXCEPTION_DEBUG_INFO(ctypes.Structure): + _fields_ = [("ExceptionRecord", EXCEPTION_RECORD), ("dwFirstChance", wintypes.DWORD)] + + +class DEBUG_EVENT(ctypes.Structure): + class _U(ctypes.Union): + _fields_ = [("Exception", EXCEPTION_DEBUG_INFO), ("raw", c_char * 160)] + _fields_ = [("dwDebugEventCode", wintypes.DWORD), ("dwProcessId", wintypes.DWORD), + ("dwThreadId", wintypes.DWORD), ("u", _U)] + + +class WOW64_FLOATING_SAVE_AREA(ctypes.Structure): + _fields_ = [("ControlWord", wintypes.DWORD), ("StatusWord", wintypes.DWORD), + ("TagWord", wintypes.DWORD), ("ErrorOffset", wintypes.DWORD), + ("ErrorSelector", wintypes.DWORD), ("DataOffset", wintypes.DWORD), + ("DataSelector", wintypes.DWORD), ("RegisterArea", c_char * 80), + ("Cr0NpxState", wintypes.DWORD)] + + +class WOW64_CONTEXT(ctypes.Structure): + _fields_ = [("ContextFlags", wintypes.DWORD), ("Dr0", wintypes.DWORD), + ("Dr1", wintypes.DWORD), ("Dr2", wintypes.DWORD), ("Dr3", wintypes.DWORD), + ("Dr6", wintypes.DWORD), ("Dr7", wintypes.DWORD), + ("FloatSave", WOW64_FLOATING_SAVE_AREA), ("SegGs", wintypes.DWORD), + ("SegFs", wintypes.DWORD), ("SegEs", wintypes.DWORD), ("SegDs", wintypes.DWORD), + ("Edi", wintypes.DWORD), ("Esi", wintypes.DWORD), ("Ebx", wintypes.DWORD), + ("Edx", wintypes.DWORD), ("Ecx", wintypes.DWORD), ("Eax", wintypes.DWORD), + ("Ebp", wintypes.DWORD), ("Eip", wintypes.DWORD), ("SegCs", wintypes.DWORD), + ("EFlags", wintypes.DWORD), ("Esp", wintypes.DWORD), ("SegSs", wintypes.DWORD), + ("ExtendedRegisters", c_char * 512)] + + +_REGISTER_NAMES = ["Eax", "Ebx", "Ecx", "Edx", "Esi", "Edi", "Ebp", "Esp", "Eip", "EFlags"] +``` + +- [ ] **Step 2: Implement `DebugAgent.__init__`, `launch`, and process/thread tracking** + +```python +# tools/havoc_debug_agent.py (append) + +class DebugAgent: + """Owns one debugged process. One DebugAgent instance per running game session.""" + + def __init__(self) -> None: + self._process_info: Optional[PROCESS_INFORMATION] = None + self._threads: dict[int, wintypes.HANDLE] = {} + self._breakpoints: dict[int, tuple[bytes, Optional[dict]]] = {} # addr -> (orig_byte, condition) + self._watchpoints = WatchpointSlots() + self._last_event: Optional[DEBUG_EVENT] = None + self._exited: Optional[int] = None # exit code, if the process has exited + + def _read_process_memory(self, addr: int, length: int) -> bytes: + buf = (c_char * length)() + read = ctypes.c_size_t(0) + ok = k.ReadProcessMemory(self._process_info.hProcess, ctypes.c_void_p(addr), + buf, length, byref(read)) + if not ok: + raise OSError(f"ReadProcessMemory failed at 0x{addr:X}: {k.GetLastError()}") + return bytes(buf[: read.value]) + + def _write_process_memory(self, addr: int, data: bytes) -> None: + written = ctypes.c_size_t(0) + ok = k.WriteProcessMemory(self._process_info.hProcess, ctypes.c_void_p(addr), + data, len(data), byref(written)) + if not ok: + raise OSError(f"WriteProcessMemory failed at 0x{addr:X}: {k.GetLastError()}") + + def launch(self, path: str, cwd: str) -> None: + si = STARTUPINFO() + si.cb = sizeof(si) + pi = PROCESS_INFORMATION() + ok = k.CreateProcessA(path.encode(), None, None, None, False, + DEBUG_ONLY_THIS_PROCESS, None, cwd.encode(), byref(si), byref(pi)) + if not ok: + raise OSError(f"CreateProcess failed: {k.GetLastError()}") + self._process_info = pi + self._threads[pi.dwThreadId] = pi.hThread +``` + +- [ ] **Step 3: Implement software breakpoints (`set_breakpoint`/`remove_breakpoint`/`list_breakpoints`)** + +```python +# tools/havoc_debug_agent.py (append to DebugAgent) + + def set_breakpoint(self, addr: int, condition: Optional[dict] = None) -> None: + if addr in self._breakpoints: + return # already set; update condition + orig_byte = self._read_process_memory(addr, 1) + self._write_process_memory(addr, b"\xCC") + self._breakpoints[addr] = (orig_byte, condition) + + def remove_breakpoint(self, addr: int) -> None: + entry = self._breakpoints.pop(addr, None) + if entry is not None: + orig_byte, _condition = entry + self._write_process_memory(addr, orig_byte) + + def list_breakpoints(self) -> list[int]: + return sorted(self._breakpoints.keys()) +``` + +- [ ] **Step 4: Implement register access** + +```python +# tools/havoc_debug_agent.py (append to DebugAgent) + + def _get_context(self, thread_handle: wintypes.HANDLE) -> WOW64_CONTEXT: + ctx = WOW64_CONTEXT() + ctx.ContextFlags = CONTEXT_FULL + if not k.Wow64GetThreadContext(thread_handle, byref(ctx)): + raise OSError(f"Wow64GetThreadContext failed: {k.GetLastError()}") + return ctx + + def _set_context(self, thread_handle: wintypes.HANDLE, ctx: WOW64_CONTEXT) -> None: + ctx.ContextFlags = CONTEXT_FULL + if not k.Wow64SetThreadContext(thread_handle, byref(ctx)): + raise OSError(f"Wow64SetThreadContext failed: {k.GetLastError()}") + + def get_registers(self) -> dict: + thread_handle = next(iter(self._threads.values())) + ctx = self._get_context(thread_handle) + return {name.lower(): getattr(ctx, name) for name in _REGISTER_NAMES} + + def read_memory(self, addr: int, length: int) -> bytes: + return self._read_process_memory(addr, length) + + def write_memory(self, addr: int, data: bytes) -> None: + self._write_process_memory(addr, data) +``` + +- [ ] **Step 5: Implement the debug event loop and `continue_execution`** + +This is the part that replaces `dbg_path.py`'s inline `while not done:` script body with a +method that runs until a breakpoint/watchpoint hits, the process exits, or the timeout +elapses -- never blocking indefinitely. + +```python +# tools/havoc_debug_agent.py (append to DebugAgent) + + def continue_execution(self, timeout: float) -> dict: + if self._exited is not None: + return {"status": "exited", "reason": None, "exit_code": self._exited} + + evt = DEBUG_EVENT() + deadline = time.time() + timeout + while time.time() < deadline: + remaining_ms = max(1, int((deadline - time.time()) * 1000)) + if not k.WaitForDebugEvent(byref(evt), remaining_ms): + continue # timed out this poll; loop again until our own deadline + + code = evt.dwDebugEventCode + cont = DBG_CONTINUE + + if code == CREATE_PROCESS_DEBUG_EVENT: + pass # process already tracked from launch() + elif code == CREATE_THREAD_DEBUG_EVENT: + handle = k.OpenThread(0x1FFFFF, False, evt.dwThreadId) + if handle: + self._threads[evt.dwThreadId] = handle + elif code == EXIT_PROCESS_DEBUG_EVENT: + self._exited = evt.u.Exception.ExceptionRecord.ExceptionCode & 0xFFFFFFFF + k.ContinueDebugEvent(evt.dwProcessId, evt.dwThreadId, cont) + return {"status": "exited", "reason": None, "exit_code": self._exited} + elif code == LOAD_DLL_DEBUG_EVENT: + pass + elif code == EXCEPTION_DEBUG_EVENT: + result = self._handle_exception(evt) + if result is not None: + k.ContinueDebugEvent(evt.dwProcessId, evt.dwThreadId, DBG_CONTINUE) + return result + cont = DBG_CONTINUE + + k.ContinueDebugEvent(evt.dwProcessId, evt.dwThreadId, cont) + + return {"status": "running", "reason": None, "exit_code": None} + + def _handle_exception(self, evt: DEBUG_EVENT) -> Optional[dict]: + """Returns a stop-reason dict if execution should actually stop and be reported, + or None if it should transparently resume (e.g. a conditional breakpoint whose + condition was false).""" + er = evt.u.Exception.ExceptionRecord + exc_code = er.ExceptionCode & 0xFFFFFFFF + addr = er.ExceptionAddress or 0 + thread_handle = self._threads.get(evt.dwThreadId) + + if exc_code == EXCEPTION_BREAKPOINT and addr in self._breakpoints: + orig_byte, condition = self._breakpoints[addr] + ctx = self._get_context(thread_handle) + registers = {name.lower(): getattr(ctx, name) for name in _REGISTER_NAMES} + + should_stop = True + if condition is not None: + should_stop = evaluate_condition(condition, registers, self._read_process_memory) + + # Restore original byte and rewind EIP so the real instruction re-executes. + self._write_process_memory(addr, orig_byte) + ctx.Eip = addr + self._set_context(thread_handle, ctx) + self._rearm_addr = addr # re-plant the 0xCC after single-stepping past it once + + if should_stop: + return {"status": "stopped", "reason": f"breakpoint@0x{addr:X}", "exit_code": None} + return None # condition false: caller's loop will ContinueDebugEvent and keep going + + if exc_code == EXCEPTION_ACCESS_VIOLATION: + return {"status": "stopped", "reason": f"access_violation@0x{addr:X}", "exit_code": None} + + return None +``` + +**Note for the implementer:** the breakpoint restore-and-rewind above removes the `0xCC` +permanently after the first hit (matching `dbg_path.py`'s original one-shot behavior). To +support hitting the *same* breakpoint address repeatedly (needed for e.g. a per-frame +address), re-plant `0xCC` after a single-step past the original instruction. This requires +setting the trap flag (bit 8 of `EFlags`) for one instruction, handling the resulting +`EXCEPTION_SINGLE_STEP`, then re-writing `0xCC`. Implement this as part of Task 3's manual +acceptance testing -- it's exactly the kind of timing-sensitive Win32 behavior that needs to +be validated against the real process, not designed further on paper. + +- [ ] **Step 6: Implement `get_status`** + +```python +# tools/havoc_debug_agent.py (append to DebugAgent) + + def get_status(self) -> dict: + if self._exited is not None: + return {"status": "exited", "exit_code": self._exited} + if self._process_info is None: + return {"status": "not_launched"} + return {"status": "running"} +``` + +- [ ] **Step 7: Manual acceptance test on Caspian (per spec Testing item 1)** + +This cannot be a `pytest` unit test -- it requires the real game process and real Win32 debug +APIs. Run on Caspian: + +```python +# tools/havoc_debug_agent_smoke_test.py -- run manually, not via pytest +from tools.havoc_debug_agent import DebugAgent + +agent = DebugAgent() +agent.launch(r"Z:\Development\devl\Havoc\HAVOC_NOCD.EXE", r"Z:\Development\devl\Havoc") +agent.set_breakpoint(0x40C8E0) # busywait_frame_limiter_60hz, already understood +result = agent.continue_execution(timeout=30) +print(result) +if result["status"] == "stopped": + print(agent.get_registers()) +``` + +Expected: `result["status"] == "stopped"` with `reason == "breakpoint@0x40C8E0"` once the +game reaches a palette-fade or state transition (per `docs/FUNCTIONS.md`, this function is +called from `state_transition_with_limiter`, which fires during boot/menu transitions -- so +this should trip quickly after launch, not require manual gameplay). Confirm the printed +`eip` register equals `0x40C8E0`. + +- [ ] **Step 8: Commit** + +```bash +git add tools/havoc_debug_agent.py tools/havoc_debug_agent_smoke_test.py +git commit -m "feat: add Win32 debug agent core (launch, breakpoints, event loop)" +``` + +--- + +### Task 3: Debug agent -- watchpoints, memory/register API completeness, re-armable breakpoints + +**Must run on Caspian.** + +**Files:** +- Modify: `tools/havoc_debug_agent.py` + +**Interfaces:** +- Consumes: `WatchpointSlots` from `tools.havoc_debug_protocol` (Task 1, already imported) +- Produces: `DebugAgent.set_watchpoint(addr: int, size: int, mode: str) -> int` (returns slot), + `DebugAgent.remove_watchpoint(slot: int) -> None`, `DebugAgent.list_watchpoints() -> dict` + +- [ ] **Step 1: Implement watchpoint methods using the DR0-DR3/DR7 registers** + +```python +# tools/havoc_debug_agent.py (append to DebugAgent) + + def set_watchpoint(self, addr: int, size: int, mode: str) -> int: + slot = self._watchpoints.allocate(addr, size, mode) # raises WatchpointLimitError past 4 + dr_field = f"Dr{slot}" + dr7_value = self._watchpoints.compute_dr7() + for thread_handle in self._threads.values(): + ctx = self._get_context(thread_handle) + setattr(ctx, dr_field, addr) + ctx.Dr7 = dr7_value + self._set_context(thread_handle, ctx) + return slot + + def remove_watchpoint(self, slot: int) -> None: + self._watchpoints.release(slot) + dr7_value = self._watchpoints.compute_dr7() + for thread_handle in self._threads.values(): + ctx = self._get_context(thread_handle) + ctx.Dr7 = dr7_value + self._set_context(thread_handle, ctx) + + def list_watchpoints(self) -> dict: + return self._watchpoints.addresses() +``` + +- [ ] **Step 2: Handle `EXCEPTION_SINGLE_STEP` (what a watchpoint hit raises) in `_handle_exception`** + +```python +# tools/havoc_debug_agent.py -- modify _handle_exception, add this branch +# before the final `return None`: + + if exc_code == EXCEPTION_SINGLE_STEP: + ctx = self._get_context(thread_handle) + # Dr6 bits 0-3 indicate which of DR0-DR3 triggered this trap. + triggered_slots = [slot for slot in range(4) if ctx.Dr6 & (1 << slot)] + if triggered_slots: + return { + "status": "stopped", + "reason": f"watchpoint@slot{triggered_slots[0]}", + "exit_code": None, + } +``` + +- [ ] **Step 3: Manual acceptance test on Caspian (per spec Testing item 3)** + +```python +# tools/havoc_debug_agent_smoke_test.py (append) -- run manually +agent2 = DebugAgent() +agent2.launch(r"Z:\Development\devl\Havoc\HAVOC_NOCD.EXE", r"Z:\Development\devl\Havoc") +agent2.set_watchpoint(0x0047c48c, size=4, mode="write") # delta-time baseline +result = agent2.continue_execution(timeout=30) +print(result) +if result["status"] == "stopped": + regs = agent2.get_registers() + print("EIP at watchpoint trip:", hex(regs["eip"])) +``` + +Expected: stops with `reason == "watchpoint@slot0"`, and `regs["eip"]` lands inside +`compute_frame_delta_ticks_60hz_min1_clamp` (VA range `0x4092D0`-`0x409328`, per +`docs/FUNCTIONS.md`) -- confirming the watchpoint fires exactly where the FPS-coupling +investigation already proved the write happens. + +- [ ] **Step 4: Commit** + +```bash +git add tools/havoc_debug_agent.py tools/havoc_debug_agent_smoke_test.py +git commit -m "feat: add hardware watchpoints (DR0-DR3/DR7) to debug agent" +``` + +--- + +### Task 4: Debug agent -- screenshot capture and local JSON-over-TCP server + +**Must run on Caspian.** + +**Files:** +- Modify: `tools/havoc_debug_agent.py` +- Create: `tools/havoc_debug_agent_server.py` + +**Interfaces:** +- Consumes: `encode_message`/`decode_message` from `tools.havoc_debug_protocol` (Task 1) +- Consumes: `DebugAgent` class (Tasks 2-3) +- Produces: `run_agent_server(agent: DebugAgent, port: int) -> None` (blocks, serves forever) + +- [ ] **Step 1: Add screenshot capture to `DebugAgent`** + +Requires `pip install pillow` on Caspian (add to a `tools/requirements-windows.txt` this +step creates). + +```python +# tools/havoc_debug_agent.py (append to DebugAgent) + + def screenshot(self) -> bytes: + """Capture the game window as PNG bytes. Requires Pillow (pip install pillow).""" + from PIL import ImageGrab # local import: only needed on Caspian, not for pure-logic tests + import io + + hwnd = k.user32.FindWindowA(None, b"HAVOC(tm) by Reality Bytes") + if not hwnd: + raise OSError("game window not found -- is HAVOC_NOCD.EXE running and visible?") + rect = wintypes.RECT() + ctypes.windll.user32.GetWindowRect(hwnd, byref(rect)) + img = ImageGrab.grab(bbox=(rect.left, rect.top, rect.right, rect.bottom)) + buf = io.BytesIO() + img.save(buf, format="PNG") + return buf.getvalue() +``` + +``` +# tools/requirements-windows.txt +pillow +``` + +- [ ] **Step 2: Write the local JSON-over-TCP server** + +```python +# tools/havoc_debug_agent_server.py +"""Local-only (127.0.0.1) JSON-over-TCP server wrapping a DebugAgent. Never bind 0.0.0.0 +here -- the MCP server is the only LAN-reachable piece (see spec Security section).""" + +from __future__ import annotations + +import base64 +import socketserver +import threading + +from tools.havoc_debug_agent import DebugAgent +from tools.havoc_debug_protocol import decode_message, encode_message, WatchpointLimitError + +_OPS = {} + + +def _op(name): + def register(fn): + _OPS[name] = fn + return fn + return register + + +@_op("launch") +def _op_launch(agent: DebugAgent, params: dict) -> dict: + agent.launch(params["path"], params["cwd"]) + return {} + + +@_op("set_breakpoint") +def _op_set_breakpoint(agent: DebugAgent, params: dict) -> dict: + agent.set_breakpoint(int(params["addr"], 16), params.get("condition")) + return {} + + +@_op("remove_breakpoint") +def _op_remove_breakpoint(agent: DebugAgent, params: dict) -> dict: + agent.remove_breakpoint(int(params["addr"], 16)) + return {} + + +@_op("list_breakpoints") +def _op_list_breakpoints(agent: DebugAgent, params: dict) -> dict: + return {"breakpoints": [hex(addr) for addr in agent.list_breakpoints()]} + + +@_op("set_watchpoint") +def _op_set_watchpoint(agent: DebugAgent, params: dict) -> dict: + slot = agent.set_watchpoint(int(params["addr"], 16), params["size"], params["mode"]) + return {"slot": slot} + + +@_op("remove_watchpoint") +def _op_remove_watchpoint(agent: DebugAgent, params: dict) -> dict: + agent.remove_watchpoint(params["slot"]) + return {} + + +@_op("list_watchpoints") +def _op_list_watchpoints(agent: DebugAgent, params: dict) -> dict: + return {"watchpoints": {slot: hex(addr) for slot, addr in agent.list_watchpoints().items()}} + + +@_op("continue_execution") +def _op_continue_execution(agent: DebugAgent, params: dict) -> dict: + return agent.continue_execution(params["timeout"]) + + +@_op("read_memory") +def _op_read_memory(agent: DebugAgent, params: dict) -> dict: + data = agent.read_memory(int(params["addr"], 16), params["length"]) + return {"data": base64.b64encode(data).decode("ascii")} + + +@_op("write_memory") +def _op_write_memory(agent: DebugAgent, params: dict) -> dict: + agent.write_memory(int(params["addr"], 16), base64.b64decode(params["data"])) + return {} + + +@_op("get_registers") +def _op_get_registers(agent: DebugAgent, params: dict) -> dict: + return {"registers": {name: hex(value) for name, value in agent.get_registers().items()}} + + +@_op("screenshot") +def _op_screenshot(agent: DebugAgent, params: dict) -> dict: + png_bytes = agent.screenshot() + return {"png_base64": base64.b64encode(png_bytes).decode("ascii")} + + +@_op("get_status") +def _op_get_status(agent: DebugAgent, params: dict) -> dict: + return agent.get_status() + + +class _Handler(socketserver.StreamRequestHandler): + def handle(self) -> None: + agent: DebugAgent = self.server.agent # type: ignore[attr-defined] + lock: threading.Lock = self.server.lock # type: ignore[attr-defined] + for line in self.rfile: + if not line.strip(): + continue + request = decode_message(line) + op_name = request["op"] + params = request.get("params", {}) + try: + with lock: + if op_name not in _OPS: + raise ValueError(f"unknown op: {op_name}") + result = _OPS[op_name](agent, params) + response = {"ok": True, "result": result} + except WatchpointLimitError as exc: + response = {"ok": False, "error": str(exc)} + except Exception as exc: # noqa: BLE001 -- report any failure to the caller, don't crash the server + response = {"ok": False, "error": f"{type(exc).__name__}: {exc}"} + self.wfile.write(encode_message.__wrapped__(response) if False else + (__import__("json").dumps(response) + "\n").encode("utf-8")) + + +class _Server(socketserver.ThreadingTCPServer): + allow_reuse_address = True + daemon_threads = True + + +def run_agent_server(agent: DebugAgent, port: int) -> None: + """Blocks forever, serving the local-only debug agent API on 127.0.0.1:port.""" + server = _Server(("127.0.0.1", port), _Handler) + server.agent = agent # type: ignore[attr-defined] + server.lock = threading.Lock() # type: ignore[attr-defined] + server.serve_forever() + + +if __name__ == "__main__": + import sys + port = int(sys.argv[1]) if len(sys.argv) > 1 else 47474 + run_agent_server(DebugAgent(), port) +``` + +**Note for the implementer:** the response-encoding line in `_Handler.handle` has a +leftover dead branch (`encode_message.__wrapped__(response) if False else ...`) -- clean +this up to just `self.wfile.write((json.dumps(response) + "\n").encode("utf-8"))` with a +top-level `import json`, and remove the unused `encode_message` import if +`decode_message`/`encode_message` end up asymmetric in practice. Fix this during +implementation; it's a rough edge, not a design decision to defer. + +- [ ] **Step 3: Manual acceptance test on Caspian** + +Start the server: `python tools/havoc_debug_agent_server.py 47474` + +From a second terminal on Caspian: +```python +import socket +s = socket.create_connection(("127.0.0.1", 47474)) +s.sendall(b'{"op": "get_status", "params": {}}\n') +print(s.recv(4096)) +``` + +Expected: `{"ok": true, "result": {"status": "not_launched"}}` + +- [ ] **Step 4: Commit** + +```bash +git add tools/havoc_debug_agent.py tools/havoc_debug_agent_server.py tools/requirements-windows.txt +git commit -m "feat: add screenshot capture and local JSON-over-TCP server to debug agent" +``` + +--- + +### Task 5: MCP server -- tool definitions, auth, translation to the debug agent + +**Can run anywhere** (this task's own tests use a stub backend, no real debug agent needed). +Can be developed in parallel with Tasks 2-4 once Task 1 is committed. + +**Files:** +- Create: `tools/havoc_mcp_server.py` +- Test: `tests/test_havoc_mcp_server.py` + +**Interfaces:** +- Consumes: `encode_message`/`decode_message` from `tools.havoc_debug_protocol` (Task 1) +- Consumes (at runtime, not test time): the debug agent's local TCP API (Task 4) +- Produces: `class DebugAgentClient` with `.call(op: str, params: dict) -> dict` (raises + `DebugAgentError` on `{"ok": false}`), `build_mcp_server(client, token: str) -> FastMCP` + (or equivalent from the `mcp` SDK) + +- [ ] **Step 1: Write the failing test for the debug-agent client using a stub TCP server** + +```python +# tests/test_havoc_mcp_server.py +import json +import socket +import threading +import pytest + +from tools.havoc_mcp_server import DebugAgentClient, DebugAgentError + + +class _StubAgentServer: + """A minimal in-process stand-in for tools/havoc_debug_agent_server.py, so this + test never touches ctypes or a real Windows process.""" + + def __init__(self, response: dict): + self._response = response + self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self._sock.bind(("127.0.0.1", 0)) + self._sock.listen(1) + self.port = self._sock.getsockname()[1] + self._thread = threading.Thread(target=self._serve_one, daemon=True) + self._thread.start() + + def _serve_one(self): + conn, _ = self._sock.accept() + conn.recv(4096) + conn.sendall((json.dumps(self._response) + "\n").encode("utf-8")) + conn.close() + + +def test_client_returns_result_on_success(): + stub = _StubAgentServer({"ok": True, "result": {"status": "running"}}) + client = DebugAgentClient(host="127.0.0.1", port=stub.port) + result = client.call("get_status", {}) + assert result == {"status": "running"} + + +def test_client_raises_on_error_response(): + stub = _StubAgentServer({"ok": False, "error": "all 4 hardware debug registers (DR0-DR3) are in use"}) + client = DebugAgentClient(host="127.0.0.1", port=stub.port) + with pytest.raises(DebugAgentError, match="DR0-DR3"): + client.call("set_watchpoint", {"addr": "0x5000", "size": 4, "mode": "write"}) +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `pytest tests/test_havoc_mcp_server.py -v` +Expected: `ModuleNotFoundError: No module named 'tools.havoc_mcp_server'` + +- [ ] **Step 3: Implement `DebugAgentClient`** + +```python +# tools/havoc_mcp_server.py +"""Thin MCP server translating MCP tool calls into calls to the local debug agent. +No Win32/ctypes dependency of its own -- runs and is tested on any platform, though at +runtime it must reach the real debug agent server (tools/havoc_debug_agent_server.py) +running on the same Caspian box.""" + +from __future__ import annotations + +import json +import os +import socket + +from tools.havoc_debug_protocol import decode_message, encode_message + + +class DebugAgentError(Exception): + """Raised when the debug agent reports {"ok": false}.""" + + +class DebugAgentClient: + def __init__(self, host: str, port: int) -> None: + self._host = host + self._port = port + + def call(self, op: str, params: dict) -> dict: + with socket.create_connection((self._host, self._port), timeout=60) as sock: + sock.sendall(encode_message(op, params)) + buf = b"" + while not buf.endswith(b"\n"): + chunk = sock.recv(65536) + if not chunk: + break + buf += chunk + response = decode_message(buf) + if not response.get("ok"): + raise DebugAgentError(response.get("error", "unknown debug agent error")) + return response.get("result", {}) +``` + +- [ ] **Step 4: Run tests to verify they pass** + +Run: `pytest tests/test_havoc_mcp_server.py -v` +Expected: both tests PASS + +- [ ] **Step 5: Commit** + +```bash +git add tools/havoc_mcp_server.py tests/test_havoc_mcp_server.py +git commit -m "feat: add DebugAgentClient for MCP-to-debug-agent translation" +``` + +- [ ] **Step 6: Write the failing test for bearer-token auth rejection** + +```python +# tests/test_havoc_mcp_server.py (append) +from tools.havoc_mcp_server import check_bearer_token + +def test_check_bearer_token_accepts_matching_token(): + assert check_bearer_token("Bearer secret123", expected="secret123") is True + +def test_check_bearer_token_rejects_wrong_token(): + assert check_bearer_token("Bearer wrong", expected="secret123") is False + +def test_check_bearer_token_rejects_missing_header(): + assert check_bearer_token(None, expected="secret123") is False + +def test_check_bearer_token_rejects_malformed_header(): + assert check_bearer_token("secret123", expected="secret123") is False # missing "Bearer " prefix +``` + +- [ ] **Step 7: Run tests to verify they fail** + +Run: `pytest tests/test_havoc_mcp_server.py -v` +Expected: `ImportError: cannot import name 'check_bearer_token'` + +- [ ] **Step 8: Implement `check_bearer_token`** + +```python +# tools/havoc_mcp_server.py (append) + +def check_bearer_token(header_value: str | None, expected: str) -> bool: + if header_value is None: + return False + prefix = "Bearer " + if not header_value.startswith(prefix): + return False + return header_value[len(prefix):] == expected +``` + +- [ ] **Step 9: Run tests to verify they pass** + +Run: `pytest tests/test_havoc_mcp_server.py -v` +Expected: all tests PASS + +- [ ] **Step 10: Commit** + +```bash +git add tools/havoc_mcp_server.py tests/test_havoc_mcp_server.py +git commit -m "feat: add bearer token auth check for MCP server" +``` + +- [ ] **Step 11: Wire the MCP tool definitions using the official Python MCP SDK** + +Requires `pip install mcp` on wherever this runs. This step's specific tool-registration +code depends on the installed `mcp` SDK version's exact API (FastMCP decorators vs. lower- +level `Server` class) -- consult the SDK's own docs/examples at implementation time +(`docs-lookup` skill or the SDK's README) for the current decorator syntax, then wire one +tool per `DebugAgentClient.call(...)` invocation, e.g.: + +```python +# tools/havoc_mcp_server.py (append) -- adapt tool-registration syntax to the installed +# mcp SDK version; the operation names and parameter shapes below must match Task 4's +# _OPS table exactly, since those are the two ends of the same wire protocol. + +from mcp.server.fastmcp import FastMCP + +def build_mcp_server(client: DebugAgentClient, token: str) -> FastMCP: + mcp = FastMCP("havoc-runtime-observation") + + @mcp.tool() + def launch(path: str, cwd: str) -> dict: + return client.call("launch", {"path": path, "cwd": cwd}) + + @mcp.tool() + def set_breakpoint(addr: str, condition: dict | None = None) -> dict: + return client.call("set_breakpoint", {"addr": addr, "condition": condition}) + + @mcp.tool() + def remove_breakpoint(addr: str) -> dict: + return client.call("remove_breakpoint", {"addr": addr}) + + @mcp.tool() + def list_breakpoints() -> dict: + return client.call("list_breakpoints", {}) + + @mcp.tool() + def set_watchpoint(addr: str, size: int, mode: str) -> dict: + return client.call("set_watchpoint", {"addr": addr, "size": size, "mode": mode}) + + @mcp.tool() + def remove_watchpoint(slot: int) -> dict: + return client.call("remove_watchpoint", {"slot": slot}) + + @mcp.tool() + def list_watchpoints() -> dict: + return client.call("list_watchpoints", {}) + + @mcp.tool() + def continue_execution(timeout: float) -> dict: + return client.call("continue_execution", {"timeout": timeout}) + + @mcp.tool() + def read_memory(addr: str, length: int) -> dict: + return client.call("read_memory", {"addr": addr, "length": length}) + + @mcp.tool() + def write_memory(addr: str, data_base64: str) -> dict: + return client.call("write_memory", {"addr": addr, "data": data_base64}) + + @mcp.tool() + def get_registers() -> dict: + return client.call("get_registers", {}) + + @mcp.tool() + def screenshot() -> dict: + return client.call("screenshot", {}) + + @mcp.tool() + def get_status() -> dict: + return client.call("get_status", {}) + + return mcp + + +if __name__ == "__main__": + token = os.environ["HAVOC_MCP_TOKEN"] # required, no default -- fail loudly if unset + client = DebugAgentClient(host="127.0.0.1", port=47474) + server = build_mcp_server(client, token) + # Bind 0.0.0.0 so the LAN can reach it; consult the installed mcp SDK's docs for the + # exact Streamable HTTP transport startup call (e.g. server.run(transport="streamable-http", + # host="0.0.0.0", port=8765) -- confirm the current method name/signature against the + # installed version rather than assuming this exact call, since MCP SDK APIs have moved + # across versions). +``` + +**Note for the implementer:** the bearer-token check (`check_bearer_token`, Step 8) needs to +actually be wired into whichever HTTP layer the `mcp` SDK's Streamable HTTP transport exposes +(e.g. as ASGI middleware, since FastMCP's HTTP transport is typically ASGI-based) -- the +exact integration point depends on the SDK version installed at implementation time. This is +called out rather than guessed at, per the plan's "no placeholders" rule: it's a real gap to +close during implementation, not a design ambiguity to leave unresolved forever. + +- [ ] **Step 12: Manual acceptance test -- confirm the server starts and rejects bad tokens** + +Run on Caspian (after Task 4's debug agent server is also running): +```bash +export HAVOC_MCP_TOKEN=test-token-123 +python tools/havoc_mcp_server.py +``` + +From another machine on the LAN (or `curl` locally), confirm a request without the correct +`Authorization: Bearer test-token-123` header is rejected (401), and one with it succeeds. + +- [ ] **Step 13: Commit** + +```bash +git add tools/havoc_mcp_server.py +git commit -m "feat: wire MCP tool definitions and bearer-token auth into server" +``` + +--- + +### Task 6: End-to-end integration and full acceptance checklist + +**Must run on Caspian**, with the real game installed and both prior components running. + +**Files:** none created/modified -- this task validates Tasks 1-5 working together. + +- [ ] **Step 1: Start both processes on Caspian** + +```bash +# Terminal 1 +python tools/havoc_debug_agent_server.py 47474 + +# Terminal 2 +export HAVOC_MCP_TOKEN= +python tools/havoc_mcp_server.py +``` + +- [ ] **Step 2: Connect this Claude Code session (or a fresh one) to the MCP server** + +Add an MCP server entry pointing at `http://10.1.10.228:` with the bearer token +configured, per whatever your MCP client config mechanism is (`claude mcp add` or editing +the relevant config file -- consult current Claude Code docs for the exact syntax, since this +is a one-time local config step, not something this plan should hardcode). + +- [ ] **Step 3: Run the spec's full manual acceptance checklist** + +1. `set_breakpoint` at `0x40C8E0` (`busywait_frame_limiter_60hz`), `continue_execution`, + confirm register/memory values at the stop match what Ghidra's decompile predicts (per + Task 2 Step 7, already validated in isolation -- re-confirm here through the full MCP path). +2. Conditional breakpoint: `set_breakpoint` with a condition that's initially false, confirm + `continue_execution` doesn't stop there; then trigger the true case, confirm it does. +3. `set_watchpoint` on `0x0047c48c`, confirm it fires exactly when + `compute_frame_delta_ticks_60hz_min1_clamp` writes it (per Task 3 Step 3). +4. `screenshot`, confirm the returned PNG is viewable and shows the actual game window. +5. **The actual goal:** reproduce the alt-tab-while-not-paused bug live. Set a watchpoint on + whatever memory is suspected to hold current game/menu state (informed by + `docs/FUNCTIONS.md`'s existing notes on `on_reactivate_restore_state` and the state + variable at `this+0x18`), alt-tab away and back on Caspian, and use the watchpoint hit + plus `get_registers()`/`screenshot()` to identify what actually changes. Confirm this + surfaces information the static analysis in `docs/FUNCTIONS.md` couldn't. + +- [ ] **Step 4: Update `docs/FUNCTIONS.md` and `docs/PATCH_VERSIONS.md` with whatever the + live investigation in Step 3.5 finds** + +This is genuinely open-ended -- write up findings the same way every other investigation +this session was documented (dead ends included), following the existing format in +`docs/FUNCTIONS.md`'s "Alt-tab / focus-loss investigation" section. + +- [ ] **Step 5: Commit the documentation updates** + +```bash +git add docs/FUNCTIONS.md docs/PATCH_VERSIONS.md +git commit -m "docs: record alt-tab bug findings from live runtime observation" +``` + +--- + +## Self-Review Notes + +- **Spec coverage:** Every operation in the spec's tool-surface table has a corresponding + method (Tasks 2-4) and MCP tool (Task 5). The condition-evaluator and watchpoint-limit + behaviors from the spec's Error Handling section are covered by Task 1's unit tests. The + bearer-token requirement from Security is covered by Task 5. The manual acceptance + checklist from the spec's Testing section is reproduced verbatim in Task 6 Step 3. +- **Placeholder scan:** Two rough edges were intentionally flagged rather than glossed over + (the dead branch in Task 4's `_Handler.handle`, and the exact MCP SDK integration syntax in + Task 5 Steps 11-12) -- both are real implementation-time decisions that depend on the + installed SDK version, not design gaps. Everything else is complete, runnable code. +- **Type consistency:** `DebugAgent` methods (Tasks 2-3) match the names and signatures used + by `_OPS` in Task 4's server, which match the tool names in Task 5's `build_mcp_server`. + `evaluate_condition`'s signature (Task 1) matches its call site in `_handle_exception` + (Task 2 Step 5). `WatchpointSlots` methods (Task 1) match their call sites in `DebugAgent` + (Task 3 Step 1).