vibefpmgui.debug_session

  1"""A debug session: a program run under the debugger (`vibefpmgui.dap`'s
  2client, with an adapter from `vibefpmgui.debug_adapters`), and what it's
  3doing: running, stopped (where, with its call stack and variables), or ended.
  4
  5The order of things, as adapters expect it: ``initialize``, then ``launch``
  6(gdb answers it only after the configuration is done), and on the
  7``initialized`` event the breakpoints, then ``configurationDone``. At each
  8stop: the threads, the stopped thread's call stack, then the scopes (Locals,
  9Arguments, module variables…) of the frame shown; variables are asked for
 10as they're expanded.
 11"""
 12
 13from __future__ import annotations
 14
 15import re
 16from collections.abc import Callable
 17from dataclasses import dataclass, field
 18from pathlib import Path
 19
 20from qtpy.QtCore import QObject, Signal
 21
 22from .dap import DapClient
 23from .debug_adapters import Adapter
 24
 25IDLE, STARTING, RUNNING, STOPPED = "idle", "starting", "running", "stopped"
 26MAX_ELEMENTS = 1000  # an array's elements asked for at once (a big one: the first ones, then "…")
 27
 28# gdb's own messages, sent as the program's output: breakpoints set before the
 29# program is loaded (they're placed when it is: see breakpoint events). With its
 30# styling on (a terminal, a UTF-8 locale), gdb puts a symbol in front: "❌️ No
 31# source file…".
 32_GDB_NOISE = re.compile(r"^[^\w\s]*\s*(No source file named .*\.|Breakpoint \d+ \(.*\) pending\.)\s*$")
 33
 34
 35@dataclass
 36class Frame:
 37    id: int
 38    name: str
 39    path: str | None  # (None: no source: a library's)
 40    line: int
 41
 42
 43@dataclass
 44class Scope:
 45    name: str
 46    reference: int
 47    expensive: bool = False
 48
 49
 50@dataclass
 51class Variable:
 52    name: str
 53    value: str
 54    reference: int = 0  # (non-zero: it has parts: an array's elements, a type's components)
 55    count: int = 0  # (its elements or components, if the adapter says)
 56    type: str = ""
 57
 58
 59@dataclass
 60class BreakpointState:
 61    verified: bool
 62    message: str = ""
 63
 64
 65@dataclass
 66class _Breakpoints:
 67    """The breakpoints sent for each file, and what the debugger said of them."""
 68    lines: dict[str, list[int]] = field(default_factory=dict)
 69    ids: dict[int, tuple[str, int]] = field(default_factory=dict)  # the debugger's id → (file, line)
 70    pending: set[tuple[str, int]] = field(default_factory=set)  # "pending" (gdb's: not placed yet)
 71
 72
 73class DebugSession(QObject):
 74    state_changed = Signal(str)
 75    stopped_at = Signal(object, str)  # [Frame] (the call stack, innermost first), why ("breakpoint", "step"…)
 76    scopes_ready = Signal(int, object)  # frame index, [Scope]
 77    output = Signal(str, str)  # category ("stdout", "stderr", "console"), text
 78    breakpoints_checked = Signal(str, object)  # file, {line: BreakpointState}
 79    ended = Signal(str)  # why ("exited with code 0", "stopped", the debugger's error)
 80
 81    def __init__(self, parent=None):
 82        super().__init__(parent)
 83        self.client = DapClient(self)
 84        self.client.event.connect(self._event)
 85        self.client.stopped.connect(self._adapter_stopped)
 86        self.adapter: Adapter | None = None
 87        self.state = IDLE
 88        self.frames: list[Frame] = []
 89        self.frame_index = 0
 90        self.thread_id: int | None = None
 91        self.exit_code: int | None = None
 92        self.hit_breakpoints: list[tuple[str, int]] = []  # (file, line as set) the last stop was at
 93        self._breakpoints = _Breakpoints()
 94        self._launch: dict = {}
 95        self._ending = False
 96
 97    # ---------------------------------------------------------- state
 98    def is_active(self) -> bool:
 99        return self.state != IDLE
