From 52253335fbfae9a42a7ffe8f1cee30a172b81433 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Sat, 12 Apr 2025 14:56:28 -0400
Subject: [PATCH 01/40] Allow pdb to attach to a running process

Signed-off-by: Matt Wozniski <mwozniski@bloomberg.net>
---
 Lib/pdb.py | 507 ++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 file changed, 505 insertions(+), 2 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 160a7043a30c55..afef3b7e079356 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -74,20 +74,28 @@
 import dis
 import code
 import glob
+import json
 import token
 import types
 import codeop
 import pprint
 import signal
+import socket
+import typing
 import asyncio
 import inspect
+import weakref
+import builtins
+import tempfile
 import textwrap
 import tokenize
 import itertools
 import traceback
 import linecache
 import _colorize
+import dataclasses
 
+from contextlib import closing
 from contextlib import contextmanager
 from rlcompleter import Completer
 from types import CodeType
@@ -2491,6 +2499,490 @@ def set_trace(*, header=None, commands=None):
         pdb.message(header)
     pdb.set_trace(sys._getframe().f_back, commands=commands)
 
+# Remote PDB
+
+@dataclasses.dataclass(frozen=True)
+class _InteractState:
+    compiler: codeop.CommandCompiler
+    ns: dict[str, typing.Any]
+
+
+class _RemotePdb(Pdb):
+    def __init__(self, sockfile, owns_sockfile=True, **kwargs):
+        self._owns_sockfile = owns_sockfile
+        self._interact_state = None
+        self._sockfile = sockfile
+        self._command_name_cache = []
+        super().__init__(**kwargs)
+
+    def _send(self, **kwargs) -> None:
+        json_payload = json.dumps(kwargs)
+        self._sockfile.write(json_payload.encode() + b"\n")
+        self._sockfile.flush()
+
+    def message(self, msg, end="\n"):
+        self._send(message=msg + end)
+
+    def error(self, msg):
+        self._send(error=msg)
+
+    def _read_command(self) -> str:
+        # Loop until we get a command for PDB or an 'interact' REPL.
+        # Process out-of-band completion messages without returning.
+        while True:
+            payload = self._sockfile.readline()
+            if not payload:
+                return "EOF"
+            try:
+                match json.loads(payload):
+                    case {"command": str(line)}:
+                        # Interact mode has been cancelled client-side,
+                        # likely by a ^D EOF at the prompt.
+                        self._interact_state = None
+                        return line
+                    case {"interact": str(lines)}:
+                        if self._interact_state:
+                            return lines
+                        # Otherwise, fall through to report an error and loop.
+                    case {
+                        "completion": {
+                            "text": str(text),
+                            "line": str(line),
+                            "begidx": int(begidx),
+                            "endidx": int(endidx),
+                        }
+                    }:
+                        items = self._complete_any(text, line, begidx, endidx)
+                        self._send(completion=items)
+                        continue
+            except json.JSONDecodeError:
+                pass
+            # Invalid JSON, or doesn't meet the schema, or wrong PDB state.
+            self.error(f"Ignoring invalid remote command: {payload}")
+
+    def _complete_any(self, text, line, begidx, endidx):
+        if begidx == 0:
+            return self.completenames(text, line, begidx, endidx)
+
+        cmd = self.parseline(line)[0]
+        if cmd:
+            compfunc = getattr(self, "complete_" + cmd, self.completedefault)
+        else:
+            compfunc = self.completedefault
+        return compfunc(text, line, begidx, endidx)
+
+    def cmdloop(self, intro=None):
+        self.preloop()
+        if intro is not None:
+            self.intro = intro
+        if self.intro:
+            self.message(str(self.intro))
+        stop = None
+        while not stop:
+            if self.cmdqueue:
+                line = self.cmdqueue.pop(0)
+            else:
+                if not self._command_name_cache:
+                    self._command_name_cache = self.completenames("", "", 0, 0)
+                    self._send(command_list=self._command_name_cache)
+
+                mode = "interact" if self._interact_state else "pdb"
+                self._send(prompt=self.prompt, mode=mode)
+                line = self._read_command()
+                if self._interact_state is not None:
+                    self._run_in_python_repl(line)
+                    continue
+            line = self.precmd(line)
+            stop = self.onecmd(line)
+            stop = self.postcmd(stop, line)
+        self.postloop()
+
+    def detach(self):
+        # Detach the debugger and close the socket without raising BdbQuit
+        self.quitting = False
+        if self._owns_sockfile:
+            self._sockfile.close()
+
+    def do_EOF(self, arg):
+        ret = super().do_EOF(arg)
+        self.detach()
+        return ret
+
+    def do_q(self, arg):
+        ret = super().do_q(arg)
+        self.detach()
+        return ret
+
+    def do_quit(self, arg):
+        ret = super().do_quit(arg)
+        self.detach()
+        return ret
+
+    def do_exit(self, arg):
+        ret = super().do_exit(arg)
+        self.detach()
+        return ret
+
+    def do_alias(self, arg):
+        # Clear our cached list of valid commands; one might be added.
+        self._command_name_cache = []
+        return super().do_alias(arg)
+
+    def do_unalias(self, arg):
+        # Clear our cached list of valid commands; one might be removed.
+        self._command_name_cache = []
+        return super().do_unalias(arg)
+
+    def do_help(self, arg):
+        # Tell the client to render the help, since it might need a pager.
+        self._send(help=arg)
+
+    do_h = do_help
+
+    def _interact_displayhook(self, obj):
+        # Like the default `sys.displayhook` except sending a socket message.
+        if obj is not None:
+            self.message(repr(obj))
+            builtins._ = obj
+
+    def _run_in_python_repl(self, lines):
+        # Run one 'interact' mode code block against an existing namespace.
+        assert self._interact_state
+        save_displayhook = sys.displayhook
+        try:
+            sys.displayhook = self._interact_displayhook
+            code_obj = self._interact_state.compiler(lines + "\n")
+            if code_obj is None:
+                raise SyntaxError("Incomplete command")
+            exec(code_obj, self._interact_state.ns)
+        except:
+            self._error_exc()
+        finally:
+            sys.displayhook = save_displayhook
+
+    def do_interact(self, arg):
+        # Prepare to run 'interact' mode code blocks, and trigger the client
+        # to start treating all input as Python commands, not PDB ones.
+        self._interact_state = _InteractState(
+            compiler=codeop.CommandCompiler(),
+            ns={**self.curframe.f_globals, **self.curframe_locals},
+        )
+
+    def do_commands(self, arg):
+        # Called with only one line in 'arg' to start collecting commands.
+        # Once the client has gathered up the full list of commands to apply,
+        # do_commands gets called again with multiple lines of input in args.
+        arg, *commands = arg.split("\n")
+        if not arg:
+            bnum = len(bdb.Breakpoint.bpbynumber) - 1
+        else:
+            try:
+                bnum = int(arg)
+            except ValueError:
+                self._print_invalid_arg(arg)
+                return
+
+        try:
+            self.get_bpbynumber(bnum)
+        except ValueError as err:
+            self.error("cannot set commands: %s" % err)
+            return
+
+        if not commands:
+            # We've only received the first line so far.
+            # Have the client enter command entry mode.
+
+            # fmt: off
+            end_cmds = [
+                "c", "cont", "continue",
+                "s", "step",
+                "n", "next",
+                "r", "return",
+                "q", "quit", "exit",
+                "j", "jump",
+            ]
+            # fmt: on
+
+            self._send(commands_entry={"bpnum": bnum, "terminators": end_cmds})
+            return
+
+        self.commands_bnum = bnum
+        self.commands[bnum] = []
+        self.commands_doprompt[bnum] = True
+        self.commands_silent[bnum] = False
+
+        for line in commands:
+            if self.handle_command_def(line):
+                break
+
+    def do_debug(self, arg):
+        # Enter a recursive _RemotePdb, telling it not to close the socket.
+        sys.settrace(None)
+        globals = self.curframe.f_globals
+        locals = self.curframe_locals
+        p = _RemotePdb(self._sockfile, owns_sockfile=False)
+        p.prompt = "(%s) " % self.prompt.strip()
+        self.message("ENTERING RECURSIVE DEBUGGER")
+        try:
+            sys.call_tracing(p.run, (arg, globals, locals))
+        except Exception:
+            self._error_exc()
+        self.message("LEAVING RECURSIVE DEBUGGER")
+        sys.settrace(self.trace_dispatch)
+        self.lastcmd = p.lastcmd
+
+    def do_run(self, arg):
+        self.error("remote PDB cannot restart the program")
+
+    do_restart = do_run
+
+    def _error_exc(self):
+        exc = sys.exception()
+        if isinstance(exc, SystemExit):
+            # If we get a SystemExit in 'interact' mode, exit the REPL.
+            self._interact_state = None
+        super()._error_exc()
+
+    def default(self, line):
+        # Unlike Pdb, don't prompt for more lines of a multi-line command.
+        # The remote needs to send us the whole block in one go.
+        try:
+            candidate = line.removeprefix("!") + "\n"
+            if not codeop.compile_command(candidate, "<stdin>", "single"):
+                raise SyntaxError("Incomplete command")
+            return super().default(candidate)
+        except:
+            self._error_exc()
+
+
+class _PdbClient:
+    def __init__(self, sockfile):
+        self.sockfile = sockfile
+        self.pdb_instance = Pdb()
+        self.pdb_commands = set()
+        self.completion_matches = []
+        self.interact_mode = False
+        self.commands_mode = False
+        self.bpnum = 0
+        self.command_list_terminators = []
+
+    def read_command(self, prompt):
+        continue_prompt = "...".ljust(len(prompt))
+
+        if self.interact_mode:
+            # Python REPL mode
+            try:
+                line = input(">>> ")
+            except EOFError:
+                print("\n*exit from pdb interact command*")
+                self.interact_mode = False
+            else:
+                continue_prompt = "...".ljust(len(">>> "))
+
+        if not self.interact_mode:
+            # PDB command entry mode
+            line = input(prompt)
+            cmd = self.pdb_instance.parseline(line)[0]
+            if cmd in self.pdb_commands or line.strip() == "":
+                # Recognized PDB command, or repeating last command
+                return line
+
+            # Otherwise, explicit or implicit exec command
+            if line.startswith("!"):
+                line = line[1:].lstrip()
+
+        # Ensure the remote won't try to use this as a PDB command.
+        prefix = "!" if not self.interact_mode else ""
+
+        if (
+            codeop.compile_command(line + "\n", "<stdin>", "single")
+            is not None
+        ):
+            # Valid single-line statement
+            return prefix + line
+
+        # Otherwise, valid first line of a multi-line statement
+        continue_prompt = "...".ljust(len(prompt))
+        buffer = line
+
+        while codeop.compile_command(buffer, "<stdin>", "single") is None:
+            buffer += "\n" + input(continue_prompt)
+
+        return prefix + buffer
+
+    @contextmanager
+    def readline_completion(self):
+        try:
+            import readline
+        except ImportError:
+            yield
+            return
+
+        old_completer = readline.get_completer()
+        try:
+            readline.set_completer(self.complete)
+            if readline.backend == "editline":
+                # libedit uses "^I" instead of "tab"
+                command_string = "bind ^I rl_complete"
+            else:
+                command_string = "tab: complete"
+            readline.parse_and_bind(command_string)
+            yield
+        finally:
+            readline.set_completer(old_completer)
+
+    def cmdloop(self):
+        with self.readline_completion():
+            while payload_bytes := self.sockfile.readline():
+                try:
+                    payload = json.loads(payload_bytes)
+                except json.JSONDecodeError:
+                    print(
+                        "***", f"Invalid JSON from remote: {payload}", flush=True
+                    )
+                    continue
+
+                self.process_payload(payload)
+
+    def process_payload(self, payload):
+        match payload:
+            case {"command_list": command_list}:
+                self.pdb_commands = set(command_list)
+            case {"message": str(msg)}:
+                print(msg, end="", flush=True)
+            case {"error": str(msg)}:
+                print("***", msg, flush=True)
+            case {"help": str(arg)}:
+                self.pdb_instance.do_help(arg)
+            case {"prompt": str(prompt), "mode": str(mode)}:
+                if mode == "interact":
+                    self.interact_mode = True
+                elif self.interact_mode and mode != "interact":
+                    print("*exit from pdb interact command*")
+                    self.interact_mode = False
+
+                if self.commands_mode:
+                    self.prompt_for_breakpoint_command_list("(com) ")
+                else:
+                    self.prompt_for_repl_command(prompt)
+            case {
+                "commands_entry": {
+                    "bpnum": int(bpnum),
+                    "terminators": list(command_list_terminators),
+                }
+            }:
+                self.bpnum = bpnum
+                self.command_list_terminators = command_list_terminators
+                self.commands_mode = True
+            case _:
+                raise RuntimeError(f"Unrecognized payload {payload}")
+
+    def prompt_for_breakpoint_command_list(self, prompt):
+        parts = [f"commands {self.bpnum}"]
+        while True:
+            try:
+                line = input(prompt).strip()
+                parts.append(line)
+                cmd = self.pdb_instance.parseline(line)[0]
+                if cmd in self.command_list_terminators:
+                    break
+            except EOFError:
+                return
+            finally:
+                self.commands_mode = False
+
+        command = "\n".join(parts)
+        self.sockfile.write((json.dumps({"command": command}) + "\n").encode())
+        self.sockfile.flush()
+
+    def prompt_for_repl_command(self, prompt):
+        while True:
+            try:
+                command = self.read_command(prompt)
+            except EOFError:
+                command = "EOF"
+            except KeyboardInterrupt:
+                print(flush=True)
+                continue
+            except Exception as exc:
+                msg = traceback.format_exception_only(exc)[-1].strip()
+                print("***", msg, flush=True)
+                continue
+
+            if self.interact_mode:
+                payload = {"interact": command}
+            else:
+                payload = {"command": command}
+            self.sockfile.write((json.dumps(payload) + "\n").encode())
+            self.sockfile.flush()
+            return
+
+    def complete(self, text, state):
+        import readline
+
+        if state == 0:
+            origline = readline.get_line_buffer()
+            line = origline.lstrip()
+            stripped = len(origline) - len(line)
+            begidx = readline.get_begidx() - stripped
+            endidx = readline.get_endidx() - stripped
+
+            msg = {
+                "completion": {
+                    "text": text,
+                    "line": line,
+                    "begidx": begidx,
+                    "endidx": endidx,
+                }
+            }
+            self.sockfile.write((json.dumps(msg) + "\n").encode())
+            self.sockfile.flush()
+
+            payload = self.sockfile.readline()
+            payload = json.loads(payload)
+            if "completion" not in payload:
+                raise RuntimeError(
+                    f"Failed to get valid completions. Got: {payload}"
+                )
+
+            self.completion_matches = payload["completion"]
+        try:
+            return self.completion_matches[state]
+        except IndexError:
+            return None
+
+
+def _connect(host, port):
+    with closing(socket.create_connection((host, port))) as conn:
+        sockfile = conn.makefile("rwb")
+
+    remote_pdb = _RemotePdb(sockfile)
+    weakref.finalize(remote_pdb, sockfile.close)
+    remote_pdb.set_trace()
+
+
+def attach(pid):
+    """Attach to a running process with the given PID."""
+    with closing(socket.create_server(("localhost", 0))) as server:
+        port = server.getsockname()[1]
+
+        with tempfile.NamedTemporaryFile("w", delete_on_close=False) as script:
+            script.write(
+                f'from pdb import _connect; _connect("localhost", {port})\n'
+            )
+            script.close()
+            sys.remote_exec(pid, script.name)
+
+            # TODO Add a timeout? Or don't bother since the user can ^C?
+            client_sock, _ = server.accept()
+
+    with closing(client_sock):
+        sockfile = client_sock.makefile("rwb")
+
+    with closing(sockfile):
+        _PdbClient(sockfile).cmdloop()
+
+
 # Post-Mortem interface
 
 def post_mortem(t=None):
@@ -2560,7 +3052,7 @@ def help():
 def main():
     import argparse
 
