diff --git a/README.md b/README.md
index 5286e86ea..df38012ab 100644
--- a/README.md
+++ b/README.md
@@ -1577,7 +1577,8 @@ supervision is available; other container and Modal paths use asynchronous HTTP
 commands so a long command does not block health checks or file requests.
 Command deadlines terminate the command process group. Commands that finish
 normally can leave background processes running. Persistent bash sessions retain
-shell state between calls and currently use SWE-ReX's session execution path.
+shell state between calls. Each session serializes shell operations in a worker
+thread, so waiting for its prompt leaves health checks and file requests responsive.
 
 ### Sandbox Modes
 
diff --git a/src/olmo_eval/harness/sandbox/_http_session.py b/src/olmo_eval/harness/sandbox/_http_session.py
new file mode 100644
index 000000000..83cb142a9
--- /dev/null
+++ b/src/olmo_eval/harness/sandbox/_http_session.py
@@ -0,0 +1,241 @@
+"""Serialize persistent shell operations without blocking the HTTP server."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+import os
+import signal
+from collections.abc import Awaitable, Callable
+from typing import Any, TypeVar
+
+from swerex.exceptions import CommandTimeoutError
+from swerex.runtime import local
+from swerex.runtime.abstract import (
+    BashAction,
+    BashInterruptAction,
+    BashObservation,
+    CloseResponse,
+    CloseSessionRequest,
+    CloseSessionResponse,
+    CreateBashSessionResponse,
+    CreateSessionRequest,
+    CreateSessionResponse,
+)
+from swerex.runtime.local import BashSession
+
+_T = TypeVar("_T")
+
+
+class ResponsiveBashSession(BashSession):
+    """Keep one pexpect reader per session while other HTTP requests run."""
+
+    def __init__(self, *args: Any, **kwargs: Any):
+        super().__init__(*args, **kwargs)
+        self._operation_lock = asyncio.Lock()
+        self._active_worker: asyncio.Task[Any] | None = None
+        self._closing = False
+        self._interrupt_task: asyncio.Task[None] | None = None
+
+    def _kill_foreground(self, *, allow_shell: bool = True) -> None:
+        with contextlib.suppress(RuntimeError, OSError):
+            foreground = os.tcgetpgrp(self.shell.child_fd)
+            if (
+                foreground > 0
+                and foreground != os.getpgrp()
+                and (allow_shell or foreground != self.shell.pid)
+            ):
+                os.killpg(foreground, signal.SIGKILL)
+
+    async def _finish_reader_timeout(self) -> None:
+        # Ownership is retained, so a non-shell foreground group still belongs
+        # to this action, even if its command advanced to another external job.
+        self._kill_foreground(allow_shell=False)
+
+        def drain_prompt() -> None:
+            self.shell.expect(self._ps1, timeout=1.0)
+
+        # Another SIGINT could flush the completed job's prompt without creating
+        # a replacement when the persistent shell ignores SIGINT.
+        try:
+            await asyncio.to_thread(drain_prompt)
+        except Exception as error:
+            self._closing = True
+            self._kill_foreground(allow_shell=False)
+
+            def close_terminal() -> None:
+                shell = self.shell
+                try:
+                    shell.close(force=True)
+                finally:
+                    if shell.isalive() and not shell.terminate(force=True):
+                        raise RuntimeError("Session terminal could not be terminated")
+
+            try:
+                await asyncio.to_thread(close_terminal)
+            except Exception as cleanup_error:
+                raise error from cleanup_error
+            raise
+
+    async def _interrupt_active(self, task: asyncio.Task[Any]) -> None:
+        if self._active_worker is not task:
+            return
+        if self._interrupt_task is None:
+            self._interrupt_task = asyncio.create_task(self._stop_active(task))
+        await asyncio.shield(self._interrupt_task)
+
+    async def _stop_active(self, task: asyncio.Task[Any]) -> None:
+        if self._active_worker is not task:
+            return
+        if task.done():
+            if not task.cancelled() and isinstance(task.exception(), CommandTimeoutError):
+                await self._finish_reader_timeout()
+            return
+        with contextlib.suppress(RuntimeError, OSError):
+            self.shell.sendintr()
+        try:
+            await asyncio.wait_for(asyncio.shield(task), 0.5)
+            return
+        except CommandTimeoutError:
+            await self._finish_reader_timeout()
+            return
+        except asyncio.TimeoutError:  # noqa: UP041 -- remote Python 3.10
+            pass
+        except Exception:
+            return
+        if self._active_worker is not task:
+            return
+        if task.done():
+            if not task.cancelled() and isinstance(task.exception(), CommandTimeoutError):
+                await self._finish_reader_timeout()
+            return
+        # A foreground job may ignore SIGINT. Signal only this terminal's
+        # foreground group, leaving the server and other sessions untouched.
+        self._kill_foreground()
+        try:
+            await asyncio.wait_for(asyncio.shield(task), 1.0)
+            return
+        except CommandTimeoutError:
+            await self._finish_reader_timeout()
+            return
+        except asyncio.TimeoutError:  # noqa: UP041 -- remote Python 3.10
+            pass
+        except Exception:
+            return
+
+        # Closing an unresponsive terminal wakes its remaining pexpect reader.
+        def close_terminal() -> None:
+            if self._active_worker is task and not task.done():
+                self.shell.close(force=True)
+
+        await asyncio.to_thread(close_terminal)
+        try:
+            await asyncio.shield(task)
+        except CommandTimeoutError:
+            await self._finish_reader_timeout()
+        except Exception:
+            pass
+
+    async def _run_worker(
+        self, operation: Callable[[], Awaitable[_T]], *, allow_closing: bool = False
+    ) -> _T:
+        async with self._operation_lock:
+            if self._closing and not allow_closing:
+                raise RuntimeError("Session is closing")
+
+            async def run_operation() -> _T:
+                return await operation()
+
+            def run() -> _T:
+                return asyncio.run(run_operation())
+
+            task = asyncio.create_task(asyncio.to_thread(run))
+            self._active_worker = task
+            cancelled = False
+            while not task.done():
+                try:
+                    await asyncio.shield(task)
+                except asyncio.CancelledError:
+                    cancelled = True
+                    # Keep the lock until the sole reader has settled.
+                    try:
+                        await self._interrupt_active(task)
+                    except asyncio.CancelledError:
+                        continue
+                    except Exception:
+                        # Failed cleanup leaves the terminal unusable; the
+                        # control-task handling below surfaces the error.
+                        self._closing = True
+                        break
+                except Exception:
+                    break
+            # A reader deadline does not stop its command. Keep ownership until
+            # the shared interrupt cleanup and its prompt read have settled.
+            control = self._interrupt_task
+            if control is not None:
+                while not control.done():
+                    try:
+                        await asyncio.shield(control)
+                    except asyncio.CancelledError:
+                        cancelled = True
+                    except Exception:
+                        break
+                error = control.exception()
+                if error is not None:
+                    self._closing = True
+            else:
+                error = None
+            self._interrupt_task = None
+            self._active_worker = None
+            if cancelled:
+                with contextlib.suppress(Exception):
+                    task.result()
+                raise asyncio.CancelledError
+            if error is not None:
+                raise error
+            return task.result()
+
+    async def start(self) -> CreateBashSessionResponse:
+        return await self._run_worker(super().start)
+
+    async def run(self, action: BashAction | BashInterruptAction) -> BashObservation:
+        if isinstance(action, BashInterruptAction) and self._active_worker is not None:
+            await self._interrupt_active(self._active_worker)
+            # The existing reader consumed the prompt. Reading it again could
+            # block when the shell itself ignores SIGINT.
+            return BashObservation(output="", exit_code=0)
+        return await self._run_worker(lambda: super(ResponsiveBashSession, self).run(action))
+
+    async def close(self) -> CloseSessionResponse:
+        self._closing = True
+        if self._active_worker is not None:
+            try:
+                await self._interrupt_active(self._active_worker)
+            except Exception as exc:
+                self.logger.warning("Session interrupt cleanup failed during close: %s", exc)
+        return await self._run_worker(super().close, allow_closing=True)
+
+
+class ResponsiveSessionRuntime(local.LocalRuntime):
+    """Serialize registry mutations across offloaded shell lifecycles."""
+
+    def __init__(self, **kwargs: Any):
+        super().__init__(**kwargs)
+        self._lifecycle_lock = asyncio.Lock()
+
+    async def create_session(self, request: CreateSessionRequest) -> CreateSessionResponse:
+        async with self._lifecycle_lock:
+            return await super().create_session(request)
+
+    async def close_session(self, request: CloseSessionRequest) -> CloseSessionResponse:
+        async with self._lifecycle_lock:
+            return await super().close_session(request)
+
+    async def close(self) -> CloseResponse:
+        async with self._lifecycle_lock:
+            return await super().close()
+
+
+def install() -> None:
+    """Install persistent sessions before the runtime creates any shell."""
+    vars(local).update(BashSession=ResponsiveBashSession, LocalRuntime=ResponsiveSessionRuntime)
diff --git a/src/olmo_eval/harness/sandbox/server_command.py b/src/olmo_eval/harness/sandbox/server_command.py
index 047e98583..4c19a2e90 100644
--- a/src/olmo_eval/harness/sandbox/server_command.py
+++ b/src/olmo_eval/harness/sandbox/server_command.py
@@ -7,12 +7,12 @@
 
 def server_command(executable: str, token: str, port: int, *, install_pipx: bool = False) -> str:
     """Preserve the server interpreter and shell environment while installing its adapter."""
-    source = Path(__file__).with_name("_http_command.py").read_bytes()
-    payload = base64.b64encode(source + b"\ninstall()\n").decode()
-    patch = (
-        f"import base64; exec(compile(base64.b64decode({payload!r}), "
-        "'<olmo-http-command>', 'exec'))\n"
-    )
+    patches = []
+    for module in ("_http_command.py", "_http_session.py"):
+        source = Path(__file__).with_name(module).read_bytes()
+        payload = base64.b64encode(source + b"\ninstall()\n").decode()
+        patches.append(f"exec(compile(base64.b64decode({payload!r}), {module!r}, 'exec'))")
+    patch = "import base64; " + "; ".join(patches) + "\n"
     encoded = base64.b64encode(patch.encode()).decode()
     # A temporary console script retains the installed entrypoint's shebang.
     # This works for an isolated SWE-ReX venv as well as an activated task venv.
diff --git a/tests/core/harness/test_http_session.py b/tests/core/harness/test_http_session.py
new file mode 100644
index 000000000..06f196c9b
--- /dev/null
+++ b/tests/core/harness/test_http_session.py
@@ -0,0 +1,563 @@
+"""Persistent shell concurrency and deadline regressions using real pexpect."""
+
+import asyncio
+
+import httpx
+import pytest
+from swerex.exceptions import CommandTimeoutError, SessionDoesNotExistError
+from swerex.runtime import local
+from swerex.runtime.abstract import (
+    BashAction,
+    BashInterruptAction,
+    CloseSessionRequest,
+    CreateBashSessionRequest,
+)
+
+from olmo_eval.harness.sandbox._http_session import ResponsiveBashSession, ResponsiveSessionRuntime
+
+
+@pytest.mark.anyio
+async def test_http_session_keeps_file_requests_responsive(monkeypatch, tmp_path):
+    import swerex.server as server
+
+    monkeypatch.setattr(local, "BashSession", ResponsiveBashSession)
+    runtime = local.LocalRuntime()
+    monkeypatch.setattr(server, "runtime", runtime)
+    marker = tmp_path / "started"
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=server.app), base_url="http://sandbox"
+    ) as client:
+        created = await client.post(
+            "/create_session", json=CreateBashSessionRequest(session="test").model_dump()
+        )
+        assert created.status_code == 200
+        task = asyncio.create_task(
+            client.post(
+                "/run_in_session",
+                json={
+                    "session": "test",
+                    "command": f"touch {marker}; sleep 1",
+                    "timeout": 3,
+                    "check": "silent",
+                    "action_type": "bash",
+                },
+            )
+        )
+        try:
+            async with asyncio.timeout(3):
+                while not marker.exists():
+                    await asyncio.sleep(0.01)
+            response = await client.post(
+                "/write_file", json={"path": str(tmp_path / "file"), "content": "ready"}
+            )
+            assert response.status_code == 200 and not task.done()
+            assert (await task).status_code == 200
+        finally:
+            await task
+            await runtime.close()
+
+
+@pytest.mark.anyio
+async def test_session_serializes_commands_and_retains_environment(tmp_path):
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    try:
+        first = asyncio.create_task(
+            session.run(
+                BashAction(
+                    command=f"sleep 0.1; cd {tmp_path}; export PERSIST=retained",
+                    timeout=3,
+                )
+            )
+        )
+        await asyncio.sleep(0.01)
+        second = asyncio.create_task(
+            session.run(BashAction(command='printf "%s:%s" "$PERSIST" "$PWD"', timeout=3))
+        )
+        await first
+        assert (await second).output == f"retained:{tmp_path}"
+    finally:
+        await session.close()
+
+
+@pytest.mark.anyio
+async def test_native_deadline_and_interrupt_recover_session():
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    try:
+        await session.run(BashAction(command="echo ready", timeout=3, check="silent"))
+        with pytest.raises(CommandTimeoutError):
+            await session.run(BashAction(command="sleep 30", timeout=0.5, check="silent"))
+        await session.run(BashInterruptAction())
+        assert (
+            "recovered"
+            in (await session.run(BashAction(command="echo recovered", timeout=3))).output
+        )
+    finally:
+        await session.close()
+
+
+@pytest.mark.anyio
+async def test_cancellation_settles_reader_before_next_command(tmp_path):
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    marker = tmp_path / "started"
+    try:
+        task = asyncio.create_task(
+            session.run(BashAction(command=f"touch {marker}; sleep 30", timeout=3))
+        )
+        async with asyncio.timeout(3):
+            while not marker.exists():
+                await asyncio.sleep(0.01)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, 5)
+        assert (
+            "recovered"
+            in (await session.run(BashAction(command="echo recovered", timeout=3))).output
+        )
+    finally:
+        await session.close()
+
+
+@pytest.mark.anyio
+async def test_failed_cancellation_cleanup_still_cancels_and_closes(monkeypatch, tmp_path):
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    marker = tmp_path / "started"
+
+    async def failing_stop(task):
+        raise RuntimeError("cleanup failed")
+
+    monkeypatch.setattr(session, "_stop_active", failing_stop)
+    try:
+        task = asyncio.create_task(
+            session.run(BashAction(command=f"touch {marker}; sleep 1", timeout=5))
+        )
+        async with asyncio.timeout(3):
+            while not marker.exists():
+                await asyncio.sleep(0.01)
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await asyncio.wait_for(task, 5)
+        assert session._active_worker is None
+        assert session._interrupt_task is None
+        with pytest.raises(RuntimeError, match="Session is closing"):
+            await session.run(BashAction(command="echo late", timeout=3))
+    finally:
+        await asyncio.sleep(1.5)
+        await session.close()
+
+
+@pytest.mark.anyio
+async def test_interrupt_can_signal_an_active_serialized_reader(tmp_path):
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    marker = tmp_path / "started"
+    try:
+        task = asyncio.create_task(
+            session.run(
+                BashAction(
+                    command=f"touch {marker}; sleep 30",
+                    timeout=3,
+                    check="silent",
+                )
+            )
+        )
+        async with asyncio.timeout(3):
+            while not marker.exists():
+                await asyncio.sleep(0.01)
+        await asyncio.wait_for(session.run(BashInterruptAction()), 3)
+        await asyncio.wait_for(task, 1)
+        response = await session.run(BashAction(command="echo recovered", timeout=3))
+        assert "recovered" in response.output
+    finally:
+        await session.close()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("operation", ["cancel", "interrupt", "close"])
+async def test_ignored_interrupt_without_deadline_does_not_lock_session(tmp_path, operation):
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    marker = tmp_path / "started"
+    task = asyncio.create_task(
+        session.run(
+            BashAction(
+                command=f"trap '' INT; touch {marker}; sleep 30",
+                timeout=None,
+                check="silent",
+            )
+        )
+    )
+    try:
+        async with asyncio.timeout(3):
+            while not marker.exists():
+                await asyncio.sleep(0.01)
+        if operation == "cancel":
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(task, 4)
+        elif operation == "interrupt":
+            await asyncio.wait_for(session.run(BashInterruptAction()), 4)
+            await task
+        else:
+            await asyncio.wait_for(session.close(), 4)
+            await task
+        assert task.done()
+        if operation != "close":
+            response = await session.run(
+                BashAction(command="trap - INT; echo recovered", timeout=3)
+            )
+            assert "recovered" in response.output
+    finally:
+        await asyncio.wait_for(session.close(), 4)
+
+
+@pytest.mark.anyio
+async def test_concurrent_close_uses_domain_error(monkeypatch):
+    monkeypatch.setattr(local, "BashSession", ResponsiveBashSession)
+    runtime = ResponsiveSessionRuntime()
+    await runtime.create_session(CreateBashSessionRequest(session="test"))
+    responses = await asyncio.gather(
+        runtime.close_session(CloseSessionRequest(session="test")),
+        runtime.close_session(CloseSessionRequest(session="test")),
+        return_exceptions=True,
+    )
+    assert sum(isinstance(value, SessionDoesNotExistError) for value in responses) == 1
+    assert sum(not isinstance(value, Exception) for value in responses) == 1
+
+
+@pytest.mark.anyio
+async def test_runtime_close_and_create_do_not_mutate_registry_during_iteration(monkeypatch):
+    monkeypatch.setattr(local, "BashSession", ResponsiveBashSession)
+    runtime = ResponsiveSessionRuntime()
+    await runtime.create_session(CreateBashSessionRequest(session="first"))
+    try:
+        closing = asyncio.create_task(runtime.close())
+        await asyncio.sleep(0)
+        creating = asyncio.create_task(
+            runtime.create_session(CreateBashSessionRequest(session="second"))
+        )
+        await asyncio.gather(closing, creating)
+        assert "second" in runtime.sessions
+    finally:
+        await runtime.close()
+
+
+@pytest.mark.anyio
+async def test_close_prevents_a_queued_unbounded_command_from_starting(tmp_path):
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    marker = tmp_path / "started"
+    first = asyncio.create_task(
+        session.run(BashAction(command=f"touch {marker}; sleep 30", timeout=None, check="silent"))
+    )
+    try:
+        async with asyncio.timeout(3):
+            while not marker.exists():
+                await asyncio.sleep(0.01)
+        queued = asyncio.create_task(session.run(BashAction(command="sleep 30", timeout=None)))
+        await asyncio.sleep(0)
+        await asyncio.wait_for(session.close(), 4)
+        await first
+        with pytest.raises(RuntimeError, match="closing"):
+            await queued
+    finally:
+        await asyncio.wait_for(session.close(), 4)
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    "operation",
+    [
+        "interrupt",
+        "cancel",
+        "both",
+        "cancel-during-cleanup",
+        "natural-exit",
+        "late-cleanup",
+        "grace-boundary",
+        "terminal-close-boundary",
+    ],
+)
+async def test_deadline_during_interrupt_grace_keeps_queued_action_safe(
+    monkeypatch, tmp_path, operation
+):
+    import contextlib
+    import os
+    import signal
+    import threading
+    import time
+
+    import pexpect
+
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    for _ in range(3):
+        ready = await session.run(BashAction(command="echo warmed-up", timeout=3, check="silent"))
+        if "warmed-up" in ready.output:
+            break
+    assert "warmed-up" in ready.output
+    marker = tmp_path / "started"
+    finish = tmp_path / "finish"
+    finished = tmp_path / "finished"
+    second_started = tmp_path / "second-started"
+    expire = threading.Event()
+    interrupted = threading.Event()
+    draining = threading.Event()
+    cancel_requested = threading.Event()
+    readers_lock = threading.Lock()
+    readers = 0
+    peak_readers = 0
+    original_expect = session.shell.expect
+    original_sendintr = session.shell.sendintr
+
+    def expect(pattern, timeout=-1, **kwargs):
+        nonlocal readers, peak_readers
+        with readers_lock:
+            readers += 1
+            peak_readers = max(readers, peak_readers)
+        try:
+            if timeout == 0.1:
+                # Start the real deadline after SIGINT, inside its grace period.
+                assert interrupted.wait(3)
+                if operation == "terminal-close-boundary":
+                    assert expire.wait(3)
+            elif timeout == 1.0 and interrupted.is_set():
+                draining.set()
+                if operation == "cancel-during-cleanup":
+                    assert cancel_requested.wait(3)
+            try:
+                return original_expect(pattern, timeout=timeout, **kwargs)
+            except pexpect.TIMEOUT:
+                if operation == "natural-exit" and timeout == 0.1:
+                    finish.touch()
+                    until = time.monotonic() + 3
+                    while not finished.exists():
+                        assert time.monotonic() < until
+                        time.sleep(0.01)
+                raise
+        finally:
+            with readers_lock:
+                readers -= 1
+
+    def sendintr():
+        original_sendintr()
+        interrupted.set()
+
+    monkeypatch.setattr(session.shell, "expect", expect)
+    monkeypatch.setattr(session.shell, "sendintr", sendintr)
+    if operation == "late-cleanup":
+        original_stop = session._stop_active
+
+        async def stop(*args):
+            # Let the real reader expire before its scheduled cleanup starts.
+            interrupted.set()
+            while not args[0].done():
+                await asyncio.sleep(0.01)
+            await original_stop(*args)
+
+        monkeypatch.setattr(session, "_stop_active", stop)
+    if operation in ("grace-boundary", "terminal-close-boundary"):
+        original_wait_for = asyncio.wait_for
+
+        async def wait_for(awaitable, timeout):
+            if operation == "terminal-close-boundary" and timeout in (0.5, 1.0):
+                awaitable.cancel()
+                raise TimeoutError
+            try:
+                return await original_wait_for(awaitable, timeout)
+            except CommandTimeoutError:
+                if timeout == 0.5:
+                    # The grace timer and reader deadline can become ready together.
+                    raise TimeoutError from None
+                raise
+
+        monkeypatch.setattr(asyncio, "wait_for", wait_for)
+    if operation == "terminal-close-boundary":
+        original_to_thread = asyncio.to_thread
+
+        async def to_thread(function, *args, **kwargs):
+            if function.__name__ == "close_terminal":
+                async with asyncio.timeout(3):
+                    while not second_started.exists():
+                        await asyncio.sleep(0.01)
+                expire.set()
+                while not session._active_worker.done():
+                    await asyncio.sleep(0.01)
+            return await original_to_thread(function, *args, **kwargs)
+
+        monkeypatch.setattr(asyncio, "to_thread", to_thread)
+    work = (
+        f"while [ ! -f {finish} ]; do sleep 0.01; done; touch {finished}"
+        if operation == "natural-exit"
+        else "sleep 30"
+    )
+    command = f"trap '' INT; /bin/sh -c 'touch {marker}; {work}'"
+    if operation == "terminal-close-boundary":
+        command += f"; /bin/sh -c 'echo $$ > {second_started}; sleep 30'"
+    first = asyncio.create_task(
+        session.run(BashAction(command=command, timeout=0.1, check="silent"))
+    )
+    foreground = None
+    queued = None
+    interrupt_task = None
+    try:
+        async with asyncio.timeout(3):
+            while not marker.exists():
+                await asyncio.sleep(0.01)
+        foreground = os.tcgetpgrp(session.shell.child_fd)
+        assert foreground != os.getpgrp()
+        queued = asyncio.create_task(
+            session.run(
+                BashAction(
+                    command="trap - INT; echo queued-recovered",
+                    timeout=2,
+                    check="silent",
+                )
+            )
+        )
+        await asyncio.sleep(0)
+        if operation in (
+            "interrupt",
+            "natural-exit",
+            "late-cleanup",
+            "grace-boundary",
+            "terminal-close-boundary",
+        ):
+            await asyncio.wait_for(session.run(BashInterruptAction()), 4)
+            with pytest.raises(CommandTimeoutError):
+                await first
+        else:
+            if operation != "cancel":
+                interrupt_task = asyncio.create_task(session.run(BashInterruptAction()))
+                signal_seen = interrupted if operation == "both" else draining
+                async with asyncio.timeout(3):
+                    while not signal_seen.is_set():
+                        await asyncio.sleep(0.01)
+            first.cancel()
+            cancel_requested.set()
+            with pytest.raises(asyncio.CancelledError):
+                await asyncio.wait_for(first, 4)
+            if interrupt_task is not None:
+                await asyncio.wait_for(interrupt_task, 4)
+        response = await asyncio.wait_for(queued, 4)
+        assert response.output.strip() == "queued-recovered"
+        assert peak_readers == 1
+    finally:
+        interrupted.set()
+        cancel_requested.set()
+        expire.set()
+        finish.touch()
+        if second_started.exists():
+            with contextlib.suppress(ProcessLookupError):
+                os.killpg(int(second_started.read_text()), signal.SIGKILL)
+        if foreground is not None:
+            with contextlib.suppress(ProcessLookupError):
+                os.killpg(foreground, signal.SIGKILL)
+        for task in (first, queued, interrupt_task):
+            if task is not None:
+                with contextlib.suppress(Exception, asyncio.CancelledError):
+                    await task
+        await session.close()
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("concurrent_close", [False, True])
+async def test_failed_prompt_drain_terminates_compound_action(
+    monkeypatch, tmp_path, concurrent_close
+):
+    import contextlib
+    import os
+    import signal
+    import subprocess
+    import threading
+
+    import pexpect
+
+    session = ResponsiveBashSession(CreateBashSessionRequest(session="test"))
+    await session.start()
+    for _ in range(3):
+        ready = await session.run(BashAction(command="echo warmed-up", timeout=3, check="silent"))
+        if "warmed-up" in ready.output:
+            break
+    assert "warmed-up" in ready.output
+    started = tmp_path / "started"
+    second_pid = tmp_path / "second-pid"
+    interrupted = threading.Event()
+    draining = threading.Event()
+    release = threading.Event()
+    original_expect = session.shell.expect
+    original_sendintr = session.shell.sendintr
+
+    def expect(pattern, timeout=-1, **kwargs):
+        if timeout == 0.1:
+            assert interrupted.wait(3)
+        elif timeout == 1.0 and interrupted.is_set():
+            draining.set()
+            assert release.wait(3)
+        return original_expect(pattern, timeout=timeout, **kwargs)
+
+    def sendintr():
+        original_sendintr()
+        interrupted.set()
+
+    monkeypatch.setattr(session.shell, "expect", expect)
+    monkeypatch.setattr(session.shell, "sendintr", sendintr)
+    action = asyncio.create_task(
+        session.run(
+            BashAction(
+                command=f"trap '' INT; /bin/sh -c 'touch {started}; sleep 30'; "
+                f"/bin/sh -c 'echo $$ > {second_pid}; sleep 30'",
+                timeout=0.1,
+                check="silent",
+            )
+        )
+    )
+    interrupt = None
+    closing = None
+    try:
+        async with asyncio.timeout(3):
+            while not started.exists():
+                await asyncio.sleep(0.01)
+        interrupt = asyncio.create_task(session.run(BashInterruptAction()))
+        async with asyncio.timeout(3):
+            while not draining.is_set() or not second_pid.exists():
+                await asyncio.sleep(0.01)
+        if concurrent_close:
+            closing = asyncio.create_task(session.close())
+            await asyncio.sleep(0)
+        release.set()
+        with pytest.raises(pexpect.TIMEOUT):
+            await asyncio.wait_for(interrupt, 4)
+        with pytest.raises(pexpect.TIMEOUT):
+            await action
+        if closing is not None:
+            await asyncio.wait_for(closing, 3)
+        else:
+            assert not session.shell.isalive()
+            await session.close()
+        with pytest.raises(RuntimeError, match="shell not initialized"):
+            _ = session.shell
+        state = subprocess.run(
+            ["ps", "-p", second_pid.read_text().strip(), "-o", "stat="],
+            capture_output=True,
+            text=True,
+            check=False,
+        ).stdout.strip()
+        assert not state or state.startswith("Z")
+    finally:
+        interrupted.set()
+        release.set()
+        if second_pid.exists():
+            with contextlib.suppress(ProcessLookupError):
+                os.killpg(int(second_pid.read_text()), signal.SIGKILL)
+        for task in (action, interrupt, closing):
+            if task is not None:
+                with contextlib.suppress(Exception, asyncio.CancelledError):
+                    await task
+        with contextlib.suppress(Exception):
+            await session.close()
+        with contextlib.suppress(RuntimeError):
+            await asyncio.to_thread(session.shell.close, force=True)