feat(debug): add Python debug visualizer and integrate with existing tools

.vscode/launch.json

@@ -37,6 +37,15 @@
                 }
             ]
         },
+        {
+            "name": "Debug list_example (Python)",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "${workspaceFolder}/python/main.py",
+            "cwd": "${workspaceFolder}/python",
+            "console": "integratedTerminal",
+            "justMyCode": true
+        },
         {
             "name": "Debug list_example (C++)",
             "type": "cppdbg",

.vscode/settings.json

@@ -5,6 +5,9 @@
     "debugVisualizer.debugAdapterConfigurations": {
         "cppdbg": {
             "expressionTemplate": "$debug_graph(${expr})"
+        },
+        "debugpy": {
+            "expressionTemplate": "__import__('debug_graph').debug_graph(${expr})"
         }
     }
 }

python/debug_graph.py

@@ -0,0 +1,196 @@
+"""Generic "reference-graph" -> Debug Visualizer JSON walker, the debugpy
+analog of gdb/debug_graph.py.
+
+Works on any plain Python object reachable from an expression, purely by
+walking `vars(obj)` -- no changes to the debuggee's source are required.
+
+Rule: every attribute in `vars(obj)` becomes one row in that node's field
+table. An attribute whose value is itself a "describable" object (has its
+own `vars()`) becomes an edge (and is recursed into); everything else
+(None, numbers, strings, ...) is just stringified in place. A visited-id
+set makes this safe for cycles, so it handles lists, trees, and arbitrary
+(possibly cyclic) graphs uniformly.
+
+Meant to be called via a debugger's "evaluate" request (VS Code's Debug
+Visualizer expressionTemplate, or a Watch box) while stopped at a
+breakpoint, e.g.:
+
+    debug_graph(lst)
+    debug_graph(top, lst=lst)   # extra name=value watched roots
+
+Besides the primary root and any explicit keyword roots, every call also
+walks and labels, as additional roots:
+  - every other local variable visible in the caller's frame -- kind "local"
+  - every module-level global visible from the caller's frame, skipping
+    dunders, modules, classes and functions -- kind "global"
+This mirrors gdb/debug_graph.py's auto-discovery: it's what lets
+`debug_graph(top)` -- a bare int with no object graph of its own -- still
+show the whole list, because `lst` (a sibling local) gets auto-walked too.
+"""
+
+import json
+import sys
+import types
+
+
+def _addr_str(obj_id):
+    return hex(obj_id)
+
+
+def _is_describable(val):
+    if val is None or isinstance(val, (bool, int, float, complex, str, bytes, type)):
+        return False
+    if isinstance(val, (types.ModuleType, types.FunctionType, types.BuiltinFunctionType)):
+        return False
+    try:
+        vars(val)
+    except TypeError:
+        return False
+    return True
+
+
+def _format_scalar(val):
+    try:
+        return str(val)
+    except Exception as e:  # noqa: BLE001 - mirror gdb's catch-all here
+        return "<error: %s>" % e
+
+
+def _type_name(val):
+    return type(val).__name__
+
+
+def _append_field(fields_out, name, value, is_pointer=False, type_hint=None):
+    field = {"name": name, "value": value, "isPointer": is_pointer}
+    if type_hint:
+        field["typeHint"] = type_hint
+    fields_out.append(field)
+
+
+def _describe_struct(val, node_id, nodes, edges, visited):
+    fields_out = []
+    for name, member_val in vars(val).items():
+        if _is_describable(member_val):
+            member_id = id(member_val)
+            _append_field(fields_out, name, _addr_str(member_id), is_pointer=True, type_hint=_type_name(member_val))
+            edges.append((node_id, _addr_str(member_id), name))
+            if member_id not in visited:
+                visited.add(member_id)
+                _describe_struct(member_val, _addr_str(member_id), nodes, edges, visited)
+        else:
+            _append_field(fields_out, name, _format_scalar(member_val))
+    nodes[node_id] = fields_out
+
+
+def _walk_primary_root(root_val, nodes, edges, visited):
+    if _is_describable(root_val):
+        node_id = _addr_str(id(root_val))
+        visited.add(id(root_val))
+        _describe_struct(root_val, node_id, nodes, edges, visited)
+        return node_id
+    if root_val is not None:
+        nodes["root"] = [{"name": "value", "value": _format_scalar(root_val), "isPointer": False}]
+    return None
+
+
+def _add_named_root(name, val, kind, nodes, edges, visited, roots_out, skip_ids):
+    if _is_describable(val):
+        obj_id = id(val)
+        addr = _addr_str(obj_id)
+        if addr in skip_ids:
+            return
+        if obj_id not in visited:
+            visited.add(obj_id)
+            _describe_struct(val, addr, nodes, edges, visited)
+        roots_out.append((name, addr, kind, _type_name(val)))
+    else:
+        roots_out.append((name, _format_scalar(val), kind, _type_name(val) if val is not None else None))
+
+
+def _is_noise_value(val):
+    if isinstance(val, (types.ModuleType, types.FunctionType, types.BuiltinFunctionType, type)):
+        return True
+    # Anything whose *class* comes from the standard library (typing
+    # constructs, __future__'s _Feature, etc.) is machinery, not user data --
+    # but only applies to describable objects: "builtins" (int, str, ...)
+    # is itself in sys.stdlib_module_names, and plain scalars are never noise.
+    if _is_describable(val) and type(val).__module__ in sys.stdlib_module_names:
+        return True
+    return False
+
+
+def _skip_local(name, val):
+    return name.startswith("_") or _is_noise_value(val)
+
+
+def _skip_global(name, val):
+    return name.startswith("__") or _is_noise_value(val)
+
+
+def build_graph_json(root_val, watched_roots, frame):
+    nodes = {}
+    edges = []
+    visited = set()
+    roots = []
+    skip_ids = set()
+    used_names = set()
+
+    primary_id = _walk_primary_root(root_val, nodes, edges, visited)
+    if primary_id is not None:
+        skip_ids.add(primary_id)
+
+    for name, val in watched_roots.items():
+        used_names.add(name)
+        _add_named_root(name, val, "watched", nodes, edges, visited, roots, skip_ids)
+
+    if frame is not None:
+        # A debugger's "evaluate" often runs the expression with locals and
+        # globals merged into one dict (so bare names resolve either way),
+        # which collapses frame.f_locals is frame.f_globals as a way to
+        # separate them. Cross-referencing the real module namespace by
+        # identity recovers an accurate local/global split regardless.
+        module = sys.modules.get(frame.f_globals.get("__name__"))
+        module_dict = module.__dict__ if module is not None else {}
+
+        for name, val in frame.f_locals.items():
+            if name in used_names or name in module_dict and module_dict[name] is val:
+                continue  # a true module global that leaked into f_locals; handled below
+            if _skip_local(name, val):
+                continue
+            used_names.add(name)
+            _add_named_root(name, val, "local", nodes, edges, visited, roots, skip_ids)
+
+        for name, val in module_dict.items():
+            if name in used_names or _skip_global(name, val):
+                continue
+            used_names.add(name)
+            _add_named_root(name, val, "global", nodes, edges, visited, roots, skip_ids)
+
+    nodes_json = [{"id": nid, "fields": fields} for nid, fields in nodes.items()]
+    edges_json = [
+        {"from": src, "to": dst, "label": label}
+        for (src, dst, label) in edges
+        if dst in nodes
+    ]
+    roots_json = [
+        {"name": name, "value": value, "kind": kind, "type": type_name}
+        if type_name
+        else {"name": name, "value": value, "kind": kind}
+        for (name, value, kind, type_name) in roots
+    ]
+
+    return {
+        "kind": {"nodeTable": True},
+        "nodes": nodes_json,
+        "edges": edges_json,
+        "roots": roots_json,
+    }
+
+
+def debug_graph(root, **watched):
+    """$debug_graph(EXPR[, name=value]...) -- render EXPR's reference graph
+    as Debug Visualizer JSON. Called from a debugger's evaluate/Watch box
+    while stopped at a breakpoint; auto-discovers sibling locals/globals in
+    the paused frame (see module docstring)."""
+    frame = sys._getframe(1)
+    return json.dumps(build_graph_json(root, watched, frame))

