Fix a writer race, the Android shell, code pages, and worker-thread shells

Fixes from another review of the pipeline changes:

- A cmd2 write could hang when a subprocess wrote through the descriptor
  relay at the same time. The write fast path checked the consumer's pipe
  for room, then wrote without a terminal lend. The relay thread writes
  to the same pipe, and could fill it in between. The write then blocked
  with no lend, while the pager, waiting for a key, stayed stopped. Once
  a relay exists, every write is now made under a lend. Without one,
  writes the pipe has room for still need none. DescriptorRelay.idle()
  had no other caller, and is gone.

- The start gate ran /bin/sh, which Android does not have, so every
  terminal pipe there failed to start. It now uses the POSIX shell that
  Popen(shell=True) itself uses: /system/bin/sh on Android.

- On Windows before Python 3.14, a console code page Python has no
  cpNNNNN codec for fell back to UTF-8, although Python knows many of
  them by other names. Pipes now try those names too: the ISO-8859 family,
  KOI8-R and KOI8-U, US-ASCII, GB18030, EUC-JP and EUC-KR. Python 3.14
  has a codec for every code page Windows supports.

- A shell command run from a worker thread joined the main thread's
  terminal pipeline. Joining relays job-control stops to the main thread,
  and may change signal handlers, which only the main thread may do: as a
  session leader, it raised ValueError. Only the main thread's shell
  commands join now.

Tests cover each, and each fails without its fix: every write is lent
once a relay exists; the gate uses Android's shell; code pages map to
their codecs on any platform; a worker thread's shell command stays out
of the pipeline. The relay tests that used idle() now use flush().

Author: tleonhardt

CHANGELOG.md

@@ -12,9 +12,10 @@
       the pipeline's job, so both processes receive Ctrl-C and Ctrl-Z
     - On Windows, fixed piped output appearing garbled in console programs such as `more`
       (`help -v | more`), a regression in 4.2.4. Pipes were written as UTF-8, but console programs
-      decode their input with the console's code page. Pipes now use that code page, and characters
-      it cannot represent are replaced rather than failing the command. Without a console, and on
-      other platforms, pipes still use UTF-8
+      decode their input with the console's code page. Pipes now use that code page, including ones
+      Python names otherwise, such as 20866 (KOI8-R) or 28591 (ISO-8859-1), and characters it cannot
+      represent are replaced rather than failing the command. Without a console, with a code page
+      Python has no codec for, and on other platforms, pipes still use UTF-8
 
 ## 4.2.4 (September 8, 2026)
 

cmd2/cmd2.py

@@ -3375,9 +3375,11 @@ def _redirect_output(self, statement: Statement) -> utils.RedirectionSavedState:
                     # user's shell as before.
                     import shlex
 
-                    user_shell = shlex.quote(kwargs.get("executable", "/bin/sh"))
+                    # The POSIX shell Popen() itself runs with shell=True
+                    posix_shell = "/system/bin/sh" if hasattr(sys, "getandroidapilevel") else "/bin/sh"
+                    user_shell = shlex.quote(kwargs.get("executable", posix_shell))
                     popen_command = f"read -r _ || exit 1; exec {user_shell} -c {shlex.quote(statement.redirect_to)}"
-                    kwargs["executable"] = "/bin/sh"
+                    kwargs["executable"] = posix_shell
 
             with contextlib.ExitStack() as terminal_stack, contextlib.ExitStack() as gate_stack:
                 if terminal_fd is not None:
@@ -5007,9 +5009,15 @@ def do_shell(self, args: argparse.Namespace) -> None:
         # the terminal per write. Run the command inside the pipeline's job instead, for as
         # long as it runs: the consumer keeps the terminal, and Ctrl-C and Ctrl-Z reach both
         # processes, as they would in a shell pipeline.
+        # A worker thread's command stays out of it: job control relays stops to the main
+        # thread, and only the main thread may change signal handlers.
         pipeline = self._cur_pipe_proc_reader
         pipeline_group = None
-        if pipeline is not None and not isinstance(self.stdout, utils.StdSim):  # type: ignore[unreachable]
+        if (
+            pipeline is not None
+            and threading.current_thread() is threading.main_thread()
+            and not isinstance(self.stdout, utils.StdSim)  # type: ignore[unreachable]
+        ):
             pipeline_group = pipeline._terminal_group
 
         # Prevent KeyboardInterrupts while in the shell process. The shell process still

cmd2/utils.py

@@ -536,22 +536,57 @@ def write(self, b: bytes) -> None:
                 self.std_sim_instance.flush()
 
 
