gh-159865: Fix profiling.sampling exception mode dropping blocked handlers

Lib/profiling/sampling/cli.py

@@ -213,6 +213,15 @@ def _parse_mode(mode_string):
     return mode_map[mode_string]
 
 
+def _skip_idle_for_mode(mode):
+    """Return whether collectors should discard idle threads for a mode."""
+    return mode not in (
+        PROFILING_MODE_WALL,
+        PROFILING_MODE_ALL,
+        PROFILING_MODE_EXCEPTION,
+    )
+
+
 def _check_process_died(process):
     """Check if process died and raise an error with stderr if available."""
     if process.poll() is None:
@@ -1186,9 +1195,7 @@ def _handle_attach(args):
     )
 
     # Determine skip_idle based on mode
-    skip_idle = (
-        mode != PROFILING_MODE_WALL if mode != PROFILING_MODE_ALL else False
-    )
+    skip_idle = _skip_idle_for_mode(mode)
 
     output_file = None
     if args.format == "binary":
@@ -1294,9 +1301,7 @@ def _handle_run(args):
     )
 
     # Determine skip_idle based on mode
-    skip_idle = (
-        mode != PROFILING_MODE_WALL if mode != PROFILING_MODE_ALL else False
-    )
+    skip_idle = _skip_idle_for_mode(mode)
 
     output_file = None
     if args.format == "binary":
@@ -1344,7 +1349,7 @@ def _handle_live_attach(args, pid):
     mode = _parse_mode(args.mode)
 
     # Determine skip_idle based on mode
-    skip_idle = mode != PROFILING_MODE_WALL
+    skip_idle = _skip_idle_for_mode(mode)
 
     # Create live collector with default settings
     collector = LiveStatsCollector(
@@ -1391,7 +1396,7 @@ def _handle_live_run(args):
     mode = _parse_mode(args.mode)
 
     # Determine skip_idle based on mode
-    skip_idle = mode != PROFILING_MODE_WALL
+    skip_idle = _skip_idle_for_mode(mode)
 
     # Create live collector with default settings
     collector = LiveStatsCollector(

Lib/test/test_profiling/test_sampling_profiler/test_cli.py

@@ -30,17 +30,72 @@
     _generate_output_filename,
     _handle_output,
     _replay_with_reader,
+    _skip_idle_for_mode,
     main,
 )
 from profiling.sampling.constants import (
     PROFILING_MODE_ALL,
     PROFILING_MODE_CPU,
+    PROFILING_MODE_EXCEPTION,
+    PROFILING_MODE_GIL,
     PROFILING_MODE_WALL,
 )
 from profiling.sampling.errors import SamplingScriptNotFoundError, SamplingModuleNotFoundError, SamplingUnknownProcessError
 from profiling.sampling.jsonl_collector import JsonlCollector
 
 class TestSampleProfilerCLI(unittest.TestCase):
+    def test_skip_idle_mode_mapping(self):
+        self.assertFalse(_skip_idle_for_mode(PROFILING_MODE_WALL))
+        self.assertTrue(_skip_idle_for_mode(PROFILING_MODE_CPU))
+        self.assertTrue(_skip_idle_for_mode(PROFILING_MODE_GIL))
+        self.assertFalse(_skip_idle_for_mode(PROFILING_MODE_ALL))
+        self.assertFalse(_skip_idle_for_mode(PROFILING_MODE_EXCEPTION))
+        # Preserve the previous default for modes unknown to this CLI.
+        self.assertTrue(_skip_idle_for_mode(999))
+
+    @requires_remote_subprocess_debugging()
+    def test_exception_mode_captures_blocked_handler(self):
+        script = """
+import time
+
+try:
+    raise ValueError("boom")
+except ValueError:
+    time.sleep(30)
+"""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            script_path = os.path.join(temp_dir, "blocked_handler.py")
+            output_path = os.path.join(temp_dir, "output.txt")
+            with open(script_path, "w", encoding="utf-8") as file:
+                file.write(script)
+
+            result = subprocess.run(
+                [
+                    sys.executable,
+                    "-m",
+                    "profiling.sampling",
+                    "run",
+                    "--mode",
+                    "exception",
+                    "--collapsed",
+                    "-r",
+                    "1000",
+                    "-d",
+                    "1",
+                    "-o",
+                    output_path,
+                    script_path,
+                ],
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
+
+            self.assertEqual(result.returncode, 0, result.stderr)
+            with open(output_path, encoding="utf-8") as file:
+                output = file.read()
+            self.assertIn("blocked_handler.py", output)
+
     def _setup_sync_mocks(self, mock_socket, mock_popen):
         """Helper to set up socket and process mocks for coordinator tests."""
         # Mock the sync socket with context manager support