fix(cli): don't change the global SIGPIPE disposition (#834)

* fix(cli): don't change the global SIGPIPE disposition

Setting SIGPIPE to SIG_DFL in Application.run() made any later socket
send() to a closed peer kill the whole process; rpyc servers died this
way when a client disconnected. Catch BrokenPipeError around the run
body instead, as the Python signal docs recommend, keeping the piped-
help fix from #727.

Assisted-by: ClaudeCode:claude-fable-5

* refactor(cli): flatten the nested try in Application.run

Move the parse/dispatch body into _parse_and_dispatch() so run() has a
single try for BrokenPipeError. Also drops a type-ignore: the subapp
result now assigns to a plain Application.

Assisted-by: ClaudeCode:claude-fable-5

* fix(cli): flush stdout before exit so a late EPIPE is caught

Help output can sit in the stdout buffer after head exits without any
write failing; the EPIPE then only surfaced at interpreter shutdown,
printing 'Exception ignored ... BrokenPipeError' to stderr (seen on
Linux CI). Flush inside the guarded region so the handler catches it,
and add a deterministic regression test for the race.

Assisted-by: ClaudeCode:claude-fable-5

Author: henryiii

plumbum/cli/application.py

@@ -918,6 +918,46 @@ def _positional_validate(
 
         return out_args
 
+    def _parse_and_dispatch(self, argv: list[str]) -> tuple[Application, int]:
+        """Parses ``argv``, runs switches and ``main()``; returns the final
+        instance (the nested subcommand's, if one ran) and the return code."""
+        inst: Application = self
+        retcode: int | None = 0
+        try:
+            swfuncs, tailargs = self._parse_args(argv)
+            ordered, tailargs = self._validate_args(swfuncs, tailargs)
+        except ShowHelp:
+            self.help()
+        except ShowHelpAll:
+            self.helpall()
+        except ShowVersion:
+            self.version()
+        except ShowCompletion:
+            info = swfuncs[self.completions.__func__]  # type: ignore[attr-defined]
+            self._print_completion(info.val[0])
+        except SwitchError as ex:
+            print(T_("Error: {0}").format(ex))
+            print(T_("------"))
+            self.help()
+            retcode = 2
+        else:
+            for f, a in ordered:
+                f(self, *a)
+
+            cleanup = None
+            if not self.nested_command or self.CALL_MAIN_IF_NESTED_COMMAND:
+                retcode = self.main(*tailargs)
+                cleanup = functools.partial(self.cleanup, retcode)
+            if not retcode and self.nested_command:
+                subapp, argv = self.nested_command
+                subapp.parent = self
+                inst, retcode = subapp.run(argv, exit=False)
+
+            if cleanup:
+                cleanup()
+
+        return inst, retcode or 0
+
     @typing.overload
     @classmethod
     def run(
@@ -955,55 +995,27 @@ def run(
            Setting ``exit`` to ``False`` is intended for testing/debugging purposes only -- do
            not override it in other situations.
         """
-        # Handle SIGPIPE to avoid BrokenPipeError when output is piped (e.g., to head)
-        # This is only available on Unix systems
-        with contextlib.suppress(ImportError, AttributeError):
-            import signal
-
-            signal.signal(signal.SIGPIPE, signal.SIG_DFL)
-
         if argv is None:
             argv = sys.argv
         cls.autocomplete(argv)
         argv = list(argv)
         inst = cls(argv.pop(0))
-        retcode = 0
         try:
-            swfuncs, tailargs = inst._parse_args(argv)
-            ordered, tailargs = inst._validate_args(swfuncs, tailargs)
-        except ShowHelp:
-            inst.help()
-        except ShowHelpAll:
-            inst.helpall()
-        except ShowVersion:
-            inst.version()
-        except ShowCompletion:
-            info = swfuncs[inst.completions.__func__]  # type: ignore[attr-defined]
-            inst._print_completion(info.val[0])
-        except SwitchError as ex:
-            print(T_("Error: {0}").format(ex))
-            print(T_("------"))
-            inst.help()
-            retcode = 2
-        else:
-            for f, a in ordered:
-                f(inst, *a)
-
-            cleanup = None
-            if not inst.nested_command or inst.CALL_MAIN_IF_NESTED_COMMAND:
-                retcode = inst.main(*tailargs)
-                cleanup = functools.partial(inst.cleanup, retcode)
-            if not retcode and inst.nested_command:
-                subapp, argv = inst.nested_command
-                subapp.parent = inst
-                inst_app, retcode = subapp.run(argv, exit=False)
-                inst = inst_app  # type: ignore[assignment]
-
-            if cleanup:
-                cleanup()
-
-            if retcode is None:
-                retcode = 0
+            inst, retcode = inst._parse_and_dispatch(argv)  # type: ignore[assignment]
+            if exit:
+                # surface an EPIPE now, while we can still handle it below
+                sys.stdout.flush()
+        except BrokenPipeError:
+            # The reader closed the pipe (e.g. output piped to ``head``).
+            # Never change the SIGPIPE disposition instead: that would make a
+            # socket send() to a closed peer kill the whole process.
+            retcode = 1
+            if exit:
+                # Point stdout at devnull so the interpreter's final flush
+                # doesn't raise on whatever is still buffered.
+                with contextlib.suppress(OSError, ValueError):
+                    devnull = os.open(os.devnull, os.O_WRONLY)
+                    os.dup2(devnull, sys.stdout.fileno())
 
         if exit:
             sys.exit(retcode)

tests/test_cli.py

@@ -1,5 +1,9 @@
 from __future__ import annotations
 
+import signal
+
+import pytest
+
 from plumbum import cli, local
 from plumbum.cli.switches import SwitchInfo
 from plumbum.cli.terminal import get_terminal_size
@@ -493,6 +497,60 @@ def main(self):
         # No traceback should be in stderr
         assert "Traceback" not in result.stderr, f"Traceback in stderr: {result.stderr}"
 
+    @pytest.mark.skipif(not hasattr(signal, "SIGPIPE"), reason="requires SIGPIPE")
+    def test_broken_pipe_at_shutdown_flush(self, tmp_path):
+        """Output still buffered when the reader is gone must not print
+        'Exception ignored ... BrokenPipeError' at interpreter shutdown."""
+        import subprocess
+        import sys
+
+        script = tmp_path / "straggler.py"
+        script.write_text(
+            f"""
+import sys, time
+sys.path.insert(0, {str(local.cwd)!r})
+from plumbum.cli import Application
+
+class App(Application):
+    def main(self):
+        print("1\\n2\\n3\\n4\\n5", flush=True)
+        time.sleep(0.5)  # let head exit
+        print("straggler")  # stays in the buffer until shutdown
+
+if __name__ == '__main__':
+    App.run()
+"""
+        )
+        result = subprocess.run(
+            f"{sys.executable} {script} | head -5",
+            shell=True,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        assert result.stderr == ""
+
+    @pytest.mark.skipif(not hasattr(signal, "SIGPIPE"), reason="requires SIGPIPE")
+    def test_run_keeps_sigpipe_disposition(self):
+        # SIG_DFL would make a socket send() to a closed peer kill the whole
+        # process (e.g. an rpyc server) instead of raising BrokenPipeError.
+        before = signal.getsignal(signal.SIGPIPE)
+        try:
+            _, rc = SimpleApp.run(["foo", "--bacon=2"], exit=False)
+        finally:
+            after = signal.getsignal(signal.SIGPIPE)
+            signal.signal(signal.SIGPIPE, before)
+        assert rc == 0
+        assert after == before
+
+    def test_run_handles_broken_pipe(self):
+        class BrokenApp(cli.Application):
+            def main(self):
+                raise BrokenPipeError
+
+        _, rc = BrokenApp.run(["app"], exit=False)
+        assert rc == 1
+
 
 class ExcludesApp(cli.Application):
     alpha = cli.Flag("--alpha")