+# Windows code pages Python names other than cpNNNNN. Before Python 3.14, which covers every
+# code page Windows supports, a console set to one of these would otherwise get UTF-8.
+_CODE_PAGE_CODECS = {
+    20127: "ascii",
+    20866: "koi8_r",
+    21866: "koi8_u",
+    28591: "latin_1",
+    28592: "iso8859_2",
+    28593: "iso8859_3",
+    28594: "iso8859_4",
+    28595: "iso8859_5",
+    28596: "iso8859_6",
+    28597: "iso8859_7",
+    28598: "iso8859_8",
+    28599: "iso8859_9",
+    28603: "iso8859_13",
+    28605: "iso8859_15",
+    51932: "euc_jp",
+    51949: "euc_kr",
+    54936: "gb18030",
+}
+
+
+def _code_page_encoding(code_page: int) -> str | None:
+    """Return the name of the Python codec for a Windows code page, or None if there is none.
+
+    :param code_page: a Windows code page identifier, such as 437
+    """
+    import codecs
+
+    for name in (f"cp{code_page}", _CODE_PAGE_CODECS.get(code_page)):
+        if name is not None:
+            with contextlib.suppress(LookupError):
+                # Normalized, so that code page 65001 is reported as utf-8
+                return codecs.lookup(name).name
+    return None
+
+
 def _pipe_encoding() -> str:
     """Return the encoding for output cmd2 pipes to a shell command.
 
     Windows console programs such as more, sort, and findstr decode piped input with the
     console's output code page, and would show UTF-8 as mojibake. Elsewhere, and on Windows
-    without a console, use UTF-8.
+    without a console or with a code page Python cannot encode, use UTF-8.
     """
     if sys.platform == "win32":
-        import codecs
         import ctypes
 
         code_page = ctypes.windll.kernel32.GetConsoleOutputCP()
         if code_page:
-            with contextlib.suppress(LookupError):
-                # Normalized, so that code page 65001 is reported as utf-8
-                return codecs.lookup(f"cp{code_page}").name
+            return _code_page_encoding(code_page) or "utf-8"
     return "utf-8"
 
 
@@ -1108,11 +1143,6 @@ def _producer_blocked(self) -> bool:
             poller.register(self.write_fd, select.POLLOUT)
             return not poller.poll(0)
 
-    def idle(self) -> bool:
-        """Whether all output written to the relay has reached the consumer's pipe."""
-        with self._lock:
-            return self._done or self._sent >= self._received + self._unread()
-
     def flush(self) -> None:
         """Wait until output already written to the relay has reached the consumer's pipe.
 
@@ -1248,8 +1278,10 @@ def write(self, b: Any) -> int:
         fd = super().fileno()
         written = 0
         try:
-            # Output a producer wrote to the relay comes first, which may need a lend.
-            if self._relay is None or self._relay.idle():
+            # Once a relay exists, its thread writes to the consumer's pipe too, and can fill it
+            # between a check for room and the write. So only without one is room checked
+            # without a lend. Output a producer wrote to the relay comes first in any case.
+            if self._relay is None:
                 # Once there is room, a write of at most PIPE_BUF bytes does not block.
                 while written < len(view) and self._poller.poll(0):
                     written += os.write(fd, view[written : written + select.PIPE_BUF])

tests/test_cmd2.py

@@ -1,8 +1,10 @@
 """Cmd2 unit/functional testing"""
 
+import contextlib
 import io
 import os
 import signal
+import subprocess
 import sys
 import tempfile
 import threading
@@ -971,6 +973,62 @@ def start_pipe(*args, **kwargs):
     assert popen.call_args.kwargs["stdin"].closed
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control")
+@pytest.mark.parametrize("android", [False, True])
+def test_terminal_pipe_start_gate_uses_the_platform_shell(redirection_app, mocker, monkeypatch, android) -> None:
+    """The start gate runs in the POSIX shell Popen() itself would use: Android has no /bin/sh."""
+    if android:
+        monkeypatch.setattr(sys, "getandroidapilevel", lambda: 30, raising=False)
+    else:
+        monkeypatch.delattr(sys, "getandroidapilevel", raising=False)
+    monkeypatch.delenv("SHELL", raising=False)
+    popen = mocker.patch("subprocess.Popen", autospec=True)
+    popen.return_value.returncode = 127
+    terminal_stream = mocker.Mock()
+    terminal_stream.isatty.return_value = True
+    terminal_stream.fileno.return_value = 10
+    redirection_app.stdout = terminal_stream
+    mocker.patch("os.tcgetpgrp", return_value=os.getpgrp())
+    mocker.patch("cmd2.utils.ProcReader")
+    redirection_app.onecmd_plus_hooks("print_output | less")
+    shell = "/system/bin/sh" if android else "/bin/sh"
+    assert popen.call_args.kwargs["executable"] == shell
+    # With SHELL unset, the gate hands the command to that shell, too.
+    assert f"exec {shell} -c less" in popen.call_args.args[0]
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX terminal job control")
+def test_shell_from_a_worker_thread_stays_out_of_the_pipeline(base_app, tmp_path) -> None:
+    """Joining a pipeline's job means relaying stops to the main thread, and may change signal handlers.
+
+    Only the main thread may do that, so a shell command run from another thread does not join.
+    """
+    lent = []
+
+    @contextlib.contextmanager
+    def _lend_terminal():
+        lent.append(True)
+        yield
+
+    spawned = []
+    real_popen = subprocess.Popen
+
+    def popen(*args, **kwargs):
+        spawned.append(kwargs)
+        return real_popen(*args, **kwargs)
+
+    base_app._cur_pipe_proc_reader = mock.Mock(_terminal_group=os.getpgrp(), _lend_terminal=_lend_terminal)
+    with (tmp_path / "output").open("w+") as output, mock.patch("subprocess.Popen", popen):
+        base_app.stdout = output
+        worker = threading.Thread(target=base_app.do_shell, args=("echo worker",))
+        worker.start()
+        worker.join(10)
+        output.seek(0)
+        assert output.read() == "worker\n"
+    assert "process_group" not in spawned[0]
+    assert not lent
+
+
 def test_restore_output_resets_pipe_state_when_the_wait_fails(base_app) -> None:
     """A failed handback while waiting for the pipe process must not leave it current.
 

