Add find_definition for LLDB using clangd (#36)

src/chatdbg/chatdbg_lldb.py

@@ -12,6 +12,7 @@
 
 from assistant.lite_assistant import LiteAssistant
 import chatdbg_utils
+import clangd_lsp_integration
 
 
 # The file produced by the panic handler if the Rust program is using the chatdbg crate.
@@ -396,13 +397,6 @@ def _instructions():
 
 
 def _make_assistant(debugger: lldb.SBDebugger, args: argparse.Namespace):
-    assistant = LiteAssistant(
-        _instructions(),
-        model=args.llm,
-        timeout=args.timeout,
-        debug=args.debug,
-    )
-
     def lldb(command: str) -> str:
         """
         {
@@ -446,14 +440,68 @@ def get_code_surrounding(filename: str, lineno: int) -> str:
         (lines, first) = llm_utils.read_lines(filename, lineno - 7, lineno + 3)
         return llm_utils.number_group_of_lines(lines, first)
 
+    clangd = clangd_lsp_integration.clangd()
+
+    def find_definition(filename: str, lineno: int, character: int) -> str:
+        """
+        {
+            "name": "find_definition",
+            "description": "Returns the definition for the symbol at the given source location.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "filename": {
+                        "type": "string",
+                        "description": "The filename the code location is from."
+                    },
+                    "lineno": {
+                        "type": "integer",
+                        "description": "The line number where the symbol is present."
+                    },
+                    "character": {
+                        "type": "integer",
+                        "description": "The column number where the symbol is present."
+                    }
+                },
+                "required": [ "filename", "lineno", "character" ]
+            }
+        }
+        """
+        clangd.didOpen(filename, "c" if filename.endswith(".c") else "cpp")
+        definition = clangd.definition(filename, lineno, character)
+        clangd.didClose(filename)
+
+        if "result" not in definition or not definition["result"]:
+            return "No definition found."
+
+        path = clangd_lsp_integration._uri_to_path(definition["result"][0]["uri"])
+        start_lineno = definition["result"][0]["range"]["start"]["line"] + 1
+        end_lineno = definition["result"][0]["range"]["end"]["line"] + 1
+        (lines, first) = llm_utils.read_lines(path, start_lineno - 5, end_lineno + 5)
+        content = llm_utils.number_group_of_lines(lines, first)
+        line_string = (
+            f"line {start_lineno}"
+            if start_lineno == end_lineno
+            else f"lines {start_lineno}-{end_lineno}"
+        )
+        return f"""File '{path}' at {line_string}:\n```\n{content}\n```"""
+
+    assistant = LiteAssistant(
+        _instructions(),
+        model=args.llm,
+        timeout=args.timeout,
+        debug=args.debug,
+    )
+
     assistant.add_function(lldb)
     assistant.add_function(get_code_surrounding)
+    assistant.add_function(find_definition)
     return assistant
 
 
 def get_frame_summary() -> str:
     target = lldb.debugger.GetSelectedTarget()
-    if not target:
+    if not target or not target.process:
         return None
 
     for thread in target.process:
@@ -491,7 +539,7 @@ def get_frame_summary() -> str:
 
 def get_error_message() -> Optional[str]:
     target = lldb.debugger.GetSelectedTarget()
-    if not target:
+    if not target or not target.process:
         return None
 
     for thread in target.process:

src/chatdbg/clangd_lsp_integration.py

@@ -0,0 +1,134 @@
+import json
+import os
+import subprocess
+import urllib.parse
+
+
+def _to_lsp_request(id, method, params):
+    request = {"jsonrpc": "2.0", "id": id, "method": method}
+    if params:
+        request["params"] = params
+
+    content = json.dumps(request)
+    header = f"Content-Length: {len(content)}\r\n\r\n"
+    return header + content
+
+
+# Same as a request, but without an id.
+def _to_lsp_notification(method, params):
+    request = {"jsonrpc": "2.0", "method": method}
+    if params:
+        request["params"] = params
+
+    content = json.dumps(request)
+    header = f"Content-Length: {len(content)}\r\n\r\n"
+    return header + content
+
+
+def _parse_lsp_response(id, file):
+    # Ignore all messages until the response with the correct id is found.
+    while True:
+        header = {}
+        while True:
+            line = file.readline().strip()
+            if not line:
+                break
+            key, value = line.split(":", 1)
+            header[key.strip()] = value.strip()
+
+        content = file.read(int(header["Content-Length"]))
+        response = json.loads(content)
+        if "id" in response and response["id"] == id:
+            return response
+
+
+def _path_to_uri(path):
+    return "file://" + os.path.abspath(path)
+
+
+def _uri_to_path(uri):
+    data = urllib.parse.urlparse(uri)
+
+    assert data.scheme == "file"
+    assert not data.netloc
+    assert not data.params
+    assert not data.query
+    assert not data.fragment
+
+    path = data.path
+    if path.startswith(os.getcwd()):
+        path = os.path.relpath(path, os.getcwd())
+    return urllib.parse.unquote(path)  # clangd seems to escape paths.
+
+
+class clangd:
+    def __init__(
+        self,
+        executable="clangd",
+        working_directory=os.getcwd(),
+        stderr=subprocess.DEVNULL,
+    ):
+        self.id = 0
+        self.process = subprocess.Popen(
+            [executable],
+            text=True,
+            stdin=subprocess.PIPE,
+            stdout=subprocess.PIPE,
+            stderr=stderr,
+            cwd=working_directory,
+        )
+        self.initialize()
+
+    def __del__(self):
+        self.process.terminate()
+
+    def initialize(self):
+        self.id += 1
+        request = _to_lsp_request(self.id, "initialize", {"processId": os.getpid()})
+        self.process.stdin.write(request)
+        self.process.stdin.flush()
+        return _parse_lsp_response(self.id, self.process.stdout)
+        # TODO: Assert there is no error.
+
+    def didOpen(self, filename, languageId):
+        with open(filename, "r") as file:
+            text = file.read()
+
+        notification = _to_lsp_notification(
+            "textDocument/didOpen",
+            {
+                "textDocument": {
+                    "uri": _path_to_uri(filename),
+                    "languageId": languageId,
+                    "version": 1,
+                    "text": text,
+                }
+            },
+        )
+        self.process.stdin.write(notification)
+        self.process.stdin.flush()
+
+    def didClose(self, filename):
+        notification = _to_lsp_notification(
+            "textDocument/didClose", {"textDocument": {"uri": _path_to_uri(filename)}}
+        )
+        self.process.stdin.write(notification)
+        self.process.stdin.flush()
+
+    def definition(self, filename, line, character):
+        self.id += 1
+        request = _to_lsp_request(
+            self.id,
+            "textDocument/definition",
+            {
+                "textDocument": {"uri": _path_to_uri(filename)},
+                "position": {
+                    # Things are 0-indexed in LSP.
+                    "line": line - 1,
+                    "character": character - 1,
+                },
+            },
+        )
+        self.process.stdin.write(request)
+        self.process.stdin.flush()
+        return _parse_lsp_response(self.id, self.process.stdout)

test/test-definition-likely.cpp

@@ -0,0 +1,25 @@
+static int* p = nullptr;
+
+struct Bob
+{
+  int ****data;
+};
+struct Adam
+{
+  Bob *b1;
+  Bob *b2;
+};
+
+int main()
+{
+  int **p2 = &p;
+  int ***p3 = &p2;
+  int ****p4 = &p3;
+
+  Bob bob1 = {p4};
+  Bob bob2 = {p4};
+  Adam adam1 = {&bob1, &bob2};
+
+  int n = ****adam1.b1->data;
+  return 0;
+}