From 95d20a6c8b8d7da7fba439fb04d39324587ca8fb Mon Sep 17 00:00:00 2001
From: westey <164392973+westey-m@users.noreply.github.com>
Date: Wed, 28 Oct 2026 12:55:11 +0000
Subject: [PATCH 1/3] Extend max output limit to stderr

---
 python/packages/tools/README.md               |  27 +-
 .../agent_framework_tools/shell/_capture.py   |  58 ++++
 .../agent_framework_tools/shell/_docker.py    | 149 +++++-----
 .../agent_framework_tools/shell/_executor.py  |  34 ++-
 .../agent_framework_tools/shell/_session.py   | 277 +++++++++---------
 .../agent_framework_tools/shell/_tool.py      |  23 +-
 .../agent_framework_tools/shell/_truncate.py  |  82 +++++-
 .../agent_framework_tools/shell/_types.py     |   4 +-
 .../tools/tests/test_docker_shell_tool.py     | 138 ++++++++-
 .../tools/tests/test_local_shell_tool.py      | 236 +++++++++++++--
 .../tools/tests/test_shell_parse_rc.py        |  52 +++-
 .../tests/test_shell_truncate_and_quote.py    |  31 +-
 .../references/harness-agent-builder.md       |   5 +-
 13 files changed, 846 insertions(+), 270 deletions(-)
 create mode 100644 python/packages/tools/agent_framework_tools/shell/_capture.py

diff --git a/python/packages/tools/README.md b/python/packages/tools/README.md
index 12fea77a9e5..539e8dbb386 100644
--- a/python/packages/tools/README.md
+++ b/python/packages/tools/README.md
@@ -62,7 +62,14 @@ Defenses (in priority order):
   Windows, a timed-out persistent session is closed even if the interrupted
   command emits its completion marker. The next command starts a fresh shell;
   shell-local variables and working-directory changes from that session are lost.
-- **Output truncation** to 64 KiB (head + tail with marker).
+- **Bounded output capture** with independent `max_stdout_bytes` and
+  `max_stderr_bytes` settings, each defaulting to 64 KiB and requiring a positive
+  value. Beginning and end portions are retained while excess bytes
+  are discarded as the streams are read. Capture uses additional bounded
+  read/decoding buffers; truncation notices and UTF-8 replacement can make the
+  returned text larger than the data allowance. This is not a whole-process
+  memory limit. Container memory limits do not cover the agent's host-side
+  output buffers; enforce host resource limits separately.
 - **Audit hook** (`on_command=…`) for SIEM / append-only logs.
 - **Optional command-pattern filter** via `ShellPolicy(denylist=[...],
   allowlist=[...])`. **Empty by default.** This is a UX pre-filter, not a
@@ -95,6 +102,24 @@ Expose this tool to the agent through `shell.as_function()` and handle its
 approval requests with human review. Calling `shell.run()` directly does not
 request approval.
 
+### Output limits
+
+Both `LocalShellTool` and `DockerShellTool` accept separate limits for normal
+output (`stdout`) and error output (`stderr`):
+
+```python
+from agent_framework.tools import DockerShellTool, LocalShellTool
+
+local_shell = LocalShellTool(max_stdout_bytes=16 * 1024, max_stderr_bytes=4 * 1024)
+docker_shell = DockerShellTool(max_stdout_bytes=16 * 1024, max_stderr_bytes=4 * 1024)
+```
+
+These settings limit retained output, not how much a command may produce.
+Persistent sessions keep the existing normal-output interruption threshold of
+four times `max_stdout_bytes`. Changing `max_stderr_bytes` does not change that
+threshold or introduce an error-output interruption rule. Stateless commands
+are not stopped solely because either output allowance is exceeded.
+
 ### Cross-OS
 
 - **Windows**: `pwsh -NoProfile -Command -` (falls back to `powershell.exe`).
diff --git a/python/packages/tools/agent_framework_tools/shell/_capture.py b/python/packages/tools/agent_framework_tools/shell/_capture.py
new file mode 100644
index 00000000000..6092ee10faf
--- /dev/null
+++ b/python/packages/tools/agent_framework_tools/shell/_capture.py
@@ -0,0 +1,58 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Bounded concurrent subprocess output capture."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from ._truncate import _OutputBuffer  # pyright: ignore[reportPrivateUsage] - shared shell capture
+
+logger = logging.getLogger(__name__)
+
+READ_CHUNK = 64 * 1024
+_DRAIN_GRACE = 2.0
+
+
+class _ProcessOutput:  # pyright: ignore[reportUnusedClass] - used by shell executors
+    """Own pipe readers independently of a command's execution timeout."""
+
+    def __init__(self, proc: asyncio.subprocess.Process, *, max_stdout_bytes: int, max_stderr_bytes: int) -> None:
+        self.stdout = _OutputBuffer(max_stdout_bytes)
+        self.stderr = _OutputBuffer(max_stderr_bytes)
+        self._readers = (
+            asyncio.create_task(self._read(proc.stdout, self.stdout)),
+            asyncio.create_task(self._read(proc.stderr, self.stderr)),
+        )
+        self._waiter = asyncio.create_task(proc.wait())
+        self._completion = asyncio.create_task(self._wait())
+
+    @staticmethod
+    async def _read(stream: asyncio.StreamReader | None, output: _OutputBuffer) -> None:
+        if stream is None:
+            return
+        while chunk := await stream.read(READ_CHUNK):
+            output.append(chunk)
+
+    async def _wait(self) -> None:
+        await asyncio.gather(*self._readers, self._waiter)
+
+    async def wait(self, timeout: float | None) -> None:
+        await asyncio.wait_for(asyncio.shield(self._completion), timeout=timeout)
+
+    async def finish(self) -> None:
+        """Drain after termination without waiting forever for inherited pipes."""
+        try:
+            await self.wait(_DRAIN_GRACE)
+        except asyncio.TimeoutError:
+            logger.warning("subprocess output did not close after termination; stopping pipe readers")
+        finally:
+            await self.close()
+
+    async def close(self) -> None:
+        tasks = (*self._readers, self._waiter, self._completion)
+        for task in tasks:
+            if not task.done():
+                task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
diff --git a/python/packages/tools/agent_framework_tools/shell/_docker.py b/python/packages/tools/agent_framework_tools/shell/_docker.py
index 20da2b78da0..f19339418b9 100644
--- a/python/packages/tools/agent_framework_tools/shell/_docker.py
+++ b/python/packages/tools/agent_framework_tools/shell/_docker.py
@@ -57,9 +57,10 @@
 from agent_framework._tools import SHELL_TOOL_KIND_VALUE
 
 from .._feature_usage import FeatureIndex
+from ._capture import _ProcessOutput  # pyright: ignore[reportPrivateUsage] - shared shell capture
 from ._policy import ShellPolicy, ShellRequest
 from ._session import ShellSession
-from ._truncate import truncate_head_tail as _truncate_bytes
+from ._truncate import _validate_output_limit  # pyright: ignore[reportPrivateUsage] - shared limit validation
 from ._types import (
     ShellCommandError,
     ShellMode,
@@ -264,7 +265,13 @@ class DockerShellTool:
             logging). Defaults to an empty policy; supply patterns
             explicitly to enable filtering.
         timeout: Per-command timeout in seconds.
-        max_output_bytes: Combined stdout/stderr byte cap before truncation.
+        max_stdout_bytes: Positive retained byte allowance for normal output.
+            Default 64 KiB, plus bounded capture overhead.
+            Truncation notices and UTF-8 replacement may enlarge returned text.
+        max_stderr_bytes: Positive retained byte allowance for error output.
+            Default 64 KiB, independent of ``max_stdout_bytes``.
+            Truncation notices and UTF-8 replacement may enlarge returned text.
+            Container memory limits do not cover host-side capture buffers.
         approval_mode: Controls the FunctionTool approval gate. Unlike
             ``LocalShellTool``, ``"never_require"`` is permitted without
             ``acknowledge_unsafe`` because the container — when launched
@@ -296,12 +303,15 @@ def __init__(
         env: Mapping[str, str] | None = None,
         policy: ShellPolicy | None = None,
         timeout: float | None = 30.0,
-        max_output_bytes: int = 64 * 1024,
+        max_stdout_bytes: int = 64 * 1024,
+        max_stderr_bytes: int = 64 * 1024,
         approval_mode: Literal["always_require", "never_require"] = "always_require",
         on_command: Callable[[str], None] | None = None,
         docker_binary: str = "docker",
         shell: str = "bash",
     ) -> None:
+        _validate_output_limit(max_stdout_bytes, name="max_stdout_bytes")
+        _validate_output_limit(max_stderr_bytes, name="max_stderr_bytes")
         if mode not in ("persistent", "stateless"):
             raise ValueError(f"mode must be 'persistent' or 'stateless', got {mode!r}")
         unsafe_extra_run_args_snapshot = tuple(unsafe_extra_run_args or ())
@@ -321,7 +331,8 @@ def __init__(
         self._env = dict(env or {})
         self._policy = policy or ShellPolicy()
         self._timeout = timeout
-        self._max_output_bytes = max_output_bytes
+        self._max_stdout_bytes = max_stdout_bytes
+        self._max_stderr_bytes = max_stderr_bytes
         self._approval_mode: Literal["always_require", "never_require"] = approval_mode
         self._on_command = on_command
         self._binary = docker_binary
@@ -361,7 +372,8 @@ async def start(self) -> None:
                 argv,
                 workdir=None,  # workdir is set on the container itself
                 env=None,
-                max_output_bytes=self._max_output_bytes,
+                max_stdout_bytes=self._max_stdout_bytes,
+                max_stderr_bytes=self._max_stderr_bytes,
             )
             await self._session.start()
 
@@ -468,88 +480,91 @@ async def _run_stateless(self, command: str, *, timeout: float | None) -> ShellR
             stderr=asyncio.subprocess.PIPE,
         )
         timed_out = False
+        output = _ProcessOutput(proc, max_stdout_bytes=self._max_stdout_bytes, max_stderr_bytes=self._max_stderr_bytes)
+        try:
+            try:
+                await output.wait(timeout)
+            except asyncio.TimeoutError:
+                timed_out = True
+                await self._terminate_stateless(proc, per_call_name)
+                await output.finish()
+            except BaseException:
+                await self._terminate_stateless(proc, per_call_name)
+                raise
+        finally:
+            await output.close()
+
+        duration_ms = int((time.monotonic() - started) * 1000)
+        stdout_str, stdout_truncated = output.stdout.format()
+        stderr_str, stderr_truncated = output.stderr.format()
+        return ShellResult(
+            stdout=stdout_str,
+            stderr=stderr_str,
+            exit_code=proc.returncode if proc.returncode is not None else -1,
+            duration_ms=duration_ms,
+            truncated=stdout_truncated or stderr_truncated,
+            timed_out=timed_out,
+        )
+
+    async def _terminate_stateless(self, proc: asyncio.subprocess.Process, name: str) -> None:
+        # Stopping the local client alone does not stop its container.
         try:
-            stdout_bytes, stderr_bytes = await asyncio.wait_for(proc.communicate(), timeout=timeout)
-        except asyncio.TimeoutError:
-            timed_out = True
-            # Kill the container by name. ``--rm`` reaps it on a clean
-            # exit; if ``docker kill`` itself hangs or returns non-zero
-            # (daemon stuck, container with unkillable kernel task) the
-            # ``--rm`` reaper never fires and the container is leaked,
-            # so we explicitly fall back to ``docker rm -f``.
             killer = await asyncio.create_subprocess_exec(
                 self._binary,
                 "kill",
                 "--signal",
                 "KILL",
-                per_call_name,
+                name,
                 stdout=asyncio.subprocess.DEVNULL,
                 stderr=asyncio.subprocess.DEVNULL,
             )
             kill_ok = False
             try:
-                rc = await asyncio.wait_for(killer.wait(), timeout=5.0)
-                kill_ok = rc == 0
-            except asyncio.TimeoutError:
-                killer.kill()
-                with contextlib.suppress(Exception):
-                    await killer.wait()
+                try:
+                    kill_ok = await asyncio.wait_for(killer.wait(), timeout=5.0) == 0
+                except asyncio.TimeoutError:
+                    killer.kill()
+                    try:
+                        await asyncio.wait_for(killer.wait(), timeout=5.0)
+                    except asyncio.TimeoutError:
+                        logger.warning("docker kill client for %s did not exit after termination", name)
+            finally:
+                if killer.returncode is None:
+                    killer.kill()
             if not kill_ok:
-                logger.warning(
-                    "docker kill of stateless container %s did not succeed; "
-                    "attempting docker rm -f to avoid a container leak",
-                    per_call_name,
-                )
+                logger.warning("docker kill of %s did not succeed; attempting docker rm -f", name)
                 reaper = await asyncio.create_subprocess_exec(
                     self._binary,
                     "rm",
                     "-f",
-                    per_call_name,
+                    name,
                     stdout=asyncio.subprocess.DEVNULL,
                     stderr=asyncio.subprocess.PIPE,
                 )
+                diagnostic = _ProcessOutput(reaper, max_stdout_bytes=4096, max_stderr_bytes=4096)
                 try:
-                    _, rerr = await asyncio.wait_for(reaper.communicate(), timeout=5.0)
-                    if reaper.returncode != 0:
-                        logger.error(
-                            "docker rm -f %s failed (rc=%s, err=%s); container may be leaked",
-                            per_call_name,
-                            reaper.returncode,
-                            rerr.decode("utf-8", errors="replace").strip(),
-                        )
-                except asyncio.TimeoutError:
-                    reaper.kill()
-                    with contextlib.suppress(Exception):
-                        await reaper.wait()
-                    logger.error(
-                        "docker rm -f %s timed out; container may be leaked",
-                        per_call_name,
-                    )
-            try:
-                stdout_bytes, stderr_bytes = await proc.communicate()
-            except Exception:
-                # Pipe/socket failures after kill leave us with no captured
-                # output. Log so operators can correlate empty results with
-                # the timeout teardown (otherwise the model just sees
-                # exit_code=-1 with no stdout/stderr).
-                logger.warning(
-                    "failed to drain stdout/stderr after killing stateless container %s",
-                    per_call_name,
-                    exc_info=True,
-                )
-                stdout_bytes, stderr_bytes = b"", b""
-
-        duration_ms = int((time.monotonic() - started) * 1000)
-        stdout_str, stdout_truncated = _truncate_bytes(stdout_bytes or b"", self._max_output_bytes)
-        stderr_str, stderr_truncated = _truncate_bytes(stderr_bytes or b"", self._max_output_bytes)
-        return ShellResult(
-            stdout=stdout_str,
-            stderr=stderr_str,
-            exit_code=proc.returncode if proc.returncode is not None else -1,
-            duration_ms=duration_ms,
-            truncated=stdout_truncated or stderr_truncated,
-            timed_out=timed_out,
-        )
+                    try:
+                        await diagnostic.wait(5.0)
+                    except asyncio.TimeoutError:
+                        reaper.kill()
+                        await diagnostic.finish()
+                        logger.error("docker rm -f %s timed out; container may be leaked", name)
+                    else:
+                        if reaper.returncode != 0:
+                            error, _ = diagnostic.stderr.format()
+                            logger.error(
+                                "docker rm -f %s failed (rc=%s, err=%s); container may be leaked",
+                                name,
+                                reaper.returncode,
+                                error.strip(),
+                            )
+                finally:
+                    if reaper.returncode is None:
+                        reaper.kill()
+                    await diagnostic.close()
+        finally:
+            if proc.returncode is None:
+                proc.kill()
 
     # ------------------------------------------------------------------ container ops
 
diff --git a/python/packages/tools/agent_framework_tools/shell/_executor.py b/python/packages/tools/agent_framework_tools/shell/_executor.py
index e12ab7da317..30f9173fcbf 100644
--- a/python/packages/tools/agent_framework_tools/shell/_executor.py
+++ b/python/packages/tools/agent_framework_tools/shell/_executor.py
@@ -17,9 +17,10 @@
 import time
 from collections.abc import Mapping, Sequence
 
+from ._capture import _ProcessOutput  # pyright: ignore[reportPrivateUsage] - shared shell capture
 from ._killtree import kill_process_tree
 from ._resolve import is_powershell
-from ._truncate import truncate_head_tail as _truncate
+from ._truncate import _validate_output_limit  # pyright: ignore[reportPrivateUsage] - shared limit validation
 from ._types import ShellResult
 
 
@@ -38,7 +39,8 @@ async def run_stateless(
     workdir: str | None,
     env: Mapping[str, str] | None,
     timeout: float | None,
-    max_output_bytes: int,
+    max_stdout_bytes: int,
+    max_stderr_bytes: int,
 ) -> ShellResult:
     """Execute ``command`` via ``argv`` + ``-c``/``-Command`` + command.
 
@@ -50,8 +52,11 @@ async def run_stateless(
         env: Environment variables, or ``None`` to inherit the current
             process environment.
         timeout: Seconds before the process tree is killed; ``None`` disables.
-        max_output_bytes: Combined byte cap per stream before truncation.
+        max_stdout_bytes: Positive retained byte allowance for normal output.
+        max_stderr_bytes: Positive retained byte allowance for error output.
     """
+    _validate_output_limit(max_stdout_bytes, name="max_stdout_bytes")
+    _validate_output_limit(max_stderr_bytes, name="max_stderr_bytes")
     # For PowerShell we prepend a UTF-8 encoding preamble so powershell.exe
     # on Windows (cp1252 by default) doesn't mojibake non-ASCII output.
     if is_powershell(argv):
@@ -69,20 +74,23 @@ async def run_stateless(
     )
 
     timed_out = False
+    output = _ProcessOutput(proc, max_stdout_bytes=max_stdout_bytes, max_stderr_bytes=max_stderr_bytes)
     try:
-        stdout_bytes, stderr_bytes = await asyncio.wait_for(proc.communicate(), timeout=timeout)
-    except asyncio.TimeoutError:
-        timed_out = True
-        await kill_process_tree(proc)
-        # Drain any queued output.
         try:
-            stdout_bytes, stderr_bytes = await proc.communicate()
-        except Exception:
-            stdout_bytes, stderr_bytes = b"", b""
+            await output.wait(timeout)
+        except asyncio.TimeoutError:
+            timed_out = True
+            await kill_process_tree(proc)
+            await output.finish()
+        except BaseException:
+            await kill_process_tree(proc)
+            raise
+    finally:
+        await output.close()
 
     duration_ms = int((time.monotonic() - started) * 1000)
