@@ -1368,6 +1368,8 @@ def communicate(self, input=None, timeout=None):
 
         if self._communication_started and input:
             raise ValueError("Cannot send input after starting communication")
+        if input and self.stdin is not None and self.stdin.closed:
+            raise ValueError("Cannot send input to closed stdin")
 
         # Optimization: If we are not worried about timeouts, we haven't
         # started communicating, and we have one or zero pipes, using select()
@@ -1376,13 +1378,14 @@ def communicate(self, input=None, timeout=None):
             [self.stdin, self.stdout, self.stderr].count(None) >= 2):
             stdout = None
             stderr = None
+            empty_output = '' if self.text_mode else b''
             if self.stdin:
                 self._stdin_write(input)
             elif self.stdout:
-                stdout = self.stdout.read()
+                stdout = self.stdout.read() if not self.stdout.closed else empty_output
                 self.stdout.close()
             elif self.stderr:
-                stderr = self.stderr.read()
+                stderr = self.stderr.read() if not self.stderr.closed else empty_output
                 self.stderr.close()
             self.wait()
         else:
@@ -1801,8 +1804,9 @@ def _wait(self, timeout):
 
 
         def _readerthread(self, fh, buffer):
-            buffer.append(fh.read())
-            fh.close()
+            if not fh.closed:
+                buffer.append(fh.read())
+                fh.close()
 
 
         def _writerthread(self, input):
@@ -1869,8 +1873,9 @@ def _communicate(self, input, endtime, orig_timeout):
                 self.stderr.close()
 
             # All data exchanged.  Translate lists into strings.
-            stdout = stdout[0] if stdout else None
-            stderr = stderr[0] if stderr else None
+            empty_output = '' if self.text_mode else b''
+            stdout = empty_output.join(stdout) if stdout is not None else None
+            stderr = empty_output.join(stderr) if stderr is not None else None
 
             return (stdout, stderr)
 

@@ -943,6 +943,43 @@ def test_communicate_stderr(self):
         self.assertEqual(stdout, None)
         self.assertEqual(stderr, b"pineapple")
 
+    def test_communicate_closed_output_pipe(self):
+        for stream, extra_pipe, text, timeout in itertools.product(
+            ('stdout', 'stderr'), (False, True), (False, True),
+            (None, support.SHORT_TIMEOUT),
+        ):
+            with self.subTest(stream=stream, extra_pipe=extra_pipe,
+                              text=text, timeout=timeout):
+                pipes = {stream: subprocess.PIPE}
+                if extra_pipe:
+                    pipes.update(stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+                with subprocess.Popen(ZERO_RETURN_CMD, text=text, **pipes) as p:
+                    getattr(p, stream).close()
+                    output = p.communicate(timeout=timeout)
+                empty = '' if text else b''
+                self.assertEqual(output, tuple(
+                    empty if name in pipes else None
+                    for name in ('stdout', 'stderr')))
+                self.assertEqual(p.returncode, 0)
+
+    def test_communicate_closed_stdin_with_input(self):
+        for extra_pipe, text, timeout in itertools.product(
+            (False, True), (False, True), (None, support.SHORT_TIMEOUT),
+        ):
+            with self.subTest(extra_pipe=extra_pipe, text=text, timeout=timeout):
+                with subprocess.Popen(
+                    ZERO_RETURN_CMD, stdin=subprocess.PIPE, text=text,
+                    stdout=subprocess.PIPE if extra_pipe else None,
+                ) as p:
+                    p.stdin.close()
+                    with self.assertRaisesRegex(ValueError, 'closed stdin'):
+                        p.communicate('unused' if text else b'unused',
+                                      timeout=timeout)
+                    output = p.communicate(timeout=timeout)
+                empty = '' if text else b''
+                self.assertEqual(output, (empty if extra_pipe else None, None))
+                self.assertEqual(p.returncode, 0)
+
     def test_communicate(self):
         p = subprocess.Popen([sys.executable, "-c",
                               'import sys,os;'
@@ -3557,7 +3594,7 @@ def test_communicate_BrokenPipeError_stdin_write(self):
         # By not setting stdout or stderr or a timeout we force the fast path
         # that just calls _stdin_write() internally due to our mock.
         proc = subprocess.Popen(ZERO_RETURN_CMD)
-        with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin:
+        with proc, mock.patch.object(proc, 'stdin', closed=False) as mock_proc_stdin:
             mock_proc_stdin.write.side_effect = BrokenPipeError
             proc.communicate(b'stuff')  # Should swallow the BrokenPipeError.
             mock_proc_stdin.write.assert_called_once_with(b'stuff')
@@ -3569,7 +3606,7 @@ def test_communicate_BrokenPipeError_stdin_flush(self):
         proc = subprocess.Popen([sys.executable, '-h'],
                                 stdin=subprocess.PIPE,
                                 stdout=subprocess.PIPE)
-        with proc, mock.patch.object(proc, 'stdin') as mock_proc_stdin, \
+        with proc, mock.patch.object(proc, 'stdin', closed=False) as mock_proc_stdin, \
                 open(os.devnull, 'wb') as dev_null:
             mock_proc_stdin.flush.side_effect = BrokenPipeError
             # because _communicate registers a selector using proc.stdin...

diff --git a/Misc/NEWS.d/next/Library/2026-10-20-04-15-00.gh-issue-131064.n4Bw8T.rst b/Misc/NEWS.d/next/Library/2026-10-20-04-15-00.gh-issue-131064.n4Bw8T.rst
@@ -0,0 +1,3 @@
+Make :meth:`subprocess.Popen.communicate` consistently handle pipes closed
+before the call. Closed output pipes produce empty output. Sending nonempty
+input to a closed input pipe raises :exc:`ValueError`.