havoc-remaster/docs/superpowers/plans/2026-07-03-runtime-observation-mcp-plan.md

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

# 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

# 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

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

# 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

# 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

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

# 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

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

# 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

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

# 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

# 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:

# 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

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

# 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

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

# 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

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

# 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

# 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:

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

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

# 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

# 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

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

# 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

# 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

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.:

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

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

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

# Terminal 1
python tools/havoc_debug_agent_server.py 47474

# Terminal 2
export HAVOC_MCP_TOKEN=<a real secret, not the test one from Task 5>
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:<mcp_port> 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

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