-    parser = argparse.ArgumentParser(usage="%(prog)s [-h] [-c command] (-m module | pyfile) [args ...]",
+    parser = argparse.ArgumentParser(usage="%(prog)s [-h] [-c command] (-m module | -p pid | pyfile) [args ...]",
                                      description=_usage,
                                      formatter_class=argparse.RawDescriptionHelpFormatter,
                                      allow_abbrev=False)
@@ -2571,6 +3063,7 @@ def main():
     parser.add_argument('-c', '--command', action='append', default=[], metavar='command', dest='commands',
                         help='pdb commands to execute as if given in a .pdbrc file')
     parser.add_argument('-m', metavar='module', dest='module')
+    parser.add_argument('-p', '--pid', type=int, help="attach to the specified PID", default=None)
 
     if len(sys.argv) == 1:
         # If no arguments were given (python -m pdb), print the whole help message.
@@ -2580,7 +3073,11 @@ def main():
 
     opts, args = parser.parse_known_args()
 
-    if opts.module:
+    if opts.pid:
+        # If attaching to a remote pid, unrecognized arguments are not allowed.
+        # This will raise an error if there are extra unrecognized arguments.
+        parser.parse_args()
+    elif opts.module:
         # If a module is being debugged, we consider the arguments after "-m module" to
         # be potential arguments to the module itself. We need to parse the arguments
         # before "-m" to check if there is any invalid argument.
@@ -2599,6 +3096,12 @@ def main():
             parser.error(f"unrecognized arguments: {' '.join(invalid_args)}")
             sys.exit(2)
 
+    if opts.pid:
+        if opts.module:
+            parser.error("argument -m: not allowed with argument --pid")
+        attach(opts.pid)
+        return
+
     if opts.module:
         file = opts.module
         target = _ModuleTarget(file)

From 78a3085161b055447e5a257e1a17862a2d898b19 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 15 Apr 2025 16:32:44 -0400
Subject: [PATCH 02/40] Remove 2 unused _RemotePdb instance attributes

These were removed from the Pdb base class in b5774603
---
 Lib/pdb.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index afef3b7e079356..d9bb4ce34826da 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2708,8 +2708,6 @@ def do_commands(self, arg):
 
         self.commands_bnum = bnum
         self.commands[bnum] = []
-        self.commands_doprompt[bnum] = True
-        self.commands_silent[bnum] = False
 
         for line in commands:
             if self.handle_command_def(line):

From 90e0a81d499634840074dcd64cf577dfc2bd57df Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 15 Apr 2025 16:43:42 -0400
Subject: [PATCH 03/40] Reduce duplication for 'debug' command

The only thing _RemotePdb needs to customize is how the recursive
debugger is constructed. Have the base class provide a customization
point that we can use.
---
 Lib/pdb.py | 23 +++++++----------------
 1 file changed, 7 insertions(+), 16 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index d9bb4ce34826da..0fe472a82caf05 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -1783,6 +1783,9 @@ def do_jump(self, arg):
                 self.error('Jump failed: %s' % e)
     do_j = do_jump
 
+    def _create_recursive_debugger(self):
+        return Pdb(self.completekey, self.stdin, self.stdout)
+
     def do_debug(self, arg):
         """debug code
 
@@ -1796,7 +1799,7 @@ def do_debug(self, arg):
         self.stop_trace()
         globals = self.curframe.f_globals
         locals = self.curframe.f_locals
-        p = Pdb(self.completekey, self.stdin, self.stdout)
+        p = self._create_recursive_debugger()
         p.prompt = "(%s) " % self.prompt.strip()
         self.message("ENTERING RECURSIVE DEBUGGER")
         try:
@@ -2713,21 +2716,9 @@ def do_commands(self, arg):
             if self.handle_command_def(line):
                 break
 
-    def do_debug(self, arg):
-        # Enter a recursive _RemotePdb, telling it not to close the socket.
-        sys.settrace(None)
-        globals = self.curframe.f_globals
-        locals = self.curframe_locals
-        p = _RemotePdb(self._sockfile, owns_sockfile=False)
-        p.prompt = "(%s) " % self.prompt.strip()
-        self.message("ENTERING RECURSIVE DEBUGGER")
-        try:
-            sys.call_tracing(p.run, (arg, globals, locals))
-        except Exception:
-            self._error_exc()
-        self.message("LEAVING RECURSIVE DEBUGGER")
-        sys.settrace(self.trace_dispatch)
-        self.lastcmd = p.lastcmd
+    @typing.override
+    def _create_recursive_debugger(self):
+        return _RemotePdb(self._sockfile, owns_sockfile=False)
 
     def do_run(self, arg):
         self.error("remote PDB cannot restart the program")

From e44a670b0c7a62642777197cc8402a71ca87114e Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 15 Apr 2025 17:17:17 -0400
Subject: [PATCH 04/40] End commands entry on 'end' and ^C and ^D

---
 Lib/pdb.py | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 0fe472a82caf05..0c78af42ee4c3e 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2706,6 +2706,7 @@ def do_commands(self, arg):
             ]
             # fmt: on
 
+            end_cmds += ["end"]  # pseudo-command
             self._send(commands_entry={"bpnum": bnum, "terminators": end_cmds})
             return
 
@@ -2875,8 +2876,10 @@ def prompt_for_breakpoint_command_list(self, prompt):
                 cmd = self.pdb_instance.parseline(line)[0]
                 if cmd in self.command_list_terminators:
                     break
-            except EOFError:
-                return
+            except (KeyboardInterrupt, EOFError):
+                print(flush=True)
+                print("command definition aborted, old commands restored")
+                break
             finally:
                 self.commands_mode = False
 

From e83724670c0df5301f6d44c51ef08ae84cdece8a Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 15 Apr 2025 17:29:06 -0400
Subject: [PATCH 05/40] Set the frame for remote pdb to stop in explicitly

This makes us robust against changes in the temporary script.
---
 Lib/pdb.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 0c78af42ee4c3e..17ae3487935358 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2944,13 +2944,13 @@ def complete(self, text, state):
             return None
 
 
-def _connect(host, port):
+def _connect(host, port, frame):
     with closing(socket.create_connection((host, port))) as conn:
         sockfile = conn.makefile("rwb")
 
     remote_pdb = _RemotePdb(sockfile)
     weakref.finalize(remote_pdb, sockfile.close)
-    remote_pdb.set_trace()
+    remote_pdb.set_trace(frame=frame)
 
 
 def attach(pid):
@@ -2960,7 +2960,8 @@ def attach(pid):
 
         with tempfile.NamedTemporaryFile("w", delete_on_close=False) as script:
             script.write(
-                f'from pdb import _connect; _connect("localhost", {port})\n'
+                f'import pdb, sys\n'
+                f'pdb._connect("localhost", {port}, sys._getframe().f_back)\n'
             )
             script.close()
             sys.remote_exec(pid, script.name)

From 27efa973275be59be2f7b5021758c296ef5e6337 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 15 Apr 2025 17:32:25 -0400
Subject: [PATCH 06/40] Fix an unbound local in an error message

---
 Lib/pdb.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 17ae3487935358..a40d6e0bf90812 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2828,7 +2828,7 @@ def cmdloop(self):
                     payload = json.loads(payload_bytes)
                 except json.JSONDecodeError:
                     print(
-                        "***", f"Invalid JSON from remote: {payload}", flush=True
+                        "***", f"Invalid JSON from remote: {payload_bytes}", flush=True
                     )
                     continue
 

From 557a7259bebe80660c84f36fd27a14543fe2eb85 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 15 Apr 2025 17:37:14 -0400
Subject: [PATCH 07/40] Clean up remote PDB detaching

---
 Lib/pdb.py | 25 +++++--------------------
 1 file changed, 5 insertions(+), 20 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index a40d6e0bf90812..470bdd25cd2c83 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2600,32 +2600,17 @@ def cmdloop(self, intro=None):
             stop = self.postcmd(stop, line)
         self.postloop()
 
+    def postloop(self):
+        super().postloop()
+        if self.quitting:
+            self.detach()
+
     def detach(self):
         # Detach the debugger and close the socket without raising BdbQuit
         self.quitting = False
         if self._owns_sockfile:
             self._sockfile.close()
 
-    def do_EOF(self, arg):
-        ret = super().do_EOF(arg)
-        self.detach()
-        return ret
-
-    def do_q(self, arg):
-        ret = super().do_q(arg)
-        self.detach()
-        return ret
-
-    def do_quit(self, arg):
-        ret = super().do_quit(arg)
-        self.detach()
-        return ret
-
-    def do_exit(self, arg):
-        ret = super().do_exit(arg)
-        self.detach()
-        return ret
-
     def do_alias(self, arg):
         # Clear our cached list of valid commands; one might be added.
         self._command_name_cache = []

From 7f7584aa5e1b9a6a569ce6bc0680d1ac80a06ef6 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Thu, 17 Apr 2025 17:35:52 -0400
Subject: [PATCH 08/40] Allow ctrl-c to interrupt a running process

---
 Lib/pdb.py | 34 ++++++++++++++++++++++++++++++----
 1 file changed, 30 insertions(+), 4 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 470bdd25cd2c83..39947641085eca 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2731,8 +2731,10 @@ def default(self, line):
 
 
 class _PdbClient:
-    def __init__(self, sockfile):
+    def __init__(self, pid, sockfile, interrupt_script):
+        self.pid = pid
         self.sockfile = sockfile
+        self.interrupt_script = interrupt_script
         self.pdb_instance = Pdb()
         self.pdb_commands = set()
         self.completion_matches = []
@@ -2808,17 +2810,32 @@ def readline_completion(self):
 
     def cmdloop(self):
         with self.readline_completion():
-            while payload_bytes := self.sockfile.readline():
+            while True:
+                try:
+                    if not (payload_bytes := self.sockfile.readline()):
+                        break
+                except KeyboardInterrupt:
+                    self.send_interrupt()
+                    continue
+
                 try:
                     payload = json.loads(payload_bytes)
                 except json.JSONDecodeError:
                     print(
-                        "***", f"Invalid JSON from remote: {payload_bytes}", flush=True
+                        f"*** Invalid JSON from remote: {payload_bytes}",
+                        flush=True,
                     )
                     continue
 
                 self.process_payload(payload)
 
+    def send_interrupt(self):
+        print(
+            "\n*** Program will stop at the next bytecode instruction."
+            " (Use 'cont' to resume)."
+        )
+        sys.remote_exec(self.pid, self.interrupt_script)
+
     def process_payload(self, payload):
         match payload:
             case {"command_list": command_list}:
@@ -2958,7 +2975,16 @@ def attach(pid):
         sockfile = client_sock.makefile("rwb")
 
     with closing(sockfile):
-        _PdbClient(sockfile).cmdloop()
+        with tempfile.NamedTemporaryFile("w", delete_on_close=False) as script:
+            script.write(
+                'import pdb, sys\n'
+                'if inst := pdb.Pdb._last_pdb_instance:\n'
+                '    inst.set_step()\n'
+                '    inst.set_trace(sys._getframe(1))\n'
+            )
+            script.close()
+
+            _PdbClient(pid, sockfile, script.name).cmdloop()
 
 
 # Post-Mortem interface

From 5666ffb4b62430a69cb6942e069e2d1d8319b6e3 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Thu, 17 Apr 2025 18:00:10 -0400
Subject: [PATCH 09/40] Automatically detach if the client dies unexpectedly

---
 Lib/pdb.py | 23 ++++++++++++++++++++---
 1 file changed, 20 insertions(+), 3 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 39947641085eca..2f593ad1a18a1d 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2516,12 +2516,21 @@ def __init__(self, sockfile, owns_sockfile=True, **kwargs):
         self._interact_state = None
         self._sockfile = sockfile
         self._command_name_cache = []
+        self._write_failed = False
         super().__init__(**kwargs)
 
     def _send(self, **kwargs) -> None:
         json_payload = json.dumps(kwargs)
-        self._sockfile.write(json_payload.encode() + b"\n")
-        self._sockfile.flush()
+        try:
+            self._sockfile.write(json_payload.encode() + b"\n")
+            self._sockfile.flush()
+        except OSError:
+            # This means that the client has abruptly disconnected, but we'll
+            # handle that the next time we try to read from the client instead
+            # of trying to handle it from everywhere _send() may be called.
+            # Track this with a flag rather than assuming readline() will ever
+            # return an empty string because the socket may be half-closed.
+            self._write_failed = True
 
     def message(self, msg, end="\n"):
         self._send(message=msg + end)
@@ -2533,9 +2542,13 @@ def _read_command(self) -> str:
         # Loop until we get a command for PDB or an 'interact' REPL.
         # Process out-of-band completion messages without returning.
         while True:
+            if self._write_failed:
+                return "EOF"
+
             payload = self._sockfile.readline()
             if not payload:
                 return "EOF"
+
             try:
                 match json.loads(payload):
                     case {"command": str(line)}:
@@ -2609,7 +2622,11 @@ def detach(self):
         # Detach the debugger and close the socket without raising BdbQuit
         self.quitting = False
         if self._owns_sockfile:
-            self._sockfile.close()
+            try:
+                self._sockfile.close()
+            except OSError:
+                # close() can fail if the connection was broken unexpectedly.
+                pass
 
     def do_alias(self, arg):
         # Clear our cached list of valid commands; one might be added.

From 325f166626840f840cc8c01e26ab6efbb9d6264a Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Thu, 17 Apr 2025 18:53:31 -0400
Subject: [PATCH 10/40] Clear _last_pdb_instance on detach

---
 Lib/pdb.py | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 2f593ad1a18a1d..a750cf41d78138 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2622,6 +2622,8 @@ def detach(self):
         # Detach the debugger and close the socket without raising BdbQuit
         self.quitting = False
         if self._owns_sockfile:
+            # Don't try to reuse this instance, it's not valid anymore.
+            Pdb._last_pdb_instance = None
             try:
                 self._sockfile.close()
             except OSError:

From 72830e2ce1bb785727bde0c10de017f81d0566b7 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Thu, 17 Apr 2025 19:03:16 -0400
Subject: [PATCH 11/40] Refuse to attach if another PDB instance is installed

---
 Lib/pdb.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index a750cf41d78138..fcf6b66dea48c2 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2971,7 +2971,11 @@ def _connect(host, port, frame):
 
     remote_pdb = _RemotePdb(sockfile)
     weakref.finalize(remote_pdb, sockfile.close)
-    remote_pdb.set_trace(frame=frame)
+
+    if Pdb._last_pdb_instance is not None:
+        remote_pdb.error("Another PDB instance is already attached.")
+    else:
+        remote_pdb.set_trace(frame=frame)
 
 
 def attach(pid):

From 27c6780160580789a780cd3b20f0e28da8ac913a Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Thu, 17 Apr 2025 20:08:31 -0400
Subject: [PATCH 12/40] Handle the confirmation prompt issued by 'clear'

---
 Lib/pdb.py | 75 ++++++++++++++++++++++++++++++++++++++++++++++++------
 1 file changed, 67 insertions(+), 8 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index fcf6b66dea48c2..fd7d23208bc811 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -89,6 +89,7 @@
 import tempfile
 import textwrap
 import tokenize
+import functools
 import itertools
 import traceback
 import linecache
@@ -1466,6 +1467,13 @@ def do_ignore(self, arg):
 
     complete_ignore = _complete_bpnumber
 
+    def _prompt_for_confirmation(self, prompt, choices, default):
+        try:
+            reply = input(prompt)
+        except EOFError:
+            reply = default
+        return reply.strip().lower()
+
     def do_clear(self, arg):
         """cl(ear) [filename:lineno | bpnumber ...]
 
@@ -1475,11 +1483,11 @@ def do_clear(self, arg):
         clear all breaks at that line in that file.
         """
         if not arg:
-            try:
-                reply = input('Clear all breaks? ')
-            except EOFError:
-                reply = 'no'
-            reply = reply.strip().lower()
+            reply = self._prompt_for_confirmation(
+                'Clear all breaks? ',
+                choices=('y', 'yes', 'n', 'no'),
+                default='no',
+            )
             if reply in ('y', 'yes'):
                 bplist = [bp for bp in bdb.Breakpoint.bpbynumber if bp]
                 self.clear_all_breaks()
@@ -2725,6 +2733,26 @@ def do_commands(self, arg):
     def _create_recursive_debugger(self):
         return _RemotePdb(self._sockfile, owns_sockfile=False)
 
+    @typing.override
+    def _prompt_for_confirmation(self, prompt, choices, default):
+        self._send(
+            confirm={"prompt": prompt, "choices": choices, "default": default}
+        )
+        payload = self._sockfile.readline()
+
+        if not payload:
+            return default
+
+        try:
+            match json.loads(payload):
+                case {"confirmation_reply": str(reply)}:
+                    return reply
+        except json.JSONDecodeError:
+            pass
+
+        self.error(f"Ignoring unexpected remote message: {payload}")
+        return default
+
     def do_run(self, arg):
         self.error("remote PDB cannot restart the program")
 
@@ -2807,7 +2835,7 @@ def read_command(self, prompt):
         return prefix + buffer
 
     @contextmanager
-    def readline_completion(self):
+    def readline_completion(self, completer):
         try:
             import readline
         except ImportError:
@@ -2816,7 +2844,7 @@ def readline_completion(self):
 
         old_completer = readline.get_completer()
         try:
-            readline.set_completer(self.complete)
+            readline.set_completer(completer)
             if readline.backend == "editline":
                 # libedit uses "^I" instead of "tab"
                 command_string = "bind ^I rl_complete"
@@ -2828,7 +2856,7 @@ def readline_completion(self):
             readline.set_completer(old_completer)
 
     def cmdloop(self):
-        with self.readline_completion():
+        with self.readline_completion(self.complete):
             while True:
                 try:
                     if not (payload_bytes := self.sockfile.readline()):
@@ -2876,6 +2904,14 @@ def process_payload(self, payload):
                     self.prompt_for_breakpoint_command_list("(com) ")
                 else:
                     self.prompt_for_repl_command(prompt)
+            case {
+                "confirm": {
+                    "prompt": str(prompt),
+                    "choices": list(choices),
+                    "default": str(default),
+                }
+            } if all(isinstance(c, str) for c in choices):
+                self.prompt_for_confirmation(prompt, choices, default)
             case {
                 "commands_entry": {
                     "bpnum": int(bpnum),
@@ -2930,6 +2966,20 @@ def prompt_for_repl_command(self, prompt):
             self.sockfile.flush()
             return
 
+    def prompt_for_confirmation(self, prompt, choices, default):
+        try:
+            with self.readline_completion(
+                functools.partial(self.complete_choices, choices=choices)
+            ):
+                reply = input(prompt).strip()
+        except (KeyboardInterrupt, EOFError):
+            print(flush=True)
+            reply = default
+
+        payload = {"confirmation_reply": reply}
+        self.sockfile.write((json.dumps(payload) + "\n").encode())
+        self.sockfile.flush()
+
     def complete(self, text, state):
         import readline
 
@@ -2964,6 +3014,15 @@ def complete(self, text, state):
         except IndexError:
             return None
 
+    def complete_choices(self, text, state, choices):
+        if state == 0:
+            self.completion_matches = [c for c in choices if c.startswith(text)]
+
+        try:
+            return choices[state]
+        except IndexError:
+            return None
+
 
 def _connect(host, port, frame):
     with closing(socket.create_connection((host, port))) as conn:

From baaf28af0f9a8d2f1bc4faeb13949a07e72f8ded Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Thu, 17 Apr 2025 20:15:54 -0400
Subject: [PATCH 13/40] Make message and error handle non-string args

The base class sometimes calls these methods with an exception object.
---
 Lib/pdb.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index fd7d23208bc811..21f58238e46f0d 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2540,11 +2540,13 @@ def _send(self, **kwargs) -> None:
             # return an empty string because the socket may be half-closed.
             self._write_failed = True
 
+    @typing.override
     def message(self, msg, end="\n"):
-        self._send(message=msg + end)
+        self._send(message=str(msg) + end)
 
+    @typing.override
     def error(self, msg):
-        self._send(error=msg)
+        self._send(error=str(msg))
 
     def _read_command(self) -> str:
         # Loop until we get a command for PDB or an 'interact' REPL.

From e61cc31fde7c7c82044388942747872f1d88cfc7 Mon Sep 17 00:00:00 2001
From: Pablo Galindo <pablogsal@gmail.com>
Date: Fri, 18 Apr 2025 01:21:11 +0100
Subject: [PATCH 14/40] Add some basic tests

---
 Lib/test/test_pyclbr.py     |   3 +-
 Lib/test/test_remote_pdb.py | 346 ++++++++++++++++++++++++++++++++++++
 2 files changed, 348 insertions(+), 1 deletion(-)
 create mode 100644 Lib/test/test_remote_pdb.py

diff --git a/Lib/test/test_pyclbr.py b/Lib/test/test_pyclbr.py
index a9ac13395a8fac..df05cd07d7e249 100644
--- a/Lib/test/test_pyclbr.py
+++ b/Lib/test/test_pyclbr.py
@@ -253,7 +253,8 @@ def test_others(self):
             cm(
                 'pdb',
                 # pyclbr does not handle elegantly `typing` or properties
-                ignore=('Union', '_ModuleTarget', '_ScriptTarget', '_ZipTarget', 'curframe_locals'),
+                ignore=('Union', '_ModuleTarget', '_ScriptTarget', '_ZipTarget', 'curframe_locals',
+                        '_InteractState'),
             )
         cm('pydoc', ignore=('input', 'output',))  # properties
 
diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
new file mode 100644
index 00000000000000..5698ad3681214c
--- /dev/null
+++ b/Lib/test/test_remote_pdb.py
@@ -0,0 +1,346 @@
+import io
+import json
+import os
+import signal
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import unittest
+import unittest.mock
+from contextlib import contextmanager
+from pathlib import Path
+from test.support import os_helper
+from test.support.os_helper import temp_dir, TESTFN, unlink
+from typing import Dict, List, Optional, Tuple, Union, Any
+
+import pdb
+from pdb import _RemotePdb, _PdbClient, _InteractState
+
+
+class MockSocketFile:
+    """Mock socket file for testing _RemotePdb without actual socket connections."""
+    
+    def __init__(self):
+        self.input_queue = []
+        self.output_buffer = []
+        
+    def write(self, data: bytes) -> None:
+        """Simulate write to socket."""
+        self.output_buffer.append(data)
+        
+    def flush(self) -> None:
+        """No-op flush implementation."""
+        pass
+        
+    def readline(self) -> bytes:
+        """Read a line from the prepared input queue."""
+        if not self.input_queue:
+            return b""
+        return self.input_queue.pop(0)
+        
+    def close(self) -> None:
+        """Close the mock socket file."""
+        pass
+        
+    def add_input(self, data: dict) -> None:
+        """Add input that will be returned by readline."""
+        self.input_queue.append(json.dumps(data).encode() + b"\n")
+        
+    def get_output(self) -> List[dict]:
+        """Get the output that was written by the object being tested."""
+        results = []
+        for data in self.output_buffer:
+            if isinstance(data, bytes) and data.endswith(b"\n"):
+                try:
+                    results.append(json.loads(data.decode().strip()))
+                except json.JSONDecodeError:
+                    pass  # Ignore non-JSON output
+        self.output_buffer = []
+        return results
+
+
+class RemotePdbTestCase(unittest.TestCase):
+    """Tests for the _RemotePdb class."""
+    
+    def setUp(self):
+        self.sockfile = MockSocketFile()
+        self.pdb = _RemotePdb(self.sockfile)
+        
+        # Create a frame for testing
+        self.test_globals = {'a': 1, 'b': 2, '__pdb_convenience_variables': {'x': 100}}
+        self.test_locals = {'c': 3, 'd': 4}
+        
+        # Create a simple test frame
+        frame_info = unittest.mock.Mock()
+        frame_info.f_globals = self.test_globals
+        frame_info.f_locals = self.test_locals
+        frame_info.f_lineno = 42
+        frame_info.f_code = unittest.mock.Mock()
+        frame_info.f_code.co_filename = "test_file.py"
+        frame_info.f_code.co_name = "test_function"
+        
+        self.pdb.curframe = frame_info
+        
+    def test_message_and_error(self):
+        """Test message and error methods send correct JSON."""
+        self.pdb.message("Test message")
+        self.pdb.error("Test error")
+        
+        outputs = self.sockfile.get_output()
+        self.assertEqual(len(outputs), 2)
+        self.assertEqual(outputs[0], {"message": "Test message\n"})
+        self.assertEqual(outputs[1], {"error": "Test error"})
+        
+    def test_read_command(self):
+        """Test reading commands from the socket."""
+        # Add test input
+        self.sockfile.add_input({"command": "help"})
+        
+        # Read the command
+        cmd = self.pdb._read_command()
+        self.assertEqual(cmd, "help")
+        
+    def test_read_command_EOF(self):
+        """Test reading EOF command."""
+        # Simulate socket closure
+        self.pdb._write_failed = True
+        cmd = self.pdb._read_command()
+        self.assertEqual(cmd, "EOF")
+        
+    def test_completion(self):
+        """Test handling completion requests."""
+        # Mock completenames to return specific values
+        with unittest.mock.patch.object(self.pdb, 'completenames', 
+                                       return_value=["continue", "clear"]):
+            
+            # Add a completion request
+            self.sockfile.add_input({
+                "completion": {
+                    "text": "c",
+                    "line": "c",
+                    "begidx": 0,
+                    "endidx": 1
+                }
+            })
+            
+            # Add a regular command to break the loop
+            self.sockfile.add_input({"command": "help"})
+            
+            # Read command - this should process the completion request first
+            cmd = self.pdb._read_command()
+            
+            # Verify completion response was sent
+            outputs = self.sockfile.get_output()
+            self.assertEqual(len(outputs), 1)
+            self.assertEqual(outputs[0], {"completion": ["continue", "clear"]})
+            
+            # The actual command should be returned
+            self.assertEqual(cmd, "help")
+            
+    def test_do_help(self):
+        """Test that do_help sends the help message."""
+        self.pdb.do_help("break")
+        
+        outputs = self.sockfile.get_output()
+        self.assertEqual(len(outputs), 1)
+        self.assertEqual(outputs[0], {"help": "break"})
+        
+    def test_interact_mode(self):
+        """Test interaction mode setup and execution."""
+        # First set up interact mode
+        self.pdb.do_interact("")
+        
+        # Verify _interact_state is properly initialized
+        self.assertIsNotNone(self.pdb._interact_state)
+        self.assertIsInstance(self.pdb._interact_state, _InteractState)
+        
+        # Test running code in interact mode
+        with unittest.mock.patch.object(self.pdb, '_error_exc') as mock_error:
+            self.pdb._run_in_python_repl("print('test')")
+            mock_error.assert_not_called()
+            
+            # Test with syntax error
+            self.pdb._run_in_python_repl("if:")
+            mock_error.assert_called_once()
+            
+    def test_do_commands(self):
+        """Test handling breakpoint commands."""
+        # Mock get_bpbynumber
+        with unittest.mock.patch.object(self.pdb, 'get_bpbynumber'):
+            # Test command entry mode initiation
+            self.pdb.do_commands("1")
+            
+            outputs = self.sockfile.get_output()
+            self.assertEqual(len(outputs), 1)
+            self.assertIn("commands_entry", outputs[0])
+            self.assertEqual(outputs[0]["commands_entry"]["bpnum"], 1)
+            
+            # Test with commands
+            self.pdb.do_commands("1\nsilent\nprint('hi')\nend")
+            
+            # Should have set up the commands for bpnum 1
+            self.assertEqual(self.pdb.commands_bnum, 1)
+            self.assertIn(1, self.pdb.commands)
+            self.assertEqual(len(self.pdb.commands[1]), 2)  # silent and print
+            
+    def test_detach(self):
+        """Test the detach method."""
+        with unittest.mock.patch.object(self.sockfile, 'close') as mock_close:
+            self.pdb.detach()
+            mock_close.assert_called_once()
+            self.assertFalse(self.pdb.quitting)
+
+    def test_cmdloop(self):
+        """Test the command loop with various commands."""
+        # Mock onecmd to track command execution
+        with unittest.mock.patch.object(self.pdb, 'onecmd', return_value=False) as mock_onecmd:
+            # Add commands to the queue
+            self.pdb.cmdqueue = ['help', 'list']
+            
+            # Add a command from the socket for when cmdqueue is empty
+            self.sockfile.add_input({"command": "next"})
+            
+            # Add a second command to break the loop
+            self.sockfile.add_input({"command": "quit"})
+            
+            # Configure onecmd to exit the loop on "quit"
+            def side_effect(line):
+                return line == 'quit'
+            mock_onecmd.side_effect = side_effect
+            
+            # Run the command loop
+            self.pdb.quitting = False # Set this by hand because we don't want to really call set_trace()
+            self.pdb.cmdloop()
+            
+            # Should have processed 4 commands: 2 from cmdqueue, 2 from socket
+            self.assertEqual(mock_onecmd.call_count, 4)
+            mock_onecmd.assert_any_call('help')
+            mock_onecmd.assert_any_call('list')
+            mock_onecmd.assert_any_call('next')
+            mock_onecmd.assert_any_call('quit')
+            
+            # Check if prompt was sent to client
+            outputs = self.sockfile.get_output()
+            prompts = [o for o in outputs if 'prompt' in o]
+            self.assertEqual(len(prompts), 2)  # Should have sent 2 prompts
+            
+
+class PdbConnectTestCase(unittest.TestCase):
+    """Tests for the _connect mechanism using direct socket communication."""
+    
+    def setUp(self):
+        # Create a server socket that will wait for the debugger to connect
+        self.server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.server_sock.bind(('127.0.0.1', 0))  # Let OS assign port
+        self.server_sock.listen(1)
+        self.port = self.server_sock.getsockname()[1]
+        
+        # Create a file for subprocess script
+        self.script_path = TESTFN + "_connect_test.py"
+        with open(self.script_path, 'w') as f:
+            f.write(f"""
+import pdb
+import sys
+import time
+
+def connect_to_debugger():
+    # Create a frame to debug
+    def dummy_function():
+        x = 42
+        # Call connect to establish connection with the test server
+        frame = sys._getframe()  # Get the current frame
+        pdb._connect('127.0.0.1', {self.port}, frame)
+        return x  # This line should not be reached in debugging
+    
+    return dummy_function()
+
+result = connect_to_debugger()
+print(f"Function returned: {{result}}")
+""")
+    
+    def tearDown(self):
+        self.server_sock.close()
+        try:
+            unlink(self.script_path)
+        except OSError:
+            pass
+    
+    def test_connect_and_basic_commands(self):
+        """Test connecting to a remote debugger and sending basic commands."""
+        # Start the subprocess that will connect to our socket
+        with subprocess.Popen(
+            [sys.executable, self.script_path],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True
+        ) as process:
+            # Accept the connection from the subprocess
+            client_sock, _ = self.server_sock.accept()
+            client_file = client_sock.makefile('rwb')
+            self.addCleanup(client_file.close)
+            self.addCleanup(client_sock.close)
+            
+            # We should receive initial data from the debugger
+            data = client_file.readline()
+            initial_data = json.loads(data.decode())
+            self.assertIn('message', initial_data)
+            self.assertIn('pdb._connect', initial_data['message'])
+
+            # First, look for command_list message
+            data = client_file.readline()
+            command_list = json.loads(data.decode())
+            self.assertIn('command_list', command_list)
+            
+            # Then, look for the first prompt
+            data = client_file.readline()
+            prompt_data = json.loads(data.decode())
+            self.assertIn('prompt', prompt_data)
+            self.assertEqual(prompt_data['mode'], 'pdb')
+            
+            # Send 'bt' (backtrace) command
+            client_file.write(json.dumps({"command": "bt"}).encode() + b"\n")
+            client_file.flush()
+            
+            # Check for response - we should get some stack frames
+            # We may get multiple messages so we need to read until we get a new prompt
+            got_stack_info = False
+            text_msg = []
+            while True:
+                data = client_file.readline()
+                if not data:
+                    break
+                    
+                msg = json.loads(data.decode())
+                if 'message' in msg and 'connect_to_debugger' in msg['message']:
+                    got_stack_info = True
+                    text_msg.append(msg['message'])
+                
+                if 'prompt' in msg:
+                    break
+                    
+            expected_stacks = [
+                "<module>",
+                "connect_to_debugger",
+            ]
+
+            for stack, msg in zip(expected_stacks, text_msg, strict=True):
+                self.assertIn(stack, msg)
+
+            self.assertTrue(got_stack_info, "Should have received stack trace information")
+            
+            # Send 'c' (continue) command to let the program finish
+            client_file.write(json.dumps({"command": "c"}).encode() + b"\n")
+            client_file.flush()
+            
+            # Wait for process to finish
+            stdout, _ = process.communicate(timeout=5)
+            
+            # Check if we got the expected output
+            self.assertIn("Function returned: 42", stdout)
+            self.assertEqual(process.returncode, 0)
+                
+
+if __name__ == "__main__":
+    unittest.main()

From 5d59ce17da1a90268bd51da4c632a5ea512b391a Mon Sep 17 00:00:00 2001
From: Pablo Galindo <pablogsal@gmail.com>
Date: Fri, 18 Apr 2025 01:22:45 +0100
Subject: [PATCH 15/40] Don't use deprecated method

---
 Lib/pdb.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 21f58238e46f0d..e8d1d5dc0313ff 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2682,7 +2682,7 @@ def do_interact(self, arg):
         # to start treating all input as Python commands, not PDB ones.
         self._interact_state = _InteractState(
             compiler=codeop.CommandCompiler(),
-            ns={**self.curframe.f_globals, **self.curframe_locals},
+            ns={**self.curframe.f_globals, **self.curframe.f_locals},
         )
 
     def do_commands(self, arg):

From 09adb2b9b0aa54db6ff8a3c3dbb98c53a09c7d9d Mon Sep 17 00:00:00 2001
From: Matt Wozniski <godlygeek@gmail.com>
Date: Fri, 18 Apr 2025 12:34:37 -0400
Subject: [PATCH 16/40] Try to prevent a PermissionError on Windows

If `NamedTemporaryFile.__exit__` tries to delete the connect script
before the remote process has closed it, a `PermissionError` is raised
on Windows. Unfortunately there is no synchronization that we can use to
ensure that the file is closed before we try to delete it. This makes
the race less likely to occur by making the `with` block contain more
code, so that more time passes before we attempt to delete the file.

Co-authored-by: Chris Eibl <138194463+chris-eibl@users.noreply.github.com>
---
 Lib/pdb.py | 32 ++++++++++++++++----------------
 1 file changed, 16 insertions(+), 16 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index e8d1d5dc0313ff..bec5b30a8bcb55 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -3044,31 +3044,31 @@ def attach(pid):
     with closing(socket.create_server(("localhost", 0))) as server:
         port = server.getsockname()[1]
 
-        with tempfile.NamedTemporaryFile("w", delete_on_close=False) as script:
-            script.write(
+        with tempfile.NamedTemporaryFile("w", delete_on_close=False) as connect_script:
+            connect_script.write(
                 f'import pdb, sys\n'
                 f'pdb._connect("localhost", {port}, sys._getframe().f_back)\n'
             )
-            script.close()
-            sys.remote_exec(pid, script.name)
+            connect_script.close()
+            sys.remote_exec(pid, connect_script.name)
 
             # TODO Add a timeout? Or don't bother since the user can ^C?
             client_sock, _ = server.accept()
 
-    with closing(client_sock):
-        sockfile = client_sock.makefile("rwb")
+            with closing(client_sock):
+                sockfile = client_sock.makefile("rwb")
 
-    with closing(sockfile):
-        with tempfile.NamedTemporaryFile("w", delete_on_close=False) as script:
-            script.write(
-                'import pdb, sys\n'
-                'if inst := pdb.Pdb._last_pdb_instance:\n'
-                '    inst.set_step()\n'
-                '    inst.set_trace(sys._getframe(1))\n'
-            )
-            script.close()
+            with closing(sockfile):
+                with tempfile.NamedTemporaryFile("w", delete_on_close=False) as interrupt_script:
+                    interrupt_script.write(
+                        'import pdb, sys\n'
+                        'if inst := pdb.Pdb._last_pdb_instance:\n'
+                        '    inst.set_step()\n'
+                        '    inst.set_trace(sys._getframe(1))\n'
+                    )
+                    interrupt_script.close()
 
-            _PdbClient(pid, sockfile, script.name).cmdloop()
+                    _PdbClient(pid, sockfile, interrupt_script.name).cmdloop()
 
 
 # Post-Mortem interface

From 600aa0504787875498faad1bee5b80697babd8cf Mon Sep 17 00:00:00 2001
From: Matt Wozniski <godlygeek@gmail.com>
Date: Sun, 20 Apr 2025 00:29:39 -0400
Subject: [PATCH 17/40] Address review comments

---
 Lib/pdb.py                  | 372 +++++++++++++-----------------------
 Lib/test/test_remote_pdb.py | 183 ++++++++++--------
 2 files changed, 234 insertions(+), 321 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index bec5b30a8bcb55..423396fc5c6db8 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -89,7 +89,6 @@
 import tempfile
 import textwrap
 import tokenize
-import functools
 import itertools
 import traceback
 import linecache
@@ -927,7 +926,7 @@ def handle_command_def(self, line):
         if cmd == 'end':
             return True  # end of cmd list
         elif cmd == 'EOF':
-            print('')
+            self.message('')
             return True  # end of cmd list
         cmdlist = self.commands[self.commands_bnum]
         if cmd == 'silent':
@@ -2542,49 +2541,64 @@ def _send(self, **kwargs) -> None:
 
     @typing.override
     def message(self, msg, end="\n"):
-        self._send(message=str(msg) + end)
+        self._send(message=str(msg) + end, type="info")
 
     @typing.override
     def error(self, msg):
-        self._send(error=str(msg))
-
-    def _read_command(self) -> str:
-        # Loop until we get a command for PDB or an 'interact' REPL.
-        # Process out-of-band completion messages without returning.
+        self._send(message=str(msg), type="error")
+
+    def _get_input(self, prompt, state) -> str:
+        # Before displaying a (Pdb) prompt, send the list of PDB commands
+        # unless we've already sent an up-to-date list.
+        if state == "pdb" and not self._command_name_cache:
+            self._command_name_cache = self.completenames("", "", 0, 0)
+            self._send(command_list=self._command_name_cache)
+        self._send(prompt=prompt, state=state)
+        return self._read_reply()
+
+    def _read_reply(self):
+        # Loop until we get a 'reply' or 'signal' from the client,
+        # processing out-of-band 'complete' requests as they arrive.
         while True:
             if self._write_failed:
-                return "EOF"
+                raise EOFError
 
-            payload = self._sockfile.readline()
-            if not payload:
-                return "EOF"
+            msg = self._sockfile.readline()
+            if not msg:
+                raise EOFError
 
             try:
-                match json.loads(payload):
-                    case {"command": str(line)}:
-                        # Interact mode has been cancelled client-side,
-                        # likely by a ^D EOF at the prompt.
-                        self._interact_state = None
-                        return line
-                    case {"interact": str(lines)}:
-                        if self._interact_state:
-                            return lines
-                        # Otherwise, fall through to report an error and loop.
-                    case {
-                        "completion": {
-                            "text": str(text),
-                            "line": str(line),
-                            "begidx": int(begidx),
-                            "endidx": int(endidx),
-                        }
-                    }:
-                        items = self._complete_any(text, line, begidx, endidx)
-                        self._send(completion=items)
-                        continue
+                payload = json.loads(msg)
             except json.JSONDecodeError:
-                pass
-            # Invalid JSON, or doesn't meet the schema, or wrong PDB state.
-            self.error(f"Ignoring invalid remote command: {payload}")
+                self.error(f"Disconnecting: client sent invalid JSON {msg}")
+                raise EOFError
+
+            match payload:
+                case {"reply": str(reply)}:
+                    return reply
+                case {"signal": str(signal)}:
+                    if signal == "INT":
+                        raise KeyboardInterrupt
+                    elif signal != "EOF":
+                        self.error(
+                            f"Received unrecognized signal: {signal}"
+                        )
+                        # Our best hope of recovering is to pretend we
+                        # got an EOF to exit whatever mode we're in.
+                    raise EOFError
+                case {
+                    "complete": {
+                        "text": str(text),
+                        "line": str(line),
+                        "begidx": int(begidx),
+                        "endidx": int(endidx),
+                    }
+                }:
+                    items = self._complete_any(text, line, begidx, endidx)
+                    self._send(completions=items)
+                    continue
+            # Valid JSON, but doesn't meet the schema.
+            self.error(f"Ignoring invalid message from client: {msg}")
 
     def _complete_any(self, text, line, begidx, endidx):
         if begidx == 0:
@@ -2605,19 +2619,28 @@ def cmdloop(self, intro=None):
             self.message(str(self.intro))
         stop = None
         while not stop:
-            if self.cmdqueue:
-                line = self.cmdqueue.pop(0)
-            else:
-                if not self._command_name_cache:
-                    self._command_name_cache = self.completenames("", "", 0, 0)
-                    self._send(command_list=self._command_name_cache)
-
-                mode = "interact" if self._interact_state else "pdb"
-                self._send(prompt=self.prompt, mode=mode)
-                line = self._read_command()
-                if self._interact_state is not None:
-                    self._run_in_python_repl(line)
-                    continue
+            if self._interact_state is not None:
+                try:
+                    reply = self._get_input(prompt=">>> ", state="interact")
+                except KeyboardInterrupt:
+                    # Match how KeyboardInterrupt is handled in a REPL
+                    self.message("\nKeyboardInterrupt")
+                except EOFError:
+                    self.message("\n*exit from pdb interact command*")
+                    self._interact_state = None
+                else:
+                    self._run_in_python_repl(reply)
+                continue
+
+            if not self.cmdqueue:
+                try:
+                    reply = self._get_input(prompt=self.prompt, state="pdb")
+                except EOFError:
+                    reply = "EOF"
+
+                self.cmdqueue.append(reply)
+
+            line = self.cmdqueue.pop(0)
             line = self.precmd(line)
             stop = self.onecmd(line)
             stop = self.postcmd(stop, line)
@@ -2640,6 +2663,12 @@ def detach(self):
                 # close() can fail if the connection was broken unexpectedly.
                 pass
 
+    def do_debug(self, arg):
+        # Clear our cached list of valid commands; the recursive debugger might
+        # send its own differing list, and so ours needs to be re-sent.
+        self._command_name_cache = []
+        return super().do_debug(arg)
+
     def do_alias(self, arg):
         # Clear our cached list of valid commands; one might be added.
         self._command_name_cache = []
@@ -2680,80 +2709,22 @@ def _run_in_python_repl(self, lines):
     def do_interact(self, arg):
         # Prepare to run 'interact' mode code blocks, and trigger the client
         # to start treating all input as Python commands, not PDB ones.
+        self.message("*pdb interact start*")
         self._interact_state = _InteractState(
             compiler=codeop.CommandCompiler(),
             ns={**self.curframe.f_globals, **self.curframe.f_locals},
         )
 
-    def do_commands(self, arg):
-        # Called with only one line in 'arg' to start collecting commands.
-        # Once the client has gathered up the full list of commands to apply,
-        # do_commands gets called again with multiple lines of input in args.
-        arg, *commands = arg.split("\n")
-        if not arg:
-            bnum = len(bdb.Breakpoint.bpbynumber) - 1
-        else:
-            try:
-                bnum = int(arg)
-            except ValueError:
-                self._print_invalid_arg(arg)
-                return
-
-        try:
-            self.get_bpbynumber(bnum)
-        except ValueError as err:
-            self.error("cannot set commands: %s" % err)
-            return
-
-        if not commands:
-            # We've only received the first line so far.
-            # Have the client enter command entry mode.
-
-            # fmt: off
-            end_cmds = [
-                "c", "cont", "continue",
-                "s", "step",
-                "n", "next",
-                "r", "return",
-                "q", "quit", "exit",
-                "j", "jump",
-            ]
-            # fmt: on
-
-            end_cmds += ["end"]  # pseudo-command
-            self._send(commands_entry={"bpnum": bnum, "terminators": end_cmds})
-            return
-
-        self.commands_bnum = bnum
-        self.commands[bnum] = []
-
-        for line in commands:
-            if self.handle_command_def(line):
-                break
-
     @typing.override
     def _create_recursive_debugger(self):
         return _RemotePdb(self._sockfile, owns_sockfile=False)
 
     @typing.override
     def _prompt_for_confirmation(self, prompt, choices, default):
-        self._send(
-            confirm={"prompt": prompt, "choices": choices, "default": default}
-        )
-        payload = self._sockfile.readline()
-
-        if not payload:
-            return default
-
         try:
-            match json.loads(payload):
-                case {"confirmation_reply": str(reply)}:
-                    return reply
-        except json.JSONDecodeError:
-            pass
-
-        self.error(f"Ignoring unexpected remote message: {payload}")
-        return default
+            return self._get_input(prompt=prompt, state="confirm")
+        except (EOFError, KeyboardInterrupt):
+            return default
 
     def do_run(self, arg):
         self.error("remote PDB cannot restart the program")
@@ -2766,6 +2737,8 @@ def _error_exc(self):
             # If we get a SystemExit in 'interact' mode, exit the REPL.
             self._interact_state = None
         super()._error_exc()
+        if isinstance(exc, SystemExit):
+            self.message("*exit from pdb interact command*")
 
     def default(self, line):
         # Unlike Pdb, don't prompt for more lines of a multi-line command.
@@ -2787,54 +2760,38 @@ def __init__(self, pid, sockfile, interrupt_script):
         self.pdb_instance = Pdb()
         self.pdb_commands = set()
         self.completion_matches = []
-        self.interact_mode = False
-        self.commands_mode = False
-        self.bpnum = 0
-        self.command_list_terminators = []
+        self.state = "dumb"
 
     def read_command(self, prompt):
-        continue_prompt = "...".ljust(len(prompt))
+        reply = input(prompt)
 
-        if self.interact_mode:
-            # Python REPL mode
-            try:
-                line = input(">>> ")
-            except EOFError:
-                print("\n*exit from pdb interact command*")
-                self.interact_mode = False
-            else:
-                continue_prompt = "...".ljust(len(">>> "))
+        if self.state == "dumb":
+            # No logic applied whatsoever, just pass the raw reply back.
+            return reply
 
-        if not self.interact_mode:
+        prefix = ""
+        if self.state == "pdb":
             # PDB command entry mode
-            line = input(prompt)
-            cmd = self.pdb_instance.parseline(line)[0]
-            if cmd in self.pdb_commands or line.strip() == "":
-                # Recognized PDB command, or repeating last command
-                return line
+            cmd = self.pdb_instance.parseline(reply)[0]
+            if cmd in self.pdb_commands or reply.strip() == "":
+                # Recognized PDB command, or blank line repeating last command
+                return reply
 
             # Otherwise, explicit or implicit exec command
-            if line.startswith("!"):
-                line = line[1:].lstrip()
-
-        # Ensure the remote won't try to use this as a PDB command.
-        prefix = "!" if not self.interact_mode else ""
+            if reply.startswith("!"):
+                prefix = "!"
+                reply = reply.removeprefix(prefix).lstrip()
 
-        if (
-            codeop.compile_command(line + "\n", "<stdin>", "single")
-            is not None
-        ):
+        if codeop.compile_command(reply + "\n", "<stdin>", "single") is not None:
             # Valid single-line statement
-            return prefix + line
+            return prefix + reply
 
         # Otherwise, valid first line of a multi-line statement
         continue_prompt = "...".ljust(len(prompt))
-        buffer = line
+        while codeop.compile_command(reply, "<stdin>", "single") is None:
+            reply += "\n" + input(continue_prompt)
 
-        while codeop.compile_command(buffer, "<stdin>", "single") is None:
-            buffer += "\n" + input(continue_prompt)
-
-        return prefix + buffer
+        return prefix + reply
 
     @contextmanager
     def readline_completion(self, completer):
@@ -2887,105 +2844,50 @@ def send_interrupt(self):
 
     def process_payload(self, payload):
         match payload:
-            case {"command_list": command_list}:
+            case {
+                "command_list": command_list
+            } if all(isinstance(c, str) for c in command_list):
                 self.pdb_commands = set(command_list)
-            case {"message": str(msg)}:
-                print(msg, end="", flush=True)
-            case {"error": str(msg)}:
-                print("***", msg, flush=True)
+            case {"message": str(msg), "type": str(msg_type)}:
+                if msg_type == "error":
+                    print("***", msg, flush=True)
+                else:
+                    print(msg, end="", flush=True)
             case {"help": str(arg)}:
                 self.pdb_instance.do_help(arg)
-            case {"prompt": str(prompt), "mode": str(mode)}:
-                if mode == "interact":
-                    self.interact_mode = True
-                elif self.interact_mode and mode != "interact":
-                    print("*exit from pdb interact command*")
-                    self.interact_mode = False
-
-                if self.commands_mode:
-                    self.prompt_for_breakpoint_command_list("(com) ")
-                else:
-                    self.prompt_for_repl_command(prompt)
-            case {
-                "confirm": {
-                    "prompt": str(prompt),
-                    "choices": list(choices),
-                    "default": str(default),
-                }
-            } if all(isinstance(c, str) for c in choices):
-                self.prompt_for_confirmation(prompt, choices, default)
-            case {
-                "commands_entry": {
-                    "bpnum": int(bpnum),
-                    "terminators": list(command_list_terminators),
-                }
-            }:
-                self.bpnum = bpnum
-                self.command_list_terminators = command_list_terminators
-                self.commands_mode = True
+            case {"prompt": str(prompt), "state": str(state)}:
+                if state not in ("pdb", "interact"):
+                    state = "dumb"
+                self.state = state
+                self.prompt_for_reply(prompt)
             case _:
                 raise RuntimeError(f"Unrecognized payload {payload}")
 
-    def prompt_for_breakpoint_command_list(self, prompt):
-        parts = [f"commands {self.bpnum}"]
+    def prompt_for_reply(self, prompt):
         while True:
             try:
-                line = input(prompt).strip()
-                parts.append(line)
-                cmd = self.pdb_instance.parseline(line)[0]
-                if cmd in self.command_list_terminators:
-                    break
-            except (KeyboardInterrupt, EOFError):
-                print(flush=True)
-                print("command definition aborted, old commands restored")
-                break
-            finally:
-                self.commands_mode = False
-
-        command = "\n".join(parts)
-        self.sockfile.write((json.dumps({"command": command}) + "\n").encode())
-        self.sockfile.flush()
-
-    def prompt_for_repl_command(self, prompt):
-        while True:
-            try:
-                command = self.read_command(prompt)
+                payload = {"reply": self.read_command(prompt)}
             except EOFError:
-                command = "EOF"
+                payload = {"signal": "EOF"}
             except KeyboardInterrupt:
-                print(flush=True)
-                continue
+                payload = {"signal": "INT"}
             except Exception as exc:
                 msg = traceback.format_exception_only(exc)[-1].strip()
                 print("***", msg, flush=True)
                 continue
 
-            if self.interact_mode:
-                payload = {"interact": command}
-            else:
-                payload = {"command": command}
             self.sockfile.write((json.dumps(payload) + "\n").encode())
             self.sockfile.flush()
             return
 
-    def prompt_for_confirmation(self, prompt, choices, default):
-        try:
-            with self.readline_completion(
-                functools.partial(self.complete_choices, choices=choices)
-            ):
-                reply = input(prompt).strip()
-        except (KeyboardInterrupt, EOFError):
-            print(flush=True)
-            reply = default
-
-        payload = {"confirmation_reply": reply}
-        self.sockfile.write((json.dumps(payload) + "\n").encode())
-        self.sockfile.flush()
-
     def complete(self, text, state):
         import readline
 
         if state == 0:
+            self.completion_matches = []
+            if self.state not in ("pdb", "interact"):
+                return None
+
             origline = readline.get_line_buffer()
             line = origline.lstrip()
             stripped = len(origline) - len(line)
@@ -2993,38 +2895,36 @@ def complete(self, text, state):
             endidx = readline.get_endidx() - stripped
 
             msg = {
-                "completion": {
+                "complete": {
                     "text": text,
                     "line": line,
                     "begidx": begidx,
                     "endidx": endidx,
                 }
             }
-            self.sockfile.write((json.dumps(msg) + "\n").encode())
-            self.sockfile.flush()
+
+            try:
+                self.sockfile.write((json.dumps(msg) + "\n").encode())
+                self.sockfile.flush()
+            except OSError:
+                return None
 
             payload = self.sockfile.readline()
+            if not payload:
+                return None
+
             payload = json.loads(payload)
-            if "completion" not in payload:
+            if "completions" not in payload:
                 raise RuntimeError(
                     f"Failed to get valid completions. Got: {payload}"
                 )
 
-            self.completion_matches = payload["completion"]
+            self.completion_matches = payload["completions"]
         try:
             return self.completion_matches[state]
         except IndexError:
             return None
 
-    def complete_choices(self, text, state, choices):
-        if state == 0:
-            self.completion_matches = [c for c in choices if c.startswith(text)]
-
-        try:
-            return choices[state]
-        except IndexError:
-            return None
-
 
 def _connect(host, port, frame):
     with closing(socket.create_connection((host, port))) as conn:
diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 5698ad3681214c..9c84c9031863ee 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -21,33 +21,33 @@
 
 class MockSocketFile:
     """Mock socket file for testing _RemotePdb without actual socket connections."""
-    
+
     def __init__(self):
         self.input_queue = []
         self.output_buffer = []
-        
+
     def write(self, data: bytes) -> None:
         """Simulate write to socket."""
         self.output_buffer.append(data)
-        
+
     def flush(self) -> None:
         """No-op flush implementation."""
         pass
-        
+
     def readline(self) -> bytes:
         """Read a line from the prepared input queue."""
         if not self.input_queue:
             return b""
         return self.input_queue.pop(0)
-        
+
     def close(self) -> None:
         """Close the mock socket file."""
         pass
-        
+
     def add_input(self, data: dict) -> None:
         """Add input that will be returned by readline."""
         self.input_queue.append(json.dumps(data).encode() + b"\n")
-        
+
     def get_output(self) -> List[dict]:
         """Get the output that was written by the object being tested."""
         results = []
@@ -63,15 +63,19 @@ def get_output(self) -> List[dict]:
 
 class RemotePdbTestCase(unittest.TestCase):
     """Tests for the _RemotePdb class."""
-    
+
     def setUp(self):
         self.sockfile = MockSocketFile()
         self.pdb = _RemotePdb(self.sockfile)
-        
+
+        # Mock some Bdb attributes that are lazily created when tracing starts
+        self.pdb.botframe = None
+        self.pdb.quitting = False
+
         # Create a frame for testing
         self.test_globals = {'a': 1, 'b': 2, '__pdb_convenience_variables': {'x': 100}}
         self.test_locals = {'c': 3, 'd': 4}
-        
+
         # Create a simple test frame
         frame_info = unittest.mock.Mock()
         frame_info.f_globals = self.test_globals
@@ -80,111 +84,120 @@ def setUp(self):
         frame_info.f_code = unittest.mock.Mock()
         frame_info.f_code.co_filename = "test_file.py"
         frame_info.f_code.co_name = "test_function"
-        
+
         self.pdb.curframe = frame_info
-        
+
     def test_message_and_error(self):
         """Test message and error methods send correct JSON."""
         self.pdb.message("Test message")
         self.pdb.error("Test error")
-        
+
         outputs = self.sockfile.get_output()
         self.assertEqual(len(outputs), 2)
-        self.assertEqual(outputs[0], {"message": "Test message\n"})
-        self.assertEqual(outputs[1], {"error": "Test error"})
-        
+        self.assertEqual(outputs[0], {"message": "Test message\n", "type": "info"})
+        self.assertEqual(outputs[1], {"message": "Test error", "type": "error"})
+
     def test_read_command(self):
         """Test reading commands from the socket."""
         # Add test input
-        self.sockfile.add_input({"command": "help"})
-        
+        self.sockfile.add_input({"reply": "help"})
+
         # Read the command
-        cmd = self.pdb._read_command()
+        cmd = self.pdb._read_reply()
         self.assertEqual(cmd, "help")
-        
+
     def test_read_command_EOF(self):
         """Test reading EOF command."""
         # Simulate socket closure
         self.pdb._write_failed = True
-        cmd = self.pdb._read_command()
-        self.assertEqual(cmd, "EOF")
-        
+        with self.assertRaises(EOFError):
+            self.pdb._read_reply()
+
     def test_completion(self):
         """Test handling completion requests."""
         # Mock completenames to return specific values
-        with unittest.mock.patch.object(self.pdb, 'completenames', 
+        with unittest.mock.patch.object(self.pdb, 'completenames',
                                        return_value=["continue", "clear"]):
-            
+
             # Add a completion request
             self.sockfile.add_input({
-                "completion": {
+                "complete": {
                     "text": "c",
                     "line": "c",
                     "begidx": 0,
                     "endidx": 1
                 }
             })
-            
+
             # Add a regular command to break the loop
-            self.sockfile.add_input({"command": "help"})
-            
+            self.sockfile.add_input({"reply": "help"})
+
             # Read command - this should process the completion request first
-            cmd = self.pdb._read_command()
-            
+            cmd = self.pdb._read_reply()
+
             # Verify completion response was sent
             outputs = self.sockfile.get_output()
             self.assertEqual(len(outputs), 1)
-            self.assertEqual(outputs[0], {"completion": ["continue", "clear"]})
-            
+            self.assertEqual(outputs[0], {"completions": ["continue", "clear"]})
+
             # The actual command should be returned
             self.assertEqual(cmd, "help")
-            
+
     def test_do_help(self):
         """Test that do_help sends the help message."""
         self.pdb.do_help("break")
-        
+
         outputs = self.sockfile.get_output()
         self.assertEqual(len(outputs), 1)
         self.assertEqual(outputs[0], {"help": "break"})
-        
+
     def test_interact_mode(self):
         """Test interaction mode setup and execution."""
         # First set up interact mode
         self.pdb.do_interact("")
-        
+
         # Verify _interact_state is properly initialized
         self.assertIsNotNone(self.pdb._interact_state)
         self.assertIsInstance(self.pdb._interact_state, _InteractState)
-        
+
         # Test running code in interact mode
         with unittest.mock.patch.object(self.pdb, '_error_exc') as mock_error:
             self.pdb._run_in_python_repl("print('test')")
             mock_error.assert_not_called()
-            
+
             # Test with syntax error
             self.pdb._run_in_python_repl("if:")
             mock_error.assert_called_once()
-            
-    def test_do_commands(self):
-        """Test handling breakpoint commands."""
+
+    def test_registering_commands(self):
+        """Test registering breakpoint commands."""
         # Mock get_bpbynumber
         with unittest.mock.patch.object(self.pdb, 'get_bpbynumber'):
-            # Test command entry mode initiation
-            self.pdb.do_commands("1")
-            
+            # Queue up some input to send
+            self.sockfile.add_input({"reply": "commands 1"})
+            self.sockfile.add_input({"reply": "silent"})
+            self.sockfile.add_input({"reply": "print('hi')"})
+            self.sockfile.add_input({"reply": "end"})
+            self.sockfile.add_input({"signal": "EOF"})
+
+            # Run the PDB command loop
+            self.pdb.cmdloop()
+
             outputs = self.sockfile.get_output()
-            self.assertEqual(len(outputs), 1)
-            self.assertIn("commands_entry", outputs[0])
-            self.assertEqual(outputs[0]["commands_entry"]["bpnum"], 1)
-            
-            # Test with commands
-            self.pdb.do_commands("1\nsilent\nprint('hi')\nend")
-            
-            # Should have set up the commands for bpnum 1
-            self.assertEqual(self.pdb.commands_bnum, 1)
-            self.assertIn(1, self.pdb.commands)
-            self.assertEqual(len(self.pdb.commands[1]), 2)  # silent and print
-            
+            self.assertIn('command_list', outputs[0])
+            self.assertEqual(outputs[1], {"prompt": "(Pdb) ", "state": "pdb"})
+            self.assertEqual(outputs[2], {"prompt": "(com) ", "state": "pdb"})
+            self.assertEqual(outputs[3], {"prompt": "(com) ", "state": "pdb"})
+            self.assertEqual(outputs[4], {"prompt": "(com) ", "state": "pdb"})
+            self.assertEqual(outputs[5], {"prompt": "(Pdb) ", "state": "pdb"})
+            self.assertEqual(outputs[6], {"message": "\n", "type": "info"})
+            self.assertEqual(len(outputs), 7)
+
+            self.assertEqual(
+                self.pdb.commands[1],
+                ["_pdbcmd_silence_frame_status", "print('hi')"],
+            )
+
     def test_detach(self):
         """Test the detach method."""
         with unittest.mock.patch.object(self.sockfile, 'close') as mock_close:
@@ -198,45 +211,45 @@ def test_cmdloop(self):
         with unittest.mock.patch.object(self.pdb, 'onecmd', return_value=False) as mock_onecmd:
             # Add commands to the queue
             self.pdb.cmdqueue = ['help', 'list']
-            
+
             # Add a command from the socket for when cmdqueue is empty
-            self.sockfile.add_input({"command": "next"})
-            
+            self.sockfile.add_input({"reply": "next"})
+
             # Add a second command to break the loop
-            self.sockfile.add_input({"command": "quit"})
-            
+            self.sockfile.add_input({"reply": "quit"})
+
             # Configure onecmd to exit the loop on "quit"
             def side_effect(line):
                 return line == 'quit'
             mock_onecmd.side_effect = side_effect
-            
+
             # Run the command loop
             self.pdb.quitting = False # Set this by hand because we don't want to really call set_trace()
             self.pdb.cmdloop()
-            
+
             # Should have processed 4 commands: 2 from cmdqueue, 2 from socket
             self.assertEqual(mock_onecmd.call_count, 4)
             mock_onecmd.assert_any_call('help')
             mock_onecmd.assert_any_call('list')
             mock_onecmd.assert_any_call('next')
             mock_onecmd.assert_any_call('quit')
-            
+
             # Check if prompt was sent to client
             outputs = self.sockfile.get_output()
             prompts = [o for o in outputs if 'prompt' in o]
             self.assertEqual(len(prompts), 2)  # Should have sent 2 prompts
-            
+
 
 class PdbConnectTestCase(unittest.TestCase):
     """Tests for the _connect mechanism using direct socket communication."""
-    
+
     def setUp(self):
         # Create a server socket that will wait for the debugger to connect
         self.server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         self.server_sock.bind(('127.0.0.1', 0))  # Let OS assign port
         self.server_sock.listen(1)
         self.port = self.server_sock.getsockname()[1]
-        
+
         # Create a file for subprocess script
         self.script_path = TESTFN + "_connect_test.py"
         with open(self.script_path, 'w') as f:
@@ -253,20 +266,20 @@ def dummy_function():
         frame = sys._getframe()  # Get the current frame
         pdb._connect('127.0.0.1', {self.port}, frame)
         return x  # This line should not be reached in debugging
-    
+
     return dummy_function()
 
 result = connect_to_debugger()
 print(f"Function returned: {{result}}")
 """)