python/main.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Generic, Optional, TypeVar
+
+T = TypeVar("T")
+
+
+# A single node containing data and a reference to the next node
+@dataclass
+class Node(Generic[T]):
+    data: T
+    next: Optional["Node[T]"]
+
+
+# The main wrapper for the linked list tracking the head node
+class LinkedList(Generic[T]):
+    def __init__(self) -> None:
+        self._head: Optional[Node[T]] = None
+
+    # Add a new element to the front of the list
+    def push_front(self, data: T) -> None:
+        self._head = Node(data, self._head)
+
+    # Remove and return the front element of the list
+    def pop_front(self) -> Optional[T]:
+        if self._head is None:
+            return None
+        data = self._head.data
+        self._head = self._head.next
+        return data
+
+    # Read the front element without removing it
+    def peek_front(self) -> Optional[T]:
+        return self._head.data if self._head else None
+
+
+def main() -> None:
+    lst: LinkedList[int] = LinkedList()
+    count = 0
+    message1 = "hello world"
+
+    # Demonstrate pushing items
+    lst.push_front(10)
+    count += 1
+    lst.push_front(20)
+    count += 1
+    lst.push_front(30)
+    count += 1
+
+    # Demonstrate peeking at the top item
+    top = lst.peek_front()
+    if top is not None:
+        print(f"Top element: {top}\ncount: {count}")  # Output: 30
+    print(message1)
+
+    # Demonstrate popping items
+    while (value := lst.pop_front()) is not None:
+        print(f"Popped: {value}")
+
+
+if __name__ == "__main__":
+    main()