100
101    def _set_state(self, state: str):
102        if state != self.state:
103            self.state = state
104            self.state_changed.emit(state)
105
106    def _end(self, why: str):
107        if self.state == IDLE:
108            return
109        self._ending = True
110        self.client.stop()
111        self._ending = False
112        self.frames, self.thread_id = [], None
113        self._set_state(IDLE)
114        self.ended.emit(why)
115
116    # ---------------------------------------------------------- starting
117    def start(self, adapter: Adapter, program: str, args: list[str], cwd: str,
118              breakpoints: dict[str, list[int]]) -> bool:
119        """Run ``program`` under the debugger, stopping at ``breakpoints``
120        (file → lines)."""
121        if self.is_active():
122            self.stop()
123        self.adapter = adapter
124        self.exit_code = None
125        self.frames, self.thread_id = [], None
126        self._breakpoints = _Breakpoints({path: sorted(set(lines)) for path, lines in breakpoints.items() if lines})
127        self._launch = adapter.launch_arguments(program, args, cwd)
128        self._set_state(STARTING)
129        self.client.start(adapter.command, cwd)
130        self.client.request("initialize", {
131            "clientID": "vibefpmgui", "clientName": "VibeFPMGUI", "adapterID": adapter.kind,
132            "linesStartAt1": True, "columnsStartAt1": True, "pathFormat": "path",
133            "supportsVariableType": True, "supportsRunInTerminalRequest": False,
134        }, self._initialized)
135        return True
136
137    def _initialized(self, success: bool, _body: dict, message: str):
138        if not success:
139            self._end(f"the debugger didn't start: {message}")
140            return
141        # (gdb answers launch only once the configuration is done: no timeout.)
142        self.client.request("launch", self._launch, self._launched, timeout=0)
143
144    def _launched(self, success: bool, _body: dict, message: str):
145        if not success and self.state != IDLE:
146            self._end(f"the program didn't start: {message or 'the debugger refused it'}")
147
148    def _configure(self):
149        """The adapter is ready for the breakpoints (the initialized event)."""
150        for path in list(self._breakpoints.lines):
151            self._send_breakpoints(path)
152        self.client.request("configurationDone", {}, self._configured)
153
154    def _configured(self, success: bool, _body: dict, message: str):
155        if not success:
156            self._end(f"the debugger couldn't start the program: {message}")
157        elif self.state == STARTING:
158            self._set_state(RUNNING)
159
160    # ---------------------------------------------------------- breakpoints
161    def set_breakpoints(self, path: str, lines: list[int]):
162        """A file's breakpoints now (sent to the debugger, if one is running)."""
163        self._breakpoints.lines[path] = sorted(set(lines))
164        if self.state in (RUNNING, STOPPED):
165            self._send_breakpoints(path)
166
167    def _send_breakpoints(self, path: str):
168        lines = self._breakpoints.lines.get(path, [])
169        self.client.request("setBreakpoints", {"source": {"path": path, "name": Path(path).name},
170                                               "breakpoints": [{"line": line} for line in lines]},
171                            lambda success, body, message, p=path, ls=lines: self._breakpoints_set(
172                                p, ls, success, body, message))
173
174    def _breakpoints_set(self, path: str, lines: list[int], success: bool, body: dict, message: str):
175        states: dict[int, BreakpointState] = {}
176        found = body.get("breakpoints") or [] if success else []
177        self._breakpoints.pending -= {(path, n) for n in self._breakpoints.lines.get(path, [])}
178        for line, answer in zip(lines, found, strict=False):
179            if isinstance(answer.get("id"), int):
180                self._breakpoints.ids[answer["id"]] = (path, line)
181            if not answer.get("verified") and answer.get("reason") == "pending":
182                self._breakpoints.pending.add((path, line))
183            states[line] = BreakpointState(_placed(answer), _why_not(answer))
184        for line in lines[len(found):]:
185            states[line] = BreakpointState(False, message or "the debugger didn't place it")
186        self.breakpoints_checked.emit(path, states)
187
188    def _breakpoint_event(self, body: dict):
189        answer = body.get("breakpoint") or {}
190        where = self._breakpoints.ids.get(answer.get("id"))
191        if where is not None and body.get("reason") in ("changed", "new"):
192            path, line = where
193            if answer.get("verified"):
194                self._breakpoints.pending.discard(where)
195            self.breakpoints_checked.emit(path, {line: BreakpointState(_placed(answer), _why_not(answer))})
196
197    # ---------------------------------------------------------- running
198    def _request_running(self, command: str, arguments: dict | None = None):
199        if self.state != STOPPED or self.thread_id is None:
200            return
201        self.frames = []
202        self._set_state(RUNNING)
203        self.client.request(command, {"threadId": self.thread_id, **(arguments or {})}, self._answered)
204
205    def _answered(self, success: bool, _body: dict, message: str):
206        if not success and message and self.state != IDLE:
207            self.output.emit("console", f"{message}\n")
208
209    def continue_(self):
210        self._request_running("continue")
211
212    def step_over(self):
213        self._request_running("next")
214
215    def step_in(self):
216        self._request_running("stepIn")
217
218    def step_out(self):
219        self._request_running("stepOut")
220
221    def pause(self):
222        if self.state == RUNNING:
223            self.client.request("pause", {"threadId": self.thread_id or 1}, self._answered)
224
225    def stop(self):
226        """End the session: the program and the debugger."""
227        if not self.is_active():
228            return
229        self.client.request("disconnect", {"terminateDebuggee": True}, timeout=1000)
230        self._end("stopped")
231
232    # ---------------------------------------------------------- events
233    def _event(self, name: str, body: dict):
234        if name == "initialized":
235            self._configure()
236        elif name == "stopped":
237            self.thread_id = body.get("threadId", self.thread_id)
238            # (Which breakpoints, as they were set: the debugger may have moved one to
239            # the next line with code, where the program stops.)
240            ids = self._breakpoints.ids
241            self.hit_breakpoints = [ids[n] for n in body.get("hitBreakpointIds") or [] if n in ids]
242            self._set_state(STOPPED)
243            self._unplaced()
244            why = str(body.get("reason") or "")
245            self.client.request("threads", {}, lambda ok, b, m: self._threads(ok, b, why))
246        elif name == "continued":
247            if self.state == STOPPED:
248                self.frames = []
249                self._set_state(RUNNING)
250        elif name == "output":
251            self._output(body)
252        elif name == "breakpoint":
253            self._breakpoint_event(body)
254        elif name == "exited":
255            self.exit_code = body.get("exitCode")
256        elif name == "terminated":
257            code = self.exit_code
258            self._end("the program ended" if code is None else f"the program exited with code {code}")
259
260    def _unplaced(self):
261        """Stopped, so the program is loaded: a breakpoint still "pending"
262        never will be placed (gdb leaves one on a line with no code so)."""
263        pending, self._breakpoints.pending = self._breakpoints.pending, set()
264        by_file: dict[str, dict[int, BreakpointState]] = {}
265        for path, line in pending:
266            by_file.setdefault(path, {})[line] = BreakpointState(False, "no code on that line: the debugger "
267                                                                        "couldn't place it")
268        for path, states in by_file.items():
269            self.breakpoints_checked.emit(path, states)
270
271    def _output(self, body: dict):
272        category, text = str(body.get("category") or "console"), str(body.get("output") or "")
273        if category == "telemetry" or not text:
274            return
275        if self.adapter is not None and self.adapter.kind == "gdb":
276            text = "".join(line for line in text.splitlines(keepends=True) if not _GDB_NOISE.match(line))
277        if text:
278            self.output.emit(category, text)
279
280    def _adapter_stopped(self, why: str):
281        if not self._ending:
282            self._end(why)
283
284    # ---------------------------------------------------------- where it stopped
285    def _threads(self, success: bool, body: dict, why: str):
286        threads = body.get("threads") or [] if success else []
287        if self.thread_id is None and threads:
288            self.thread_id = threads[0].get("id")
289        if self.thread_id is None:
290            return
291        self.client.request("stackTrace", {"threadId": self.thread_id, "levels": 100},
292                            lambda ok, b, m, w=why: self._stack(ok, b, w))
293
294    def _stack(self, success: bool, body: dict, why: str):
295        if self.state != STOPPED:
296            return
297        self.frames = [Frame(int(f.get("id", 0)), str(f.get("name") or "?"),
298                             (f.get("source") or {}).get("path"), int(f.get("line") or 0))
299                       for f in (body.get("stackFrames") or [] if success else [])]
300        # The first frame with source (not a library's): where the program is, for the user.
301        self.frame_index = next((i for i, f in enumerate(self.frames) if f.path), 0)
302        self.stopped_at.emit(self.frames, why)
303        self.select_frame(self.frame_index)
304
305    def select_frame(self, index: int):
306        """Show a frame of the call stack: its scopes (scopes_ready)."""
307        if self.state != STOPPED or not 0 <= index < len(self.frames):
308            return
309        self.frame_index = index
310        self.client.request("scopes", {"frameId": self.frames[index].id},
311                            lambda ok, b, m, i=index: self._scopes(i, ok, b))
312
313    def _scopes(self, index: int, success: bool, body: dict):
314        scopes = [Scope(str(s.get("name") or ""), int(s.get("variablesReference") or 0), bool(s.get("expensive")))
315                  for s in (body.get("scopes") or [] if success else [])
316                  # (A processor's registers: not what a Fortran programmer is after.)
317                  if s.get("presentationHint") != "registers" and str(s.get("name")).lower() != "registers"]
318        self.scopes_ready.emit(index, scopes)
319
320    def variables(self, reference: int, callback: Callable[[list[Variable], bool], None]):
321        """The variables of a scope, or the parts of one (an array's
322        elements, a type's components): ``callback(variables, more)``, more:
323        there were more than MAX_ELEMENTS."""
324        if self.state != STOPPED:
325            callback([], False)
326            return
327        self.client.request("variables", {"variablesReference": reference, "start": 0, "count": MAX_ELEMENTS},
328                            lambda ok, b, m: callback(*_variables(ok, b)))
329
330    def evaluate(self, expression: str, callback: Callable[[bool, str, int], None], context: str = "watch"):
331        """An expression's value in the frame shown: ``callback(ok, text, reference)``.
332        ("watch": an expression; gdb takes "repl" as one of its commands.)"""
333        frame = self.frames[self.frame_index].id if self.frames and self.state == STOPPED else None
334        arguments = {"expression": expression, "context": context}
335        if frame is not None:
336            arguments["frameId"] = frame
337        self.client.request("evaluate", arguments, lambda ok, b, m: callback(
338            ok, str(b.get("result", "")) if ok else (m or "can't evaluate that"), int(b.get("variablesReference") or 0)))
339
340
341def _placed(answer: dict) -> bool:
342    """Placed, or as good as: "pending" (gdb's, until the program is loaded,
343    when it says where it went), not shown as a problem meanwhile."""
344    return bool(answer.get("verified")) or answer.get("reason") == "pending"
345
346
347def _why_not(answer: dict) -> str:
348    if _placed(answer):
349        return ""
350    return str(answer.get("message") or "the debugger couldn't place it (no code on that line?)")
351
352
353def _variables(success: bool, body: dict) -> tuple[list[Variable], bool]:
354    found = body.get("variables") or [] if success else []
355    variables = [Variable(str(v.get("name") or ""), str(v.get("value") or ""), int(v.get("variablesReference") or 0),
356                          int(v.get("indexedVariables") or v.get("namedVariables") or 0), str(v.get("type") or ""))
357                 for v in found]
358    return variables, len(found) >= MAX_ELEMENTS