-    
+
     def tearDown(self):
         self.server_sock.close()
         try:
             unlink(self.script_path)
         except OSError:
             pass
-    
+
     def test_connect_and_basic_commands(self):
         """Test connecting to a remote debugger and sending basic commands."""
         # Start the subprocess that will connect to our socket
@@ -281,7 +294,7 @@ def test_connect_and_basic_commands(self):
             client_file = client_sock.makefile('rwb')
             self.addCleanup(client_file.close)
             self.addCleanup(client_sock.close)
-            
+
             # We should receive initial data from the debugger
             data = client_file.readline()
             initial_data = json.loads(data.decode())
@@ -292,17 +305,17 @@ def test_connect_and_basic_commands(self):
             data = client_file.readline()
             command_list = json.loads(data.decode())
             self.assertIn('command_list', command_list)
-            
+
             # Then, look for the first prompt
             data = client_file.readline()
             prompt_data = json.loads(data.decode())
             self.assertIn('prompt', prompt_data)
-            self.assertEqual(prompt_data['mode'], 'pdb')
-            
+            self.assertEqual(prompt_data['state'], 'pdb')
+
             # Send 'bt' (backtrace) command
-            client_file.write(json.dumps({"command": "bt"}).encode() + b"\n")
+            client_file.write(json.dumps({"reply": "bt"}).encode() + b"\n")
             client_file.flush()