scripts/dump_debug_graph.py

@@ -1,22 +1,23 @@
 #!/usr/bin/env python3
-"""Drives both the C++ and Rust debug builds under gdb to their equivalent
-breakpoint (right after the 30/20/10 list is built), evaluates
-$debug_graph(list) exactly as VS Code's "Debug Visualizer" extension would
-(via -data-evaluate-expression), and replays the extension's own
-parseEvaluationResultFromGenericDebugAdapter algorithm on the raw result --
-so we can tell, without going through the VS Code UI, whether our JSON will
-actually be recognized or whether the extension will fall back to its
-generic byte-walk visualization.
+"""Drives the C++ and Rust debug builds under gdb, and the Python build
+under debugpy, to their equivalent breakpoint (right after the 30/20/10
+list is built), evaluates the JSON-producing debug-graph helper exactly as
+VS Code's "Debug Visualizer" extension would (via -data-evaluate-expression
+for gdb, an `evaluate` DAP request for debugpy), and replays the
+extension's own parseEvaluationResultFromGenericDebugAdapter algorithm on
+the raw result -- so we can tell, without going through the VS Code UI,
+whether our JSON will actually be recognized or whether the extension will
+fall back to its generic byte-walk visualization.
 
 The parsing port below mirrors:
   external/vscode-debug-visualizer/extension/src/VisualizationBackend/
     parseEvaluationResultFromGenericDebugAdapter.ts
 Keep it in sync if that file changes upstream.
 
-Writes debug_cpp.json / debug_rust.json into the output directory (default
-~/lg, matching the style of the extension's own captured dumps): either the
-resolved data object on success, or a {"PARSE_ERROR": ..., "raw": ...}
-diagnostic on failure.
+Writes debug_cpp.json / debug_rust.json / debug_python.json into the
+output directory (default ~/lg, matching the style of the extension's own
+captured dumps): either the resolved data object on success, or a
+{"PARSE_ERROR": ..., "raw": ...} diagnostic on failure.
 
 For the C++ build specifically, the breakpoint sits inside
 `if (const int* top = list.peek_front())`, so `top` is a live local pointing
@@ -31,11 +32,22 @@ relies on gdb/debug_graph.py's auto-discovery of sibling locals (here,
 `list`, a struct type it *can* walk) merging into the same node graph as
 `top`'s address, rather than requiring `top` to be passed as an explicit
 watched root.
