# 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).