-            
+
             # Check for response - we should get some stack frames
             # We may get multiple messages so we need to read until we get a new prompt
             got_stack_info = False
@@ -311,15 +324,15 @@ def test_connect_and_basic_commands(self):
                 data = client_file.readline()
                 if not data:
                     break
-                    
+
                 msg = json.loads(data.decode())
                 if 'message' in msg and 'connect_to_debugger' in msg['message']:
                     got_stack_info = True
                     text_msg.append(msg['message'])
-                
+
                 if 'prompt' in msg:
                     break
-                    
+
             expected_stacks = [
                 "<module>",
                 "connect_to_debugger",
@@ -329,18 +342,18 @@ def test_connect_and_basic_commands(self):
                 self.assertIn(stack, msg)
 
             self.assertTrue(got_stack_info, "Should have received stack trace information")
-            
+
             # Send 'c' (continue) command to let the program finish
-            client_file.write(json.dumps({"command": "c"}).encode() + b"\n")
+            client_file.write(json.dumps({"reply": "c"}).encode() + b"\n")
             client_file.flush()
-            
+
             # Wait for process to finish
             stdout, _ = process.communicate(timeout=5)
-            
+
             # Check if we got the expected output
             self.assertIn("Function returned: 42", stdout)
             self.assertEqual(process.returncode, 0)
-                
+
 
 if __name__ == "__main__":
     unittest.main()

From 0601f10be5f39f7e609e1da04bd0b018b8133893 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <godlygeek@gmail.com>
Date: Sun, 20 Apr 2025 18:02:54 -0400
Subject: [PATCH 18/40] Add protocol versioning and support -c commands

---
 Lib/pdb.py | 43 ++++++++++++++++++++++++++++++++++++++-----
 1 file changed, 38 insertions(+), 5 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 423396fc5c6db8..bd923857a4440b 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2526,6 +2526,19 @@ def __init__(self, sockfile, owns_sockfile=True, **kwargs):
         self._write_failed = False
         super().__init__(**kwargs)
 
+    @staticmethod
+    def protocol_version():
+        # By default, assume a client and server are compatible if they run
+        # the same Python major.minor version. We'll try to keep backwards
+        # compatibility between patch versions of a minor version if possible.
+        # If we do need to change the protocol in a patch version, we'll change
+        # `revision` to the patch version where the protocol changed.
+        # We can ignore compatibility for pre-release versions; sys.remote_exec
+        # can't attach to a pre-release version except from that same version.
+        v = sys.version_info
+        revision = 0
+        return int(f"{v.major:02X}{v.minor:02X}{revision:02X}F0", 16)
+
     def _send(self, **kwargs) -> None:
         json_payload = json.dumps(kwargs)
         try:
@@ -2926,7 +2939,7 @@ def complete(self, text, state):
             return None
 
 
-def _connect(host, port, frame):
+def _connect(host, port, frame, commands, version):
     with closing(socket.create_connection((host, port))) as conn:
         sockfile = conn.makefile("rwb")
 
@@ -2935,19 +2948,39 @@ def _connect(host, port, frame):
 
     if Pdb._last_pdb_instance is not None:
         remote_pdb.error("Another PDB instance is already attached.")
+    elif version != remote_pdb.protocol_version():
+        target_ver = f"0x{remote_pdb.protocol_version():08X}"
+        attach_ver = f"0x{version:08X}"
+        remote_pdb.error(
+            f"The target process is running a Python version that is"
+            f" incompatible with this PDB module."
+            f"\nTarget process pdb protocol version: {target_ver}"
+            f"\nLocal pdb module's protocol version: {attach_ver}"
+        )
     else:
+        remote_pdb.rcLines.extend(commands.splitlines())
         remote_pdb.set_trace(frame=frame)
 
 
-def attach(pid):
+def attach(pid, commands=()):
     """Attach to a running process with the given PID."""
     with closing(socket.create_server(("localhost", 0))) as server:
         port = server.getsockname()[1]
 
         with tempfile.NamedTemporaryFile("w", delete_on_close=False) as connect_script:
             connect_script.write(
-                f'import pdb, sys\n'
-                f'pdb._connect("localhost", {port}, sys._getframe().f_back)\n'
+                textwrap.dedent(
+                    f"""
+                    import pdb, sys
+                    pdb._connect(
+                        host="localhost",
+                        port={port},
+                        frame=sys._getframe(1),
+                        commands={json.dumps("\n".join(commands))},
+                        version={_RemotePdb.protocol_version()},
+                    )
+                    """
+                )
             )
             connect_script.close()
             sys.remote_exec(pid, connect_script.name)
@@ -3087,7 +3120,7 @@ def main():
     if opts.pid:
         if opts.module:
             parser.error("argument -m: not allowed with argument --pid")
-        attach(opts.pid)
+        attach(opts.pid, opts.commands)
         return
 
     if opts.module:

From 5a1755b91c5f103549a8a766d32eeff4b0a8375c Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Sun, 20 Apr 2025 22:27:44 -0400
Subject: [PATCH 19/40] Fix tests to match new _connect signature for protocol
 versioning/commands

---
 Lib/test/test_remote_pdb.py | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 9c84c9031863ee..ac6e03370b2a22 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -264,7 +264,13 @@ def dummy_function():
         x = 42
         # Call connect to establish connection with the test server
         frame = sys._getframe()  # Get the current frame
-        pdb._connect('127.0.0.1', {self.port}, frame)
+        pdb._connect(
+            host='127.0.0.1',
+            port={self.port},
+            frame=frame,
+            commands="",
+            version=pdb._RemotePdb.protocol_version(),
+        )
         return x  # This line should not be reached in debugging
 
     return dummy_function()

From f184e4e1a94f959a6c835708e2289fec856b4e8e Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Sun, 20 Apr 2025 23:12:46 -0400
Subject: [PATCH 20/40] Add some comments describing our protocol

Ensure all sent messages conform to the protocol. Tell future
maintainers what to update if the protocol changes.
---
 Lib/pdb.py | 99 +++++++++++++++++++++++++++++++++++++++++++++++++-----
 1 file changed, 91 insertions(+), 8 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index bd923857a4440b..b6d6208bd6e8ca 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2539,7 +2539,48 @@ def protocol_version():
         revision = 0
         return int(f"{v.major:02X}{v.minor:02X}{revision:02X}F0", 16)
 
-    def _send(self, **kwargs) -> None:
+    def _ensure_valid_message(self, msg):
+        # Ensure the message conforms to our protocol.
+        # If anything needs to be changed here for a patch release of Python,
+        # the 'revision' in protocol_version() should be updated.
+        match msg:
+            case {"message": str(), "type": str()}:
+                # Have the client show a message. The client chooses how to
+                # format the message based on its type. The currently defined
+                # types are "info" and "error". If a message has a type the
+                # client doesn't recognize, it must be treated as "info".
+                pass
+            case {"help": str()}:
+                # Have the client show the help for a given argument.
+                pass
+            case {"prompt": str(), "state": str()}:
+                # Have the client display the given prompt and wait for a reply
+                # from the user. If the client recognizes the state it may
+                # enable mode-specific features like multi-line editing.
+                # If it doesn't recognize the state it must prompt for a single
+                # line only and send it directly to the server. A server won't
+                # progress until it gets a "reply" or "signal" message, but can
+                # process "complete" requests while waiting for the reply.
+                pass
+            case {
+                "completions": list(completions)
+            } if all(isinstance(c, str) for c in completions):
+                # Return valid completions for a client's "complete" request.
+                pass
+            case {
+                "command_list": list(command_list)
+            } if all(isinstance(c, str) for c in command_list):
+                # Report the list of legal PDB commands to the client.
+                # Due to aliases this list is not static, but the client
+                # needs to know it for multi-line editing.
+                pass
+            case _:
+                raise AssertionError(
+                    f"PDB message doesn't follow the schema! {msg}"
+                )
+
+    def _send(self, **kwargs):
+        self._ensure_valid_message(kwargs)
         json_payload = json.dumps(kwargs)
         try:
             self._sockfile.write(json_payload.encode() + b"\n")
@@ -2774,6 +2815,51 @@ def __init__(self, pid, sockfile, interrupt_script):
         self.pdb_commands = set()
         self.completion_matches = []
         self.state = "dumb"
+        self.write_failed = False
+
+    def _ensure_valid_message(self, msg):
+        # Ensure the message conforms to our protocol.
+        # If anything needs to be changed here for a patch release of Python,
+        # the 'revision' in protocol_version() should be updated.
+        match msg:
+            case {"reply": str()}:
+                # Send input typed by a user at a prompt to the remote PDB.
+                pass
+            case {"signal": "EOF"}:
+                # Tell the remote PDB that the user pressed ^D at a prompt.
+                pass
+            case {"signal": "INT"}:
+                # Tell the remote PDB that the user pressed ^C at a prompt.
+                pass
+            case {
+                "complete": {
+                    "text": str(),
+                    "line": str(),
+                    "begidx": int(),
+                    "endidx": int(),
+                }
+            }:
+                # Ask the remote PDB what completions are valid for the given
+                # parameters, using readline's completion protocol.
+                pass
+            case _:
+                raise AssertionError(
+                    f"PDB message doesn't follow the schema! {msg}"
+                )
+
+    def _send(self, **kwargs):
+        self._ensure_valid_message(kwargs)
+        json_payload = json.dumps(kwargs)
+        try:
+            self.sockfile.write(json_payload.encode() + b"\n")
+            self.sockfile.flush()
+        except OSError:
+            # This means that the client has abruptly disconnected, but we'll
+            # handle that the next time we try to read from the client instead
+            # of trying to handle it from everywhere _send() may be called.
+            # Track this with a flag rather than assuming readline() will ever
+            # return an empty string because the socket may be half-closed.
+            self.write_failed = True
 
     def read_command(self, prompt):
         reply = input(prompt)
@@ -2829,7 +2915,7 @@ def readline_completion(self, completer):
 
     def cmdloop(self):
         with self.readline_completion(self.complete):
-            while True:
+            while not self.write_failed:
                 try:
                     if not (payload_bytes := self.sockfile.readline()):
                         break
@@ -2889,8 +2975,7 @@ def prompt_for_reply(self, prompt):
                 print("***", msg, flush=True)
                 continue
 
-            self.sockfile.write((json.dumps(payload) + "\n").encode())
-            self.sockfile.flush()
+            self._send(**payload)
             return
 
     def complete(self, text, state):
@@ -2916,10 +3001,8 @@ def complete(self, text, state):
                 }
             }
 
