gh-47798: run_pipeline: reject session/group kwargs and pipeline-level stderr=STDOUT

start_new_session and process_group: each command is spawned as a
sibling child of the calling process, so applying these per command
yields N separate sessions/groups rather than a single group spanning
the pipeline.

stderr=STDOUT at the pipeline level: would merge each non-final
command's stderr into the next command's stdin.  Per-command
PipelineCommand(stderr=STDOUT) covers the legitimate use.

Leave a note at Popen.__init__ reminding future kwarg additions to
consider run_pipeline forwarding.

Author: gpshead

Doc/library/subprocess.rst

@@ -316,6 +316,14 @@ underlying :class:`Popen` interface can be used directly.
    injection surface this function exists to avoid. When one command
    genuinely needs shell interpretation (a glob, or a shell builtin),
    wrap it in a :class:`PipelineCommand` with ``shell=True``.
+   ``start_new_session`` and ``process_group`` are also rejected: each
+   command is spawned as a sibling child of the calling process, so
+   applying these per command does not produce a single process group
+   spanning the pipeline. ``stderr=STDOUT`` at the pipeline level is
+   rejected because it would merge each non-final command's stderr into
+   the next command's stdin; use a :class:`PipelineCommand` with
+   ``stderr=STDOUT`` for the one command that needs it, or
+   ``capture_output=True`` to capture stderr from every command.
 
    .. rubric:: Standard error handling
 
@@ -338,11 +346,8 @@ underlying :class:`Popen` interface can be used directly.
 
    To exempt one command from the shared stderr pipe, wrap it in a
    :class:`PipelineCommand` with ``stderr=DEVNULL`` (discard) or
-   ``stderr=STDOUT`` (merge into that command's stdout). Note that
-   passing ``stderr=STDOUT`` at the *pipeline* level redirects every
-   command's stderr to its own stdout, which for non-final commands
-   means stderr feeds the next command's stdin -- rarely what callers
-   want.
+   ``stderr=STDOUT`` (merge into that command's stdout).
+   ``stderr=STDOUT`` at the *pipeline* level is rejected.
 
    Examples::
 

Lib/subprocess.py

@@ -1108,6 +1108,26 @@ def run_pipeline(*commands, input=None, capture_output=False, timeout=None,
         raise ValueError(
             'executable= is not supported by run_pipeline')
 
+    if kwargs.get('stderr') is STDOUT:
+        raise ValueError(
+            'stderr=STDOUT at the run_pipeline level would merge each '
+            "non-final command's stderr into the next command's stdin.  "
+            'Use PipelineCommand(cmd, stderr=STDOUT) for a single command, '
+            'or capture_output=True to capture stderr from every command.')
+
+    if kwargs.get('start_new_session') or kwargs.get('process_group') is not None:
+        # run_pipeline spawns each command as a sibling child of this
+        # process, so a per-command session/group does not give the shell
+        # "one process group per pipeline" semantic that callers passing
+        # these almost certainly want.  Reject for now; a feature that
+        # places every command in a single new group is a possible
+        # follow-on.
+        raise ValueError(
+            'start_new_session and process_group are not supported by '
+            'run_pipeline; each command is spawned as a sibling child, '
+            'so a per-command session or group does not yield a single '
+            'process group for the pipeline')
+
     commands = tuple(c if isinstance(c, PipelineCommand) else PipelineCommand(c)
                      for c in commands)
 
@@ -1581,6 +1601,9 @@ class Popen:
     """
     _child_created = False  # Set here since __del__ checks it
 
+    # When adding a new keyword here, consider whether forwarding it to
+    # every command in run_pipeline() makes sense; if not, reject or
+    # special-case it there.
     def __init__(self, args, bufsize=-1, executable=None,
                  stdin=None, stdout=None, stderr=None,
                  preexec_fn=None, close_fds=True,

Lib/test/test_subprocess.py

@@ -2208,6 +2208,23 @@ def test_pipeline_capture_output_conflict(self):
             )
         self.assertIn('capture_output', str(cm.exception))
 
+    def test_pipeline_session_group_rejected(self):
+        """start_new_session= and process_group= are rejected.
+
+        Each command is spawned as a sibling child of this process, so
+        per-command sessions/groups would not yield a single process
+        group spanning the pipeline.
+        """
+        for kw in ({'start_new_session': True}, {'process_group': 0}):
+            with self.subTest(kw=kw):
+                with self.assertRaises(ValueError) as cm:
+                    subprocess.run_pipeline(
+                        [sys.executable, '-c', 'pass'],
+                        [sys.executable, '-c', 'pass'],
+                        **kw,
+                    )
+                self.assertIn('process group', str(cm.exception))
+
     def test_pipeline_close_fds_false_rejected(self):
         """Test that close_fds=False is rejected (would deadlock)"""
         with self.assertRaises(ValueError) as cm:
@@ -2538,19 +2555,20 @@ def test_pipeline_check_true_success(self):
         self.assertEqual(result.returncodes, (0, 0))
         self.assertEqual(result.stdout.strip(), 'ok')
 
-    def test_pipeline_stderr_to_stdout(self):
-        """stderr=STDOUT routes the final process's stderr to stdout"""
-        result = subprocess.run_pipeline(
-            [sys.executable, '-c', 'print("data")'],
-            [sys.executable, '-c',
-             'import sys; sys.stdout.write(sys.stdin.read()); '
-             'sys.stderr.write("ERR\\n")'],
-            stdout=subprocess.PIPE, stderr=subprocess.STDOUT
-        )
-        self.assertEqual(result.returncodes, [0, 0])
-        self.assertIn(b'data', result.stdout)
-        self.assertIn(b'ERR', result.stdout)
-        self.assertIsNone(result.stderr)
+    def test_pipeline_stderr_to_stdout_rejected(self):
+        """stderr=STDOUT at the pipeline level is rejected.
+
+        It would merge each non-final command's stderr into the next
+        command's stdin; PipelineCommand(stderr=STDOUT) is the
+        per-command spelling.
+        """
+        with self.assertRaises(ValueError) as cm:
+            subprocess.run_pipeline(
+                [sys.executable, '-c', 'pass'],
+                [sys.executable, '-c', 'pass'],
+                stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+            )
+        self.assertIn('PipelineCommand', str(cm.exception))
 
     def test_pipeline_intermediate_stdout_closed_in_parent(self):
         """Parent closes intermediate stdout so an early-exiting consumer