chore: add exploratory test for timeout

Author: cuinixam

pyproject.toml

@@ -96,6 +96,9 @@ addopts = """\
     --cov-report=xml
     """
 pythonpath = [ "src" ]
+markers = [
+    "exploratory: local-only exploratory tests; skipped by default (run with --run-exploratory)",
+]
 
 [tool.coverage.run]
 branch = true

src/py_app_dev/core/subprocess.py

@@ -119,8 +119,6 @@ def execute(self, handle_errors: bool = True) -> subprocess.CompletedProcess[Any
                             time.sleep(0.01)
                         # TIMEOUT CHECK
                         if self.timeout is not None and (time.monotonic() - start_time) > self.timeout:
-                            self.logger.error(f"Process timed out after {self.timeout} seconds. Terminating process tree (PID={process.pid}).")
-                            self._terminate_process_tree(process.pid)
                             raise subprocess.TimeoutExpired(
                                 cmd=self.command_str,
                                 timeout=self.timeout,
@@ -130,20 +128,10 @@ def execute(self, handle_errors: bool = True) -> subprocess.CompletedProcess[Any
                     process.wait()
                 # NON-STREAMING MODE (communicate with native timeout)
                 elif self.capture_output and not self.print_output:
-                    try:
-                        stdout, stderr = process.communicate(timeout=self.timeout)
-                    except subprocess.TimeoutExpired:
-                        self.logger.error(f"Process timed out after {self.timeout} seconds. Terminating process tree (PID={process.pid}).")
-                        self._terminate_process_tree(process.pid)
-                        raise
+                    stdout, stderr = process.communicate(timeout=self.timeout)
                 else:
                     # No output capturing; just wait for completion or timeout
-                    try:
-                        process.wait(timeout=self.timeout)
-                    except subprocess.TimeoutExpired:
-                        self.logger.error(f"Process timed out after {self.timeout} seconds. Terminating process tree (PID={process.pid}).")
-                        self._terminate_process_tree(process.pid)
-                        raise
+                    process.wait(timeout=self.timeout)
             # HANDLE RETURN CODES AND ERROR INTERPRETATION
             if handle_errors:
                 # Check return code
@@ -152,6 +140,7 @@ def execute(self, handle_errors: bool = True) -> subprocess.CompletedProcess[Any
             else:
                 completed_process = subprocess.CompletedProcess(process.args, process.returncode, stdout, stderr)
         except subprocess.TimeoutExpired:
+            self._terminate_process_tree(process.pid)
             raise UserNotificationException(f"Command '{self.command_str}' timed out after {self.timeout} seconds and was forcefully terminated.") from None
         except subprocess.CalledProcessError as e:
             raise UserNotificationException(f"Command '{self.command_str}' execution failed with return code {e.returncode}") from None

tests/test_subprocess.py

@@ -215,6 +215,31 @@ def test_timeout_kills_process_tree(self, tmp_path: Path) -> None:
         child_pid = int(pid_file.read_text().strip())
         assert not psutil.pid_exists(child_pid), f"Child process {child_pid} should have been killed by _terminate_process_tree"
 
+    @pytest.mark.exploratory
+    def test_streaming_timeout_blocked_by_silent_process(self) -> None:
+        """
+        Reproduces the bug where readline() in the streaming output loop blocks
+        indefinitely when the process is alive but writes nothing to stdout.
+
+        Expected: UserNotificationException raised within ~timeout seconds.
+        Actual:   readline() blocks until the process exits (~10 s), so the
+                  timeout check is never reached on time.
+        """
+        executor = SubprocessExecutor(
+            ["python", "-c", "import time; time.sleep(10)"],
+            capture_output=True,
+            print_output=True,  # triggers the streaming readline() path
+            timeout=2,
+        )
+        start = time.monotonic()
+        with pytest.raises(UserNotificationException, match="timed out"):
+            executor.execute()
+        elapsed = time.monotonic() - start
+        # BUG: elapsed will be ~10 s instead of ~2 s because readline() blocks
+        # waiting for a newline or EOF; the timeout check is only reached after
+        # the process finally exits.
+        assert elapsed < 5, f"Timeout should fire within ~2 s but took {elapsed:.1f} s"
+
     @pytest.mark.parametrize(
         "capture_output, print_output",
         [