-            try:
-                self.sockfile.write((json.dumps(msg) + "\n").encode())
-                self.sockfile.flush()
-            except OSError:
+            self._send(**msg)
+            if self.write_failed:
                 return None
 
             payload = self.sockfile.readline()

From 82b71f88649f7941d02f20bb649b0cede71149eb Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 17:26:52 -0400
Subject: [PATCH 21/40] Use the 'commands' state for '(com)' prompts

This prevents all multi-line editing, instead sending each line directly
to the server so that it can decide when the command list is terminated.
---
 Lib/pdb.py                  | 3 ++-
 Lib/test/test_remote_pdb.py | 6 +++---
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index b6d6208bd6e8ca..34e37ca4557b65 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2688,7 +2688,8 @@ def cmdloop(self, intro=None):
 
             if not self.cmdqueue:
                 try:
-                    reply = self._get_input(prompt=self.prompt, state="pdb")
+                    state = "commands" if self.commands_defining else "pdb"
+                    reply = self._get_input(prompt=self.prompt, state=state)
                 except EOFError:
                     reply = "EOF"
 
diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index ac6e03370b2a22..0a7ca458386a27 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -186,9 +186,9 @@ def test_registering_commands(self):
             outputs = self.sockfile.get_output()
             self.assertIn('command_list', outputs[0])
             self.assertEqual(outputs[1], {"prompt": "(Pdb) ", "state": "pdb"})