+
+The Python build has no pointers -- `top` there is just a plain int, not a
+value interior to any struct -- but python/debug_graph.py's own
+auto-discovery of sibling locals (here `lst`) means evaluating
+debug_graph(top) still delves into the whole list for the same underlying
+reason: verify_top_delves_into_list_py exercises that.
 """
+import ast
 import json
+import os
 import re
+import shutil
+import socket
 import subprocess
 import sys
+import time
 from pathlib import Path
 
 REPO = Path(__file__).resolve().parent.parent
@@ -94,6 +106,207 @@ def parse_like_extension(mi_escaped_value):
         return {"ok": False, "error": str(e), "raw": result_text}
 
 
+def find_debugpy_python():
+    """Locates a Python interpreter with `debugpy` importable: the debugpy
+    analog of picking `gdb`/`rust-gdb` off the PATH above. Checked in order:
+    an explicit override, the interpreter running this script, then a
+    couple of common places a dev might have installed it, since debugpy
+    isn't always on the system Python (this repo's system python3 is on a
+    version too new for the Debian-packaged debugpy, so it typically lives
+    in a venv)."""
+    candidates = []
+    override = os.environ.get("DUMP_DEBUG_GRAPH_PYTHON")
+    if override:
+        candidates.append(override)
+    candidates.append(sys.executable)
+    candidates.append(str(Path.home() / "py314" / "bin" / "python3"))
+    which = shutil.which("python3")
+    if which:
+        candidates.append(which)
+
+    for candidate in candidates:
+        if not candidate:
+            continue
+        try:
+            subprocess.run(
+                [candidate, "-c", "import debugpy"],
+                check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=10,
+            )
+        except Exception:
+            continue
+        return candidate
+    raise RuntimeError(
+        "no python interpreter with debugpy found; set DUMP_DEBUG_GRAPH_PYTHON=/path/to/python"
+    )
+
+
+class _DapClient:
+    """Minimal Debug Adapter Protocol client: just enough of the wire
+    protocol (Content-Length-framed JSON over a TCP socket) to drive
+    debugpy the same way -interpreter=mi drives gdb above -- no IDE, no
+    `debugpy` package needed on this side, since debugpy itself only needs
+    to be importable by the *debuggee* interpreter."""
+
+    def __init__(self, sock):
+        self._sock = sock
+        self._buf = b""
+        self._seq = 0
+
+    def send(self, command, arguments=None):
+        self._seq += 1
+        msg = {"seq": self._seq, "type": "request", "command": command}
+        if arguments is not None:
+            msg["arguments"] = arguments
+        body = json.dumps(msg).encode("utf-8")
+        self._sock.sendall(("Content-Length: %d\r\n\r\n" % len(body)).encode("ascii") + body)
+        return self._seq
+
+    def _read_message(self):
+        while b"\r\n\r\n" not in self._buf:
+            chunk = self._sock.recv(4096)
+            if not chunk:
+                raise EOFError("debugpy closed the connection")
+            self._buf += chunk
+        header, _, rest = self._buf.partition(b"\r\n\r\n")
+        length = int(header.split(b":")[1].strip())
+        while len(rest) < length:
+            chunk = self._sock.recv(4096)
+            if not chunk:
+                raise EOFError("debugpy closed the connection")
+            rest += chunk
+        body, self._buf = rest[:length], rest[length:]
+        return json.loads(body.decode("utf-8"))
+
+    def wait_for(self, predicate, timeout=15):
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            msg = self._read_message()
+            if predicate(msg):
+                return msg
+        raise TimeoutError("timed out waiting for a matching DAP message")
+
+
+def run_debugpy(python_bin, script_path, break_line, expr):
+    """Launches `script_path` under debugpy (paused via --wait-for-client),
+    connects a plain DAP client, sets one breakpoint, waits for it to hit,
+    and evaluates `expr` in that frame -- the debugpy equivalent of
+    run_gdb()'s -data-evaluate-expression. `context: "watch"` mirrors what
+    VS Code sends for a Watch-panel entry (rather than "repl"), matching
+    how the Debug Visualizer extension's expressionTemplate is actually
+    invoked. Returns the raw `result` string from the evaluate response
+    (debugpy's repr() of whatever the expression returned)."""
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as probe:
+        probe.bind(("127.0.0.1", 0))
+        port = probe.getsockname()[1]
+
+    proc = subprocess.Popen(
+        [python_bin, "-m", "debugpy", "--listen", "127.0.0.1:%d" % port, "--wait-for-client", str(script_path)],
+        cwd=str(Path(script_path).parent),
+        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+    try:
+        sock = None
+        for _ in range(50):
+            try:
+                sock = socket.create_connection(("127.0.0.1", port), timeout=1)
+                break
+            except OSError:
+                time.sleep(0.2)
+        if sock is None:
+            raise RuntimeError("debugpy never opened its listen socket")
+
+        with sock:
+            client = _DapClient(sock)
+            client.send("initialize", {
+                "clientID": "dump-debug-graph", "adapterID": "debugpy", "pathFormat": "path",
+                "linesStartAt1": True, "columnsStartAt1": True,
+            })
+            client.wait_for(lambda m: m.get("type") == "response" and m.get("command") == "initialize")
+
+            client.send("attach", {"justMyCode": False})
+            client.wait_for(lambda m: m.get("type") == "event" and m.get("event") == "initialized")
+
+            client.send("setBreakpoints", {
+                "source": {"path": str(script_path)},
+                "breakpoints": [{"line": break_line}],
+            })
+            client.wait_for(lambda m: m.get("type") == "response" and m.get("command") == "setBreakpoints")
+
+            client.send("configurationDone")
+            client.wait_for(lambda m: m.get("type") == "response" and m.get("command") == "configurationDone")
+            client.wait_for(lambda m: m.get("type") == "response" and m.get("command") == "attach")
+
+            stopped = client.wait_for(lambda m: m.get("type") == "event" and m.get("event") == "stopped")
+            thread_id = stopped["body"]["threadId"]
+
+            st_seq = client.send("stackTrace", {"threadId": thread_id})
+            frames = client.wait_for(lambda m: m.get("type") == "response" and m.get("request_seq") == st_seq)
+            frame_id = frames["body"]["stackFrames"][0]["id"]
+
+            ev_seq = client.send("evaluate", {"expression": expr, "frameId": frame_id, "context": "watch"})
+            evaluated = client.wait_for(lambda m: m.get("type") == "response" and m.get("request_seq") == ev_seq)
+            if not evaluated.get("success"):
+                raise RuntimeError("evaluate failed: %r" % (evaluated.get("message"),))
+            result = evaluated["body"]["result"]
+
+            client.send("continue", {"threadId": thread_id})
+        return result
+    finally:
+        try:
+            proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.wait(timeout=10)
+
+
+def parse_python_result(raw_result):
+    """debugpy's evaluate response reports a string return value as
+    Python's own repr() of it (e.g. `'{"kind": ...}'`), not raw text --
+    ast.literal_eval() is the exact inverse of that repr(), so it recovers
+    our debug_graph()-produced JSON string before the normal json.loads()."""
+    try:
+        json_text = ast.literal_eval(raw_result)
+        result_obj = json.loads(json_text)
+        return {"ok": True, "data": result_obj}
+    except Exception as e:
+        return {"ok": False, "error": str(e), "raw": raw_result}
+
+
+def verify_top_delves_into_list_py(data):
+    """Python analog of verify_top_delves_into_list: checks that evaluating
+    bare `top` (a plain int with no reference graph of its own) still shows
+    the whole 30/20/10 list because `lst`, a sibling local at the same
+    breakpoint, gets auto-discovered and walked too."""
+    nodes = data.get("nodes", [])
+    edges = data.get("edges", [])
+    roots_by_name = {r.get("name"): r for r in data.get("roots", [])}
+
+    node_shaped = [
+        n for n in nodes
+        if [f.get("name") for f in n.get("fields", [])] == ["data", "next"]
+    ]
+    if not node_shaped:
+        print("  VERIFY FAIL: debug_graph(top) produced no data+next Node-shaped node -- "
+              "nodes=%r" % [n.get("id") for n in nodes])
+        return False
+
+    next_edges = [e for e in edges if e.get("label") == "next"]
+    if len(next_edges) < 2:
+        print("  VERIFY FAIL: expected 2 'next' edges chaining the 30/20/10 nodes, found %d "
+              "(edges=%r)" % (len(next_edges), edges))
+        return False
+
+    if "lst" not in roots_by_name:
+        print("  VERIFY FAIL: 'lst' was not auto-discovered as a sibling local alongside "
+              "top (roots=%r)" % data.get("roots"))
+        return False
+
+    print("  VERIFY OK: debug_graph(top) delves into the full list -- %d nodes, %d 'next' "
+          "edges, 'lst' auto-discovered as a %r root"
+          % (len(nodes), len(next_edges), roots_by_name["lst"].get("kind")))
+    return True
+
+
 def verify_top_delves_into_list(data):
     """Checks that evaluating bare `top` -- i.e. $debug_graph(top), exactly
     what VS Code's expressionTemplate produces for a Watch box containing
@@ -134,6 +347,12 @@ def verify_top_delves_into_list(data):
     return True
 
 
+def _write_dump(label, out_dir, result):
+    out_path = out_dir / ("debug_%s.json" % label)
+    out_path.write_text(json.dumps(result, indent=4))
+    print("Wrote %s (%s)" % (out_path, "OK" if "PARSE_ERROR" not in result else "PARSE_ERROR"))
+
+
 def dump(label, gdb_bin, binary, break_file, break_line, out_dir, expr="$debug_graph(list)", verify=None):
     ok = True
     try:
@@ -156,9 +375,34 @@ def dump(label, gdb_bin, binary, break_file, break_line, out_dir, expr="$debug_g
             }
             ok = False
 
-    out_path = out_dir / ("debug_%s.json" % label)
-    out_path.write_text(json.dumps(result, indent=4))
-    print("Wrote %s (%s)" % (out_path, "OK" if "PARSE_ERROR" not in result else "PARSE_ERROR"))
+    _write_dump(label, out_dir, result)
+    return ok
+
+
+def dump_python(label, python_bin, script_path, break_line, out_dir,
+                expr="__import__('debug_graph').debug_graph(lst)", verify=None):
+    ok = True
+    try:
+        raw_result = run_debugpy(python_bin, script_path, break_line, expr)
+    except Exception as e:
+        result = {"PARSE_ERROR": "debugpy invocation failed", "detail": str(e)}
+        ok = False
+    else:
+        parsed = parse_python_result(raw_result)
+        if parsed["ok"]:
+            result = parsed["data"]
+            if verify is not None:
+                ok = verify(result)
+        else:
+            result = {
+                "PARSE_ERROR": parsed["error"],
+                "raw": parsed["raw"],
+                "note": "This is what the Debug Visualizer extension would fail to parse, "
+                "triggering its generic byte-walk fallback (constructGraphFromVariablesReference).",
+            }
+            ok = False
+
+    _write_dump(label, out_dir, result)
     return ok
 
 
@@ -174,7 +418,16 @@ def main():
     )
     ok_rust = dump("rust", "rust-gdb", str(REPO / "target" / "debug" / "list_example"), "src/main.rs", 51, out_dir)
 
-    if not (ok_cpp and ok_cpp_top and ok_rust):
+    python_bin = find_debugpy_python()
+    python_script = REPO / "python" / "main.py"
+    ok_python = dump_python("python", python_bin, python_script, 54, out_dir)
+    ok_python_top = dump_python(
+        "python_top", python_bin, python_script, 54, out_dir,
+        expr="__import__('debug_graph').debug_graph(top)",
+        verify=verify_top_delves_into_list_py,
+    )
+
+    if not (ok_cpp and ok_cpp_top and ok_rust and ok_python and ok_python_top):
         sys.exit(1)