fix: apply _rewrite_compound_background in spawn_local to prevent worker deadlock on server backgrounding

Issue #68915: when the agent runs a compound command with trailing & (e.g.
`cd /app && node server.js &`), bash parses it as `(A && B) &` — a subshell
that holds the stdout pipe open forever when B is a long-running server.
The existing _rewrite_compound_background in terminal_tool.py correctly
rewrites this to `A && { B & }` to avoid the subshell fork, but it was only
applied in the foreground execute() path (tools/environments/base.py).

The background spawn_local() path bypasses base.py entirely and passed the
raw command directly to Popen/PTY, leaving the deadlock unmitigated.

Fix: apply _rewrite_compound_background in spawn_local() before the command
is passed to Popen or PTY spawn. Uses a lazy import to avoid circular
dependency (terminal_tool imports process_registry).

- PTY spawn path: now uses safe_command (rewritten)
- Popen spawn path: now uses safe_command (rewritten)
- Session.command still stores the original (unrewritten) command for display
- Simple `cmd &` is left unchanged (no subshell bug)

Tests: 4 regression tests verifying (1) compound is rewritten, (2) simple bg
is preserved, (3) multi-line compounds are rewritten, (4) session.command
stores original.

tests/tools/test_process_registry.py

@@ -911,6 +911,126 @@ class TestPopenLeakOnSetupFailure:
         assert session.pid == 7777
 
 
+# =========================================================================
+# Spawn rewrite regression (issue #68915)
+# =========================================================================
+
+
+class TestSpawnRewriteCompoundBackground:
+    """Verify that spawn_local rewrites `A && B &` patterns to avoid subshell deadlocks.
+
+    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.
+    """
+
+    def test_compound_and_background_gets_rewritten(self, registry):
+        """A && B & must be rewritten to A && { B & } before Popen."""
+        captured_cmd = []
+
+        def fake_popen(args, **kwargs):
+            captured_cmd.append(args)
+            proc = MagicMock()
+            proc.pid = 1111
+            proc.stdout = MagicMock()
+            return proc
+
+        fake_thread = MagicMock()
+        fake_thread.daemon = False
+
+        with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \
+             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] or \
+               "&& { node" in shell_cmd[2]
+
+    def test_simple_background_preserved(self, registry):
+        """Simple cmd & (no &&) must NOT be rewritten — no subshell bug."""
+        captured_cmd = []
+
+        def fake_popen(args, **kwargs):
+            captured_cmd.append(args)
+            proc = MagicMock()
+            proc.pid = 2222
+            proc.stdout = MagicMock()
+            return proc
+
+        fake_thread = MagicMock()
+        fake_thread.daemon = False
+
+        with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \
+             patch("subprocess.Popen", side_effect=fake_popen), \
+             patch("threading.Thread", return_value=fake_thread), \
+             patch.object(registry, "_write_checkpoint"):
+            registry.spawn_local("sleep 5 &", cwd="/tmp")
+
+        assert len(captured_cmd) == 1
+        shell_cmd = captured_cmd[0][2]
+        # Simple background must remain as-is
+        assert "sleep 5 &" in shell_cmd
+
+    def test_multi_line_compound_background(self, registry):
+        """Multi-line cd + server start must be rewritten."""
+        captured_cmd = []
+
+        def fake_popen(args, **kwargs):
+            captured_cmd.append(args)
+            proc = MagicMock()
+            proc.pid = 3333
+            proc.stdout = MagicMock()
+            return proc
+
+        fake_thread = MagicMock()
+        fake_thread.daemon = False
+
+        with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \
+             patch("subprocess.Popen", side_effect=fake_popen), \
+             patch("threading.Thread", return_value=fake_thread), \
+             patch.object(registry, "_write_checkpoint"):
+            registry.spawn_local(
+                "cd /app && python3 -m http.server &\nsleep 1\ncurl http://localhost:8000/",
+                cwd="/tmp",
+            )
+
+        assert len(captured_cmd) == 1
+        shell_cmd = captured_cmd[0][2]
+        # First line's compound should be rewritten; rest is preserved
+        assert "&& { python3 -m http.server & }" in shell_cmd or \
+               "&& { python3" in shell_cmd
+        assert "sleep 1" in shell_cmd
+        assert "curl http://localhost:8000/" in shell_cmd
+
+    def test_session_stores_original_command(self, registry):
+        """Session.command must store the ORIGINAL (unrewritten) command."""
+        captured = []
+
+        def fake_popen(args, **kwargs):
+            proc = MagicMock()
+            proc.pid = 4444
+            proc.stdout = MagicMock()
+            captured.append(args)
+            return proc
+
+        fake_thread = MagicMock()
+        fake_thread.daemon = False
+
+        with patch("tools.process_registry._find_shell", return_value="/bin/bash"), \
+             patch("subprocess.Popen", side_effect=fake_popen), \
+             patch("threading.Thread", return_value=fake_thread), \
+             patch.object(registry, "_write_checkpoint"):
+            session = registry.spawn_local("A && B &", cwd="/tmp")
+
+        assert session.command == "A && B &"
+        assert "{ B" in captured[0][2]  # rewritten in Popen args
+
+
 # =========================================================================
 # Checkpoint
 # =========================================================================

tools/process_registry.py

@@ -705,6 +705,15 @@ class ProcessRegistry:
                      CLI tools (Codex, Claude Code, Python REPL). Falls back to
                      subprocess.Popen if ptyprocess is not installed.
         """
+        # Guard against the `A && B &` subshell-wait trap (issue #68915).
+        # Bash parses ``A && B &`` as ``(A && B) &`` — a subshell that holds
+        # the stdout pipe open forever when B is a long-running server.
+        # The rewriter wraps it to ``A && { B & }`` so no subshell fork.
+        # Lazy import avoids circular dependency (terminal_tool imports this).
+        from tools.terminal_tool import _rewrite_compound_background as _rewrite_bg
+
+        safe_command = _rewrite_bg(command)
+
         session = ProcessSession(
             id=f"proc_{uuid.uuid4().hex[:12]}",
             command=command,
@@ -725,7 +734,7 @@ class ProcessRegistry:
                 pty_env = _sanitize_subprocess_env(os.environ, env_vars)
                 pty_env["PYTHONUNBUFFERED"] = "1"
                 pty_proc = _PtyProcessCls.spawn(
-                    [user_shell, "-lic", f"set +m; {command}"],
+                    [user_shell, "-lic", f"set +m; {safe_command}"],
                     cwd=session.cwd,
                     env=pty_env,
                     dimensions=(30, 120),
@@ -769,7 +778,7 @@ class ProcessRegistry:
         _popen_kwargs = {"creationflags": windows_hide_flags()} if _IS_WINDOWS else {}
 
         proc = subprocess.Popen(
-            [user_shell, "-lic", f"set +m; {command}"],
+            [user_shell, "-lic", f"set +m; {safe_command}"],
             text=True,
             cwd=session.cwd,
             env=bg_env,