-            self.assertEqual(outputs[2], {"prompt": "(com) ", "state": "pdb"})
-            self.assertEqual(outputs[3], {"prompt": "(com) ", "state": "pdb"})
-            self.assertEqual(outputs[4], {"prompt": "(com) ", "state": "pdb"})
+            self.assertEqual(outputs[2], {"prompt": "(com) ", "state": "commands"})
+            self.assertEqual(outputs[3], {"prompt": "(com) ", "state": "commands"})
+            self.assertEqual(outputs[4], {"prompt": "(com) ", "state": "commands"})
             self.assertEqual(outputs[5], {"prompt": "(Pdb) ", "state": "pdb"})
             self.assertEqual(outputs[6], {"message": "\n", "type": "info"})
             self.assertEqual(len(outputs), 7)

From 3986c17f787673c8b71a4c0d65bf91536fa9401b Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:27:54 -0400
Subject: [PATCH 22/40] Remove choices parameter from _prompt_for_confirmation

---
 Lib/pdb.py | 5 ++---
 1 file changed, 2 insertions(+), 3 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 34e37ca4557b65..d1d32dc1a513a9 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -1466,7 +1466,7 @@ def do_ignore(self, arg):
 
     complete_ignore = _complete_bpnumber
 
-    def _prompt_for_confirmation(self, prompt, choices, default):
+    def _prompt_for_confirmation(self, prompt, default):
         try:
             reply = input(prompt)
         except EOFError:
@@ -1484,7 +1484,6 @@ def do_clear(self, arg):
         if not arg:
             reply = self._prompt_for_confirmation(
                 'Clear all breaks? ',
-                choices=('y', 'yes', 'n', 'no'),
                 default='no',
             )
             if reply in ('y', 'yes'):
@@ -2775,7 +2774,7 @@ def _create_recursive_debugger(self):
         return _RemotePdb(self._sockfile, owns_sockfile=False)
 
     @typing.override
-    def _prompt_for_confirmation(self, prompt, choices, default):
+    def _prompt_for_confirmation(self, prompt, default):
         try:
             return self._get_input(prompt=prompt, state="confirm")
         except (EOFError, KeyboardInterrupt):

From 2e6966783326d2d86982e381fe0cf9350db7823c Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:30:00 -0400
Subject: [PATCH 23/40] Rename _RemotePdb to _PdbServer

---
 Lib/pdb.py                  |  8 ++++----
 Lib/test/test_remote_pdb.py | 10 +++++-----
 2 files changed, 9 insertions(+), 9 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index d1d32dc1a513a9..d71949a802040e 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2516,7 +2516,7 @@ class _InteractState:
     ns: dict[str, typing.Any]
 
 
-class _RemotePdb(Pdb):
+class _PdbServer(Pdb):
     def __init__(self, sockfile, owns_sockfile=True, **kwargs):
         self._owns_sockfile = owns_sockfile
         self._interact_state = None
@@ -2771,7 +2771,7 @@ def do_interact(self, arg):
 
     @typing.override
     def _create_recursive_debugger(self):
-        return _RemotePdb(self._sockfile, owns_sockfile=False)
+        return _PdbServer(self._sockfile, owns_sockfile=False)
 
     @typing.override
     def _prompt_for_confirmation(self, prompt, default):
@@ -3026,7 +3026,7 @@ def _connect(host, port, frame, commands, version):
     with closing(socket.create_connection((host, port))) as conn:
         sockfile = conn.makefile("rwb")
 
-    remote_pdb = _RemotePdb(sockfile)
+    remote_pdb = _PdbServer(sockfile)
     weakref.finalize(remote_pdb, sockfile.close)
 
     if Pdb._last_pdb_instance is not None:
@@ -3060,7 +3060,7 @@ def attach(pid, commands=()):
                         port={port},
                         frame=sys._getframe(1),
                         commands={json.dumps("\n".join(commands))},
