feat(coding-agent): add async bash() to IPython kernel

packages/coding-agent/CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## [Unreleased]
 
+- Added an async `bash()` function to the IPython kernel so `await bash("...")` runs shell commands without blocking the kernel event loop.
 - Fixed URLs not opening on click in fullscreen mode on terminals such as Ghostty; clicking a link in the transcript, dock, or overlays now opens it in the browser.
 
 ## [0.7.2] - 2026-08-11

packages/coding-agent/src/core/prompts/rlm.ts

@@ -16,7 +16,7 @@ const IPYTHON_CONTROL_PROMPT = [
 	"",
 	"Do not assume IPython is the native runtime of the external thing being investigated. A repository, package, service, dataset, paper, website, benchmark, or API may have its own environment and normal interface. Evaluate external systems through their own interface, then use IPython to coordinate the process and analyze what comes back.",
 	"",
-	"When running shell commands from IPython, use `%%bash` cells. If you use `%%bash`, it must be the first line of the code cell: no comments, spaces, blank lines, imports, or Python statements before it. Avoid `!cmd` shell escapes for project commands so shell behavior is explicit and multi-line commands share one shell context.",
+	'When running shell commands from IPython, prefer `await bash("...")` over `%%bash` cells: it keeps the kernel event loop responsive to interrupts, cleans up its managed process group on interruption, and bounds captured stdout/stderr. `%%bash` cells block the kernel until the command finishes. If you do use `%%bash`, it must be the first line of the code cell: no comments, spaces, blank lines, imports, or Python statements before it. Avoid `!cmd` shell escapes for project commands so shell behavior is explicit and multi-line commands share one shell context.',
 	"",
 	"Important: do not install dependencies into the IPython kernel just to make an external project import or run there. If a project import, test, script, CLI, or dependency check is needed, run it through that project's own environment and normal command interface. For example, in a Python repo use its documented commands, `uv run ...`, `.venv/bin/python ...`, or the active project interpreter from the repo root. Treat failures from that native environment as the relevant result.",
 	"",

packages/coding-agent/src/core/tools/ipython.ts

@@ -34,6 +34,230 @@ try:
 except Exception:
     pass
 
+import asyncio as _prime_agent_asyncio
+import subprocess as _prime_agent_subprocess
+import shutil as _prime_agent_shutil
+import signal as _prime_agent_signal
+import os as _prime_agent_bash_os
+
+_PRIME_AGENT_BASH_CAPTURE_LIMIT = 32 * 1024
+_PRIME_AGENT_BASH_READ_CHUNK = 64 * 1024
+_PRIME_AGENT_BASH_STOP_GRACE = 0.25
+
+
+class _PrimeAgentBoundedBytes:
+    """Tail buffer with a fixed memory ceiling and total-byte accounting."""
+    def __init__(self, limit: int):
+        self.limit = limit
+        self.data = bytearray()
+        self.total = 0
+        self.render_truncated = False
+
+    def append(self, chunk: bytes) -> None:
+        self.total += len(chunk)
+        if len(chunk) >= self.limit:
+            self.data[:] = chunk[-self.limit:]
+            return
+        overflow = len(self.data) + len(chunk) - self.limit
+        if overflow > 0:
+            del self.data[:overflow]
+        self.data.extend(chunk)
+
+    @property
+    def truncated(self) -> bool:
+        return self.total > len(self.data)
+
+    def text(self) -> str:
+        text = bytes(self.data).decode("utf-8", errors="replace")
+        rendered = text.encode("utf-8")
+        if len(rendered) <= self.limit:
+            return text
+        # Replacement characters can expand malformed bytes. Bound the actual
+        # UTF-8 payload IPython will serialize, trimming only from the head.
+        self.render_truncated = True
+        return rendered[-self.limit:].decode("utf-8", errors="ignore")
+
+
+class _PrimeAgentBashResult:
+    """Result of an async bash command with bounded captured output."""
+    def __init__(
+        self,
+        stdout: str,
+        stderr: str,
+        returncode: int,
+        *,
+        stdout_truncated: bool = False,
+        stderr_truncated: bool = False,
+        stdout_total_bytes: int = 0,
+        stderr_total_bytes: int = 0,
+    ):
+        self.stdout = stdout
+        self.stderr = stderr
+        self.returncode = returncode
+        self.stdout_truncated = stdout_truncated
+        self.stderr_truncated = stderr_truncated
+        self.stdout_total_bytes = stdout_total_bytes
+        self.stderr_total_bytes = stderr_total_bytes
+
+    def __str__(self) -> str:
+        parts = []
+        if self.stdout:
+            parts.append(self.stdout)
+        if self.stdout_truncated:
+            parts.append(f"[stdout truncated; captured {self.stdout_total_bytes} raw bytes; showing a bounded decoded tail]")
+        if self.stderr:
+            parts.append(self.stderr)
+        if self.stderr_truncated:
+            parts.append(f"[stderr truncated; captured {self.stderr_total_bytes} raw bytes; showing a bounded decoded tail]")
+        text = chr(10).join(parts)
+        if self.returncode != 0:
+            text += chr(10) + chr(10) + f"Command exited with code {self.returncode}"
+        return text
+
+    def __repr__(self) -> str:
+        # IPython renders a cell's final expression with repr(), so expose the
+        # captured command output instead of hiding it behind a metadata summary.
+        return str(self)
+
+
+async def _prime_agent_bash_drain(stream, capture: _PrimeAgentBoundedBytes) -> None:
+    while True:
+        chunk = await stream.read(_PRIME_AGENT_BASH_READ_CHUNK)
+        if not chunk:
+            return
+        capture.append(chunk)
+
+
+async def _prime_agent_bash_collect(proc, stdout_task, stderr_task) -> None:
+    # asyncio's process wait can remain pending while descendants hold inherited
+    # pipes, even after the shell's returncode is set. Observe that state directly.
+    while proc.returncode is None:
+        await _prime_agent_asyncio.sleep(0.01)
+    # A successful shell may leave background descendants in its session. Stop
+    # them before waiting for inherited pipe descriptors to close.
+    _prime_agent_bash_signal_group(proc, _prime_agent_signal.SIGTERM)
+    deadline = _prime_agent_asyncio.get_running_loop().time() + _PRIME_AGENT_BASH_STOP_GRACE
+    while _prime_agent_bash_group_exists(proc) and _prime_agent_asyncio.get_running_loop().time() < deadline:
+        await _prime_agent_asyncio.sleep(0.01)
+    if _prime_agent_bash_group_exists(proc):
+        _prime_agent_bash_signal_group(proc, _prime_agent_signal.SIGKILL)
+    await _prime_agent_asyncio.gather(stdout_task, stderr_task)
+
+
+def _prime_agent_bash_signal_group(proc, sig) -> None:
+    # The shell may have exited while descendants still hold its pipes open, so
+    # always signal the session by the original leader PID.
+    try:
+        _prime_agent_bash_os.killpg(proc.pid, sig)
+    except (ProcessLookupError, PermissionError):
+        pass
+
+
+def _prime_agent_bash_group_exists(proc) -> bool:
+    try:
+        _prime_agent_bash_os.killpg(proc.pid, 0)
+        return True
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+
+
+
+
+async def _prime_agent_bash_stop(proc, collector, stdout_task, stderr_task) -> None:
+    """Stop the command group and finish or cancel every pipe-drain task."""
+    _prime_agent_bash_signal_group(proc, _prime_agent_signal.SIGTERM)
+    deadline = _prime_agent_asyncio.get_running_loop().time() + _PRIME_AGENT_BASH_STOP_GRACE
+    while _prime_agent_bash_group_exists(proc) and _prime_agent_asyncio.get_running_loop().time() < deadline:
+        await _prime_agent_asyncio.sleep(0.01)
+    if _prime_agent_bash_group_exists(proc):
+        _prime_agent_bash_signal_group(proc, _prime_agent_signal.SIGKILL)
+
+    try:
+        await _prime_agent_asyncio.wait_for(_prime_agent_asyncio.shield(collector), timeout=2)
+        return
+    except _prime_agent_asyncio.TimeoutError:
+        if proc.returncode is None:
+            proc.kill()
+    finally:
+        if not collector.done():
+            collector.cancel()
+        for task in (stdout_task, stderr_task):
+            if not task.done():
+                task.cancel()
+        await _prime_agent_asyncio.gather(collector, stdout_task, stderr_task, return_exceptions=True)
+
+
+async def bash(
+    command: str,
+    *,
+    timeout: float | None = None,
+    cwd: str | None = None,
+    max_output_bytes: int = _PRIME_AGENT_BASH_CAPTURE_LIMIT,
+) -> _PrimeAgentBashResult:
+    """Run a shell command without blocking the IPython kernel event loop.
+
+    At most max_output_bytes from the tail of each output stream is retained;
+    both pipes are continuously drained so large output cannot deadlock or grow
+    the persistent kernel without bound. Truncation and total byte counts are
+    reported on the result.
+    """
+    if isinstance(max_output_bytes, bool) or not isinstance(max_output_bytes, int) or max_output_bytes <= 0:
+        raise ValueError("max_output_bytes must be a positive integer")
+    shell = _prime_agent_shutil.which("bash") or "/bin/bash"
+    env = dict(_prime_agent_bash_os.environ)
+    work_dir = cwd or _prime_agent_bash_os.getcwd()
+    spawn_options = {"start_new_session": True}
+
+    proc = await _prime_agent_asyncio.create_subprocess_exec(
+        shell, "-c", command,
+        stdin=_prime_agent_subprocess.DEVNULL,
+        stdout=_prime_agent_subprocess.PIPE,
+        stderr=_prime_agent_subprocess.PIPE,
+        cwd=work_dir,
+        env=env,
+        limit=_PRIME_AGENT_BASH_READ_CHUNK,
+        **spawn_options,
+    )
+    stdout_capture = _PrimeAgentBoundedBytes(max_output_bytes)
+    stderr_capture = _PrimeAgentBoundedBytes(max_output_bytes)
+    stdout_task = _prime_agent_asyncio.create_task(_prime_agent_bash_drain(proc.stdout, stdout_capture))
+    stderr_task = _prime_agent_asyncio.create_task(_prime_agent_bash_drain(proc.stderr, stderr_capture))
+    collector = _prime_agent_asyncio.create_task(_prime_agent_bash_collect(proc, stdout_task, stderr_task))
+
+    try:
+        await _prime_agent_asyncio.wait_for(_prime_agent_asyncio.shield(collector), timeout=timeout)
+    except _prime_agent_asyncio.TimeoutError:
+        cleanup = _prime_agent_asyncio.create_task(
+            _prime_agent_bash_stop(proc, collector, stdout_task, stderr_task)
+        )
+        await _prime_agent_asyncio.shield(cleanup)
+        raise TimeoutError(f"bash command timed out after {timeout}s") from None
+    except BaseException:
+        cleanup = _prime_agent_asyncio.create_task(
+            _prime_agent_bash_stop(proc, collector, stdout_task, stderr_task)
+        )
+        await _prime_agent_asyncio.shield(cleanup)
+        raise
+    finally:
+        if not collector.done() and proc.returncode is not None:
+            collector.cancel()
+            await _prime_agent_asyncio.gather(collector, return_exceptions=True)
+
+    stdout_text = stdout_capture.text()
+    stderr_text = stderr_capture.text()
+    return _PrimeAgentBashResult(
+        stdout=stdout_text,
+        stderr=stderr_text,
+        returncode=proc.returncode if proc.returncode is not None else -1,
+        stdout_truncated=stdout_capture.truncated or stdout_capture.render_truncated,
+        stderr_truncated=stderr_capture.truncated or stderr_capture.render_truncated,
+        stdout_total_bytes=stdout_capture.total,
+        stderr_total_bytes=stderr_capture.total,
+    )
+
+
 try:
     import rlm as _prime_agent_rlm_module
     rlm = _prime_agent_rlm_module.rlm