@@ -19,6 +19,7 @@
 import sys
 from textwrap import dedent
 import threading
+import time
 import warnings
 
 from git.compat import defenc, force_bytes, safe_decode
@@ -99,6 +100,42 @@
 ## @{
 
 
+def _kill_process(pid: int) -> bool:
+    """Kill a POSIX process and its descendants, returning whether it was killed."""
+    # Collect descendants before signalling, while their parent PIDs still identify them.
+    pids = [pid]
+    for parent_pid in pids:
+        try:
+            try:
+                p = Popen(["pgrep", "-P", str(parent_pid)], stdout=PIPE)
+            except FileNotFoundError:
+                # POSIX ps does not support selecting by parent PID.
+                with Popen(["ps", "-A", "-o", "pid=", "-o", "ppid="], stdout=PIPE) as p:
+                    if p.stdout is not None:
+                        for line in p.stdout:
+                            fields = line.split()
+                            if len(fields) == 2 and all(field.isdigit() for field in fields):
+                                if int(fields[1]) == parent_pid:
+                                    pids.append(int(fields[0]))
+            else:
+                with p:
+                    if p.stdout is not None:
+                        for line in p.stdout:
+                            if line.strip().isdigit():
+                                pids.append(int(line))
+        except OSError as ex:
+            _logger.info("Unable to enumerate child processes: %r", ex)
+    killed = False
+    if sys.platform != "win32":
+        for process_pid in pids:
+            try:
+                os.kill(process_pid, signal.SIGKILL)
+                killed = killed or process_pid == pid
+            except OSError:
+                pass
+    return killed
+
+
 def handle_process_output(
     process: Union["Git.AutoInterrupt", Popen],
     stdout_handler: Union[
@@ -165,7 +202,7 @@ def pump_stream(
 
         except Exception as ex:
             _logger.error(f"Pumping {name!r} of cmd({remove_password_if_present(cmdline)}) failed due to: {ex!r}")
-            if "I/O operation on closed file" not in str(ex):
+            if not (isinstance(ex, ValueError) and stream.closed):
                 # Only reraise if the error was not due to the stream closing.
                 raise CommandError([f"<{name}-pump>"] + remove_password_if_present(cmdline), ex) from ex
         finally:
@@ -200,27 +237,23 @@ def pump_stream(
         threads.append(t)
 
     # FIXME: Why join? Will block if stdin needs feeding...
+    deadline = None if kill_after_timeout is None else time.monotonic() + kill_after_timeout
     for t in threads:
-        t.join(timeout=kill_after_timeout)
+        t.join(timeout=None if deadline is None else max(0, deadline - time.monotonic()))
         if t.is_alive():
             if isinstance(process, Git.AutoInterrupt):
+                if sys.platform != "win32" and process.proc is not None:
+                    _kill_process(process.proc.pid)
                 process._terminate()
             else:  # Don't want to deal with the other case.
                 raise RuntimeError(
                     "Thread join() timed out in cmd.handle_process_output()."
                     f" kill_after_timeout={kill_after_timeout} seconds"
                 )
-            if stderr_handler:
-                error_str: Union[str, bytes] = (
-                    f"error: process killed because it timed out. kill_after_timeout={kill_after_timeout} seconds"
-                )
-                if not decode_streams and isinstance(p_stderr, BinaryIO):
-                    # Assume stderr_handler needs binary input.
-                    error_str = cast(str, error_str)
-                    error_str = error_str.encode()
-                # We ignore typing on the next line because mypy does not like the way
-                # we inferred that stderr takes str or bytes.
-                stderr_handler(error_str)  # type: ignore[arg-type]
+            process._timeout_error = (
+                f"error: process killed because it timed out. kill_after_timeout={kill_after_timeout} seconds"
+            )
+            break
 
     if finalizer:
         finalizer(process)
@@ -326,7 +359,7 @@ class _AutoInterrupt:
     raise.
     """
 
-    __slots__ = ("proc", "args", "status")
+    __slots__ = ("proc", "args", "status", "_timeout_error")
 
     # If this is non-zero it will override any status code during _terminate, used
     # to prevent race conditions in testing.
@@ -336,6 +369,7 @@ def __init__(self, proc: Union[None, subprocess.Popen], args: Any) -> None:
         self.proc = proc
         self.args = args
         self.status: Union[int, None] = None
+        self._timeout_error: Optional[str] = None
 
     def _terminate(self) -> None:
         """Terminate the underlying process."""
@@ -344,37 +378,45 @@ def _terminate(self) -> None:
 
         proc = self.proc
         self.proc = None
-        if proc.stdin:
-            proc.stdin.close()
-        if proc.stdout:
-            proc.stdout.close()
-        if proc.stderr:
-            proc.stderr.close()
-        # Did the process finish already so we have a return code?
         try:
-            if proc.poll() is not None:
-                self.status = self._status_code_if_terminate or proc.poll()
+            if proc.stdin:
+                # A timed-out process may already have exited before input is flushed.
+                with contextlib.suppress(BrokenPipeError):
+                    proc.stdin.close()
+            # Did the process finish already so we have a return code?
+            try:
+                if proc.poll() is not None:
+                    self.status = self._status_code_if_terminate or proc.poll()
+                    return
+            except OSError as ex:
+                _logger.info("Ignored error after process had died: %r", ex)
+
+            # It can be that nothing really exists anymore...
+            if os is None or getattr(os, "kill", None) is None:
                 return
-        except OSError as ex:
-            _logger.info("Ignored error after process had died: %r", ex)
 
-        # It can be that nothing really exists anymore...
-        if os is None or getattr(os, "kill", None) is None:
-            return
+            # Try to kill it.
+            try:
+                proc.terminate()
+            except (OSError, AttributeError) as ex:
+                # On interpreter shutdown (notably on Windows), parts of the stdlib used by
+                # subprocess can already be torn down (e.g. `subprocess._winapi` becomes None),
+                # which can cause AttributeError during terminate(). In that case, we prefer
+                # to silently ignore to avoid noisy "Exception ignored in: __del__" messages.
+                _logger.info("Ignored error while terminating process: %r", ex)
+                return
+            # END exception handling
+        finally:
+            if proc.stdout:
+                proc.stdout.close()
+            if proc.stderr:
+                proc.stderr.close()
 
-        # Try to kill it.
         try:
-            proc.terminate()
-            status = proc.wait()  # Ensure the process goes away.
-
+            status = proc.wait()
             self.status = self._status_code_if_terminate or status
         except (OSError, AttributeError) as ex:
-            # On interpreter shutdown (notably on Windows), parts of the stdlib used by
-            # subprocess can already be torn down (e.g. `subprocess._winapi` becomes None),
-            # which can cause AttributeError during terminate(). In that case, we prefer
-            # to silently ignore to avoid noisy "Exception ignored in: __del__" messages.
-            _logger.info("Ignored error while terminating process: %r", ex)
-        # END exception handling
+            _logger.info("Ignored error while waiting for terminated process: %r", ex)
 
     def __del__(self) -> None:
         self._terminate()
@@ -395,6 +437,8 @@ def wait(self, stderr: Union[None, str, bytes] = b"") -> int:
         :raise git.exc.GitCommandError:
             If the return status is not 0.
         """
+        if self._timeout_error is not None:
+            stderr = self._timeout_error
         stderr_b = force_bytes(data=stderr, encoding="utf-8") or b""
         status: Union[int, None]
         if self.proc is not None:
@@ -1382,8 +1426,8 @@ def execute(
             2. Enumerating child processes requires ``pgrep -P``, or a ``ps`` command
                supporting the POSIX ``-A`` and ``-o`` options if ``pgrep`` is not
                installed. Effectiveness may vary on systems without these commands.
-            3. Deeper descendants do not receive signals, though they may sometimes
-               terminate as a consequence of their parent processes being killed.
+            3. Descendants are enumerated before signalling. Processes that detach
+               or spawn after enumeration may not receive signals.
             4. `kill_after_timeout` uses ``SIGKILL``, which can have negative side
                effects on a repository. For example, stale locks in case of
                :manpage:`git-gc(1)` could render the repository incapable of accepting
@@ -1537,43 +1581,9 @@ def execute(
             timeout = kill_after_timeout
 
             def kill_process(pid: int) -> None:
-                """Callback to kill a process.
-
-                This callback implementation would be ineffective and unsafe on Windows.
-                """
-                child_pids = []
-                try:
-                    p = Popen(["pgrep", "-P", str(pid)], stdout=PIPE)
-                except FileNotFoundError:
-                    # POSIX ps does not support selecting by parent PID.
-                    with Popen(["ps", "-A", "-o", "pid=", "-o", "ppid="], stdout=PIPE) as p:
-                        if p.stdout is not None:
-                            for line in p.stdout:
-                                fields = line.split()
-                                if len(fields) == 2 and all(field.isdigit() for field in fields):
-                                    if int(fields[1]) == pid:
-                                        child_pids.append(int(fields[0]))
-                else:
-                    with p:
-                        if p.stdout is not None:
-                            for line in p.stdout:
-                                if line.strip().isdigit():
-                                    child_pids.append(int(line))
-                try:
-                    os.kill(pid, signal.SIGKILL)
-                    for child_pid in child_pids:
-                        try:
-                            os.kill(child_pid, signal.SIGKILL)
-                        except OSError:
-                            pass
-                    # Tell the main routine that the process was killed.
+                if _kill_process(pid):
                     assert kill_check is not None
                     kill_check.set()
-                except OSError:
-                    # It is possible that the process gets completed in the duration
-                    # after timeout happens and before we try to kill the process.
-                    pass
-                return
 
             def make_timeout_error() -> Union[str, bytes]:
                 err = f'Timeout: the command "{" ".join(redacted_command)}" did not complete in {timeout:g} secs.'
 
@@ -988,7 +988,7 @@ def stdout_handler(line: str) -> None:
             # even if there is an output).
             if not output:
                 raise
-            elif stderr_text:
+            elif stderr_text or proc._timeout_error:
                 _logger.warning("Error lines received while fetching: %s", stderr_text)
                 output.error = e
 
 
@@ -1,3 +1,9 @@
+from subprocess import PIPE
+import sys
+import threading
+
+import pytest
+
 from git.cmd import Git
 
 
@@ -31,3 +37,36 @@ def test_autointerrupt_terminate_ignores_attributeerror():
 
     # Ensure the reference is cleared to avoid repeated attempts.
     assert ai.proc is None
+
+
+def test_autointerrupt_terminate_closes_buffered_stdin():
+    ai = Git().hash_object("--stdin", as_process=True, istream=PIPE)
+    proc = ai.proc
+    proc.stdin.write(b"buffered input")
+    try:
+        ai._terminate()
+        assert proc.stdin.closed
+        assert proc.stdout.closed
+        assert proc.stderr.closed
+    finally:
+        proc.stdout.close()
+        proc.stderr.close()
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX SIGTERM handling")
+def test_autointerrupt_terminate_closes_pipes_before_waiting():
+    script = (
+        "import signal, sys; signal.signal(signal.SIGTERM, signal.SIG_IGN); "
+        "print('ready', flush=True); sys.stdin.read(); sys.stdout.buffer.write(b'x' * 1000000)"
+    )
+    ai = Git().execute([sys.executable, "-c", script], as_process=True, istream=PIPE)
+    proc = ai.proc
+    assert proc.stdout.readline() == b"ready\n"
+    watchdog = threading.Timer(5, proc.kill)
+    watchdog.start()
+    try:
+        ai._terminate()
+        assert proc.returncode != -9, "cleanup waited for exit with undrained output pipes"
+    finally:
+        watchdog.cancel()
+        watchdog.join()