tests/test_utils.py

@@ -212,6 +212,31 @@ def test_pipe_encoding_follows_the_console_code_page(monkeypatch, code_page, enc
     assert cu._pipe_encoding() == encoding
 
 
+@pytest.mark.parametrize(
+    ("code_page", "sample", "codec"),
+    [
+        (437, "─", "cp437"),
+        # chcp 65001
+        (65001, "─", "utf-8"),
+        # Code pages Python names otherwise. Python 3.14 on Windows has a cpNNNNN codec for them too.
+        (20866, "Ж", "koi8_r"),
+        (28591, "é", "latin_1"),
+        (20127, "a", "ascii"),
+    ],
+)
+def test_code_page_encoding(code_page, sample, codec) -> None:
+    encoding = cu._code_page_encoding(code_page)
+    assert encoding is not None
+    assert sample.encode(encoding) == sample.encode(codec)
+
+
+@pytest.mark.parametrize("code_page", [50220, 1])
+def test_code_page_encoding_without_a_codec(code_page) -> None:
+    if sys.platform == "win32" and sys.version_info >= (3, 14) and code_page == 50220:
+        pytest.skip("Python 3.14 on Windows may support every code page Windows does")
+    assert cu._code_page_encoding(code_page) is None
+
+
 @pytest.fixture
 def pr_none():
     import subprocess
@@ -498,12 +523,13 @@ def slow_read(fd, size):
             release.wait(5)
         return data
 
-    answers = []
+    flushed = threading.Event()
     asking = threading.Event()
 
     def ask() -> None:
         asking.set()
-        answers.append(relay.idle())
+        relay.flush()
+        flushed.set()
 
     # The relay thread must start inside the patch, or it is already in the real read.
     with mock.patch("os.read", side_effect=slow_read):
@@ -515,14 +541,12 @@ def ask() -> None:
             asker = threading.Thread(target=ask)
             asker.start()
             assert asking.wait(5)
-            asker.join(0.3)
-            # While the output is in the relay's hands, idle() waits for the relay or says the
-            # output is still pending. Once released, the relay passes it on, so a later answer
-            # may rightly be that nothing is left.
-            answered_during_read = list(answers)
+            # While the output is in the relay's hands, flush() must wait for it.
+            assert not flushed.wait(0.3)
             release.set()
+            # Once released, the relay passes the output on, and flush() returns.
+            assert flushed.wait(5)
             asker.join(5)
-        assert answered_during_read in ([], [False])
     finally:
         release.set()
         relay.close_write_fd()
@@ -544,11 +568,38 @@ def test_descriptor_relay_that_finished_has_nothing_pending() -> None:
             time.sleep(0.01)
         assert relay._done
         relay.flush()
-        assert relay.idle()
     finally:
         os.close(read_fd)
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="POSIX pipes")
+@pytest.mark.parametrize("relaying", [False, True])
+def test_pipeline_writer_lends_every_write_once_a_relay_exists(relaying) -> None:
+    """The relay's thread writes to the consumer's pipe too. It could fill the pipe between a check for room and a write.
+
+    So once a relay exists, even a write the pipe has room for is made with the terminal lent,
+    in case it blocks. Without one, such a write needs no lend.
+    """
+    read_fd, write_fd = os.pipe()
+    lends = []
+
+    @contextlib.contextmanager
+    def lend_terminal():
+        lends.append(1)
+        yield
+
+    writer = cu._PipelineWriter(write_fd, mock.Mock(_lend_terminal=lend_terminal))
+    try:
+        if relaying:
+            writer.fileno()
+        assert writer.write(b"fits") == 4
+        assert lends == ([1] if relaying else [])
+    finally:
+        writer.close()
+        assert os.read(read_fd, 100) == b"fits"
+        os.close(read_fd)
+
+
 @pytest.mark.skipif(sys.platform == "win32", reason="POSIX pipes")
 def test_pipeline_writer_relay_passes_consumer_exit_to_the_producer() -> None:
     import subprocess