-    stdout_str, stdout_truncated = _truncate(stdout_bytes or b"", max_output_bytes)
-    stderr_str, stderr_truncated = _truncate(stderr_bytes or b"", max_output_bytes)
+    stdout_str, stdout_truncated = output.stdout.format()
+    stderr_str, stderr_truncated = output.stderr.format()
 
     return ShellResult(
         stdout=stdout_str,
diff --git a/python/packages/tools/agent_framework_tools/shell/_session.py b/python/packages/tools/agent_framework_tools/shell/_session.py
index f83f1c8e98e..0c295be15fb 100644
--- a/python/packages/tools/agent_framework_tools/shell/_session.py
+++ b/python/packages/tools/agent_framework_tools/shell/_session.py
@@ -30,8 +30,8 @@
 * ``$LASTEXITCODE`` only tracks external-process exits, so the rc is
   also derived from ``$?`` and caught exceptions.
 * stdout and stderr are consumed by **persistent reader tasks** that
-  run for the lifetime of the session. Each ``run()`` snapshots buffer
-  offsets before writing the command and scans forward from there.
+  run for the lifetime of the session. Each ``run()`` installs bounded
+  output capture and a separate incremental completion-marker parser.
   This avoids ``read() called while another coroutine is already
   waiting`` errors from per-call ``wait_for(stream.read())`` loops and
   prevents late stderr from being attributed to the next command.
@@ -48,18 +48,70 @@
 import time
 from collections.abc import Mapping, Sequence
 
+from ._capture import READ_CHUNK
 from ._killtree import kill_process_tree
 from ._resolve import is_powershell
-from ._truncate import truncate_head_tail as _truncate_bytes
-from ._truncate import truncate_text_head_tail as _truncate_text
+from ._truncate import (
+    _TextOutputBuffer,  # pyright: ignore[reportPrivateUsage] - shared shell capture
+    _validate_output_limit,  # pyright: ignore[reportPrivateUsage] - shared limit validation
+)
 from ._types import ShellResult
 
-_READ_CHUNK = 64 * 1024
 _SHUTDOWN_GRACE = 2.0
 # Extra grace window after the sentinel arrives to let late stderr drain.
 _STDERR_QUIESCENCE = 0.05
 
 
+class _CommandOutput:
+    """Separate bounded command output from completion-marker recognition."""
+
+    def __init__(self, sentinel: bytes, *, max_stdout_bytes: int, max_stderr_bytes: int) -> None:
+        self.stdout = _TextOutputBuffer(max_stdout_bytes, strip_newlines=True)
+        self.stderr = _TextOutputBuffer(max_stderr_bytes)
+        self.stdout_bytes = 0
+        self.exit_code: int | None = None
+        self.marker_seen_at: float | None = None
+        self._sentinel = sentinel
+        self._pending = bytearray()
+
+    def read_stdout(self, chunk: bytes) -> None:
+        if self.exit_code is not None:
+            return
+        self._pending.extend(chunk)
+        if self.marker_seen_at is None:
+            idx = self._pending.find(self._sentinel)
+            if idx == -1:
+                keep = min(len(self._pending), len(self._sentinel) - 1)
+                while keep and not self._pending.endswith(self._sentinel[:keep]):
+                    keep -= 1
+                emit = len(self._pending) - keep
+                self.stdout_bytes += emit
+                self.stdout.append(bytes(self._pending[:emit]))
+                del self._pending[:emit]
+                return
+            self.stdout_bytes += idx
+            self.stdout.append(bytes(self._pending[:idx]), final=True)
+            del self._pending[: idx + len(self._sentinel)]
+            self.marker_seen_at = time.monotonic()
+        newline = self._pending.find(b"\n")
+        # A shell exit status fits comfortably in 32 bytes. Never retain an
+        # arbitrary suffix while waiting for the marker's newline.
+        if (newline if newline != -1 else len(self._pending)) > 32:
+            raise RuntimeError("shell completion marker has an oversized exit status")
+        if newline != -1:
+            self.exit_code = _parse_rc(bytes(self._pending[:newline]))
+            self._pending.clear()
+
+    def finish_stdout(self) -> None:
+        if self.exit_code is not None:
+            return
+        if self.marker_seen_at is not None:
+            self.exit_code = _parse_rc(bytes(self._pending))
+        else:
+            self.stdout.append(bytes(self._pending), final=True)
+        self._pending.clear()
+
+
 class ShellSession:
     """A long-lived shell subprocess that executes commands via sentinels.
 
@@ -73,12 +125,16 @@ def __init__(
         *,
         workdir: str | None = None,
         env: Mapping[str, str] | None = None,
-        max_output_bytes: int = 64 * 1024,
+        max_stdout_bytes: int = 64 * 1024,
+        max_stderr_bytes: int = 64 * 1024,
     ) -> None:
+        _validate_output_limit(max_stdout_bytes, name="max_stdout_bytes")
+        _validate_output_limit(max_stderr_bytes, name="max_stderr_bytes")
         self._argv = list(argv)
         self._workdir = workdir
         self._env = dict(env) if env is not None else None
-        self._max_output_bytes = max_output_bytes
+        self._max_stdout_bytes = max_stdout_bytes
+        self._max_stderr_bytes = max_stderr_bytes
         self._proc: asyncio.subprocess.Process | None = None
         # Serialises per-command execution onto the single stdin/stdout
         # pipe. This is an ordering primitive within one owning session;
@@ -94,10 +150,8 @@ def __init__(
         self._sentinel_tag = secrets.token_hex(8)
         self._is_pwsh = is_powershell(argv)
 
-        # Persistent reader state. The reader tasks append into these
-        # buffers; _run_locked scans forward from a per-call offset.
-        self._stdout_buf = bytearray()
-        self._stderr_buf = bytearray()
+        self._output: _CommandOutput | None = None
+        self._reader_error: Exception | None = None
         self._stdout_event = asyncio.Event()
         self._stdout_reader: asyncio.Task[None] | None = None
         self._stderr_reader: asyncio.Task[None] | None = None
@@ -131,18 +185,15 @@ async def start(self) -> None:
                 **popen_kwargs,  # type: ignore[arg-type]
             )
 
-            # Reset buffer state in case this is a restart after close().
-            self._stdout_buf.clear()
-            self._stderr_buf.clear()
+            self._output = None
+            self._reader_error = None
             self._stdout_event = asyncio.Event()
             self._stdout_closed = False
 
             if self._proc.stdout is None or self._proc.stderr is None:
                 raise RuntimeError("subprocess pipes were not created; stdout/stderr unavailable")
-            self._stdout_reader = asyncio.create_task(self._reader(self._proc.stdout, self._stdout_buf, is_stdout=True))
-            self._stderr_reader = asyncio.create_task(
-                self._reader(self._proc.stderr, self._stderr_buf, is_stdout=False)
-            )
+            self._stdout_reader = asyncio.create_task(self._reader(self._proc.stdout, is_stdout=True))
+            self._stderr_reader = asyncio.create_task(self._reader(self._proc.stderr, is_stdout=False))
 
             # Best-effort: make PowerShell emit UTF-8 and fail loudly on errors.
             if self._is_pwsh:
@@ -229,137 +280,85 @@ async def _run_locked(self, command: str, *, timeout: float | None) -> ShellResu
 
         sentinel = f"__AF_END_{self._sentinel_tag}_{secrets.token_hex(4)}__"
         script = self._build_script(command, sentinel)
-        # Snapshot current buffer positions so we only attribute output
-        # produced *after* the command is submitted.
-        stdout_offset = len(self._stdout_buf)
-        stderr_offset = len(self._stderr_buf)
+        output = _CommandOutput(
+            sentinel.encode("utf-8"),
+            max_stdout_bytes=self._max_stdout_bytes,
+            max_stderr_bytes=self._max_stderr_bytes,
+        )
+        self._output = output
         self._stdout_event.clear()
 
         started = time.monotonic()
-        try:
-            self._proc.stdin.write(script.encode("utf-8"))
-            await self._proc.stdin.drain()
-        except (ConnectionResetError, BrokenPipeError) as exc:
-            raise RuntimeError("persistent shell session is no longer alive") from exc
-
-        needle = sentinel.encode("utf-8")
         timed_out = False
-        hard_cap = self._max_output_bytes * 4
+        overflow = False
+        hard_cap = self._max_stdout_bytes * 4
 
-        async def _wait_for_sentinel() -> tuple[int, int]:
-            """Return (sentinel_start_index, exit_code) once seen."""
+        async def _wait_for_sentinel() -> int:
             while True:
-                idx = self._stdout_buf.find(needle, stdout_offset)
-                if idx != -1:
-                    # Parse trailing ``_<digits>``.
-                    tail_start = idx + len(needle)
-                    # Wait briefly for the rc digits + newline to arrive.
-                    deadline = time.monotonic() + 1.0
-                    while time.monotonic() < deadline:
-                        after = bytes(self._stdout_buf[tail_start:])
-                        nl = after.find(b"\n")
-                        if nl != -1:
-                            break
-                        if self._stdout_closed:
-                            break
-                        self._stdout_event.clear()
-                        try:
-                            await asyncio.wait_for(self._stdout_event.wait(), timeout=0.1)
-                        except asyncio.TimeoutError:
-                            pass
-                    after = bytes(self._stdout_buf[tail_start:])
-                    rc = _parse_rc(after)
-                    return idx, rc
-                if self._stdout_closed:
-                    raise RuntimeError("shell closed stdout before emitting sentinel")
-                if len(self._stdout_buf) - stdout_offset > hard_cap:
-                    raise _SentinelOverflow
+                if self._reader_error is not None:
+                    raise RuntimeError("failed to read persistent shell output") from self._reader_error
+                if output.exit_code is not None:
+                    return output.exit_code
+                if output.marker_seen_at is not None:
+                    if self._stdout_closed or time.monotonic() - output.marker_seen_at >= 1.0:
+                        output.finish_stdout()
+                        return output.exit_code if output.exit_code is not None else -1
+                else:
+                    if self._stdout_closed:
+                        raise RuntimeError("shell closed stdout before emitting sentinel")
+                    if output.stdout_bytes > hard_cap:
+                        raise _SentinelOverflow
                 self._stdout_event.clear()
                 try:
-                    await asyncio.wait_for(self._stdout_event.wait(), timeout=0.5)
+                    await asyncio.wait_for(self._stdout_event.wait(), timeout=0.1)
                 except asyncio.TimeoutError:
-                    # Keep spinning; timeout is enforced at the wait_for below.
                     pass
 
-        sentinel_idx: int
         exit_code: int
         try:
-            sentinel_idx, exit_code = await asyncio.wait_for(_wait_for_sentinel(), timeout=timeout)
-        except asyncio.TimeoutError:
-            timed_out = True
-            await self._interrupt_current_command()
             try:
-                sentinel_idx, exit_code = await asyncio.wait_for(_wait_for_sentinel(), timeout=_SHUTDOWN_GRACE)
-            except (asyncio.TimeoutError, RuntimeError, _SentinelOverflow):
-                # Session is unrecoverable; tear it down so the next call
-                # gets a fresh subprocess.
-                await self._close_process()
-                duration_ms = int((time.monotonic() - started) * 1000)
-                stdout_bytes = bytes(self._stdout_buf[stdout_offset:])
-                stderr_bytes = bytes(self._stderr_buf[stderr_offset:])
-                stdout_str, so_trunc = _truncate_bytes(stdout_bytes, self._max_output_bytes)
-                stderr_str, se_trunc = _truncate_bytes(stderr_bytes, self._max_output_bytes)
-                return ShellResult(
-                    stdout=stdout_str,
-                    stderr=stderr_str,
-                    exit_code=124,
-                    duration_ms=duration_ms,
-                    truncated=so_trunc or se_trunc,
-                    timed_out=True,
-                )
-            if sys.platform == "win32":
-                # CTRL_BREAK can flush PowerShell's finally sentinel before
-                # the process finishes exiting. Do not dispatch the next
-                # command into that interrupted session, even with a sentinel.
+                self._proc.stdin.write(script.encode("utf-8"))
+                await self._proc.stdin.drain()
+            except (ConnectionResetError, BrokenPipeError) as exc:
+                raise RuntimeError("persistent shell session is no longer alive") from exc
+            try:
+                exit_code = await asyncio.wait_for(_wait_for_sentinel(), timeout=timeout)
+            except asyncio.TimeoutError:
+                timed_out = True
+                await self._interrupt_current_command()
+                try:
+                    exit_code = await asyncio.wait_for(_wait_for_sentinel(), timeout=_SHUTDOWN_GRACE)
+                except (asyncio.TimeoutError, RuntimeError, _SentinelOverflow):
+                    await self._close_process()
+                    exit_code = 124
+                if sys.platform == "win32":
+                    await self._close_process()
+            except _SentinelOverflow:
+                overflow = True
+                await self._interrupt_current_command()
                 await self._close_process()
-        except _SentinelOverflow:
-            # Runaway output; recover by interrupting and restarting.
-            await self._interrupt_current_command()
-            await self._close_process()
-            duration_ms = int((time.monotonic() - started) * 1000)
-            stdout_bytes = bytes(self._stdout_buf[stdout_offset : stdout_offset + hard_cap])
-            stderr_bytes = bytes(self._stderr_buf[stderr_offset:])
-            stdout_str, _ = _truncate_bytes(stdout_bytes, self._max_output_bytes)
-            stderr_str, _ = _truncate_bytes(stderr_bytes, self._max_output_bytes)
+                exit_code = -1
+
+            await asyncio.sleep(_STDERR_QUIESCENCE)
+            if self._reader_error is not None:
+                raise RuntimeError("failed to read persistent shell output") from self._reader_error
+            output.finish_stdout()
+            output.stderr.append(b"", final=True)
+            stdout_str, stdout_truncated = output.stdout.format()
+            stderr_str, stderr_truncated = output.stderr.format()
             return ShellResult(
                 stdout=stdout_str,
                 stderr=stderr_str,
-                exit_code=-1,
-                duration_ms=duration_ms,
-                truncated=True,
-                timed_out=False,
+                exit_code=exit_code,
+                duration_ms=int((time.monotonic() - started) * 1000),
+                truncated=overflow or stdout_truncated or stderr_truncated,
+                timed_out=timed_out,
             )
-
-        # Let stderr quiesce briefly — late writes from the completing
-        # command otherwise leak into the *next* run().
-        await asyncio.sleep(_STDERR_QUIESCENCE)
-
-        duration_ms = int((time.monotonic() - started) * 1000)
-        stdout_raw = bytes(self._stdout_buf[stdout_offset:sentinel_idx])
-        stderr_raw = bytes(self._stderr_buf[stderr_offset:])
-
-        stdout_text = stdout_raw.decode("utf-8", errors="replace").rstrip("\r\n")
-        stderr_text = stderr_raw.decode("utf-8", errors="replace")
-
-        stdout_str, stdout_truncated = _truncate_text(stdout_text, self._max_output_bytes)
-        stderr_str, stderr_truncated = _truncate_text(stderr_text, self._max_output_bytes)
-
-        # Trim the persistent buffers: everything we needed has been copied
-        # into stdout_raw/stderr_raw above, so discarding now keeps the
-        # session's memory bounded across many commands. The reader tasks
-        # only ever ``extend()`` these buffers (no offset bookkeeping
-        # outside this method), so resetting them here is safe.
-        del self._stdout_buf[:]
-        del self._stderr_buf[:]
-
-        return ShellResult(
-            stdout=stdout_str,
-            stderr=stderr_str,
-            exit_code=exit_code,
-            duration_ms=duration_ms,
-            truncated=stdout_truncated or stderr_truncated,
-            timed_out=timed_out,
-        )
+        except BaseException:
+            await self._close_process()
+            raise
+        finally:
+            self._output = None
 
     # ------------------------------------------------------------------ helpers
 
@@ -424,28 +423,30 @@ async def _write_raw(self, text: str) -> None:
     async def _reader(
         self,
         stream: asyncio.StreamReader,
-        buf: bytearray,
         *,
         is_stdout: bool,
     ) -> None:
-        """Persistent reader task: drains ``stream`` into ``buf`` until EOF."""
+        """Drain continuously, discarding output when no command owns it."""
         try:
             while True:
-                chunk = await stream.read(_READ_CHUNK)
+                chunk = await stream.read(READ_CHUNK)
                 if not chunk:
                     if is_stdout:
                         self._stdout_closed = True
                         self._stdout_event.set()
                     return
-                buf.extend(chunk)
+                if self._output is not None:
+                    if is_stdout:
+                        self._output.read_stdout(chunk)
+                    else:
+                        self._output.stderr.append(chunk)
                 if is_stdout:
                     self._stdout_event.set()
         except asyncio.CancelledError:
             raise
-        except Exception:
-            if is_stdout:
-                self._stdout_closed = True
-                self._stdout_event.set()
+        except Exception as exc:
+            self._reader_error = exc
+            self._stdout_event.set()
 
     async def _interrupt_current_command(self) -> None:
         if self._proc is None or self._proc.returncode is not None:
diff --git a/python/packages/tools/agent_framework_tools/shell/_tool.py b/python/packages/tools/agent_framework_tools/shell/_tool.py
index d85beeebd03..a2ab434187d 100644
--- a/python/packages/tools/agent_framework_tools/shell/_tool.py
+++ b/python/packages/tools/agent_framework_tools/shell/_tool.py
@@ -19,6 +19,7 @@
 from ._policy import ShellPolicy, ShellRequest
 from ._resolve import is_powershell, resolve_shell
 from ._session import ShellSession
+from ._truncate import _validate_output_limit  # pyright: ignore[reportPrivateUsage] - shared limit validation
 from ._types import (
     ShellCommandError,
     ShellMode,
@@ -129,8 +130,12 @@ class LocalShellTool:
             gating + sandbox tier are the real defenses.
         timeout: Per-command timeout in seconds. ``None`` disables. Default
             30 s.
-        max_output_bytes: Combined stdout/stderr byte cap before truncation.
-            Default 64 KiB.
+        max_stdout_bytes: Positive retained byte allowance for normal output.
+            Default 64 KiB, plus bounded capture overhead.
+            Truncation notices and UTF-8 replacement may enlarge returned text.
+        max_stderr_bytes: Positive retained byte allowance for error output.
+            Default 64 KiB, independent of ``max_stdout_bytes``.
+            Truncation notices and UTF-8 replacement may enlarge returned text.
         approval_mode: ``"always_require"`` (default) or ``"never_require"``.
             Controls the ``FunctionTool.approval_mode`` on the returned
             function, which the framework uses to gate execution via
@@ -156,11 +161,14 @@ def __init__(
         clean_env: bool = False,
         policy: ShellPolicy | None = None,
         timeout: float | None = 30.0,
-        max_output_bytes: int = 64 * 1024,
+        max_stdout_bytes: int = 64 * 1024,
+        max_stderr_bytes: int = 64 * 1024,
         approval_mode: Literal["always_require", "never_require"] = "always_require",
         acknowledge_unsafe: bool = False,
         on_command: Callable[[str], None] | None = None,
     ) -> None:
+        _validate_output_limit(max_stdout_bytes, name="max_stdout_bytes")
+        _validate_output_limit(max_stderr_bytes, name="max_stderr_bytes")
         if mode not in ("persistent", "stateless"):
             raise ValueError(f"mode must be 'persistent' or 'stateless', got {mode!r}")
         if approval_mode == "never_require" and not acknowledge_unsafe:
@@ -178,7 +186,8 @@ def __init__(
         self._confine_workdir = confine_workdir
         self._policy = policy or ShellPolicy()
         self._timeout = timeout
-        self._max_output_bytes = max_output_bytes
+        self._max_stdout_bytes = max_stdout_bytes
+        self._max_stderr_bytes = max_stderr_bytes
         self._approval_mode: Literal["always_require", "never_require"] = approval_mode
         self._on_command = on_command
 
@@ -214,7 +223,8 @@ async def start(self) -> None:
                     self._interactive_argv,
                     workdir=self._workdir,
                     env=self._env,
-                    max_output_bytes=self._max_output_bytes,
+                    max_stdout_bytes=self._max_stdout_bytes,
+                    max_stderr_bytes=self._max_stderr_bytes,
                 )
             await self._session.start()
 
@@ -280,7 +290,8 @@ async def run(self, command: str, *, timeout: float | None = None) -> ShellResul
             workdir=self._workdir,
             env=self._env,
             timeout=effective_timeout,
-            max_output_bytes=self._max_output_bytes,
+            max_stdout_bytes=self._max_stdout_bytes,
+            max_stderr_bytes=self._max_stderr_bytes,
         )
 
     # ------------------------------------------------------------------ AF wiring
diff --git a/python/packages/tools/agent_framework_tools/shell/_truncate.py b/python/packages/tools/agent_framework_tools/shell/_truncate.py
index ac48fe88207..6366114e490 100644
--- a/python/packages/tools/agent_framework_tools/shell/_truncate.py
+++ b/python/packages/tools/agent_framework_tools/shell/_truncate.py
@@ -4,6 +4,75 @@
 
 from __future__ import annotations
 
+import codecs
+
+
+def _validate_output_limit(cap: int, *, name: str = "cap") -> None:
+    if cap <= 0:
+        raise ValueError(f"{name} must be positive; got {cap}")
+
+
+class _OutputBuffer:
+    """Keep only the beginning and end of a stream, counting omitted bytes."""
+
+    def __init__(self, cap: int) -> None:
+        _validate_output_limit(cap)
+        self._cap = cap
+        self._head = bytearray()
+        self._tail = bytearray()
+        self.total_bytes = 0
+
+    def append(self, data: bytes) -> None:
+        self.total_bytes += len(data)
+        head_remaining = self._cap // 2 - len(self._head)
+        if head_remaining:
+            self._head.extend(data[:head_remaining])
+            data = data[head_remaining:]
+        tail_cap = self._cap - self._cap // 2
+        self._tail.extend(data[-tail_cap:])
+        if len(self._tail) > tail_cap:
+            del self._tail[:-tail_cap]
+
+    def append_buffer(self, other: _OutputBuffer) -> None:
+        """Append another equally sized capture without reconstructing its gaps."""
+        self.append(bytes(other._head))
+        self.total_bytes += other.total_bytes - len(other._head) - len(other._tail)
+        self.append(bytes(other._tail))
+
+    def format(self) -> tuple[str, bool]:
+        if self.total_bytes <= self._cap:
+            return (self._head + self._tail).decode("utf-8", errors="replace"), False
+        head = self._head.decode("utf-8", errors="replace")
+        tail = self._tail.decode("utf-8", errors="replace")
+        dropped = self.total_bytes - self._cap
+        return f"{head}\n[... truncated {dropped} bytes ...]\n{tail}", True
+
+
+class _TextOutputBuffer:  # pyright: ignore[reportUnusedClass] - used by persistent sessions
+    """Incrementally preserve persistent-session decoding and newline stripping."""
+
+    def __init__(self, cap: int, *, strip_newlines: bool = False) -> None:
+        self._cap = cap
+        self._output = _OutputBuffer(cap)
+        self._pending_newlines = _OutputBuffer(cap)
+        self._decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
+        self._strip_newlines = strip_newlines
+
+    def append(self, data: bytes, *, final: bool = False) -> None:
+        text = self._decoder.decode(data, final=final)
+        if not self._strip_newlines:
+            self._output.append(text.encode("utf-8"))
+            return
+        stripped = text.rstrip("\r\n")
+        if stripped:
+            self._output.append_buffer(self._pending_newlines)
+            self._pending_newlines = _OutputBuffer(self._cap)
+            self._output.append(stripped.encode("utf-8"))
+        self._pending_newlines.append(text[len(stripped) :].encode("utf-8"))
+
+    def format(self) -> tuple[str, bool]:
+        return self._output.format()
+
 
 def truncate_head_tail(data: bytes, cap: int) -> tuple[str, bool]:
     """Truncate ``data`` to ``cap`` bytes, keeping a head and a tail slice.
@@ -19,16 +88,9 @@ def truncate_head_tail(data: bytes, cap: int) -> tuple[str, bool]:
     cap has no consistent meaning here and silently treating it as
     "unlimited" would defeat output-capping assumptions in callers.
     """
-    if cap <= 0:
-        raise ValueError(f"cap must be positive; got {cap}")
-    if len(data) <= cap:
-        return data.decode("utf-8", errors="replace"), False
-    head_cap = cap // 2
-    tail_cap = cap - head_cap
-    head = data[:head_cap].decode("utf-8", errors="replace")
-    tail = data[len(data) - tail_cap :].decode("utf-8", errors="replace")
-    dropped = len(data) - cap
-    return f"{head}\n[... truncated {dropped} bytes ...]\n{tail}", True
+    output = _OutputBuffer(cap)
+    output.append(data)
+    return output.format()
 
 
 def truncate_text_head_tail(text: str, cap: int) -> tuple[str, bool]:
diff --git a/python/packages/tools/agent_framework_tools/shell/_types.py b/python/packages/tools/agent_framework_tools/shell/_types.py
index 4fbc7dfe5f4..1c2e5a3d68a 100644
--- a/python/packages/tools/agent_framework_tools/shell/_types.py
+++ b/python/packages/tools/agent_framework_tools/shell/_types.py
@@ -21,8 +21,8 @@ class ShellResult:
         stderr: Captured standard error, possibly truncated.
         exit_code: The exit status reported by the shell or subprocess.
         duration_ms: How long the command took, in milliseconds.
-        truncated: ``True`` when stdout or stderr was truncated to fit
-            ``max_output_bytes``.
+        truncated: ``True`` when stdout or stderr was truncated to fit its
+            respective ``max_stdout_bytes`` or ``max_stderr_bytes`` allowance.
         timed_out: ``True`` when the command was killed because it exceeded
             the configured timeout.
     """
diff --git a/python/packages/tools/tests/test_docker_shell_tool.py b/python/packages/tools/tests/test_docker_shell_tool.py
index 58a145a8894..159c8687dcf 100644
--- a/python/packages/tools/tests/test_docker_shell_tool.py
+++ b/python/packages/tools/tests/test_docker_shell_tool.py
@@ -46,11 +46,19 @@ def __init__(
     ) -> None:
         self.pid = pid
         self.returncode = returncode
-        self.stdout = object()
-        self.stderr = object()
+        self.stdout = asyncio.StreamReader()
+        self.stderr = asyncio.StreamReader()
         self.killed = False
         self._communicate_results = list(communicate_results or [(b"", b"")])
-        self._wait_results = list(wait_results or [returncode])
+        first = self._communicate_results[0]
+        self._wait_results = list(
+            wait_results or ([first, returncode] if isinstance(first, BaseException) else [returncode])
+        )
+        output = next((item for item in self._communicate_results if isinstance(item, tuple)), (b"", b""))
+        self.stdout.feed_data(output[0])
+        self.stderr.feed_data(output[1])
+        self.stdout.feed_eof()
+        self.stderr.feed_eof()
 
     async def communicate(self) -> tuple[bytes, bytes]:
         result = self._communicate_results.pop(0)
@@ -60,7 +68,7 @@ async def communicate(self) -> tuple[bytes, bytes]:
         return stdout, stderr
 
     async def wait(self) -> int | None:
-        result = self._wait_results.pop(0)
+        result = self._wait_results.pop(0) if self._wait_results else self.returncode
         if isinstance(result, BaseException):
             raise result
         self.returncode = result
@@ -309,7 +317,7 @@ async def test_start_and_close_are_noops_in_stateless_mode() -> None:
 
 
 async def test_start_creates_and_reuses_persistent_session() -> None:
-    tool = DockerShellTool(docker_binary="podman", shell="sh")
+    tool = DockerShellTool(docker_binary="podman", shell="sh", max_stdout_bytes=1024, max_stderr_bytes=2048)
     session = AsyncMock()
 
     with (
@@ -324,7 +332,8 @@ async def test_start_creates_and_reuses_persistent_session() -> None:
         ["podman", "exec", "-i", tool._container_name, "sh"],
         workdir=None,
         env=None,
-        max_output_bytes=tool._max_output_bytes,
+        max_stdout_bytes=1024,
+        max_stderr_bytes=2048,
     )
     assert session.start.await_count == 2
 
@@ -438,6 +447,123 @@ async def test_run_stateless_builds_expected_argv() -> None:
     assert result.timed_out is False
 
 
+@pytest.mark.parametrize(("max_stdout_bytes", "max_stderr_bytes"), [(17, 33), (33, 17)])
+async def test_docker_stateless_bounds_both_host_output_buffers(
+    max_stdout_bytes: int, max_stderr_bytes: int, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from agent_framework_tools.shell._truncate import _OutputBuffer, truncate_head_tail
+
+    append = _OutputBuffer.append
+    observed: list[int] = []
+
+    def checked_append(self: _OutputBuffer, data: bytes) -> None:
+        append(self, data)
+        retained = len(self._head) + len(self._tail)
+        assert self._cap in (max_stdout_bytes, max_stderr_bytes)
+        assert retained <= self._cap
+        observed.append(retained)
+
+    monkeypatch.setattr(_OutputBuffer, "append", checked_append)
+    tool = DockerShellTool(mode="stateless", max_stdout_bytes=max_stdout_bytes, max_stderr_bytes=max_stderr_bytes)
+    proc = _FakeProcess(communicate_results=[(b"a" * 65536, b"b" * 65536)])
+    with patch("agent_framework_tools.shell._docker.asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+        result = await tool.run("unused")
+    assert result.truncated and result.exit_code == 0
+    assert result.stdout == truncate_head_tail(b"a" * 65536, max_stdout_bytes)[0]
+    assert result.stderr == truncate_head_tail(b"b" * 65536, max_stderr_bytes)[0]
+    assert observed
+
+
+@pytest.mark.parametrize("failure", ["cancel", "reader"])
+async def test_docker_stateless_failure_stops_container_and_local_client(failure: str) -> None:
+    tool = DockerShellTool(mode="stateless")
+    proc = _FakeProcess(returncode=None)
+    killer = _FakeProcess()
+    waiting = asyncio.Event()
+
+    async def wait() -> int:
+        waiting.set()
+        await asyncio.Event().wait()
+        return 0
+
+    if failure == "reader":
+        proc.stderr.set_exception(OSError("pipe failed"))
+    with (
+        patch.object(proc, "wait", wait),
+        patch(
+            "agent_framework_tools.shell._docker.asyncio.create_subprocess_exec",
+            AsyncMock(side_effect=[proc, killer]),
+        ) as create_process,
+    ):
+        task = asyncio.create_task(tool.run("unused", timeout=5))
+        await waiting.wait()
+        if failure == "cancel":
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+        else:
+            with pytest.raises(OSError, match="pipe failed"):
+                await task
+    assert proc.killed
+    assert create_process.await_count == 2
+    assert create_process.await_args_list[1].args[1:4] == ("kill", "--signal", "KILL")
+
+
+@pytest.mark.parametrize("cap", [0, -1])
+@pytest.mark.parametrize("stream", ["stdout", "stderr"])
+def test_docker_rejects_invalid_output_limit_before_execution(cap: int, stream: str) -> None:
+    with pytest.raises(ValueError, match=f"max_{stream}_bytes must be positive"):
+        DockerShellTool(
+            max_stdout_bytes=cap if stream == "stdout" else 1024,
+            max_stderr_bytes=cap if stream == "stderr" else 1024,
+        )
+
+
+async def test_docker_cleanup_reader_failure_stops_cleanup_process() -> None:
+    tool = DockerShellTool(mode="stateless")
+    proc = _FakeProcess(returncode=None, wait_results=[asyncio.TimeoutError()])
+    killer = _FakeProcess(returncode=1)
+    reaper = _FakeProcess(returncode=None)
+    reaper.stderr.set_exception(OSError("cleanup pipe failed"))
+    with (
+        patch(
+            "agent_framework_tools.shell._docker.asyncio.create_subprocess_exec",
+            AsyncMock(side_effect=[proc, killer, reaper]),
+        ),
+        pytest.raises(OSError, match="cleanup pipe failed"),
+    ):
+        await tool.run("unused", timeout=5)
+    assert reaper.killed and proc.killed
+
+
+@pytest.mark.parametrize("cleanup_command", ["kill", "rm"])
+async def test_docker_cancellation_stops_cleanup_process(cleanup_command: str) -> None:
+    tool = DockerShellTool(mode="stateless")
+    proc = _FakeProcess(returncode=None, wait_results=[asyncio.TimeoutError()])
+    cleanup = _FakeProcess(returncode=None)
+    waiting = asyncio.Event()
+
+    async def wait() -> int:
+        waiting.set()
+        await asyncio.Event().wait()
+        return 0
+
+    processes = [proc, cleanup] if cleanup_command == "kill" else [proc, _FakeProcess(returncode=1), cleanup]
+    with (
+        patch.object(cleanup, "wait", wait),
+        patch(
+            "agent_framework_tools.shell._docker.asyncio.create_subprocess_exec",
+            AsyncMock(side_effect=processes),
+        ),
+    ):
+        task = asyncio.create_task(tool.run("unused", timeout=5))
+        await waiting.wait()
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+    assert cleanup.killed and proc.killed
+
+
 async def test_run_stateless_places_unsafe_extra_args_after_isolation_defaults() -> None:
     tool = DockerShellTool(
         mode="stateless",
diff --git a/python/packages/tools/tests/test_local_shell_tool.py b/python/packages/tools/tests/test_local_shell_tool.py
index 24ecbaaeb3b..1461ebb9d5a 100644
--- a/python/packages/tools/tests/test_local_shell_tool.py
+++ b/python/packages/tools/tests/test_local_shell_tool.py
@@ -26,6 +26,8 @@
 from agent_framework_tools._feature_usage import FeatureIndex
 from agent_framework_tools.shell import LocalShellTool, ShellCommandError, ShellPolicy, ShellResult
 from agent_framework_tools.shell._executor import _popen_kwargs_for_group, run_stateless
+from agent_framework_tools.shell._truncate import _OutputBuffer
+from agent_framework_tools.shell._types import ShellMode
 
 _TEST_SHELL = "agent-framework-test-shell"
 _APPROVED_COMMAND = "printf '%s' approved-value"
@@ -41,16 +43,17 @@ def __init__(
         communicate_results: list[tuple[bytes, bytes] | BaseException] | None = None,
     ) -> None:
         self.returncode = returncode
-        self.stdout = object()
-        self.stderr = object()
-        self._communicate_results = list(communicate_results or [(b"", b"")])
+        self.stdout = asyncio.StreamReader()
+        self.stderr = asyncio.StreamReader()
+        result = (communicate_results or [(b"", b"")])[0]
+        assert isinstance(result, tuple)
+        self.stdout.feed_data(result[0])
+        self.stderr.feed_data(result[1])
+        self.stdout.feed_eof()
+        self.stderr.feed_eof()
 
-    async def communicate(self) -> tuple[bytes, bytes]:
-        result = self._communicate_results.pop(0)
-        if isinstance(result, BaseException):
-            raise result
-        stdout, stderr = result
-        return stdout, stderr
+    async def wait(self) -> int:
+        return self.returncode if self.returncode is not None else -1
 
 
 class _ScriptedFunctionClient(FunctionInvocationLayer, BaseChatClient):
@@ -128,6 +131,184 @@ async def test_stateless_echo() -> None:
     assert result.timed_out is False
 
 
+@pytest.mark.parametrize("mode", ["persistent", "stateless"])
+@pytest.mark.parametrize("stream", ["stdout", "stderr", "both"])
+@pytest.mark.parametrize(("max_stdout_bytes", "max_stderr_bytes"), [(32768, 49152), (49152, 32768), (32768, 1024)])
+async def test_output_is_bounded_during_finite_command(
+    mode: ShellMode,
+    stream: str,
+    max_stdout_bytes: int,
+    max_stderr_bytes: int,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    caps = {"stdout": max_stdout_bytes, "stderr": max_stderr_bytes}
+    observed: list[int] = []
+    append = _OutputBuffer.append
+
+    def checked_append(self: _OutputBuffer, data: bytes) -> None:
+        append(self, data)
+        retained = len(self._head) + len(self._tail)
+        assert self._cap in caps.values()
+        assert retained <= self._cap
+        observed.append(retained)
+
+    monkeypatch.setattr(_OutputBuffer, "append", checked_append)
+    if sys.platform == "win32":
+        command = "; ".join(
+            f"[Console]::{'Out' if name == 'stdout' else 'Error'}.Write('{char}' * 65536)"
+            for name, char in (("stdout", "x"), ("stderr", "z"))
+            if stream in (name, "both")
+        )
+        command += "; [System.Threading.Thread]::Sleep(200)"
+        followup = "Write-Output healthy"
+    else:
+        script = "; ".join(
+            f"sys.{name}.write('{char}' * 65536); sys.{name}.flush()"
+            for name, char in (("stdout", "x"), ("stderr", "z"))
+            if stream in (name, "both")
+        )
+        command = f"{shlex.quote(sys.executable)} -c {shlex.quote('import sys, time; ' + script + '; time.sleep(0.2)')}"
+        followup = "printf healthy"
+    async with LocalShellTool(
+        mode=mode,
+        max_stdout_bytes=max_stdout_bytes,
+        max_stderr_bytes=max_stderr_bytes,
+        timeout=5,
+    ) as tool:
+        original_process = tool._session._proc if tool._session else None
+        result = await tool.run(command)
+        assert result.truncated and not result.timed_out
+        for name, char in (("stdout", "x"), ("stderr", "z")):
+            text = getattr(result, name)
+            if stream in (name, "both"):
+                assert text.count(char) == caps[name]
+            else:
+                assert not text
+        if mode == "persistent":
+            assert tool._session is not None
+            assert tool._session._proc is original_process
+            assert result.exit_code == 0
+        healthy = await tool.run(followup)
+        assert healthy.stdout.strip() == "healthy"
+        assert not healthy.stderr and healthy.exit_code == 0
+    assert observed
+
+
+@pytest.mark.parametrize("cap", [0, -1])
+@pytest.mark.parametrize("stream", ["stdout", "stderr"])
+def test_local_shell_rejects_invalid_output_limit_before_execution(cap: int, stream: str) -> None:
+    with pytest.raises(ValueError, match=f"max_{stream}_bytes must be positive"):
+        LocalShellTool(
+            max_stdout_bytes=cap if stream == "stdout" else 1024,
+            max_stderr_bytes=cap if stream == "stderr" else 1024,
+        )
+
+
+@pytest.mark.parametrize("cancel", [False, True])
+async def test_stateless_timeout_or_cancellation_preserves_reader_ownership(cancel: bool) -> None:
+    from agent_framework_tools.shell._capture import _ProcessOutput
+
+    proc = _FakeExecProcess(returncode=None, communicate_results=[(b"partial", b"error")])
+    waiting = asyncio.Event()
+    stopped = asyncio.Event()
+
+    async def wait() -> int:
+        waiting.set()
+        await stopped.wait()
+        proc.returncode = -9
+        return -9
+
+    async def stop(_proc: object) -> None:
+        stopped.set()
+
+    created: list[_ProcessOutput] = []
+    output_class = _ProcessOutput
+
+    def capture(process: asyncio.subprocess.Process, *, max_stdout_bytes: int, max_stderr_bytes: int) -> _ProcessOutput:
+        output = output_class(process, max_stdout_bytes=max_stdout_bytes, max_stderr_bytes=max_stderr_bytes)
+        created.append(output)
+        return output
+
+    with (
+        patch.object(proc, "wait", wait),
+        patch("agent_framework_tools.shell._executor.asyncio.create_subprocess_exec", AsyncMock(return_value=proc)),
+        patch("agent_framework_tools.shell._executor.kill_process_tree", side_effect=stop) as terminate,
+        patch("agent_framework_tools.shell._executor._ProcessOutput", side_effect=capture),
+    ):
+        task = asyncio.create_task(
+            run_stateless(
+                ["unused"], "unused", workdir=None, env=None, timeout=0.02, max_stdout_bytes=8, max_stderr_bytes=16
+            )
+        )
+        await waiting.wait()
+        if cancel:
+            task.cancel()
+            with pytest.raises(asyncio.CancelledError):
+                await task
+        else:
+            result = await task
+            assert result.timed_out
+            assert result.stdout == "partial" and result.stderr == "error"
+        terminate.assert_awaited_once()
+    assert all(task.done() for output in created for task in (*output._readers, output._waiter, output._completion))
+
+
+async def test_persistent_discards_idle_output_before_next_command() -> None:
+    from agent_framework_tools.shell._session import ShellSession, _CommandOutput
+
+    session = ShellSession(["unused"], max_stdout_bytes=16, max_stderr_bytes=32)
+    for is_stdout in (True, False):
+        stream = asyncio.StreamReader()
+        stream.feed_data(b"idle" * 16384)
+        stream.feed_eof()
+        await session._reader(stream, is_stdout=is_stdout)
+        assert session._output is None
+    output = _CommandOutput(b"marker", max_stdout_bytes=16, max_stderr_bytes=32)
+    session._output = output
+    stream = asyncio.StreamReader()
+    stream.feed_data(b"next\nmarker_0\n")
+    stream.feed_eof()
+    await session._reader(stream, is_stdout=True)
+    assert output.stdout.format() == ("next", False)
+    assert output.stderr.format() == ("", False)
+    assert output.exit_code == 0
+
+
+async def test_capture_cleanup_does_not_wait_forever_for_inherited_pipe(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    from agent_framework_tools.shell import _capture
+
+    proc = _FakeExecProcess()
+    # Model an exited parent whose child still holds both pipe handles.
+    proc.stdout = asyncio.StreamReader()
+    proc.stderr = asyncio.StreamReader()
+    proc.stdout.feed_data(b"partial")
+    monkeypatch.setattr(_capture, "_DRAIN_GRACE", 0.01)
+    output = _capture._ProcessOutput(
+        proc,  # type: ignore[arg-type]  # ty: ignore[invalid-argument-type]
+        max_stdout_bytes=16,
+        max_stderr_bytes=32,
+    )
+    await output.finish()
+    assert output.stdout.format() == ("partial", False)
+    assert all(task.done() for task in (*output._readers, output._waiter, output._completion))
+    assert "did not close after termination" in caplog.text
+
+
+async def test_persistent_reader_failure_stops_session_and_is_reported() -> None:
+    async with LocalShellTool(timeout=5) as tool:
+        session = tool._session
+        assert session is not None and session._proc is not None
+        proc = session._proc
+        assert proc.stderr is not None
+        proc.stderr.set_exception(OSError("pipe failed"))
+        with pytest.raises(RuntimeError, match="failed to read persistent shell output"):
+            await tool.run("Write-Output hello" if sys.platform == "win32" else "printf hello")
+        assert session._proc is None
+        assert proc.returncode is not None
+
+
 async def test_stateless_exit_code_propagates() -> None:
     tool = LocalShellTool(mode="stateless", approval_mode="never_require", acknowledge_unsafe=True)
     cmd = "exit 7" if sys.platform == "win32" else "sh -c 'exit 7'"
@@ -267,7 +448,8 @@ async def test_run_stateless_adds_powershell_encoding_preamble() -> None:
             workdir=None,
             env=None,
             timeout=1.0,
-            max_output_bytes=1024,
+            max_stdout_bytes=1024,
+            max_stderr_bytes=2048,
         )
 
     assert result.stdout == "ok"
@@ -275,26 +457,26 @@ async def test_run_stateless_adds_powershell_encoding_preamble() -> None:
     assert create_proc.await_args.args[-1].startswith("$OutputEncoding = [Console]::OutputEncoding")
 
 
-async def test_run_stateless_timeout_returns_empty_output_if_drain_fails() -> None:
-    proc = _FakeExecProcess(returncode=None, communicate_results=[asyncio.TimeoutError(), RuntimeError("drain failed")])
+async def test_run_stateless_reader_failure_is_reported_and_cleans_up() -> None:
+    proc = _FakeExecProcess()
+    proc.stderr.set_exception(RuntimeError("drain failed"))
 
     with (
         patch("agent_framework_tools.shell._executor.asyncio.create_subprocess_exec", AsyncMock(return_value=proc)),
         patch("agent_framework_tools.shell._executor.kill_process_tree", AsyncMock()) as kill_tree,
+        pytest.raises(RuntimeError, match="drain failed"),
     ):
-        result = await run_stateless(
+        await run_stateless(
             ["/bin/sh", "-c"],
             "sleep 5",
             workdir=None,
             env=None,
             timeout=0.01,
-            max_output_bytes=1024,
+            max_stdout_bytes=1024,
+            max_stderr_bytes=2048,
         )
 
     kill_tree.assert_awaited_once_with(proc)
-    assert result.timed_out is True
-    assert result.stdout == ""
-    assert result.stderr == ""
 
 
 @pytest.mark.skipif(sys.platform == "win32", reason="persistent-mode sentinel on POSIX")
@@ -568,25 +750,29 @@ async def test_persistent_confines_workdir_by_default(tmp_path: os.PathLike[str]
         assert os.path.realpath(pwd.stdout.strip()) == os.path.realpath(str(tmp_path))
 
 
-@pytest.mark.parametrize("failure", ["timeout", "unrecoverable_timeout", "overflow"])
-async def test_persistent_queued_command_restarts_after_session_recovery(failure: str) -> None:
+@pytest.mark.parametrize(
+    ("failure", "max_stderr_bytes"),
+    [("timeout", 128), ("unrecoverable_timeout", 128), ("overflow", 1), ("overflow", 8192)],
+)
+async def test_persistent_queued_command_restarts_after_session_recovery(failure: str, max_stderr_bytes: int) -> None:
     is_timeout = failure != "overflow"
     if sys.platform == "win32":
         command = (
             "[System.Threading.Thread]::Sleep(30000)"
             if is_timeout
-            else "while ($true) { [Console]::Write('overflow') }"
+            else "[Console]::Write('overflow' * 1024); [System.Threading.Thread]::Sleep(1000)"
         )
         followup = "Write-Output recovered"
     else:
-        command = "trap '' INT; sleep 30" if is_timeout else "while :; do printf 'overflow'; done"
+        command = "trap '' INT; sleep 30" if is_timeout else "printf '%8192s' overflow; sleep 1"
         followup = "printf 'recovered'"
 
     async with LocalShellTool(
         mode="persistent",
         approval_mode="never_require",
         acknowledge_unsafe=True,
-        max_output_bytes=128,
+        max_stdout_bytes=128,
+        max_stderr_bytes=max_stderr_bytes,
     ) as tool:
         session = tool._session
         assert session is not None
@@ -661,7 +847,9 @@ async def wait_for_exit() -> None:
             )
 
             async def wait_for_replacement() -> None:
-                while session._proc is original_process or b"started" not in session._stdout_buf:  # noqa: ASYNC110
+                while (  # noqa: ASYNC110 - synchronize against the replacement command's first output
+                    session._proc is original_process or session._output is None or session._output.stdout_bytes == 0
+                ):
                     await asyncio.sleep(0)
 
             await asyncio.wait_for(wait_for_replacement(), timeout=10.0)
@@ -758,7 +946,7 @@ async def test_persistent_close_rejects_queued_commands(queue_during_close: bool
     try:
 
         async def wait_until_running() -> None:
-            while b"running" not in session._stdout_buf:
+            while session._output is None or session._output.stdout_bytes == 0:
                 session._stdout_event.clear()
                 await session._stdout_event.wait()
 
diff --git a/python/packages/tools/tests/test_shell_parse_rc.py b/python/packages/tools/tests/test_shell_parse_rc.py
index 8befc8393ea..76e6c91983e 100644
--- a/python/packages/tools/tests/test_shell_parse_rc.py
+++ b/python/packages/tools/tests/test_shell_parse_rc.py
@@ -9,7 +9,57 @@
 
 from __future__ import annotations
 
-from agent_framework_tools.shell._session import _parse_rc
+import pytest
+
+from agent_framework_tools.shell._session import _CommandOutput, _parse_rc
+
+
+@pytest.mark.parametrize("split", range(1, 21))
+def test_completion_marker_survives_read_boundaries_with_tiny_allowance(split: int) -> None:
+    sentinel = b"__AF_END_test_tag__"
+    output = _CommandOutput(sentinel, max_stdout_bytes=1, max_stderr_bytes=8)
+    data = b"prefix\r\n" + sentinel + b"_42\r\nignored"
+    for offset in range(0, len(data), split):
+        output.read_stdout(data[offset : offset + split])
+        assert len(output._pending) <= len(sentinel) + 32
+    assert output.exit_code == 42
+    assert output.stdout.format() == ("\n[... truncated 5 bytes ...]\nx", True)
+
+
+def test_completion_parser_bounds_output_before_and_after_marker() -> None:
+    sentinel = b"__AF_END_test__"
+    output = _CommandOutput(sentinel, max_stdout_bytes=16, max_stderr_bytes=8)
+    for _ in range(1000):
+        output.read_stdout(b"not-a-marker" * 100)
+        assert len(output._pending) < len(sentinel)
+        assert len(output.stdout._output._head) + len(output.stdout._output._tail) <= 16
+    output.read_stdout(sentinel + b"_-7\n" + b"discard" * 1000)
+    assert output.exit_code == -7
+    assert not output._pending
+
+
+def test_completion_parser_rejects_unbounded_exit_status() -> None:
+    output = _CommandOutput(b"marker", max_stdout_bytes=1, max_stderr_bytes=8)
+    with pytest.raises(RuntimeError, match="oversized exit status"):
+        output.read_stdout(b"marker_" + b"1" * 1000)
+
+
+def test_completion_parser_finishes_partial_exit_status() -> None:
+    output = _CommandOutput(b"marker", max_stdout_bytes=8, max_stderr_bytes=1)
+    output.read_stdout(b"hello\nmarker_12")
+    output.finish_stdout()
+    assert output.exit_code == 12
+    assert output.stdout.format() == ("hello", False)
+
+
+def test_partial_completion_marker_does_not_count_toward_output_overflow() -> None:
+    sentinel = b"__AF_END_test_tag__"
+    output = _CommandOutput(sentinel, max_stdout_bytes=1, max_stderr_bytes=8)
+    for value in sentinel:
+        output.read_stdout(bytes([value]))
+        assert output.stdout_bytes == 0
+    output.read_stdout(b"_0\n")
+    assert output.exit_code == 0
 
 
 def test_parse_rc_zero() -> None:
diff --git a/python/packages/tools/tests/test_shell_truncate_and_quote.py b/python/packages/tools/tests/test_shell_truncate_and_quote.py
index 071e6f4e2e6..526de6710f0 100644
--- a/python/packages/tools/tests/test_shell_truncate_and_quote.py
+++ b/python/packages/tools/tests/test_shell_truncate_and_quote.py
@@ -7,7 +7,36 @@
 import pytest
 
 from agent_framework_tools.shell._tool import _quote_posix, _quote_powershell
-from agent_framework_tools.shell._truncate import truncate_head_tail, truncate_text_head_tail
+from agent_framework_tools.shell._truncate import (
+    _OutputBuffer,
+    _TextOutputBuffer,
+    truncate_head_tail,
+    truncate_text_head_tail,
+)
+
+
+@pytest.mark.parametrize("cap", [1, 2, 5, 64])
+@pytest.mark.parametrize("chunk_size", [1, 7, 1024])
+def test_streaming_capture_matches_truncation_with_bounded_storage(cap: int, chunk_size: int) -> None:
+    data = bytes(range(128)) * 32
+    output = _OutputBuffer(cap)
+    for offset in range(0, len(data), chunk_size):
+        output.append(data[offset : offset + chunk_size])
+        assert len(output._head) + len(output._tail) <= cap
+    assert output.format() == truncate_head_tail(data, cap)
+
+
+@pytest.mark.parametrize("cap", [1, 5, 64])
+def test_text_capture_preserves_decoding_and_strips_unbounded_trailing_newlines(cap: int) -> None:
+    data = b"head\n" + "caf\u00e9".encode() + b"\xff" + b"\r\n" * 100 + b"tail" + b"\r\n" * 100
+    output = _TextOutputBuffer(cap, strip_newlines=True)
+    for value in data:
+        output.append(bytes([value]))
+        for buffer in (output._output, output._pending_newlines):
+            assert len(buffer._head) + len(buffer._tail) <= cap
+    output.append(b"", final=True)
+    expected = data.decode("utf-8", errors="replace").rstrip("\r\n")
+    assert output.format() == truncate_text_head_tail(expected, cap)
 
 
 def test_truncate_under_cap_returns_original() -> None:
diff --git a/skills/agent-framework-builder-python/references/harness-agent-builder.md b/skills/agent-framework-builder-python/references/harness-agent-builder.md
index ad992ce0a44..fb92cdab26a 100644
--- a/skills/agent-framework-builder-python/references/harness-agent-builder.md
+++ b/skills/agent-framework-builder-python/references/harness-agent-builder.md
@@ -36,7 +36,10 @@ complete option list.
 - Load skills and background agents only from trusted sources; their content crosses the model and tool boundary.
 - Bound autonomous loops and stop for unresolved approvals or user input.
 - Treat shell command filters as a usability guard, not a security boundary. The host owns sandboxing and executor
-  lifecycle.
+  lifecycle. Apply resource limits to the host agent process as well as the shell container; container memory
+  limits do not cover output captured outside the container. `LocalShellTool` and `DockerShellTool` independently
+  limit retained normal and error output through `max_stdout_bytes` and `max_stderr_bytes`, each defaulting to
+  64 KiB. Both settings must be positive; they do not impose a total process-memory limit.
 
 ## Validation
 

From f5cef56c4dc1322ee158e50946cee14b66a3ffbe Mon Sep 17 00:00:00 2001
From: westey <164392973+westey-m@users.noreply.github.com>
Date: Wed, 28 Oct 2026 18:42:06 +0000
Subject: [PATCH 2/3] Address PR comments

---
 .../agent_framework_tools/shell/_capture.py   | 16 +++-
 .../tools/tests/test_docker_shell_tool.py     |  6 ++
 .../tools/tests/test_local_shell_tool.py      | 89 ++++++++++++++++++-
 3 files changed, 108 insertions(+), 3 deletions(-)

diff --git a/python/packages/tools/agent_framework_tools/shell/_capture.py b/python/packages/tools/agent_framework_tools/shell/_capture.py
index 6092ee10faf..e693f1e308a 100644
--- a/python/packages/tools/agent_framework_tools/shell/_capture.py
+++ b/python/packages/tools/agent_framework_tools/shell/_capture.py
@@ -6,6 +6,7 @@
 
 import asyncio
 import logging
+from typing import cast
 
 from ._truncate import _OutputBuffer  # pyright: ignore[reportPrivateUsage] - shared shell capture
 
@@ -19,6 +20,7 @@ class _ProcessOutput:  # pyright: ignore[reportUnusedClass] - used by shell exec
     """Own pipe readers independently of a command's execution timeout."""
 
     def __init__(self, proc: asyncio.subprocess.Process, *, max_stdout_bytes: int, max_stderr_bytes: int) -> None:
+        self._proc = proc
         self.stdout = _OutputBuffer(max_stdout_bytes)
         self.stderr = _OutputBuffer(max_stderr_bytes)
         self._readers = (
@@ -55,4 +57,16 @@ async def close(self) -> None:
         for task in tasks:
             if not task.done():
                 task.cancel()
-        await asyncio.gather(*tasks, return_exceptions=True)
+        try:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        finally:
+            # Process has no public read-pipe close method. Use its transport
+            # only to close owned output pipes, not to terminate the process.
+            transport = cast(
+                asyncio.SubprocessTransport,
+                self._proc._transport,  # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType] - absent from typeshed
+            )
+            for fd in (1, 2):
+                pipe = transport.get_pipe_transport(fd)
+                if pipe is not None:
+                    pipe.close()
diff --git a/python/packages/tools/tests/test_docker_shell_tool.py b/python/packages/tools/tests/test_docker_shell_tool.py
index 159c8687dcf..f6f80c6854c 100644
--- a/python/packages/tools/tests/test_docker_shell_tool.py
+++ b/python/packages/tools/tests/test_docker_shell_tool.py
@@ -46,6 +46,10 @@ def __init__(
     ) -> None:
         self.pid = pid
         self.returncode = returncode
+        self._transport = MagicMock(spec=asyncio.SubprocessTransport)
+        self.stdout_transport = MagicMock(spec=asyncio.ReadTransport)
+        self.stderr_transport = MagicMock(spec=asyncio.ReadTransport)
+        self._transport.get_pipe_transport.side_effect = {1: self.stdout_transport, 2: self.stderr_transport}.get
         self.stdout = asyncio.StreamReader()
         self.stderr = asyncio.StreamReader()
         self.killed = False
@@ -505,6 +509,8 @@ async def wait() -> int:
             with pytest.raises(OSError, match="pipe failed"):
                 await task
     assert proc.killed
+    proc.stdout_transport.close.assert_called()
+    proc.stderr_transport.close.assert_called()
     assert create_process.await_count == 2
     assert create_process.await_args_list[1].args[1:4] == ("kill", "--signal", "KILL")
 
diff --git a/python/packages/tools/tests/test_local_shell_tool.py b/python/packages/tools/tests/test_local_shell_tool.py
index 1461ebb9d5a..fe7316a5e9f 100644
--- a/python/packages/tools/tests/test_local_shell_tool.py
+++ b/python/packages/tools/tests/test_local_shell_tool.py
@@ -8,12 +8,13 @@
 import signal
 import sys
 from collections.abc import Awaitable, Mapping, Sequence
-from contextlib import nullcontext
+from contextlib import nullcontext, suppress
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
-from unittest.mock import AsyncMock, patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
+import psutil
 import pytest
 from agent_framework import Agent, AgentSession, BaseChatClient, ChatResponse, Content, FunctionInvocationLayer, Message
 from agent_framework.security import (
@@ -43,6 +44,10 @@ def __init__(
         communicate_results: list[tuple[bytes, bytes] | BaseException] | None = None,
     ) -> None:
         self.returncode = returncode
+        self._transport = MagicMock(spec=asyncio.SubprocessTransport)
+        self.stdout_transport = MagicMock(spec=asyncio.ReadTransport)
+        self.stderr_transport = MagicMock(spec=asyncio.ReadTransport)
+        self._transport.get_pipe_transport.side_effect = {1: self.stdout_transport, 2: self.stderr_transport}.get
         self.stdout = asyncio.StreamReader()
         self.stderr = asyncio.StreamReader()
         result = (communicate_results or [(b"", b"")])[0]
@@ -251,6 +256,8 @@ def capture(process: asyncio.subprocess.Process, *, max_stdout_bytes: int, max_s
             assert result.stdout == "partial" and result.stderr == "error"
         terminate.assert_awaited_once()
     assert all(task.done() for output in created for task in (*output._readers, output._waiter, output._completion))
+    proc.stdout_transport.close.assert_called()
+    proc.stderr_transport.close.assert_called()
 
 
 async def test_persistent_discards_idle_output_before_next_command() -> None:
@@ -294,6 +301,82 @@ async def test_capture_cleanup_does_not_wait_forever_for_inherited_pipe(
     assert output.stdout.format() == ("partial", False)
     assert all(task.done() for task in (*output._readers, output._waiter, output._completion))
     assert "did not close after termination" in caplog.text
+    proc.stdout_transport.close.assert_called()
+    proc.stderr_transport.close.assert_called()
+    await output.close()
+
+
+async def test_capture_closes_real_inherited_pipes_repeatedly(monkeypatch: pytest.MonkeyPatch) -> None:
+    from agent_framework_tools.shell import _capture
+
+    monkeypatch.setattr(_capture, "_DRAIN_GRACE", 0.01)
+    script = (
+        "import subprocess, sys; "
+        "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(10)']); "
+        "print(child.pid, flush=True)"
+    )
+    for _ in range(3):
+        proc = await asyncio.create_subprocess_exec(
+            sys.executable, "-c", script, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+        )
+        output = _capture._ProcessOutput(proc, max_stdout_bytes=128, max_stderr_bytes=128)
+        child: psutil.Process | None = None
+        try:
+
+            async def wait_for_parent(
+                proc: asyncio.subprocess.Process = proc, output: _capture._ProcessOutput = output
+            ) -> None:
+                while proc.returncode is None or not output.stdout.format()[0]:  # noqa: ASYNC110
+                    await asyncio.sleep(0.01)
+
+            await asyncio.wait_for(wait_for_parent(), timeout=5)
+            child = psutil.Process(int(output.stdout.format()[0]))
+            transport = proc._transport  # type: ignore[attr-defined]  # ty: ignore[unresolved-attribute]  # pyright: ignore[reportAttributeAccessIssue]
+            pipes = [transport.get_pipe_transport(fd) for fd in (1, 2)]
+            assert all(pipe is not None for pipe in pipes)
+            handles = [pipe.get_extra_info("pipe") for pipe in pipes if pipe is not None]
+            await output.finish()
+            await output.close()
+
+            async def wait_for_closed_handles(handles: list[Any] = handles) -> None:
+                while not all(handle.closed for handle in handles):  # noqa: ASYNC110
+                    await asyncio.sleep(0)
+
+            await asyncio.wait_for(wait_for_closed_handles(), timeout=2)
+            assert all(pipe.is_closing() for pipe in pipes if pipe is not None)
+            assert child.is_running()
+            assert all(task.done() for task in (*output._readers, output._waiter, output._completion))
+        finally:
+            await output.close()
+            if proc.returncode is None:
+                proc.kill()
+            if child is not None:
+                with suppress(psutil.NoSuchProcess):
+                    child.kill()
+            await asyncio.wait_for(proc.wait(), timeout=5)
+
+
+@pytest.mark.parametrize("missing", ["stdout", "stderr"])
+async def test_capture_closes_output_with_other_stream_discarded(missing: str) -> None:
+    from agent_framework_tools.shell._capture import _ProcessOutput
+
+    proc = await asyncio.create_subprocess_exec(
+        sys.executable,
+        "-c",
+        "import sys; print('normal'); print('error', file=sys.stderr)",
+        stdout=asyncio.subprocess.DEVNULL if missing == "stdout" else asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.DEVNULL if missing == "stderr" else asyncio.subprocess.PIPE,
+    )
+    output = _ProcessOutput(proc, max_stdout_bytes=128, max_stderr_bytes=128)
+    try:
+        await output.wait(5)
+        assert output.stdout.format()[0].strip() == ("" if missing == "stdout" else "normal")
+        assert output.stderr.format()[0].strip() == ("" if missing == "stderr" else "error")
+    finally:
+        if proc.returncode is None:
+            proc.kill()
+        await output.close()
+        await asyncio.wait_for(proc.wait(), timeout=5)
 
 
 async def test_persistent_reader_failure_stops_session_and_is_reported() -> None:
@@ -477,6 +560,8 @@ async def test_run_stateless_reader_failure_is_reported_and_cleans_up() -> None:
         )
 
     kill_tree.assert_awaited_once_with(proc)
+    proc.stdout_transport.close.assert_called()
+    proc.stderr_transport.close.assert_called()
 
 
 @pytest.mark.skipif(sys.platform == "win32", reason="persistent-mode sentinel on POSIX")

From 6448b7f6c2cc66583d637cd0ec2d028d6c4e6675 Mon Sep 17 00:00:00 2001
From: westey <164392973+westey-m@users.noreply.github.com>
Date: Wed, 28 Oct 2026 19:17:59 +0000
Subject: [PATCH 3/3] Fix build error

---
 python/packages/tools/tests/test_local_shell_tool.py | 11 ++++++++++-
 1 file changed, 10 insertions(+), 1 deletion(-)

diff --git a/python/packages/tools/tests/test_local_shell_tool.py b/python/packages/tools/tests/test_local_shell_tool.py
index fe7316a5e9f..b24c99a85f6 100644
--- a/python/packages/tools/tests/test_local_shell_tool.py
+++ b/python/packages/tools/tests/test_local_shell_tool.py
@@ -315,6 +315,15 @@ async def test_capture_closes_real_inherited_pipes_repeatedly(monkeypatch: pytes
         "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(10)']); "
         "print(child.pid, flush=True)"
     )
+
+    def handle_is_closed(handle: Any) -> bool:
+        # Windows PipeHandle has no .closed; both platforms reject fileno() after closure.
+        try:
+            handle.fileno()
+        except ValueError:
+            return True
+        return False
+
     for _ in range(3):
         proc = await asyncio.create_subprocess_exec(
             sys.executable, "-c", script, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
@@ -339,7 +348,7 @@ async def wait_for_parent(
             await output.close()
 
             async def wait_for_closed_handles(handles: list[Any] = handles) -> None:
-                while not all(handle.closed for handle in handles):  # noqa: ASYNC110
+                while not all(handle_is_closed(handle) for handle in handles):  # noqa: ASYNC110
                     await asyncio.sleep(0)
 
             await asyncio.wait_for(wait_for_closed_handles(), timeout=2)