-                        version={_RemotePdb.protocol_version()},
+                        version={_PdbServer.protocol_version()},
                     )
                     """
                 )
diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 0a7ca458386a27..5e66473c26148c 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -16,11 +16,11 @@
 from typing import Dict, List, Optional, Tuple, Union, Any
 
 import pdb
-from pdb import _RemotePdb, _PdbClient, _InteractState
+from pdb import _PdbServer, _PdbClient, _InteractState
 
 
 class MockSocketFile:
-    """Mock socket file for testing _RemotePdb without actual socket connections."""
+    """Mock socket file for testing _PdbServer without actual socket connections."""
 
     def __init__(self):
         self.input_queue = []
@@ -62,11 +62,11 @@ def get_output(self) -> List[dict]:
 
 
 class RemotePdbTestCase(unittest.TestCase):
-    """Tests for the _RemotePdb class."""
+    """Tests for the _PdbServer class."""
 
     def setUp(self):
         self.sockfile = MockSocketFile()
-        self.pdb = _RemotePdb(self.sockfile)
+        self.pdb = _PdbServer(self.sockfile)
 
         # Mock some Bdb attributes that are lazily created when tracing starts
         self.pdb.botframe = None
@@ -269,7 +269,7 @@ def dummy_function():
             port={self.port},
             frame=frame,
             commands="",
-            version=pdb._RemotePdb.protocol_version(),
+            version=pdb._PdbServer.protocol_version(),
         )
         return x  # This line should not be reached in debugging
 

From 55adbcc08b577d495fe02d2b9bf7e9da93ff0ddf Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:30:41 -0400
Subject: [PATCH 24/40] Avoid fallthrough in signal handling

---
 Lib/pdb.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index d71949a802040e..331d52d3b342e5 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2632,13 +2632,15 @@ def _read_reply(self):
                 case {"signal": str(signal)}:
                     if signal == "INT":
                         raise KeyboardInterrupt
-                    elif signal != "EOF":
+                    elif signal == "EOF":
+                        raise EOFError
+                    else:
                         self.error(
                             f"Received unrecognized signal: {signal}"
                         )
                         # Our best hope of recovering is to pretend we
                         # got an EOF to exit whatever mode we're in.
-                    raise EOFError
+                        raise EOFError
                 case {
                     "complete": {
                         "text": str(text),

From 0bda5c2f636d2667657bf428da3bd98631624707 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:31:30 -0400
Subject: [PATCH 25/40] Fix handling of a SystemExit raised in normal pdb mode

---
 Lib/pdb.py | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 331d52d3b342e5..a19bada54fbc1d 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2788,13 +2788,14 @@ def do_run(self, arg):
     do_restart = do_run
 
     def _error_exc(self):
-        exc = sys.exception()
-        if isinstance(exc, SystemExit):
+        if self._interact_state and isinstance(sys.exception(), SystemExit):
             # If we get a SystemExit in 'interact' mode, exit the REPL.
             self._interact_state = None
-        super()._error_exc()
-        if isinstance(exc, SystemExit):
+            ret = super()._error_exc()
             self.message("*exit from pdb interact command*")
+            return ret
+        else:
+            return super()._error_exc()
 
     def default(self, line):
         # Unlike Pdb, don't prompt for more lines of a multi-line command.

From 5e93247bc2a06cebc16eb8782e3ea42e06152896 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:31:44 -0400
Subject: [PATCH 26/40] Address nit

---
 Lib/pdb.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index a19bada54fbc1d..6b4f57374b89d2 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -2802,7 +2802,7 @@ def default(self, line):
         # The remote needs to send us the whole block in one go.
         try:
             candidate = line.removeprefix("!") + "\n"
-            if not codeop.compile_command(candidate, "<stdin>", "single"):
+            if codeop.compile_command(candidate, "<stdin>", "single") is None:
                 raise SyntaxError("Incomplete command")
             return super().default(candidate)
         except:

From f799e8325de36bb053e5018c455357a9c0dd5401 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:34:10 -0400
Subject: [PATCH 27/40] Use textwrap.dedent for test readability

---
 Lib/test/test_remote_pdb.py | 55 ++++++++++++++++++++-----------------
 1 file changed, 30 insertions(+), 25 deletions(-)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 5e66473c26148c..02674270af6d11 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -253,31 +253,36 @@ def setUp(self):
         # Create a file for subprocess script
         self.script_path = TESTFN + "_connect_test.py"
         with open(self.script_path, 'w') as f:
-            f.write(f"""
-import pdb
-import sys
-import time
-
-def connect_to_debugger():
-    # Create a frame to debug
-    def dummy_function():
-        x = 42
-        # Call connect to establish connection with the test server
-        frame = sys._getframe()  # Get the current frame
-        pdb._connect(
-            host='127.0.0.1',
-            port={self.port},
-            frame=frame,
-            commands="",
-            version=pdb._PdbServer.protocol_version(),
-        )
-        return x  # This line should not be reached in debugging
-
-    return dummy_function()
-
-result = connect_to_debugger()
-print(f"Function returned: {{result}}")
-""")
+            f.write(
+                textwrap.dedent(
+                    f"""
+                    import pdb
+                    import sys
+                    import time
+
+                    def connect_to_debugger():
+                        # Create a frame to debug
+                        def dummy_function():
+                            x = 42
+                            # Call connect to establish connection
+                            # with the test server
+                            frame = sys._getframe()  # Get the current frame
+                            pdb._connect(
+                                host='127.0.0.1',
+                                port={self.port},
+                                frame=frame,
+                                commands="",
+                                version=pdb._PdbServer.protocol_version(),
+                            )
+                            return x  # This line won't be reached in debugging
+
+                        return dummy_function()
+
+                    result = connect_to_debugger()
+                    print(f"Function returned: {{result}}")
+                    """
+                )
+            )
 
     def tearDown(self):
         self.server_sock.close()

From 46fb219fd37248783358043daf20760337296e48 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:37:00 -0400
Subject: [PATCH 28/40] Drop dataclasses dependency

---
 Lib/pdb.py | 13 +++----------
 1 file changed, 3 insertions(+), 10 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 6b4f57374b89d2..a300e9851507b6 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -93,7 +93,6 @@
 import traceback
 import linecache
 import _colorize
-import dataclasses
 
 from contextlib import closing
 from contextlib import contextmanager
@@ -2510,12 +2509,6 @@ def set_trace(*, header=None, commands=None):
 
 # Remote PDB
 
-@dataclasses.dataclass(frozen=True)
-class _InteractState:
-    compiler: codeop.CommandCompiler
-    ns: dict[str, typing.Any]
-
-
 class _PdbServer(Pdb):
     def __init__(self, sockfile, owns_sockfile=True, **kwargs):
         self._owns_sockfile = owns_sockfile
@@ -2753,10 +2746,10 @@ def _run_in_python_repl(self, lines):
         save_displayhook = sys.displayhook
         try:
             sys.displayhook = self._interact_displayhook
-            code_obj = self._interact_state.compiler(lines + "\n")
+            code_obj = self._interact_state["compiler"](lines + "\n")
             if code_obj is None:
                 raise SyntaxError("Incomplete command")
-            exec(code_obj, self._interact_state.ns)
+            exec(code_obj, self._interact_state["ns"])
         except:
             self._error_exc()
         finally:
@@ -2766,7 +2759,7 @@ def do_interact(self, arg):
         # Prepare to run 'interact' mode code blocks, and trigger the client
         # to start treating all input as Python commands, not PDB ones.
         self.message("*pdb interact start*")
-        self._interact_state = _InteractState(
+        self._interact_state = dict(
             compiler=codeop.CommandCompiler(),
             ns={**self.curframe.f_globals, **self.curframe.f_locals},
         )

From 1ec9475fd2ef15dcba4355d06888c02399080707 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:41:54 -0400
Subject: [PATCH 29/40] Combine the two blocks for handling -p PID into one

---
 Lib/pdb.py | 12 +++++-------
 1 file changed, 5 insertions(+), 7 deletions(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index a300e9851507b6..24a7ec7bf803bb 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -3176,7 +3176,11 @@ def main():
     if opts.pid:
         # If attaching to a remote pid, unrecognized arguments are not allowed.
         # This will raise an error if there are extra unrecognized arguments.
-        parser.parse_args()
+        opts = parser.parse_args()
+        if opts.module:
+            parser.error("argument -m: not allowed with argument --pid")
+        attach(opts.pid, opts.commands)
+        return
     elif opts.module:
         # If a module is being debugged, we consider the arguments after "-m module" to
         # be potential arguments to the module itself. We need to parse the arguments
@@ -3196,12 +3200,6 @@ def main():
             parser.error(f"unrecognized arguments: {' '.join(invalid_args)}")
             sys.exit(2)
 
-    if opts.pid:
-        if opts.module:
-            parser.error("argument -m: not allowed with argument --pid")
-        attach(opts.pid, opts.commands)
-        return
-
     if opts.module:
         file = opts.module
         target = _ModuleTarget(file)

From 662c7eb29eeecb841c8ae59915dbab42bc7bc6c8 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:45:57 -0400
Subject: [PATCH 30/40] Add a news entry

---
 .../Library/2025-04-22-19-45-46.gh-issue-132451.eIzMvE.rst     | 3 +++
 1 file changed, 3 insertions(+)
 create mode 100644 Misc/NEWS.d/next/Library/2025-04-22-19-45-46.gh-issue-132451.eIzMvE.rst

diff --git a/Misc/NEWS.d/next/Library/2025-04-22-19-45-46.gh-issue-132451.eIzMvE.rst b/Misc/NEWS.d/next/Library/2025-04-22-19-45-46.gh-issue-132451.eIzMvE.rst
new file mode 100644
index 00000000000000..01ca64868531e6
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2025-04-22-19-45-46.gh-issue-132451.eIzMvE.rst
@@ -0,0 +1,3 @@
+The CLI for the PDB debugger now accepts a ``-p PID`` argument to allow
+attaching to a running process. The process must be running the same version
+of Python as the one running PDB.

From ac36d7dd333b8174bf2c36b6fe8641d11fbf7243 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:49:43 -0400
Subject: [PATCH 31/40] Skip remote PDB integration test on WASI

---
 Lib/test/test_remote_pdb.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 02674270af6d11..b2e58e3177e9e4 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -11,7 +11,7 @@
 import unittest.mock
 from contextlib import contextmanager
 from pathlib import Path
-from test.support import os_helper
+from test.support import is_wasi, os_helper
 from test.support.os_helper import temp_dir, TESTFN, unlink
 from typing import Dict, List, Optional, Tuple, Union, Any
 
@@ -240,6 +240,7 @@ def side_effect(line):
             self.assertEqual(len(prompts), 2)  # Should have sent 2 prompts
 
 
+@unittest.skipIf(is_wasi, "WASI does not support TCP sockets")
 class PdbConnectTestCase(unittest.TestCase):
     """Tests for the _connect mechanism using direct socket communication."""
 

From f06d9c252afddf16469626cb4a78890239cfe362 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:51:01 -0400
Subject: [PATCH 32/40] Two small things missed in the previous fixes

---
 Lib/test/test_remote_pdb.py | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index b2e58e3177e9e4..2b4aae96aad9bc 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -6,6 +6,7 @@
 import subprocess
 import sys
 import tempfile
+import textwrap
 import threading
 import unittest
 import unittest.mock
@@ -16,7 +17,7 @@
 from typing import Dict, List, Optional, Tuple, Union, Any
 
 import pdb
-from pdb import _PdbServer, _PdbClient, _InteractState
+from pdb import _PdbServer, _PdbClient
 
 
 class MockSocketFile:
@@ -158,7 +159,7 @@ def test_interact_mode(self):
 
         # Verify _interact_state is properly initialized
         self.assertIsNotNone(self.pdb._interact_state)
-        self.assertIsInstance(self.pdb._interact_state, _InteractState)
+        self.assertIsInstance(self.pdb._interact_state, dict)
 
         # Test running code in interact mode
         with unittest.mock.patch.object(self.pdb, '_error_exc') as mock_error:

From 715af272e76833620f487d66c226f3526c7f09cc Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Tue, 22 Apr 2025 19:59:06 -0400
Subject: [PATCH 33/40] Remove call to set_step in interrupt handler

---
 Lib/pdb.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/Lib/pdb.py b/Lib/pdb.py
index 24a7ec7bf803bb..328599d17e69d2 100644
--- a/Lib/pdb.py
+++ b/Lib/pdb.py
@@ -3075,7 +3075,6 @@ def attach(pid, commands=()):
                     interrupt_script.write(
                         'import pdb, sys\n'
                         'if inst := pdb.Pdb._last_pdb_instance:\n'
-                        '    inst.set_step()\n'
                         '    inst.set_trace(sys._getframe(1))\n'
                     )
                     interrupt_script.close()

From c654fdfcfe544e90e9103963bec54508cf14307e Mon Sep 17 00:00:00 2001
From: Pablo Galindo <pablogsal@gmail.com>
Date: Wed, 23 Apr 2025 16:48:49 +0100
Subject: [PATCH 34/40] More tests

---
 Lib/test/test_remote_pdb.py | 249 +++++++++++++++++++++++++++---------
 1 file changed, 191 insertions(+), 58 deletions(-)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 2b4aae96aad9bc..bc24741dd8f5d0 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -252,39 +252,48 @@ def setUp(self):
         self.server_sock.listen(1)
         self.port = self.server_sock.getsockname()[1]
 
+    def _create_script(self, script=None):
         # Create a file for subprocess script
+        if script is None:
+            script = textwrap.dedent(
+                f"""
+                import pdb
+                import sys
+                import time
+
+                def foo():
+                    x = 42
+                    return bar()
+                
+                def bar():
+                    return 42
+
+                def connect_to_debugger():
+                    # Create a frame to debug
+                    def dummy_function():
+                        x = 42
+                        # Call connect to establish connection
+                        # with the test server
+                        frame = sys._getframe()  # Get the current frame
+                        pdb._connect(
+                            host='127.0.0.1',
+                            port={self.port},
+                            frame=frame,
+                            commands="",
+                            version=pdb._PdbServer.protocol_version(),
+                        )
+                        return x  # This line won't be reached in debugging
+
+                    return dummy_function()
+
+                result = connect_to_debugger()
+                foo()
+                print(f"Function returned: {{result}}")
+                """)
+
         self.script_path = TESTFN + "_connect_test.py"
         with open(self.script_path, 'w') as f:
-            f.write(
-                textwrap.dedent(
-                    f"""
-                    import pdb
-                    import sys
-                    import time
-
-                    def connect_to_debugger():
-                        # Create a frame to debug
-                        def dummy_function():
-                            x = 42
-                            # Call connect to establish connection
-                            # with the test server
-                            frame = sys._getframe()  # Get the current frame
-                            pdb._connect(
-                                host='127.0.0.1',
-                                port={self.port},
-                                frame=frame,
-                                commands="",
-                                version=pdb._PdbServer.protocol_version(),
-                            )
-                            return x  # This line won't be reached in debugging
-
-                        return dummy_function()
-
-                    result = connect_to_debugger()
-                    print(f"Function returned: {{result}}")
-                    """
-                )
-            )
+            f.write(script)
 
     def tearDown(self):
         self.server_sock.close()
@@ -293,21 +302,62 @@ def tearDown(self):
         except OSError:
             pass
 
-    def test_connect_and_basic_commands(self):
-        """Test connecting to a remote debugger and sending basic commands."""
+    def _connect_and_get_client_file(self):
+        """Helper to start subprocess and get connected client file."""
         # Start the subprocess that will connect to our socket
-        with subprocess.Popen(
+        process = subprocess.Popen(
             [sys.executable, self.script_path],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True
-        ) as process:
-            # Accept the connection from the subprocess
-            client_sock, _ = self.server_sock.accept()
-            client_file = client_sock.makefile('rwb')
-            self.addCleanup(client_file.close)
-            self.addCleanup(client_sock.close)
+        )
+        
+        # Accept the connection from the subprocess
+        client_sock, _ = self.server_sock.accept()
+        client_file = client_sock.makefile('rwb')
+        self.addCleanup(client_file.close)
+        self.addCleanup(client_sock.close)
+        
+        return process, client_file
+
+    def _read_until_prompt(self, client_file):
+        """Helper to read messages until a prompt is received."""
+        messages = []
+        while True:
+            data = client_file.readline()
+            if not data:
+                break
+            msg = json.loads(data.decode())
+            messages.append(msg)
+            if 'prompt' in msg:
+                break
+        return messages
+
+    def _send_command(self, client_file, command):
+        """Helper to send a command to the debugger."""
+        client_file.write(json.dumps({"reply": command}).encode() + b"\n")
+        client_file.flush()
+    
+    def _send_interrupt(self, pid):
+        """Helper to send an interrupt signal to the debugger."""
+        # with tempfile.NamedTemporaryFile("w", delete_on_close=False) as interrupt_script:
+        interrupt_script = TESTFN + "_interrupt_script.py"
+        with open(interrupt_script, 'w') as f:
+            f.write(
+                'import pdb, sys\n'
+                'print("Hello, world!")\n'
+                'if inst := pdb.Pdb._last_pdb_instance:\n'
+                '    inst.set_trace(sys._getframe(1))\n'
+            )
+        sys.remote_exec(pid, interrupt_script)
+        self.addCleanup(unlink, interrupt_script)
 
+    def test_connect_and_basic_commands(self):
+        """Test connecting to a remote debugger and sending basic commands."""
+        self._create_script()
+        process, client_file = self._connect_and_get_client_file()
+
+        with process:
             # We should receive initial data from the debugger
             data = client_file.readline()
             initial_data = json.loads(data.decode())
@@ -326,25 +376,15 @@ def test_connect_and_basic_commands(self):
             self.assertEqual(prompt_data['state'], 'pdb')
 
             # Send 'bt' (backtrace) command
-            client_file.write(json.dumps({"reply": "bt"}).encode() + b"\n")
-            client_file.flush()
+            self._send_command(client_file, "bt")
 
             # Check for response - we should get some stack frames
-            # We may get multiple messages so we need to read until we get a new prompt
-            got_stack_info = False
-            text_msg = []
-            while True:
-                data = client_file.readline()
-                if not data:
-                    break
-
-                msg = json.loads(data.decode())
-                if 'message' in msg and 'connect_to_debugger' in msg['message']:
-                    got_stack_info = True
-                    text_msg.append(msg['message'])
-
-                if 'prompt' in msg:
-                    break
+            messages = self._read_until_prompt(client_file)
+            
+            # Extract text messages containing stack info
+            text_msg = [msg['message'] for msg in messages 
+                    if 'message' in msg and 'connect_to_debugger' in msg['message']]
+            got_stack_info = bool(text_msg)
 
             expected_stacks = [
                 "<module>",
@@ -357,8 +397,7 @@ def test_connect_and_basic_commands(self):
             self.assertTrue(got_stack_info, "Should have received stack trace information")
 
             # Send 'c' (continue) command to let the program finish
-            client_file.write(json.dumps({"reply": "c"}).encode() + b"\n")
-            client_file.flush()
+            self._send_command(client_file, "c")
 
             # Wait for process to finish
             stdout, _ = process.communicate(timeout=5)
@@ -367,6 +406,100 @@ def test_connect_and_basic_commands(self):
             self.assertIn("Function returned: 42", stdout)
             self.assertEqual(process.returncode, 0)
 
+    def test_breakpoints(self):
+        """Test setting and hitting breakpoints."""
+        self._create_script()
+        process, client_file = self._connect_and_get_client_file()
+        with process:
+            # Skip initial messages until we get to the prompt
+            self._read_until_prompt(client_file)
+
+            # Set a breakpoint at the return statement
+            self._send_command(client_file, "break bar")
+            messages = self._read_until_prompt(client_file)
+            bp_msg = next(msg['message'] for msg in messages if 'message' in msg)
+            self.assertIn("Breakpoint", bp_msg)
+
+            # Continue execution until breakpoint
+            self._send_command(client_file, "c")
+            messages = self._read_until_prompt(client_file)
+            
+            # Verify we hit the breakpoint
+            hit_msg = next(msg['message'] for msg in messages if 'message' in msg)
+            self.assertIn("bar()", hit_msg)
+
+            # Check breakpoint list
+            self._send_command(client_file, "b")
+            messages = self._read_until_prompt(client_file)
+            list_msg = next(msg['message'] for msg in reversed(messages) if 'message' in msg)
+            self.assertIn("1   breakpoint", list_msg)
+            self.assertIn("breakpoint already hit 1 time", list_msg)
+
+            # Clear breakpoint
+            self._send_command(client_file, "clear 1")
+            messages = self._read_until_prompt(client_file)
+            clear_msg = next(msg['message'] for msg in reversed(messages) if 'message' in msg)
+            self.assertIn("Deleted breakpoint", clear_msg)
+
+            # Continue to end
+            self._send_command(client_file, "c")
+            stdout, _ = process.communicate(timeout=5)
+
+            self.assertIn("Function returned: 42", stdout)
+            self.assertEqual(process.returncode, 0)
+
+    def test_keyboard_interrupt(self):
+        """Test that sending keyboard interrupt breaks into pdb."""
+        script = f"""
+import time
+import sys
+import pdb
+def bar():
+    frame = sys._getframe()  # Get the current frame
+    pdb._connect(
+        host='127.0.0.1',
+        port={self.port},
+        frame=frame,
+        commands="",
+        version=pdb._PdbServer.protocol_version(),
+    )
+    print("Connected to debugger")
+    iterations = 10
+    while iterations > 0:
+        print("Iteration", iterations)
+        time.sleep(1)
+        iterations -= 1
+    return 42
+
+if __name__ == "__main__":
+    print("Function returned:", bar())
+"""
+        self._create_script(script=script)
+        process, client_file = self._connect_and_get_client_file()
+
+        with process:
+
+            # Skip initial messages until we get to the prompt
+            self._read_until_prompt(client_file)
+
+            # Continue execution
+            self._send_command(client_file, "c")
+
+            # Send keyboard interrupt signal
+            self._send_command(client_file, json.dumps({"signal": "INT"}))
+            self._send_interrupt(process.pid)
+            messages = self._read_until_prompt(client_file)
+
+            # Verify we got the keyboard interrupt message
+            interrupt_msg = next(msg['message'] for msg in messages if 'message' in msg)
+            self.assertIn("bar()", interrupt_msg)
+
+            # Continue to end
+            self._send_command(client_file, "iterations = 0")
+            self._send_command(client_file, "c")
+            stdout, _ = process.communicate(timeout=5)
+            self.assertIn("Function returned: 42", stdout)
+            self.assertEqual(process.returncode, 0)
 
 if __name__ == "__main__":
     unittest.main()

From 205bc552b68f145a26e0263b38051a399afdba59 Mon Sep 17 00:00:00 2001
From: Pablo Galindo <pablogsal@gmail.com>
Date: Wed, 23 Apr 2025 16:48:49 +0100
Subject: [PATCH 35/40] More tests

---
 Lib/test/test_remote_pdb.py | 168 ++++++++++++++++++++++++++++++++++++
 1 file changed, 168 insertions(+)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index bc24741dd8f5d0..7b040f1596a9d7 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -501,5 +501,173 @@ def bar():
             self.assertIn("Function returned: 42", stdout)
             self.assertEqual(process.returncode, 0)
 
+    def test_handle_eof(self):
+        """Test that EOF signal properly exits the debugger."""
+        self._create_script()
+        process, client_file = self._connect_and_get_client_file()
+
+        with process:
+            # Skip initial messages until we get to the prompt
+            self._read_until_prompt(client_file)
+
+            # Send EOF signal to exit the debugger
+            client_file.write(json.dumps({"signal": "EOF"}).encode() + b"\n")
+            client_file.flush()
+
+            # The process should complete normally after receiving EOF
+            stdout, stderr = process.communicate(timeout=5)
+
+            # Verify process completed correctly
+            self.assertIn("Function returned: 42", stdout)
+            self.assertEqual(process.returncode, 0)
+            self.assertEqual(stderr, "")
+
+    def test_protocol_version(self):
+        """Test that incompatible protocol versions are properly detected."""
+        # Create a script using an incompatible protocol version
+        script = f"""
+import sys
+import pdb
+
+def run_test():
+    frame = sys._getframe()
+
+    # Use a fake version number that's definitely incompatible
+    fake_version = 0x01010101 # A fake version that doesn't match any real Python version
+
+    # Connect with the wrong version
+    pdb._connect(
+        host='127.0.0.1',
+        port={self.port},
+        frame=frame,
+        commands="",
+        version=fake_version,
+    )
+
+    # This should print if the debugger detaches correctly
+    print("Debugger properly detected version mismatch")
+    return True
+
+if __name__ == "__main__":
+    print("Test result:", run_test())
+"""
+        self._create_script(script=script)
+        process, client_file = self._connect_and_get_client_file()
+
+        with process:
+            # First message should be an error about protocol version mismatch
+            data = client_file.readline()
+            message = json.loads(data.decode())
+
+            self.assertIn('message', message)
+            self.assertEqual(message['type'], 'error')
+            self.assertIn('incompatible', message['message'])
+            self.assertIn('protocol version', message['message'])
+
+            # The process should complete normally
+            stdout, stderr = process.communicate(timeout=5)
+
+            # Verify the process completed successfully
+            self.assertIn("Test result: True", stdout)
+            self.assertIn("Debugger properly detected version mismatch", stdout)
+            self.assertEqual(process.returncode, 0)
+
+    def test_help_system(self):
+        """Test that the help system properly sends help text to the client."""
+        self._create_script()
+        process, client_file = self._connect_and_get_client_file()
+
+        with process:
+            # Skip initial messages until we get to the prompt
+            self._read_until_prompt(client_file)
+
+            # Request help for different commands
+            help_commands = ["help", "help break", "help continue", "help pdb"]
+
+            for cmd in help_commands:
+                self._send_command(client_file, cmd)
+
+                # Look for help message
+                data = client_file.readline()
+                message = json.loads(data.decode())
+
+                self.assertIn('help', message)
+
+                if cmd == "help":
+                    # Should just contain the command itself
+                    self.assertEqual(message['help'], "")
+                else:
+                    # Should contain the specific command we asked for help with
+                    command = cmd.split()[1]
+                    self.assertEqual(message['help'], command)
+
+                # Skip to the next prompt
+                self._read_until_prompt(client_file)
+
+            # Continue execution to finish the program
+            self._send_command(client_file, "c")
+
+            stdout, stderr = process.communicate(timeout=5)
+            self.assertIn("Function returned: 42", stdout)
+            self.assertEqual(process.returncode, 0)
+
+    def test_multi_line_commands(self):
+        """Test that multi-line commands work properly over remote connection."""
+        self._create_script()
+        process, client_file = self._connect_and_get_client_file()
+
+        with process:
+            # Skip initial messages until we get to the prompt
+            self._read_until_prompt(client_file)
+
+            # Send a multi-line command
+            multi_line_commands = [
+                # Define a function
+                "def test_func():\n    return 42",
+
+                # For loop
+                "for i in range(3):\n    print(i)",
+
+                # If statement
+                "if True:\n    x = 42\nelse:\n    x = 0",
+
+                # Try/except
+                "try:\n    result = 10/2\n    print(result)\nexcept ZeroDivisionError:\n    print('Error')",
+
+                # Class definition
+                "class TestClass:\n    def __init__(self):\n        self.value = 100\n    def get_value(self):\n        return self.value"
+            ]
+
+            for cmd in multi_line_commands:
+                self._send_command(client_file, cmd)
+                self._read_until_prompt(client_file)
+
+            # Test executing the defined function
+            self._send_command(client_file, "test_func()")
+            messages = self._read_until_prompt(client_file)
+
+            # Find the result message
+            result_msg = next(msg['message'] for msg in messages if 'message' in msg)
+            self.assertIn("42", result_msg)
+
+            # Test creating an instance of the defined class
+            self._send_command(client_file, "obj = TestClass()")
+            self._read_until_prompt(client_file)
+
+            # Test calling a method on the instance
+            self._send_command(client_file, "obj.get_value()")
+            messages = self._read_until_prompt(client_file)
+
+            # Find the result message
+            result_msg = next(msg['message'] for msg in messages if 'message' in msg)
+            self.assertIn("100", result_msg)
+
+            # Continue execution to finish
+            self._send_command(client_file, "c")
+
+            stdout, stderr = process.communicate(timeout=5)
+            self.assertIn("Function returned: 42", stdout)
+            self.assertEqual(process.returncode, 0)
+
 if __name__ == "__main__":
     unittest.main()

From bbe784bd100a8df87e388829e2be69a293b214ec Mon Sep 17 00:00:00 2001
From: Pablo Galindo <pablogsal@gmail.com>
Date: Wed, 23 Apr 2025 18:50:24 +0100
Subject: [PATCH 36/40] More tests

---
 Lib/test/test_remote_pdb.py | 19 +++++++++++--------
 1 file changed, 11 insertions(+), 8 deletions(-)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 7b040f1596a9d7..fed4b74aa9aa0d 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -264,7 +264,7 @@ def _create_script(self, script=None):
                 def foo():
                     x = 42
                     return bar()
-                
+
                 def bar():
                     return 42
 
@@ -311,13 +311,13 @@ def _connect_and_get_client_file(self):
             stderr=subprocess.PIPE,
             text=True
         )
-        
+
         # Accept the connection from the subprocess
         client_sock, _ = self.server_sock.accept()
         client_file = client_sock.makefile('rwb')
         self.addCleanup(client_file.close)
         self.addCleanup(client_sock.close)
-        
+
         return process, client_file
 
     def _read_until_prompt(self, client_file):
@@ -337,7 +337,7 @@ def _send_command(self, client_file, command):
         """Helper to send a command to the debugger."""
         client_file.write(json.dumps({"reply": command}).encode() + b"\n")
         client_file.flush()
-    
+
     def _send_interrupt(self, pid):
         """Helper to send an interrupt signal to the debugger."""
         # with tempfile.NamedTemporaryFile("w", delete_on_close=False) as interrupt_script:
@@ -349,7 +349,10 @@ def _send_interrupt(self, pid):
                 'if inst := pdb.Pdb._last_pdb_instance:\n'
                 '    inst.set_trace(sys._getframe(1))\n'
             )
-        sys.remote_exec(pid, interrupt_script)
+        try:
+            sys.remote_exec(pid, interrupt_script)
+        except PermissionError:
+            self.skipTest("Insufficient permissions to execute code in remote process")
         self.addCleanup(unlink, interrupt_script)
 
     def test_connect_and_basic_commands(self):
@@ -380,9 +383,9 @@ def test_connect_and_basic_commands(self):
 
             # Check for response - we should get some stack frames
             messages = self._read_until_prompt(client_file)
-            
+
             # Extract text messages containing stack info
-            text_msg = [msg['message'] for msg in messages 
+            text_msg = [msg['message'] for msg in messages
                     if 'message' in msg and 'connect_to_debugger' in msg['message']]
             got_stack_info = bool(text_msg)
 
@@ -423,7 +426,7 @@ def test_breakpoints(self):
             # Continue execution until breakpoint
             self._send_command(client_file, "c")
             messages = self._read_until_prompt(client_file)
-            
+
             # Verify we hit the breakpoint
             hit_msg = next(msg['message'] for msg in messages if 'message' in msg)
             self.assertIn("bar()", hit_msg)

From 30cb5371ed4313ed560e11f72f778f5f158a189f Mon Sep 17 00:00:00 2001
From: Pablo Galindo <pablogsal@gmail.com>
Date: Wed, 23 Apr 2025 19:00:50 +0100
Subject: [PATCH 37/40] Add what's new entry

---
 Doc/whatsnew/3.14.rst | 30 +++++++++++++++++++++++++++++-
 1 file changed, 29 insertions(+), 1 deletion(-)

diff --git a/Doc/whatsnew/3.14.rst b/Doc/whatsnew/3.14.rst
index 762d53eeb2df1a..181612b4e171ab 100644
--- a/Doc/whatsnew/3.14.rst
+++ b/Doc/whatsnew/3.14.rst
@@ -99,7 +99,9 @@ PEP 768: Safe external debugger interface for CPython
 
 :pep:`768` introduces a zero-overhead debugging interface that allows debuggers and profilers
 to safely attach to running Python processes. This is a significant enhancement to Python's
-debugging capabilities allowing debuggers to forego unsafe alternatives.
+debugging capabilities allowing debuggers to forego unsafe alternatives. See
+:ref:`below <whatsnew314-remote-pdb>` for how this feature is leveraged to
+implement the new :mod:`pdb` module's remote attaching capabilities.
 
 The new interface provides safe execution points for attaching debugger code without modifying
 the interpreter's normal execution path or adding runtime overhead. This enables tools to
@@ -149,6 +151,32 @@ See :pep:`768` for more details.
 
 (Contributed by Pablo Galindo Salgado, Matt Wozniski, and Ivona Stojanovic in :gh:`131591`.)
 
+
+.. _whatsnew314-remote-pdb:
+
+Remote attaching to a running Python process with PDB
+-----------------------------------------------------
+
+The :mod:`pdb` module now supports remote attaching to a running Python process
+using a new ``-p PID`` command-line option:
+
+.. code-block:: sh
+
+   python -m pdb -p 1234
+
+This will connect to the Python process with the given PID and allow you to
+debug it interactively. Notice that due to how the Python interpreter works
+attaching to a remote process that is blocked in a system call or waiting for
+I/O will only work once the next bytecode instruction is executed or when the
+process receives a signal.
+
+This feature leverages :pep:`768` and the :func:`sys.remote_exec` function
+to attach to the remote process and send the PDB commands to it.
+
+
+(Contributed by Matt Wozniski and Pablo Galindo in :gh:`131591`.)
+
+
 .. _whatsnew314-pep758:
 
 PEP 758 – Allow except and except* expressions without parentheses

From 659556f5050ffbacc7d0bd77b1bfc3d9d7e50212 Mon Sep 17 00:00:00 2001
From: Pablo Galindo <pablogsal@gmail.com>
Date: Thu, 24 Apr 2025 00:09:28 +0100
Subject: [PATCH 38/40] use dedent

---
 Lib/test/test_remote_pdb.py | 102 ++++++++++++++++++------------------
 1 file changed, 51 insertions(+), 51 deletions(-)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index fed4b74aa9aa0d..6301f249f09c05 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -349,11 +349,11 @@ def _send_interrupt(self, pid):
                 'if inst := pdb.Pdb._last_pdb_instance:\n'
                 '    inst.set_trace(sys._getframe(1))\n'
             )
+        self.addCleanup(unlink, interrupt_script)
         try:
             sys.remote_exec(pid, interrupt_script)
         except PermissionError:
             self.skipTest("Insufficient permissions to execute code in remote process")
-        self.addCleanup(unlink, interrupt_script)
 
     def test_connect_and_basic_commands(self):
         """Test connecting to a remote debugger and sending basic commands."""
@@ -453,30 +453,30 @@ def test_breakpoints(self):
 
     def test_keyboard_interrupt(self):
         """Test that sending keyboard interrupt breaks into pdb."""
-        script = f"""
-import time
-import sys
-import pdb
-def bar():
-    frame = sys._getframe()  # Get the current frame
-    pdb._connect(
-        host='127.0.0.1',
-        port={self.port},
-        frame=frame,
-        commands="",
-        version=pdb._PdbServer.protocol_version(),
-    )
-    print("Connected to debugger")
-    iterations = 10
-    while iterations > 0:
-        print("Iteration", iterations)
-        time.sleep(1)
-        iterations -= 1
-    return 42
-
-if __name__ == "__main__":
-    print("Function returned:", bar())
-"""
+        script = textwrap.dedent(f"""
+            import time
+            import sys
+            import pdb
+            def bar():
+                frame = sys._getframe()  # Get the current frame
+                pdb._connect(
+                    host='127.0.0.1',
+                    port={self.port},
+                    frame=frame,
+                    commands="",
+                    version=pdb._PdbServer.protocol_version(),
+                )
+                print("Connected to debugger")
+                iterations = 10
+                while iterations > 0:
+                    print("Iteration", iterations)
+                    time.sleep(1)
+                    iterations -= 1
+                return 42
+
+            if __name__ == "__main__":
+                print("Function returned:", bar())
+            """)
         self._create_script(script=script)
         process, client_file = self._connect_and_get_client_file()
 
@@ -528,32 +528,32 @@ def test_handle_eof(self):
     def test_protocol_version(self):
         """Test that incompatible protocol versions are properly detected."""
         # Create a script using an incompatible protocol version
-        script = f"""
-import sys
-import pdb
-
-def run_test():
-    frame = sys._getframe()
-
-    # Use a fake version number that's definitely incompatible
-    fake_version = 0x01010101 # A fake version that doesn't match any real Python version
-
-    # Connect with the wrong version
-    pdb._connect(
-        host='127.0.0.1',
-        port={self.port},
-        frame=frame,
-        commands="",
-        version=fake_version,
-    )
-
-    # This should print if the debugger detaches correctly
-    print("Debugger properly detected version mismatch")
-    return True
-
-if __name__ == "__main__":
-    print("Test result:", run_test())
-"""
+        script = textwrap.dedent(f'''
+            import sys
+            import pdb
+
+            def run_test():
+                frame = sys._getframe()
+
+                # Use a fake version number that's definitely incompatible
+                fake_version = 0x01010101 # A fake version that doesn't match any real Python version
+
+                # Connect with the wrong version
+                pdb._connect(
+                    host='127.0.0.1',
+                    port={self.port},
+                    frame=frame,
+                    commands="",
+                    version=fake_version,
+                )
+
+                # This should print if the debugger detaches correctly
+                print("Debugger properly detected version mismatch")
+                return True
+
+            if __name__ == "__main__":
+                print("Test result:", run_test())
+            ''')
         self._create_script(script=script)
         process, client_file = self._connect_and_get_client_file()
 

From 6c2d970b823603946696f958a0d8cf20accc4a2f Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Thu, 24 Apr 2025 16:52:06 -0400
Subject: [PATCH 39/40] Add synchronization to test_keyboard_interrupt

---
 Lib/test/test_remote_pdb.py | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 6301f249f09c05..72e29ea918ea62 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -453,9 +453,17 @@ def test_breakpoints(self):
 
     def test_keyboard_interrupt(self):
         """Test that sending keyboard interrupt breaks into pdb."""
+        synchronizer_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        synchronizer_sock.bind(('127.0.0.1', 0))  # Let OS assign port
+        synchronizer_sock.settimeout(5)
+        synchronizer_sock.listen(1)
+        self.addCleanup(synchronizer_sock.close)
+        sync_port = synchronizer_sock.getsockname()[1]
+
         script = textwrap.dedent(f"""
             import time
             import sys
