Coverage for vibefpmgui/debug_session.py: 86%
242 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-01 20:51 +0000
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.
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"""
13from __future__ import annotations
15import re
16from collections.abc import Callable
17from dataclasses import dataclass, field
18from pathlib import Path
20from qtpy.QtCore import QObject, Signal
22from .dap import DapClient
23from .debug_adapters import Adapter
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 "…")
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*$")
35@dataclass
36class Frame:
37 id: int
38 name: str
39 path: str | None # (None: no source: a library's)
40 line: int
43@dataclass
44class Scope:
45 name: str
46 reference: int
47 expensive: bool = False
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 = ""
59@dataclass
60class BreakpointState:
61 verified: bool
62 message: str = ""
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)
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)
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
97 # ---------------------------------------------------------- state
98 def is_active(self) -> bool:
99 return self.state != IDLE
101 def _set_state(self, state: str):
102 if state != self.state: 102 ↛ exitline 102 didn't return from function '_set_state' because the condition on line 102 was always true
103 self.state = state
104 self.state_changed.emit(state)
106 def _end(self, why: str):
107 if self.state == IDLE: 107 ↛ 108line 107 didn't jump to line 108 because the condition on line 107 was never true
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)
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(): 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true
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
137 def _initialized(self, success: bool, _body: dict, message: str):
138 if not success: 138 ↛ 139line 138 didn't jump to line 139 because the condition on line 138 was never true
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)
144 def _launched(self, success: bool, _body: dict, message: str):
145 if not success and self.state != IDLE: 145 ↛ 146line 145 didn't jump to line 146 because the condition on line 145 was never true
146 self._end(f"the program didn't start: {message or 'the debugger refused it'}")
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)
154 def _configured(self, success: bool, _body: dict, message: str):
155 if not success: 155 ↛ 156line 155 didn't jump to line 156 because the condition on line 155 was never true
156 self._end(f"the debugger couldn't start the program: {message}")
157 elif self.state == STARTING: 157 ↛ exitline 157 didn't return from function '_configured' because the condition on line 157 was always true
158 self._set_state(RUNNING)
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)
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))
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): 179 ↛ 181line 179 didn't jump to line 181 because the condition on line 179 was always true
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):]: 184 ↛ 185line 184 didn't jump to line 185 because the loop on line 184 never started
185 states[line] = BreakpointState(False, message or "the debugger didn't place it")
186 self.breakpoints_checked.emit(path, states)
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"): 191 ↛ exitline 191 didn't return from function '_breakpoint_event' because the condition on line 191 was always true
192 path, line = where
193 if answer.get("verified"): 193 ↛ 195line 193 didn't jump to line 195 because the condition on line 193 was always true
194 self._breakpoints.pending.discard(where)
195 self.breakpoints_checked.emit(path, {line: BreakpointState(_placed(answer), _why_not(answer))})
197 # ---------------------------------------------------------- running
198 def _request_running(self, command: str, arguments: dict | None = None):
199 if self.state != STOPPED or self.thread_id is None: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 return
201 self.frames = []
202 self._set_state(RUNNING)
203 self.client.request(command, {"threadId": self.thread_id, **(arguments or {})}, self._answered)
205 def _answered(self, success: bool, _body: dict, message: str):
206 if not success and message and self.state != IDLE: 206 ↛ 207line 206 didn't jump to line 207 because the condition on line 206 was never true
207 self.output.emit("console", f"{message}\n")
209 def continue_(self):
210 self._request_running("continue")
212 def step_over(self):
213 self._request_running("next")
215 def step_in(self):
216 self._request_running("stepIn")
218 def step_out(self):
219 self._request_running("stepOut")
221 def pause(self):
222 if self.state == RUNNING:
223 self.client.request("pause", {"threadId": self.thread_id or 1}, self._answered)
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")
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: 247 ↛ 248line 247 didn't jump to line 248 because the condition on line 247 was never true
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}")
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)
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: 273 ↛ 274line 273 didn't jump to line 274 because the condition on line 273 was never true
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)
280 def _adapter_stopped(self, why: str):
281 if not self._ending:
282 self._end(why)
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: 287 ↛ 288line 287 didn't jump to line 288 because the condition on line 287 was never true
288 self.thread_id = threads[0].get("id")
289 if self.thread_id is None: 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
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))
294 def _stack(self, success: bool, body: dict, why: str):
295 if self.state != STOPPED: 295 ↛ 296line 295 didn't jump to line 296 because the condition on line 295 was never true
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)
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))
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)
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: 324 ↛ 325line 324 didn't jump to line 325 because the condition on line 324 was never true
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)))
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: 335 ↛ 337line 335 didn't jump to line 337 because the condition on line 335 was always true
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)))
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"
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?)")
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