From 6752e1f334274ec6c484e82ecb794154ac5b2f0f Mon Sep 17 00:00:00 2001
From: Sora-bluesky <sora.bluesky.dev@gmail.com>
Date: Sun, 26 Jul 2026 19:37:19 +0900
Subject: [PATCH 1/3] fix(terminal): retire the compound-background rewriter
 instead of patching its scanner again
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

_rewrite_compound_background rewrote `A && B &` into `A && { B & }` to stop
a backgrounded compound subshell from wedging the worker on its held stdout
pipe (#68915, the vela/sal/combiagent leaks). Two things changed since:

1. #71008 fixed the hang at the process layer (orphan-held stdout pipes),
   and #68915 is closed. The rewrite no longer guards anything critical.
2. Review of the rewriter (#68948) kept producing inputs where the textual
   scan turns valid bash into invalid bash or silently changes program data:
   backtick substitutions, ${...} expansions, [[ ]] conditionals, $[ ]
   legacy arithmetic, array subscripts, heredoc payloads, $'...' ANSI-C
   strings — and `false && echo B &` observably changes $? even in the
   intended case. Each scanner marker added for one class surfaced the
   next, and runtime-created syntax (alias, eval) is out of reach of any
   pre-execution textual check.

A transform that risks corrupting arbitrary LLM-generated commands to save
one leaked subshell is a bad trade, so this removes the rewrite at both
call sites (BaseEnvironment.execute, ProcessRegistry.spawn_local) along
with the execute() opt-out parameter. The regression suite now pins the
retirement: every previously-corrupted input class must reach bash
byte-identical.

What this gives up: a long-running `A && B &` leaks one subshell in
wait4 until B exits (resource hygiene, not a hang). If that cost matters,
the sound replacement is a parser-backed rewrite, not another marker.

tests/tools/test_persist_on_release.py no longer patches the removed
_rewrite_compound_background.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/tools/test_persist_on_release.py        |   2 -
 tests/tools/test_process_registry.py          |  87 ++--
 .../test_terminal_compound_background.py      | 377 +++++++++---------
 tools/environments/base.py                    |  10 +-
 tools/environments/managed_modal.py           |   9 +-
 tools/process_registry.py                     |  13 +-
 tools/terminal_tool_sudo.py                   |  55 ---
 7 files changed, 258 insertions(+), 295 deletions(-)

diff --git a/tests/tools/test_persist_on_release.py b/tests/tools/test_persist_on_release.py
index 4375097917ce5..0c3694fd484f1 100644
--- a/tests/tools/test_persist_on_release.py
+++ b/tests/tools/test_persist_on_release.py
@@ -125,7 +125,6 @@ def test_spawn_local_stamps_persist_on_release(registry, monkeypatch, tmp_path):
     ProcessSession so every kill filter can see it (#41225)."""
     import os
 
-    from tools import terminal_tool_sudo
 
     # Stay off the real hermes home: the spawn-path env sanitizer resolves the
     # real console-script install (_resolve_hermes_bin_dir), which the test
@@ -134,7 +133,6 @@ def test_spawn_local_stamps_persist_on_release(registry, monkeypatch, tmp_path):
     from tools.environments import local as local_env
     monkeypatch.setattr(local_env, "_resolve_hermes_bin_dir", lambda: None)
     monkeypatch.setattr(registry, "_track_started", lambda *a, **k: None)
-    monkeypatch.setattr(terminal_tool_sudo, "_rewrite_compound_background", lambda c: c)
     monkeypatch.setattr(ProcessRegistry, "_scope_argv", lambda *a, **k: None)
     fake_popen = MagicMock()
     fake_popen.pid = 4242
diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py
index 8faa5cf9a39ce..9c06da0fa7fbb 100644
--- a/tests/tools/test_process_registry.py
+++ b/tests/tools/test_process_registry.py
@@ -937,6 +937,32 @@ def execute(self, command, **kwargs):
         # A failed launch must not be exposed as a running/tracked session.
         assert session.id not in registry._running
 
+    def test_spawn_via_env_wrapper_reaches_env_untransformed(self, registry):
+        # The compound-background rewriter is retired; execute() must receive
+        # the bg wrapper exactly as spawn_via_env built it, with no rewrite
+        # opt-out kwarg (the parameter no longer exists).
+        class FakeEnv:
+            def __init__(self):
+                self.commands = []
+
+            def get_temp_dir(self):
+                return "/tmp"
+
+            def execute(self, command, **kwargs):
+                self.commands.append((command, kwargs))
+                return {"output": "4321\n", "returncode": 0}
+
+        env = FakeEnv()
+        fake_thread = MagicMock()
+
+        with patch("tools.process_registry.threading.Thread", return_value=fake_thread), \
+            patch.object(registry, "_write_checkpoint"):
+            registry.spawn_via_env(env, "echo hello")
+
+        command_str, kwargs = env.commands[0]
+        assert "rewrite_compound_background" not in kwargs
+        assert "nohup" in command_str
+
     def test_env_poller_quotes_temp_paths_with_spaces(self, registry):
         session = _make_session(sid="proc_space")
         session.exited = False
@@ -1165,21 +1191,21 @@ def boom(*args, **kwargs):
 # =========================================================================
 
 
-class TestSpawnRewriteCompoundBackground:
-    """Verify that spawn_local rewrites `A && B &` patterns to avoid subshell deadlocks.
+class TestSpawnCommandVerbatim:
+    """The compound-background rewriter is retired (#68948): spawn_local must
+    hand the shell the caller's command byte-identical.
 
-    Issue #68915: when bash parses ``A && B &`` it forks a subshell ``(A && B) &``.
-    If B is a long-running server, the subshell never exits and holds the stdout
-    pipe open, causing a permanent deadlock. The rewriter wraps the tail to
-    ``A && { B & }`` so no subshell fork occurs.
+    The rewriter existed for the ``A && B &`` subshell-wait trap (#68915);
+    that hang was fixed at the process layer in #71008, and the textual
+    rewrite kept corrupting valid bash (backticks, ``${...}``, ``[[``,
+    heredoc payloads -- see tests/tools/test_terminal_compound_background.py).
     """
 
-    def test_compound_and_background_gets_rewritten(self, registry):
-        """A && B & must be rewritten to A && { B & } before Popen."""
-        captured_cmd = []
+    def _spawn_and_capture(self, registry, command):
+        captured = []
 
         def fake_popen(args, **kwargs):
-            captured_cmd.append(args)
+            captured.append(args)
             proc = MagicMock()
             proc.pid = 1111
             proc.stdout = MagicMock()
@@ -1192,16 +1218,31 @@ def fake_popen(args, **kwargs):
              patch("subprocess.Popen", side_effect=fake_popen), \
              patch("threading.Thread", return_value=fake_thread), \
              patch.object(registry, "_write_checkpoint"):
-            registry.spawn_local("cd /app && node server.js &>/tmp/srv.log &", cwd="/tmp")
-
-        assert len(captured_cmd) == 1
-        shell_cmd = captured_cmd[0]
-        # The command passed to Popen should be the REWRITTEN version
-        assert "&& { node server.js &>/tmp/srv.log & }" in shell_cmd[2]
-
-
-    def test_pty_path_uses_rewritten_command(self, registry):
-        """PTY spawn path must also use the rewritten command (issue #68915)."""
+            session = registry.spawn_local(command, cwd="/tmp")
+
+        assert len(captured) == 1
+        return session, captured[0][2]
+
+    def test_compound_and_background_not_rewritten(self, registry):
+        cmd = "cd /app && node server.js &>/tmp/srv.log &"
+        session, shell_cmd = self._spawn_and_capture(registry, cmd)
+        assert shell_cmd == f"set +m; {cmd}"
+        assert session.command == cmd
+
+    def test_backtick_compound_untouched(self, registry):
+        # The exact input class the retired rewriter corrupted into invalid
+        # bash (unmatched backtick, #68948 review).
+        cmd = "echo `A && B` &"
+        _, shell_cmd = self._spawn_and_capture(registry, cmd)
+        assert shell_cmd == f"set +m; {cmd}"
+
+    def test_multi_line_command_untouched(self, registry):
+        cmd = "cd /app && python3 -m http.server &\nsleep 1\ncurl http://localhost:8000/"
+        _, shell_cmd = self._spawn_and_capture(registry, cmd)
+        assert shell_cmd == f"set +m; {cmd}"
+
+    @pytest.mark.platforms("posix")
+    def test_pty_path_uses_verbatim_command(self, registry):
         mock_pty_proc = MagicMock()
         mock_pty_proc.pid = 5555
 
@@ -1221,11 +1262,9 @@ def test_pty_path_uses_rewritten_command(self, registry):
                 use_pty=True,
             )
 
-        assert mock_pty_module.PtyProcess.spawn.called, \
-            "PTY spawn should have been attempted"
+        assert mock_pty_module.PtyProcess.spawn.called
         pty_args = mock_pty_module.PtyProcess.spawn.call_args[0][0]
-        assert "&& { node server.js & }" in pty_args[2], \
-            f"PTY path should use rewritten command, got: {pty_args[2]}"
+        assert pty_args[2] == "set +m; cd /app && node server.js &"
         assert session.command == "cd /app && node server.js &"
 
 
diff --git a/tests/tools/test_terminal_compound_background.py b/tests/tools/test_terminal_compound_background.py
index aac682e2721bd..c286ac69790ca 100644
--- a/tests/tools/test_terminal_compound_background.py
+++ b/tests/tools/test_terminal_compound_background.py
@@ -1,203 +1,198 @@
-"""Regression tests for _rewrite_compound_background.
-
-Context: bash parses ``A && B &`` as ``(A && B) &`` — it forks a subshell
-for the compound and backgrounds the subshell. Inside the subshell, B
-runs foreground, so the subshell waits for B. When B never exits on its
-own (HTTP servers, ``yes > /dev/null``, etc.), the subshell is stuck in
-``wait4`` forever and leaks as an orphan process. Pre-fix, we saw this
-pattern leak processes across the fleet (vela, sal, combiagent).
-
-The rewriter fixes this by wrapping the tail in a brace group —
-``A && { B & }`` — so B runs as a simple backgrounded command inside
-the current shell. No subshell fork, no wait.
+"""The compound-background rewriter is retired; commands must reach bash verbatim.
+
+``_rewrite_compound_background`` textually rewrote ``A && B &`` into
+``A && { B & }`` so a backgrounded compound couldn't leak a subshell stuck in
+``wait4`` on a long-running B (the vela/sal/combiagent fleet leaks; #68915).
+The worker hang that made the leak urgent was fixed at the process layer in
+#71008 (orphan-held stdout pipes), and review of the rewriter (#68948) kept
+finding inputs where the textual scan turned valid bash into invalid bash or
+silently changed program data:
+
+- ``echo `A && B` &``        -> unmatched-backtick syntax error
+- ``echo ${x:-A&&B} &``      -> broken expansion
+- ``[[ -n x && -n y ]] &``   -> broken conditional
+- ``echo $[1&&2] &``         -> broken legacy arithmetic
+- ``a[1&&2]=x &``            -> broken array subscript
+- a heredoc payload containing ``A && B &``  -> payload data changed
+- ``$'...'`` ANSI-C strings with ``\\'``      -> string data changed
+- ``false && echo B &``      -> observable ``$?`` changed (0 -> 1)
+
+Every scanner marker added for one of these surfaced the next; syntax created
+at runtime (alias expansion, ``eval``) is out of reach of ANY pre-execution
+textual check.  So the rewrite is removed instead of patched again.  These
+tests pin the retirement at two depths: a seam probe (nothing transforms the
+command before ``_wrap_command``), and a ``subprocess.Popen`` capture on the
+concrete local backends (the exact argv bash receives) so a rewrite hidden
+inside ``_wrap_command`` or ``_run_bash`` cannot slip past either.
 """
 
-import shutil
-import subprocess
+import inspect
+import os
 
 import pytest
 
-from tools.terminal_tool_sudo import _rewrite_compound_background as rewrite
+import tools.process_registry as process_registry
+import tools.terminal_tool as terminal_tool
+import tools.terminal_tool_sudo as terminal_tool_sudo
+from tools.environments import base as env_base
+from tools.environments import local as env_local
 
+# Inputs the retired rewriter provably corrupted (syntax or data), plus the
+# ``A && B &`` shape it was built to transform.  If any transformation
+# reappears on the execute path, at least one of these identity assertions
+# fails and points here.
+CORRUPTION_CLASS = [
+    "A && B &",
+    "A || B &",
+    "echo `A && B` &",
+    "echo ${x:-A&&B} &",
+    "[[ -n x && -n y ]] &",
+    "echo $[1&&2] &",
+    "a[1&&2]=x &",
+    'echo "x`printf "%s && %s" A B`y" &',
+    "read -r x <<'EOF'\nA && B &\nEOF\nprintf '<%s>\\n' \"$x\"",
+    "printf '%s\\n' $'prefix\\' A && B &\nsuffix'",
+    "false && echo B &\nprintf 'status=%s\\n' \"$?\"\nwait",
+]
 
-class TestRewrites:
-    """Commands that trigger the subshell-wait bug MUST be rewritten."""
 
-    def test_simple_and_background(self):
-        assert rewrite("A && B &") == "A && { B & }"
+def test_rewriter_is_gone():
+    assert not hasattr(terminal_tool, "_rewrite_compound_background")
+    assert not hasattr(terminal_tool_sudo, "_rewrite_compound_background")
 
-    def test_or_background(self):
-        assert rewrite("A || B &") == "A || { B & }"
 
+def test_execute_has_no_rewrite_parameter():
+    sig = inspect.signature(env_base.BaseEnvironment.execute)
+    assert "rewrite_compound_background" not in sig.parameters
 
-    def test_multiple_rewrites_in_one_script(self):
-        cmd = "A && B &\nfalse || C &"
-        assert rewrite(cmd) == "A && { B & }\nfalse || { C & }"
 
-
-class TestPreserved:
-    """Commands that DON'T have the bug MUST pass through unchanged."""
-
-    def test_simple_background(self):
-        # No compound — just background a single command. Works fine as-is.
-        assert rewrite("sleep 5 &") == "sleep 5 &"
-
-
-
-    def test_whitespace_only(self):
-        assert rewrite("   \n\t") == "   \n\t"
-
-
-class TestRedirectsNotConfused:
-    """``&>``, ``2>&1``, ``>&2`` must not be mistaken for background ``&``."""
-
-    def test_amp_gt_redirect_alone(self):
-        assert rewrite("echo hi &>/dev/null") == "echo hi &>/dev/null"
-
-
-    def test_gt_amp_inside_compound(self):
-        cmd = "A && B 2>&1 &"
-        assert rewrite(cmd) == "A && { B 2>&1 & }"
-
-
-class TestQuotingAndParens:
-    """Shell metacharacters inside quotes/parens must not be parsed as operators."""
-
-    def test_and_and_inside_single_quotes(self):
-        cmd = "echo 'A && B &'"
-        assert rewrite(cmd) == "echo 'A && B &'"
-
-
-    def test_backslash_escaped_ampersand(self):
-        # Escaped & is not a background operator.
-        cmd = r"echo A \&\& B"
-        assert rewrite(cmd) == cmd
-
-    def test_comment_line_not_rewritten(self):
-        cmd = "# A && B &\nC"
-        assert rewrite(cmd) == "# A && B &\nC"
-
-
-class TestIdempotence:
-    """Running the rewriter twice should be a no-op on its own output."""
-
-    def test_already_rewritten(self):
-        once = rewrite("A && B &")
-        twice = rewrite(once)
-        assert once == twice
-        assert twice == "A && { B & }"
-
-    def test_multiline_idempotent(self):
-        once = rewrite("cd /tmp && server &\nsleep 1")
-        assert rewrite(once) == once
-
-
-class TestEdgeCases:
-    def test_only_chain_op_no_second_command(self):
-        # Malformed input: bash would error, we shouldn't crash or rewrite.
-        cmd = "A && &"
-        # Don't assert a specific output; just don't raise.
-        rewrite(cmd)
-
-
-    def test_tabs_between_tokens(self):
-        assert rewrite("A\t&&\tB\t&") == "A\t&&\t{ B\t& }"
-
-
-class TestTrailingStatementSeparator:
-    """A statement after the backgrounded compound on the SAME line.
-
-    In ``A && B & C`` the trailing ``&`` is both the background operator and
-    the separator between the compound and ``C``. The rewrite consumes that
-    ``&`` into the brace group; without restoring a separator the result is
-    ``A && { B & } C`` — a bash syntax error (a brace group must be terminated
-    by ``;``, ``&``, ``|``, a newline, or ``)``/``}`` before the next command).
-    That mangles a valid command into one that fails entirely.
-    """
-
-    def test_trailing_command_gets_separator(self):
-        assert rewrite("echo hi && sleep 5 & echo done") == (
-            "echo hi && { sleep 5 & } ; echo done"
-        )
-
-    def test_trailing_chain_gets_separator(self):
-        assert rewrite("a && b & c && d") == "a && { b & } ; c && d"
-
-    def test_redirect_then_trailing_command(self):
-        assert rewrite("echo hi && sleep 5 &>/dev/null & echo done") == (
-            "echo hi && { sleep 5 &>/dev/null & } ; echo done"
-        )
-
-    def test_existing_semicolon_separator_untouched(self):
-        # An explicit `;` already separates the group; don't add a second one.
-        assert rewrite("a && b &; c") == "a && { b & }; c"
-
-    def test_newline_separator_untouched(self):
-        # A newline already terminates the brace group — no `;` needed.
-        assert rewrite("a && b &\necho next") == "a && { b & }\necho next"
-
-    def test_pipe_after_group_untouched(self):
-        # `{ ...; } | cmd` is valid; the pipe is its own terminator.
-        assert rewrite("a && b & | cat") == "a && { b & } | cat"
-
-    def test_redirect_prefix_on_trailing_command_gets_separator(self):
-        # `&>` after the group is a redirect for the NEXT command, not a
-        # terminator: `{ b & } &>/dev/null c` is a syntax error.
-        assert rewrite("a && b & &>/dev/null c") == "a && { b & } ; &>/dev/null c"
-
-    def test_case_arm_terminator_untouched(self):
-        # `;;` already terminates the arm; adding `;` would leave an empty
-        # command between `;` and `;;`, which bash rejects.
-        assert rewrite("case $x in p) b && c & ;; esac") == "case $x in p) b && { c & } ;; esac"
-
-
-    def test_second_background_then_trailing(self):
-        assert rewrite("echo a && sleep 5 & echo b & echo c") == (
-            "echo a && { sleep 5 & } ; echo b & echo c"
-        )
-
-
-@pytest.mark.skipif(shutil.which("bash") is None, reason="bash not available")
-class TestRewriteIsValidBash:
-    """The rewrite must always produce syntactically valid bash.
-
-    This is the crux of the trailing-statement bug: a mangled command fails
-    with a confusing syntax error and neither half runs. ``bash -n`` parses
-    without executing, so it catches the corruption directly.
-    """
-
-    @pytest.mark.parametrize(
-        "command",
-        [
-            "echo hi && sleep 5 & echo done",
-            "a && b & c && d",
-            "echo hi && sleep 5 &>/dev/null & echo done",
-            "echo a && sleep 5 & echo b & echo c",
-            "A && B &",
-            "A && B &; C",
-            "A && B &\nC",
-            "cd /tmp && python3 -m http.server 0 &>/dev/null & curl localhost",
-            "a && b & &>/dev/null c",
-            "case $x in p) b && c & ;; esac",
-            "A && B & echo x\nC && D & echo y && E & echo z",
-        ],
-    )
-    def test_rewrite_parses(self, command):
-        rewritten = rewrite(command)
-        result = subprocess.run(
-            ["bash", "-n", "-c", rewritten],
-            capture_output=True,
-            text=True,
-        )
-        assert result.returncode == 0, (
-            f"rewrite produced invalid bash: {rewritten!r}\n{result.stderr}"
-        )
-
-    def test_trailing_statement_actually_runs(self):
-        # End-to-end: the command after the backgrounded compound must run.
-        rewritten = rewrite("echo first && true & echo SECOND_RAN")
-        result = subprocess.run(
-            ["bash", "-c", rewritten], capture_output=True, text=True
-        )
-        assert result.returncode == 0
-        assert "SECOND_RAN" in result.stdout
+class _ProbeEnv(env_base.BaseEnvironment):
+    """Concrete environment that records what reaches ``_wrap_command`` --
+    the exact seam the retired rewriter used to sit in front of."""
+
+    def __init__(self):
+        self.timeout = 5
+        self.cwd = ""
+        self._stdin_mode = "none"
+        self._snapshot_ready = True
+        self._prefer_nonlogin = False
+        self.seen = []
+
+    def _before_execute(self):
+        pass
+
+    def _prepare_command(self, command):
+        return command, None
+
+    def _wrap_command(self, command, cwd):
+        self.seen.append(command)
+        return command
+
+    def _run_bash(self, command, *, login=False, timeout=None, stdin_data=None):
+        return None
+
+    def _wait_for_process(self, proc, *, timeout=None, output=None, **_ignored):
+        return {"output": "", "returncode": 0}
+
+    def _update_cwd(self, result):
+        pass
+
+    def cleanup(self):
+        pass
+
+
+@pytest.mark.parametrize("command", CORRUPTION_CLASS)
+def test_execute_passes_command_verbatim(command):
+    """execute() must hand the prepared command to _wrap_command
+    byte-identical: nothing may transform it on the way."""
+    env = _ProbeEnv()
+    env.execute(command)
+    assert env.seen == [command]
+
+
+class _ArgvRecordingProc:
+    """Popen stand-in: satisfies the minimal lifecycle execute()/spawn_local()
+    drive after spawning (poll/wait/reader), so the test can assert on the
+    captured argv without running a real shell."""
+
+    def __init__(self):
+        self.pid = 4242
+        self.stdout = None
+        self.returncode = 0
+
+    def poll(self):
+        return 0
+
+    def wait(self, timeout=None):
+        return 0
+
+    def kill(self):
+        pass
+
+
+@pytest.fixture
+def _argv_capture(monkeypatch):
+    """Capture the final subprocess.Popen argv on both local backends.
+
+    Only the two shell-invocation shapes under test are intercepted;
+    everything else (Windows shell/ASLR probes, _find_bash checks) is
+    delegated to the real Popen so their module-level caches stay truthful."""
+    seen = []
+    real_popen = env_local.subprocess.Popen
+
+    def _fake_popen(args, **kwargs):
+        argv = list(args) if isinstance(args, (list, tuple)) else [args]
+        if len(argv) == 3 and argv[1] in ("-c", "-lic"):
+            seen.append(argv)
+            return _ArgvRecordingProc()
+        return real_popen(args, **kwargs)
+
+    monkeypatch.setattr(env_local.subprocess, "Popen", _fake_popen)
+    monkeypatch.setattr(process_registry.subprocess, "Popen", _fake_popen)
+    return seen
+
+
+@pytest.fixture
+def _local_env(monkeypatch):
+    """A real LocalEnvironment minus the login-shell snapshot bootstrap.
+
+    init_session is stubbed out (it spawns a real login bash); with
+    ``_prefer_nonlogin`` set, execute() takes the plain ``bash -c`` path with
+    no init-file prepend, so the wrapped script is fully deterministic."""
+    monkeypatch.setattr(env_local.LocalEnvironment, "init_session", lambda self: None)
+    env = env_local.LocalEnvironment(cwd=os.getcwd())
+    env._snapshot_ready = False
+    env._prefer_nonlogin = True
+    return env
+
+
+@pytest.mark.parametrize("command", CORRUPTION_CLASS)
+def test_local_execute_final_bash_argv_is_verbatim(command, _argv_capture, _local_env):
+    """The argv LocalEnvironment hands to Popen is what bash receives — the
+    boundary the retired rewriter can no longer sit in front of.  The wrapper
+    embeds the user command as ``eval '<escaped>'`` where the only permitted
+    transformation is the documented single-quote escape; asserting that exact
+    payload pins the command body byte-identical through _prepare_command,
+    _wrap_command, and _run_bash at once."""
+    _local_env.execute(command)
+    assert len(_argv_capture) == 1
+    args = _argv_capture[0]
+    assert len(args) == 3 and args[1] == "-c"  # plain non-login foreground shape
+    escaped = command.replace("'", "'\\''")
+    assert f"eval '{escaped}'" in args[2]
+
+
+@pytest.mark.parametrize("command", CORRUPTION_CLASS)
+def test_spawn_local_final_shell_argv_is_verbatim(command, _argv_capture, monkeypatch, tmp_path):
+    """spawn_local's contract is ``[shell, -lic, "set +m; <command>"]`` with the
+    command verbatim — full argv equality, so ANY reintroduced transformation
+    (including substring-preserving wrappers) fails here."""
+    # CHECKPOINT_PATH is resolved at import time, before conftest's per-test
+    # HERMES_HOME redirect — repoint it so the test never touches the real one.
+    monkeypatch.setattr(process_registry, "CHECKPOINT_PATH", tmp_path / "processes.json")
+    reg = process_registry.ProcessRegistry()
+    session = reg.spawn_local(command)
+    assert len(_argv_capture) == 1
+    args = _argv_capture[0]
+    assert args[1:] == ["-lic", f"set +m; {command}"]
+    assert session.command == command
diff --git a/tools/environments/base.py b/tools/environments/base.py
index 4bc9f330899f5..a838dacfca6b8 100644
--- a/tools/environments/base.py
+++ b/tools/environments/base.py
@@ -300,7 +300,7 @@ def fetch_file(self, remote_path: str, local_dest: Path, *, max_bytes: int) -> N
         # ``readlink -f`` output before any bytes move.
         result = self.execute(
             f"[ -f {quoted} ] && echo {marker} && head -c {max_bytes + 1} < {quoted} | base64 && echo {marker}",
-            timeout=_FETCH_TIMEOUT_SECONDS, rewrite_compound_background=False)
+            timeout=_FETCH_TIMEOUT_SECONDS)
         output = result.get("output") or ""
         first, last = output.find(marker), output.rfind(marker)
         if int(result.get("returncode") or 0) != 0 or first == -1 or last <= first:
@@ -315,7 +315,7 @@ def fetch_file(self, remote_path: str, local_dest: Path, *, max_bytes: int) -> N
 
     def fetch_realpath(self, remote_path: str) -> str | None:
         """``readlink -f`` inside the backend, or None when it cannot be resolved."""
-        result = self.execute(f"readlink -f {shlex.quote(remote_path)} 2>/dev/null", rewrite_compound_background=False)
+        result = self.execute(f"readlink -f {shlex.quote(remote_path)} 2>/dev/null")
         if int(result.get("returncode") or 0) != 0:
             return None
         return next((ln.strip() for ln in reversed((result.get("output") or "").splitlines()) if ln.strip().startswith("/")), None)
@@ -591,7 +591,6 @@ def execute(
         *,
         timeout: int | None = None,
         stdin_data: str | None = None,
-        rewrite_compound_background: bool = True,
         bounded_capture: bool = False,
         yield_handler: Callable[[ProcessHandle, str], dict] | None = None) -> dict:
         """Execute a command, return {"output": str, "returncode": int}. ``bounded_capture=True``
@@ -608,11 +607,6 @@ def execute(
         self._before_execute()
 
         exec_command, sudo_stdin = self._prepare_command(command)
-        # Guard against the `A && B &` subshell-wait trap by default; callers
-        # that already produce shell-safe wrappers (spawn_via_env) pass False.
-        if rewrite_compound_background:
-            from tools.terminal_tool_sudo import _rewrite_compound_background
-            exec_command = _rewrite_compound_background(exec_command)
         effective_timeout = timeout or self.timeout
         effective_cwd = cwd or self.cwd
 
diff --git a/tools/environments/managed_modal.py b/tools/environments/managed_modal.py
index 66936ea097198..6c79a46ce858c 100644
--- a/tools/environments/managed_modal.py
+++ b/tools/environments/managed_modal.py
@@ -76,12 +76,11 @@ def __init__(self, image: str, cwd: str = "/root", timeout: int = 60,
         self._sandbox_id = self._create_sandbox()
 
     def execute(self, command: str, cwd: str = "", *, timeout: int | None = None, stdin_data: str | None = None,
-                rewrite_compound_background: bool = True, bounded_capture: bool = False) -> dict:
+                bounded_capture: bool = False) -> dict:
         # Signature parity with BaseEnvironment.execute only: the gateway runs commands
-        # explicitly (no shell background rewriting) and returns the remote result in one
-        # payload, so streaming-time bounding does not apply (the terminal tool's final
-        # truncation still caps it).
-        del rewrite_compound_background, bounded_capture
+        # explicitly and returns the remote result in one payload, so streaming-time
+        # bounding does not apply (the terminal tool's final truncation still caps it).
+        del bounded_capture
         exec_command, sudo_stdin = self._prepare_command(command)
         if sudo_stdin is not None:
             # Feed sudo via a shell pipe: the transport has no direct stdin piping.
diff --git a/tools/process_registry.py b/tools/process_registry.py
index 89a2f360efc1b..79c05cc4186a5 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -1220,19 +1220,12 @@ def spawn_local(
         spawn_via_env()). ``use_pty`` requests a pseudo-terminal via ptyprocess/pywinpty
         for interactive CLIs, falling back to a plain pipe when unavailable or failing.
         ``persist_on_release`` keeps the process out of agent-lifecycle kill sweeps (#41225)."""
-        # Bash parses ``A && B &`` as ``(A && B) &`` — a subshell that holds our stdout
-        # pipe open forever when B is a long-running server. The rewriter turns it into
-        # ``A && { B & }``. Lazy import: terminal_tool imports this module.
-        # Guard against the `A && B &` subshell-wait trap (issue #68915).
-        from tools.terminal_tool_sudo import _rewrite_compound_background as _rewrite_bg
-
-        safe_command = _rewrite_bg(command)
         session = self._new_session(command, task_id, owner_task_id, session_key, _resolve_safe_cwd(cwd or os.getcwd()),
                                     persist_on_release=persist_on_release)
         pty_scope_attempted = False
         if use_pty:
             try:
-                return self._spawn_local_pty(session, safe_command, env_vars)
+                return self._spawn_local_pty(session, command, env_vars)
             except ImportError:
                 logger.warning("ptyprocess not installed, falling back to pipe mode")
             except Exception as e:
@@ -1248,7 +1241,7 @@ def spawn_local(
         # Pipe path (non-PTY or PTY fallback).
         _popen_kwargs = {"creationflags": windows_hide_flags()} if _IS_WINDOWS else {}
         unit_suffix = f"{session.id}-pipe-fallback" if pty_scope_attempted else session.id
-        spawn_argv = self._scope_argv(session, safe_command, unit_suffix, "Local")
+        spawn_argv = self._scope_argv(session, command, unit_suffix, "Local")
         spawn_env = self._spawn_env(env_vars)
         if session.systemd_unit:
             spawn_env = systemd_user_bus_env(spawn_env)
@@ -1332,7 +1325,7 @@ def spawn_via_env(
             f"rc=$?; printf '%s\\n' \"$rc\" > {q(exit_path)} ) & "
             f"echo $! > {q(pid_path)} && cat {q(pid_path)}")
         try:
-            result = env.execute(bg_command, timeout=timeout, rewrite_compound_background=False)
+            result = env.execute(bg_command, timeout=timeout)
             output = result.get("output", "").strip()
             session.pid = next((int(ln) for ln in map(str.strip, output.splitlines()) if ln.isdigit()), None)
             # No PID from the wrapper (syntax error, broken redirect): a failed launch,
diff --git a/tools/terminal_tool_sudo.py b/tools/terminal_tool_sudo.py
index 4f1b78949b776..24a58866aa7ba 100644
--- a/tools/terminal_tool_sudo.py
+++ b/tools/terminal_tool_sudo.py
@@ -374,61 +374,6 @@ def _count_real_sudo_invocations(command: str) -> int:
     return _rewrite_real_sudo_invocations(command)[1]
 
 
-def _rewrite_compound_background(command: str) -> str:
-    """Wrap `A && B &` (or `A || B &`) to `A && { B & }` at depth 0. Bash binds `&&` tighter
-    than `&`, so `A && B &` backgrounds a subshell that runs B in the foreground and waits for
-    it; a long-running B leaves that subshell stuck in ``wait4`` forever, and its open stdout
-    pipe can keep the terminal tool from returning. The brace group keeps `&&`'s
-    skip-B-on-failure semantics without a fork: bash backgrounds B as a simple command and
-    exits immediately, orphaning B normally. Redirects (``&>``, ``2>&1``), quoted strings,
-    comments and ``(...)``/``{ ... }`` bodies never count as the backgrounding ``&`` (see
-    ``_scan_shell``); tracking brace depth also makes the rewrite idempotent. `(...)` subshells
-    have the same bug class but are not the common agent pattern; left for a follow-up.
-    Simple ``cmd &`` is left alone — it doesn't have the subshell-wait bug."""
-    chain_end = -1  # just after the last depth-0 `&&`/`||` of this statement; -1 = none active
-    rewrites: list[tuple[int, int]] = []  # (chain_op_end, amp_pos)
-    for kind, start, end, _ in _scan_shell(command, background=True):
-        text = command[start:end]
-        if kind == "op" and text in ("&&", "||"):
-            chain_end = end
-        elif kind == "ws" and text == "\n" or kind == "op" and text in (";", "|", "}"):
-            # Newline / `;` end a statement, `|` starts a pipeline stage, `}` closes a group.
-            chain_end = -1
-        elif kind == "op" and text == "&":
-            # `&&` and `&>` never reach here; a `>&` / `<&` fd target (look back past
-            # whitespace) is a redirect, anything else is the real background operator.
-            j = start - 1
-            while j >= 0 and command[j].isspace():
-                j -= 1
-            if j >= 0 and command[j] in "<>":
-                continue
-            if chain_end >= 0:
-                rewrites.append((chain_end, start))
-            chain_end = -1
-
-    # Apply rewrites back-to-front so earlier indices remain valid.
-    result = command
-    for chain_end, amp_pos in reversed(rewrites):
-        # Skip whitespace right after the `&&`/`||` so the brace group opens flush against
-        # the inner command. `{` needs a trailing space in bash; the closing `}` needs to be
-        # preceded by `;` or `&` — we're providing `&` from the backgrounding.
-        insert_pos = chain_end
-        while insert_pos < amp_pos and result[insert_pos].isspace():
-            insert_pos += 1
-        # The consumed `&` also separated the compound from any statement that followed
-        # on the same line (`A && B & C`); `{ B & } C` is a syntax error, so restore a `;`
-        # when the suffix resumes with command text. No separator when the suffix already
-        # starts with a terminator (`;` `&` `|` newline `)` `}`) — except `&>`, which is a
-        # redirect prefix for the NEXT command, not a terminator. Strip only spaces/tabs:
-        # a newline already terminates the group.
-        suffix = result[amp_pos + 1 :]
-        tail = suffix.lstrip(" \t")
-        needs_separator = bool(tail) and (tail[0] not in ";\n&|)}" or tail.startswith("&>"))
-        separator = " ;" if needs_separator else ""
-        result = result[:insert_pos] + "{ " + result[insert_pos:amp_pos] + "& }" + separator + suffix
-    return result
-
-
 def _transform_sudo_command(
     command: str | None,
     sudo_nopasswd_check: Callable[[], bool] | None = None,

From 7728c60085d77dc03499af70e8d449298a01ec1d Mon Sep 17 00:00:00 2001
From: Sora-bluesky <sora.bluesky.dev@gmail.com>
Date: Sat, 26 Sep 2026 09:13:15 +0900
Subject: [PATCH 2/3] fix(windows): mirror the child-exit-aware reader on
 PeekNamedPipe

The #71008 reader fix is select()-based and select() does not work on
pipe fds on Windows, so _reader_loop kept the historical blocking
read1() there. Any command that leaves a background grandchild holding
the stdout pipe parked the reader thread: session.exited never flipped
on its own and notify_on_complete never fired until the grandchild
died. Measured on Windows 11 with spawn_local("true && sleep 30
>/dev/null 2>&1 &"): exited stayed False for the full sleep; with this
change it flips at ~1.1s, right after the direct child exits.

PeekNamedPipe works on anonymous pipes and reports buffered bytes
without blocking, so the Windows branch now mirrors the POSIX select()
loop exactly: read only when bytes are available, otherwise check the
direct child and stop after the same short idle grace. Like the select()
loop it honours _reader_finish_requested and marks the session
_reader_selectable, so _reconcile_local_exit now hands the final drain
and the completion to a live reader on Windows as it already does on
POSIX, and test_list_leaves_live_reader_as_completion_owner runs on
every OS.
Streams without a real fd still use the blocking fallback.

The contract tests fake msvcrt/_winapi/time and run on Windows under
platforms("windows") rather than faking the module's OS flag
(AGENTS.md, "Don't fake the host OS"); the integration test runs the
real pipeline there. The Windows stdout fake gains close() for the
finished-handle release.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/tools/test_process_registry.py          | 319 ++++++++++++++++++
 .../tools/test_process_registry_list_exit.py  |   2 +-
 tools/process_registry.py                     |  49 ++-
 3 files changed, 361 insertions(+), 9 deletions(-)

diff --git a/tests/tools/test_process_registry.py b/tests/tools/test_process_registry.py
index 9c06da0fa7fbb..69f92ab703b21 100644
--- a/tests/tools/test_process_registry.py
+++ b/tests/tools/test_process_registry.py
@@ -2169,6 +2169,35 @@ def _run():
             except (ProcessLookupError, PermissionError):
                 pass
 
+    def test_reader_still_streams_full_output_to_eof(self, registry):
+        """No-orphan case: the reader must still capture ALL output through
+        true EOF (the early-exit path must not race away buffered tail)."""
+        script = (
+            "for i in 1 2 3 4 5; do echo line-$i; done; "
+            "sleep 0.3; echo tail-after-sleep"
+        )
+        proc = subprocess.Popen(
+            ["sh", "-c", script],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            encoding="utf-8",
+            errors="replace",
+            preexec_fn=os.setsid,
+        )
+        s = _make_session(sid="proc_orphan_fulldrain")
+        s.process = proc
+        s.pid = proc.pid
+        registry._running[s.id] = s
+
+        registry._reader_loop(s)
+
+        assert s.exited is True
+        assert s.exit_code == 0
+        for i in range(1, 6):
+            assert f"line-{i}" in s.output_buffer
+        assert "tail-after-sleep" in s.output_buffer
+
 # =========================================================================
 # systemd cgroup isolation for gateway-spawned local executors (#70716)
 # =========================================================================
@@ -2904,6 +2933,296 @@ def test_stop_systemd_unit_treats_absent_unit_as_clean(self, monkeypatch):
 
 
 
+# =========================================================================
+# Reader loop, Windows branch: PeekNamedPipe instead of select() (#68948)
+# =========================================================================
+
+class _FakeWinStdout:
+    """stdout stand-in exposing the buffered raw interface + a real-int fd."""
+
+    def __init__(self, chunks):
+        self._chunks = list(chunks)
+        self.read1_calls = 0
+        # _release_finished_handles closes the child's streams on the finish
+        # path and only suppresses OSError/ValueError (29b981c846), so a real
+        # stream interface must include close().
+        self.close_calls = 0
+
+        outer = self
+
+        class _Buffer:
+            def read1(self, n):
+                outer.read1_calls += 1
+                if outer._chunks:
+                    return outer._chunks.pop(0)
+                return b""
+
+        self.buffer = _Buffer()
+
+    def fileno(self):
+        return 7
+
+    def close(self):
+        self.close_calls += 1
+
+
+@pytest.mark.platforms("windows")
+class TestReaderLoopWindowsPeekBranch:
+    """Pins the PeekNamedPipe loop's cadence with fake msvcrt, _winapi,
+    and time on a Windows host."""
+
+    class _TimeProxy:
+        """time-module stand-in scoped to process_registry: records the
+        arguments the reader passes to sleep() so tests can pin the 0.2s
+        cadence itself, not just coarse wall-clock bounds."""
+
+        def __init__(self, real, log):
+            self._real = real
+            self._log = log
+
+        def sleep(self, seconds):
+            self._log.append(seconds)
+            self._real.sleep(seconds)
+
+        def __getattr__(self, name):
+            return getattr(self._real, name)
+
+    def _patch_windows(self, monkeypatch, peek_results, tmp_path):
+        """Install fake msvcrt/_winapi; ``peek_results`` items are either
+        (n_avail, n_left) tuples or exceptions to raise, last one sticky.
+        Also repoints the import-time CHECKPOINT_PATH: _move_to_finished()
+        writes a checkpoint, and without the redirect these tests would
+        overwrite the developer's real processes.json. Returns the list the
+        sleep spy appends to."""
+        import tools.process_registry as pr_mod
+
+        monkeypatch.setattr(pr_mod, "CHECKPOINT_PATH", tmp_path / "processes.json")
+
+        sleeps: list = []
+        monkeypatch.setattr(pr_mod, "time", self._TimeProxy(pr_mod.time, sleeps))
+
+        fake_msvcrt = MagicMock()
+        fake_msvcrt.get_osfhandle.return_value = 1234
+
+        results = list(peek_results)
+
+        def _peek(handle):
+            assert handle == 1234
+            item = results.pop(0) if len(results) > 1 else results[0]
+            if isinstance(item, BaseException):
+                raise item
+            return item
+
+        fake_winapi = MagicMock()
+        fake_winapi.PeekNamedPipe.side_effect = _peek
+
+
+        monkeypatch.setattr(pr_mod, "msvcrt", fake_msvcrt)
+        monkeypatch.setattr(pr_mod, "_winapi", fake_winapi)
+        return sleeps
+
+    def test_reader_exits_when_pipe_stays_open_after_child_exit(
+        self, registry, monkeypatch, tmp_path
+    ):
+        """Windows #68915 semantics: the direct child is gone, a grandchild
+        still holds the pipe (peek forever reports 0 bytes) — the reader must
+        stop after the idle grace instead of parking in a blocking read."""
+        sleeps = self._patch_windows(monkeypatch, [(0, 0)], tmp_path)
+
+        import tools.process_registry as pr_mod
+
+        proc = MagicMock()
+        proc.stdout = _FakeWinStdout([])
+        proc.poll.return_value = 0
+        proc.wait.return_value = 0
+        proc.returncode = 0
+
+        s = _make_session(sid="proc_win_peek_orphan")
+        s.process = proc
+        s.notify_on_complete = True
+        registry._running[s.id] = s
+
+        done = threading.Event()
+
+        def _run():
+            registry._reader_loop(s)
+            done.set()
+
+        started = time.monotonic()
+        t = threading.Thread(target=_run, daemon=True)
+        t.start()
+        assert done.wait(timeout=10.0), (
+            "_reader_loop parked on a grandchild-held pipe on the Windows "
+            "branch — session.exited would never flip on its own"
+        )
+        elapsed = time.monotonic() - started
+        assert s.exited is True
+        assert s.exit_code == 0
+        assert s.completion_reason == "exited"
+        assert s.id in registry._finished
+        # The autonomous completion event is the point of the exit-aware
+        # reader — assert it actually fired.
+        item = registry.completion_queue.get_nowait()
+        assert item["type"] == "completion"
+        assert item["session_id"] == s.id
+        # The whole point of peeking: never call read1() while no bytes are
+        # reported, because that call is what blocks forever.
+        assert proc.stdout.read1_calls == 0
+        # The finish path released the pipe handle exactly once.
+        assert proc.stdout.close_calls == 1
+        # Pin the polling cadence to the POSIX select() branch: three idle
+        # windows, each a real 0.2s sleep followed by a re-peek (2 peeks per
+        # window). The sleep spy pins the interval itself — a 0.6s sleep
+        # would pass any coarse wall-clock bound — and the elapsed floor
+        # stays as the busy-loop guard (dropped sleeps finish in
+        # microseconds).
+        assert pr_mod._winapi.PeekNamedPipe.call_count == 6
+        assert sleeps == [0.2, 0.2, 0.2]
+        assert elapsed >= 0.5
+
+    def test_reader_keeps_same_grace_when_child_exits_mid_loop(
+        self, registry, monkeypatch, tmp_path
+    ):
+        """running→exited transition: while the child is alive the loop just
+        waits (no idle counting); after it exits, exactly the same three
+        0.2s windows run as in the already-exited case."""
+        sleeps = self._patch_windows(monkeypatch, [(0, 0)], tmp_path)
+
+        polls = {"n": 0}
+
+        def _poll():
+            polls["n"] += 1
+            return None if polls["n"] == 1 else 0
+
+        proc = MagicMock()
+        proc.stdout = _FakeWinStdout([])
+        proc.poll.side_effect = _poll
+        proc.wait.return_value = 0
+        proc.returncode = 0
+
+        s = _make_session(sid="proc_win_peek_transition")
+        s.process = proc
+        registry._running[s.id] = s
+
+        done = threading.Event()
+
+        def _run():
+            registry._reader_loop(s)
+            done.set()
+
+        threading.Thread(target=_run, daemon=True).start()
+        assert done.wait(timeout=10.0)
+        assert s.exited is True
+        # One alive observation (a full window, no idle count) plus the
+        # three idle windows: four 0.2s sleeps, never more.
+        assert sleeps == [0.2, 0.2, 0.2, 0.2]
+
+    def test_reader_drains_buffered_tail_then_stops_on_broken_pipe(
+        self, registry, monkeypatch, tmp_path
+    ):
+        """Buffered output is drained through read1() while peek reports
+        bytes; a broken pipe (all writers closed, buffer empty) is EOF."""
+        self._patch_windows(
+            monkeypatch, [(5, 0), OSError(109, "broken pipe")], tmp_path
+        )
+
+        proc = MagicMock()
+        proc.stdout = _FakeWinStdout([b"tail\n"])
+        proc.poll.return_value = 0
+        proc.wait.return_value = 0
+        proc.returncode = 0
+
+        s = _make_session(sid="proc_win_peek_tail")
+        s.process = proc
+        registry._running[s.id] = s
+
+        registry._reader_loop(s)
+
+        assert s.exited is True
+        assert "tail" in s.output_buffer
+        assert proc.stdout.read1_calls == 1
+
+
+    def test_finish_request_stops_one_window_while_child_alive(
+        self, registry, monkeypatch, tmp_path
+    ):
+        """Alive child, peek stuck at 0; the first fake sleep sets the finish request."""
+        sleeps = self._patch_windows(monkeypatch, [(0, 0)], tmp_path)
+        import tools.process_registry as pr_mod
+        proc = MagicMock()
+        proc.stdout = _FakeWinStdout([])
+        proc.poll.return_value = None
+        proc.wait.return_value = proc.returncode = 0
+        s = _make_session(sid="proc_win_peek_finish")
+        s.process, s.notify_on_complete = proc, True
+        registry._running[s.id] = s
+        proxy = pr_mod.time
+        orig = proxy.sleep
+
+        def _sleep(seconds):
+            if not proxy._log:
+                s._reader_finish_requested.set()
+            orig(seconds)
+
+        monkeypatch.setattr(proxy, "sleep", _sleep)
+        done = threading.Event()
+        threading.Thread(
+            target=lambda: (registry._reader_loop(s), done.set()), daemon=True,
+        ).start()
+        assert done.wait(timeout=10.0)
+        assert s.exited is True
+        assert proc.stdout.read1_calls == 0
+        assert sleeps == [0.2]
+        got = registry.completion_queue.get_nowait()
+        assert (
+            got["type"] == "completion"
+            and got["session_id"] == s.id
+            and registry.completion_queue.empty()
+        )
+
+@pytest.mark.platforms("windows")
+class TestReaderLoopWindowsIntegration:
+    """Real-process twin of TestReaderLoopOrphanedPipe for Windows, driven
+    through spawn_local like the terminal tool does. The redirected compound
+    background (``true && B >/dev/null 2>&1 &``) is the shape where the
+    intermediate ``(A && B)`` subshell holds the reader's pipe while B runs
+    detached — before the peek branch this parked the reader until B died."""
+
+    def test_spawn_local_completes_despite_redirected_compound_background(
+        self, registry, monkeypatch, tmp_path
+    ):
+        import tools.process_registry as pr_mod
+
+        # Redirect the import-time checkpoint path for the WHOLE test: the
+        # reader thread calls _write_checkpoint via _move_to_finished after
+        # spawn_local returns, so a with-scoped patch would not cover it.
+        monkeypatch.setattr(pr_mod, "CHECKPOINT_PATH", tmp_path / "processes.json")
+        s = registry.spawn_local("true && sleep 15 >/dev/null 2>&1 &")
+        try:
+            # Lifecycle observation only — no poll()/wait() calls, so the
+            # lazy _reconcile_local_exit safety net cannot mask the reader.
+            deadline = time.time() + 8
+            while time.time() < deadline and not s.exited:
+                time.sleep(0.1)
+            assert s.exited is True, (
+                "spawn_local session never completed on its own while a "
+                "redirected background grandchild outlived the direct child"
+            )
+            assert s.exit_code == 0
+            # exited flips BEFORE _move_to_finished writes the checkpoint;
+            # join the reader so that write lands while CHECKPOINT_PATH is
+            # still monkeypatched, not after teardown restored the real one.
+            s._reader_thread.join(timeout=5)
+            assert not s._reader_thread.is_alive()
+        finally:
+            try:
+                if s.process is not None and s.process.poll() is None:
+                    s.process.kill()
+            except Exception:
+                pass
+
+
+
 
 class TestNotificationRedaction:
     """Background-process notification delivery (completion_queue) applies the
diff --git a/tests/tools/test_process_registry_list_exit.py b/tests/tools/test_process_registry_list_exit.py
index 0d5d7243a02df..7a322caca3769 100644
--- a/tests/tools/test_process_registry_list_exit.py
+++ b/tests/tools/test_process_registry_list_exit.py
@@ -15,7 +15,7 @@
 import pytest
 
 
-@pytest.mark.platforms("posix")
+@pytest.mark.platforms("any")
 def test_list_leaves_live_reader_as_completion_owner():
     from tools.process_registry import ProcessRegistry, ProcessSession
 
diff --git a/tools/process_registry.py b/tools/process_registry.py
index 79c05cc4186a5..830983ac9cb12 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -21,6 +21,12 @@
 from pathlib import Path
 
 _IS_WINDOWS = platform.system() == "Windows"
+try:
+    import msvcrt
+    import _winapi
+except ImportError:
+    msvcrt = None
+    _winapi = None
 # systemd transient scopes exist only on Linux; gate every scope-path branch on this
 # (not merely "not Windows") so macOS and other POSIX platforms never touch systemd.
 # See #70716.
@@ -1354,10 +1360,11 @@ def _reader_loop(self, session: ProcessSession):
         end so EOF never arrives while it lives, which would park this thread and never
         fire ``notify_on_complete``; on POSIX we ``select()`` and stop draining shortly
         after the direct child exits (mirrors ``environments/base.py::_wait_for_process``).
-        Windows pipes lack select(), so the lazy ``_reconcile_local_exit`` is the net.
-
-        Windows pipes don't support select(); the blocking path is kept there and the lazy reconcile in
-        poll()/wait() remains the safety net. See #68915, #8340.
+        Windows pipes lack select(), so the same loop runs on ``PeekNamedPipe``:
+        read only when bytes are available, otherwise check the direct child and stop
+        after the same short idle grace (``environments/base_output.py::_drain_fd_windows``
+        is the foreground twin). Streams without a real OS fd still use the
+        blocking fallback. See #68915, #8340.
         """
         # ``bash -lic`` without a tty writes its startup warnings one write() per line, so the
         # reader can wake between them; strip leading noise from every chunk until the
@@ -1394,14 +1401,24 @@ def _read_once():
             # select() needs a real OS fd; mocked streams (tests, adapters) may lack
             # fileno() and use the blocking read instead.
             try:
-                fd = stdout.fileno() if raw_read is not None and not _IS_WINDOWS else None
+                fd = stdout.fileno() if raw_read is not None else None
             except Exception:
                 fd = None
             if not (isinstance(fd, int) and fd >= 0):
                 fd = None
+            # select() cannot poll pipe fds on Windows; PeekNamedPipe reports buffered bytes without
+            # blocking. No usable handle -> the blocking fallback below (fd None).
+            peek_handle = None
+            if fd is not None and _IS_WINDOWS:
+                if msvcrt is not None and _winapi is not None:
+                    with suppress(OSError):
+                        peek_handle = msvcrt.get_osfhandle(fd)
+                fd = None
             if fd is not None:
                 import select as _select
                 session._reader_selectable = True
+            elif peek_handle is not None:
+                session._reader_selectable = True   # peek loop honors _reader_finish_requested (see below)
             idle_after_exit = 0
             while True:
                 if fd is not None:
@@ -1421,6 +1438,23 @@ def _read_once():
                         if idle_after_exit >= 3:
                             break
                         continue
+                elif peek_handle is not None:
+                    try:
+                        n_avail = _winapi.PeekNamedPipe(peek_handle)[0]
+                        if n_avail == 0:
+                            time.sleep(0.2)          # select()'s bounded wait
+                            n_avail = _winapi.PeekNamedPipe(peek_handle)[0]
+                    except (OSError, ValueError):    # BrokenPipeError is an OSError: all writers closed, buffer drained
+                        break
+                    if n_avail == 0:
+                        if session._reader_finish_requested.is_set():
+                            break
+                        # Same idle grace as the select() branch: direct child gone and pipe idle.
+                        if proc.poll() is not None:
+                            idle_after_exit += 1
+                        if idle_after_exit >= 3:
+                            break
+                        continue
                 chunk = _read_once()
                 if chunk is None:
                     break  # true EOF — all writers closed
@@ -1938,8 +1972,7 @@ def _reconcile_local_exit(self, session: "ProcessSession") -> None:
             return  # Direct child still running — reader block is legitimate.
         reader = session._reader_thread
         if (
-            not _IS_WINDOWS
-            and session._reader_selectable
+            session._reader_selectable
             and reader is not None
             and reader.is_alive()
         ):
@@ -1947,7 +1980,7 @@ def _reconcile_local_exit(self, session: "ProcessSession") -> None:
             # finish avoids a competing TextIOWrapper read here racing the
             # reader, publishing an empty owner-stamped result, then closing
             # the pipe before the buffered tail is ingested. It wakes within
-            # the reader's bounded select interval (or after one final chunk).
+            # the reader's bounded select()/PeekNamedPipe interval (or after one final chunk).
             session._reader_finish_requested.set()
             with session._lock:
                 session.mark_exited(rc)

From 1c384a73a49c9f0590930c5216ae5d9d3dd61d4a Mon Sep 17 00:00:00 2001
From: Sora-bluesky <sora.bluesky.dev@gmail.com>
Date: Sat, 26 Sep 2026 09:14:42 +0900
Subject: [PATCH 3/3] test(terminal): pin the Docker execute wrapper and the
 remote-kernel spawn path

Acceptance contract from #98222: pin DockerEnvironment.execute so a
reintroduced rewrite option cannot ride its **kwargs forwarder, and
drive the code_kernel_remote spawn template through a dependency-light
fake of the shared execute() path asserting the un-rewritten command,
a real PID, persistent state on a second call, and a failed-spawn
negative control kept separate from reader lifecycle concerns.

The probe stubs take `_wait_for_process(proc, *, timeout=None,
output=None, **_ignored)`, since execute() now also forwards
watch_interrupt_tid and output.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/tools/test_code_kernel_remote_spawn.py  | 242 ++++++++++++++++++
 .../test_terminal_compound_background.py      |  40 +++
 2 files changed, 282 insertions(+)
 create mode 100644 tests/tools/test_code_kernel_remote_spawn.py

diff --git a/tests/tools/test_code_kernel_remote_spawn.py b/tests/tools/test_code_kernel_remote_spawn.py
new file mode 100644
index 0000000000000..ac0bf6235efea
--- /dev/null
+++ b/tests/tools/test_code_kernel_remote_spawn.py
@@ -0,0 +1,242 @@
+"""Regression coverage for the remote-kernel spawn broken by #68948.
+
+The fake environment inherits the shared BaseEnvironment.execute() path while
+replacing only its shell/process boundary. No Docker daemon, SSH connection,
+Modal client, subprocess, or POSIX shell is required.
+"""
+
+import json
+import shlex
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import tools.code_kernel_remote as remote_kernel
+from tools.environments.base import BaseEnvironment
+
+
+class _SharedExecuteEnv(BaseEnvironment):
+    """Dependency-light remote whose commands still traverse shared execute()."""
+
+    def __init__(self, *, spawn_success=True):
+        self.timeout = 30
+        self.cwd = ""
+        self._stdin_mode = "none"
+        self._snapshot_ready = True
+        self._prefer_nonlogin = False
+
+        self.spawn_success = spawn_success
+        self.background_pid = "4242" if spawn_success else None
+        self.runner_alive = False
+        self.shell_commands = []
+        self.shipped = {}
+        self.cell_codes = []
+        self.cell_payloads = []
+        self.namespace = {}
+        self.execution_count = 0
+        self.spawn_attempts = 0
+        self.spawn_returncodes = []
+
+    def get_temp_dir(self):
+        return "/tmp"
+
+    def ship_file(self, path, content):
+        self.shipped[path] = content
+        if "/cells/cell_req_" not in path:
+            return
+
+        request = json.loads(content)
+        code = request["code"]
+        self.cell_codes.append(code)
+        self.execution_count += 1
+
+        status = "ok"
+        stdout = ""
+        trace = ""
+        if code == "counter = 41":
+            self.namespace["counter"] = 41
+        elif code == "print(counter)" and "counter" in self.namespace:
+            stdout = f"{self.namespace['counter']}\n"
+        else:
+            status = "error"
+            trace = "NameError: unsupported fake cell"
+
+        self.cell_payloads.append({
+            "id": request["id"],
+            "status": status,
+            "stdout": stdout,
+            "stderr": "",
+            "stdout_clipped": False,
+            "stderr_clipped": False,
+            "traceback": trace,
+            "execution_count": self.execution_count,
+        })
+
+    def _before_execute(self):
+        pass
+
+    def _prepare_command(self, command):
+        return command, None
+
+    def _wrap_command(self, command, cwd):
+        return command
+
+    def _run_bash(self, command, *, login=False, timeout=None, stdin_data=None):
+        self.shell_commands.append(command)
+        return command
+
+    def _wait_for_process(self, proc, *, timeout=None, output=None, **_ignored):
+        command = proc
+
+        if "nohup env " in command:
+            self.spawn_attempts += 1
+            if self.spawn_success:
+                self.runner_alive = True
+                self.spawn_returncodes.append(0)
+                return {
+                    "output": f"PID:{self.background_pid}\n",
+                    "returncode": 0,
+                }
+            self.spawn_returncodes.append(1)
+            return {"output": "sh: cannot fork\n", "returncode": 1}
+
+        if command.startswith("kill -0 "):
+            if self.runner_alive:
+                return {"output": "ALIVE\n", "returncode": 0}
+            return {"output": "", "returncode": 1}
+
+        if command.startswith("cat ") and "cell_res_" in command:
+            if self.cell_payloads:
+                return {
+                    "output": json.dumps(self.cell_payloads.pop(0)),
+                    "returncode": 0,
+                }
+            return {"output": "", "returncode": 1}
+
+        return {"output": "", "returncode": 0}
+
+    def _update_cwd(self, result):
+        pass
+
+    def cleanup(self):
+        pass
+
+
+def _ship_to_fake(env, path, content):
+    env.ship_file(path, content)
+
+
+def _execute_code(env, code, *, task):
+    return remote_kernel.execute_in_remote_kernel(
+        code,
+        env=env,
+        env_type="ssh",
+        task_env_id=task,
+        sandbox_tools=frozenset({"read_file"}),
+        timeout=10,
+        max_tool_calls=5,
+        reset=False,
+    )
+
+
+def test_remote_spawn_command_pid_persistence_and_failure():
+    remote_kernel.shutdown_all_remote_kernels()
+    healthy = _SharedExecuteEnv()
+    failed_env = _SharedExecuteEnv(spawn_success=False)
+    fixed_uuid = SimpleNamespace(hex="0123456789abcdef0123456789abcdef")
+    fixed_token = "fixed-token"
+
+    try:
+        with patch.object(
+            remote_kernel.uuid,
+            "uuid4",
+            return_value=fixed_uuid,
+        ), patch(
+            "secrets.token_urlsafe",
+            return_value=fixed_token,
+        ), patch(
+            "tools.code_execution_tool._ship_file_to_remote",
+            side_effect=_ship_to_fake,
+        ), patch(
+            "tools.code_execution_tool._rpc_poll_loop",
+            new=lambda *args, **kwargs: None,
+        ):
+            first = _execute_code(
+                healthy,
+                "counter = 41",
+                task="spawn-regression",
+            )
+            second = _execute_code(
+                healthy,
+                "print(counter)",
+                task="spawn-regression",
+            )
+
+            runner_path = next(
+                path for path in healthy.shipped
+                if path.endswith("/kernel_runner.py")
+            )
+            kernel_dir = runner_path.rsplit("/", 1)[0]
+            q_dir = shlex.quote(kernel_dir)
+            env_prefix = (
+                f"HERMES_KERNEL_DIR={q_dir} "
+                f"HERMES_RPC_DIR={shlex.quote(kernel_dir + '/rpc')} "
+                f"HERMES_RPC_TOKEN={shlex.quote(fixed_token)} "
+                f"PYTHONDONTWRITEBYTECODE=1 PYTHONPATH={q_dir}"
+            )
+            expected_spawn = (
+                f"cd {q_dir} && nohup env {env_prefix} "
+                f"python3 kernel_runner.py > {q_dir}/runner.log 2>&1 "
+                f"& echo PID:$!"
+            )
+            spawn_commands = [
+                command for command in healthy.shell_commands
+                if "nohup env " in command
+            ]
+
+            # Command identity is independent of PID parsing and registration.
+            assert spawn_commands == [expected_spawn]
+            assert spawn_commands[0].endswith(" & echo PID:$!")
+            assert "&& { nohup" not in spawn_commands[0]
+            assert "& } echo PID:$!" not in spawn_commands[0]
+
+            registered = [
+                kernel for kernel in remote_kernel._REMOTE_KERNELS.values()
+                if kernel.env is healthy
+            ]
+            assert len(registered) == 1
+            assert registered[0].pid is not None
+            assert registered[0].pid == healthy.background_pid
+            assert registered[0].pid == "4242"
+
+            # Persistence evidence is separate from command and PID evidence.
+            assert first is not None
+            assert first["status"] == "success"
+            assert first["kernel"]["reused"] is False
+            assert second is not None
+            assert second["status"] == "success"
+            assert second["kernel"]["reused"] is True
+            assert second["stdout"] == "41\n"
+            assert second["kernel"]["execution_count"] == 2
+            assert healthy.cell_codes == ["counter = 41", "print(counter)"]
+            assert healthy.spawn_attempts == 1
+
+            remote_kernel.shutdown_all_remote_kernels()
+
+            failed = _execute_code(
+                failed_env,
+                "counter = 41",
+                task="failed-spawn-regression",
+            )
+
+            # A real nonzero spawn with no PID marker must fail open and must
+            # never become indistinguishable from the registered healthy run.
+            assert failed_env.spawn_attempts == 1
+            assert failed_env.spawn_returncodes == [1]
+            assert failed is None
+            assert not [
+                kernel for kernel in remote_kernel._REMOTE_KERNELS.values()
+                if kernel.env is failed_env
+            ]
+            assert len(remote_kernel._REMOTE_KERNELS) == 0
+    finally:
+        remote_kernel.shutdown_all_remote_kernels()
diff --git a/tests/tools/test_terminal_compound_background.py b/tests/tools/test_terminal_compound_background.py
index c286ac69790ca..c428fe3b807c4 100644
--- a/tests/tools/test_terminal_compound_background.py
+++ b/tests/tools/test_terminal_compound_background.py
@@ -35,6 +35,7 @@
 import tools.terminal_tool as terminal_tool
 import tools.terminal_tool_sudo as terminal_tool_sudo
 from tools.environments import base as env_base
+from tools.environments import docker as env_docker
 from tools.environments import local as env_local
 
 # Inputs the retired rewriter provably corrupted (syntax or data), plus the
@@ -66,6 +67,45 @@ def test_execute_has_no_rewrite_parameter():
     assert "rewrite_compound_background" not in sig.parameters
 
 
+class _DockerWrapperProbe(env_docker.DockerEnvironment):
+    """Docker execute() probe that bypasses container setup entirely."""
+
+    def __init__(self):
+        self.timeout = 5
+        self.cwd = ""
+        self._stdin_mode = "none"
+        self._snapshot_ready = True
+        self._prefer_nonlogin = False
+
+    def _before_execute(self):
+        pass
+
+    def _prepare_command(self, command):
+        return command, None
+
+    def _wrap_command(self, command, cwd):
+        return command
+
+    def _run_bash(self, command, *, login=False, timeout=None, stdin_data=None):
+        return None
+
+    def _wait_for_process(self, proc, *, timeout=None, output=None, **_ignored):
+        return {"output": "", "returncode": 0}
+
+    def _update_cwd(self, result):
+        pass
+
+    def cleanup(self):
+        pass
+
+
+def test_docker_execute_rejects_rewrite_parameter():
+    """Docker's **kwargs forwarder must not reopen the retired option."""
+    env = _DockerWrapperProbe()
+    with pytest.raises(TypeError, match="rewrite_compound_background"):
+        env.execute("true", rewrite_compound_background=False)
+
+
 class _ProbeEnv(env_base.BaseEnvironment):
     """Concrete environment that records what reaches ``_wrap_command`` --
     the exact seam the retired rewriter used to sit in front of."""