+            import socket
             import pdb
             def bar():
                 frame = sys._getframe()  # Get the current frame
@@ -468,6 +476,7 @@ def bar():
                 )
                 print("Connected to debugger")
                 iterations = 10
+                socket.create_connection(('127.0.0.1', {sync_port})).close()
                 while iterations > 0:
                     print("Iteration", iterations)
                     time.sleep(1)
@@ -488,6 +497,9 @@ def bar():
             # Continue execution
             self._send_command(client_file, "c")
 
+            # Wait until execution has continued
+            synchronizer_sock.accept()[0].close()
+
             # Send keyboard interrupt signal
             self._send_command(client_file, json.dumps({"signal": "INT"}))
             self._send_interrupt(process.pid)

From 100be4482281981135df5d7bdf2aca6bd8a73574 Mon Sep 17 00:00:00 2001
From: Matt Wozniski <mwozniski@bloomberg.net>
Date: Thu, 24 Apr 2025 16:53:37 -0400
Subject: [PATCH 40/40] Stop sending a "signal" message in
 test_keyboard_interrupt"

---
 Lib/test/test_remote_pdb.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/Lib/test/test_remote_pdb.py b/Lib/test/test_remote_pdb.py
index 72e29ea918ea62..cc0ada12814afd 100644
--- a/Lib/test/test_remote_pdb.py
+++ b/Lib/test/test_remote_pdb.py
@@ -500,8 +500,7 @@ def bar():
             # Wait until execution has continued
             synchronizer_sock.accept()[0].close()
 
-            # Send keyboard interrupt signal
-            self._send_command(client_file, json.dumps({"signal": "INT"}))
+            # Inject a script to interrupt the running process
             self._send_interrupt(process.pid)
             messages = self._read_until_prompt(client_file)