From 48934b27e4b2c19fd9a12b7690fe3dedab28c65b Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Tue, 1 Sep 2026 18:05:26 -0600
Subject: [PATCH 1/6] feat(py): exec-shaped backend seam for the execution
 worker

The first piece of M6 (kata f11e). Everything the execution driver needs
from a process host is one call, so hosting the worker somewhere else --
a Connect container, most likely -- becomes another implementation of
ExecBackend rather than an edit to the driver above it.

Three behaviours here are the ones Inspect's subprocess utilities got
right and are easy to get wrong:

Input goes in on stdin, never as an argument, so there is no escaping to
mishandle and no command-line length limit. stdin is written without a
drain: a worker that dies during startup leaves nobody reading the pipe,
and that is a failed call to report rather than a BrokenPipeError out of
the plumbing.

Output past the cap keeps the tail and lets the process finish. Killing
on the cap would discard a result the code had already computed, and
simply not reading would deadlock the child against a full pipe. The
head is what gets dropped, since the result is usually last.

Shutdown escalates rather than going straight to SIGKILL, so a worker
that handles SIGTERM gets to clean up. After SIGKILL the wait is bounded:
the child watcher can miss an exit, and a killed process is gone whether
or not we observe it go.

env replaces the parent's environment rather than extending it. The
allowlist that decides what belongs in it is a separate task; this is
only the mechanism that makes an allowlist possible at all.

Tests drive real subprocesses. The one stand-in is for the missed-exit
race, which cannot be provoked on demand.
---
 pkg-py/src/commons/_execution/__init__.py |   1 +
 pkg-py/src/commons/_execution/_backend.py | 156 +++++++++++++++
 pkg-py/tests/test_execution_backend.py    | 220 ++++++++++++++++++++++
 3 files changed, 377 insertions(+)
 create mode 100644 pkg-py/src/commons/_execution/__init__.py
 create mode 100644 pkg-py/src/commons/_execution/_backend.py
 create mode 100644 pkg-py/tests/test_execution_backend.py

diff --git a/pkg-py/src/commons/_execution/__init__.py b/pkg-py/src/commons/_execution/__init__.py
new file mode 100644
index 00000000..48df49b2
--- /dev/null
+++ b/pkg-py/src/commons/_execution/__init__.py
@@ -0,0 +1 @@
+"""Running model-written code in a worker process."""
diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py
new file mode 100644
index 00000000..74e627f1
--- /dev/null
+++ b/pkg-py/src/commons/_execution/_backend.py
@@ -0,0 +1,156 @@
+"""The seam between the execution driver and whatever runs the worker.
+
+Everything above this line talks to a single ``exec``-shaped call, so a
+container-hosted backend can be added later as another implementation rather
+than as an edit to the driver.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from typing import Protocol, runtime_checkable
+
+__all__ = ["ExecBackend", "ExecResult", "ExecTimeoutError", "LocalBackend"]
+
+# Enough for a generous amount of printed output without letting a runaway
+# loop hold the whole of it in memory.
+DEFAULT_OUTPUT_LIMIT = 1024 * 1024
+
+# How long to be patient with a process being shut down: first for it to
+# honour SIGTERM, then for its exit to be observed after SIGKILL.
+TERMINATE_GRACE = 2.0
+
+
+class ExecTimeoutError(TimeoutError):
+    """The command ran past its deadline and was killed."""
+
+
+@dataclass(frozen=True)
+class ExecResult:
+    returncode: int
+    stdout: str
+    stderr: str
+    stdout_truncated: bool = False
+    stderr_truncated: bool = False
+
+
+@runtime_checkable
+class ExecBackend(Protocol):
+    """What the driver needs from whatever runs the worker.
+
+    Kept to one call so that hosting the worker somewhere else — a container,
+    say — is a new implementation of this, not a change to the driver.
+    """
+
+    async def exec(
+        self,
+        cmd: Sequence[str],
+        *,
+        input: str | None = None,
+        cwd: str | None = None,
+        env: Mapping[str, str] | None = None,
+        timeout: float | None = None,
+    ) -> ExecResult:
+        """Run ``cmd``, feeding ``input`` on stdin, and collect its output.
+
+        Raises ``ExecTimeoutError`` if ``timeout`` passes before the command
+        finishes, having first made sure the process is gone.
+        """
+        ...
+
+
+async def _read_tail(
+    stream: asyncio.StreamReader | None, limit: int
+) -> tuple[bytes, bool]:
+    """Drain ``stream``, keeping only its last ``limit`` bytes.
+
+    Draining is the point: a process whose output nobody reads blocks forever
+    on a full pipe. Dropping the head rather than the tail keeps the part of
+    the output most likely to hold the result.
+    """
+    if stream is None:
+        return b"", False
+    kept = bytearray()
+    truncated = False
+    while True:
+        chunk = await stream.read(64 * 1024)
+        if not chunk:
+            return bytes(kept), truncated
+        kept += chunk
+        if len(kept) > limit:
+            del kept[: len(kept) - limit]
+            truncated = True
+
+
+class LocalBackend:
+    """Runs the worker as a child of this process, with no isolation."""
+
+    def __init__(
+        self,
+        output_limit: int = DEFAULT_OUTPUT_LIMIT,
+        terminate_grace: float = TERMINATE_GRACE,
+    ) -> None:
+        self._output_limit = output_limit
+        self._terminate_grace = terminate_grace
+
+    async def exec(
+        self,
+        cmd: Sequence[str],
+        *,
+        input: str | None = None,
+        cwd: str | None = None,
+        env: Mapping[str, str] | None = None,
+        timeout: float | None = None,
+    ) -> ExecResult:
+        process = await asyncio.create_subprocess_exec(
+            *cmd,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            cwd=cwd,
+            env=None if env is None else dict(env),
+        )
+        if process.stdin is not None:
+            if input is not None:
+                process.stdin.write(input.encode())
+            process.stdin.close()
+        reading = asyncio.gather(
+            _read_tail(process.stdout, self._output_limit),
+            _read_tail(process.stderr, self._output_limit),
+        )
+        try:
+            stdout, stderr = await asyncio.wait_for(reading, timeout)
+        except TimeoutError:
+            await _terminate(process, self._terminate_grace)
+            raise ExecTimeoutError(
+                f"the command exceeded its {timeout}-second time limit"
+            ) from None
+        await process.wait()
+        return ExecResult(
+            returncode=process.returncode or 0,
+            stdout=stdout[0].decode(errors="replace"),
+            stderr=stderr[0].decode(errors="replace"),
+            stdout_truncated=stdout[1],
+            stderr_truncated=stderr[1],
+        )
+
+
+async def _terminate(process: asyncio.subprocess.Process, grace: float) -> None:
+    """Ask the process to exit, then insist."""
+    if process.returncode is not None:
+        return
+    process.terminate()
+    try:
+        await asyncio.wait_for(asyncio.shield(process.wait()), grace)
+        return
+    except TimeoutError:
+        pass
+    process.kill()
+    # The exit can go unobserved if the child watcher misses it, and a killed
+    # process is gone whether or not we see it go. Wait, but not forever.
+    try:
+        await asyncio.wait_for(asyncio.shield(process.wait()), grace)
+    except TimeoutError:
+        pass
diff --git a/pkg-py/tests/test_execution_backend.py b/pkg-py/tests/test_execution_backend.py
new file mode 100644
index 00000000..a33be288
--- /dev/null
+++ b/pkg-py/tests/test_execution_backend.py
@@ -0,0 +1,220 @@
+"""The exec-shaped seam the execution subsystem sits on.
+
+Tests drive real subprocesses rather than mocks: the behaviours that matter
+here (stdin delivery, output caps, kill escalation) are properties of process
+handling, and a mock would only restate the implementation.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import sys
+from typing import Any, cast
+
+import pytest
+
+from commons._execution._backend import (
+    ExecBackend,
+    ExecTimeoutError,
+    LocalBackend,
+    _terminate,
+)
+
+
+async def test_runs_a_command_and_returns_its_output() -> None:
+    backend = LocalBackend()
+
+    result = await backend.exec([sys.executable, "-c", "print('hello')"])
+
+    assert result.returncode == 0
+    assert result.stdout == "hello\n"
+    assert result.stderr == ""
+
+
+async def test_input_reaches_the_process_on_stdin() -> None:
+    # Code goes in on stdin rather than as an argument: no escaping to get
+    # wrong and no command-line length limit.
+    backend = LocalBackend()
+
+    result = await backend.exec(
+        [sys.executable, "-c", "import sys; sys.stdout.write(sys.stdin.read())"],
+        input="model-written code\n",
+    )
+
+    assert result.stdout == "model-written code\n"
+
+
+async def test_the_process_starts_in_the_given_working_directory(tmp_path) -> None:
+    backend = LocalBackend()
+
+    result = await backend.exec(
+        [sys.executable, "-c", "import os; print(os.getcwd())"],
+        cwd=str(tmp_path),
+    )
+
+    assert result.stdout.strip() == os.path.realpath(tmp_path)
+
+
+async def test_the_given_environment_replaces_the_parents_rather_than_extending_it(
+    monkeypatch,
+) -> None:
+    # The parent holds credentials a child has no business seeing, and a
+    # subprocess inherits the whole environment by default. Passing `env` has
+    # to mean "exactly this", not "this as well".
+    monkeypatch.setenv("COMMONS_TEST_SECRET", "sk-not-a-real-key")
+    backend = LocalBackend()
+
+    result = await backend.exec(
+        [
+            sys.executable,
+            "-c",
+            "import os; print(os.environ.get('COMMONS_TEST_SECRET'))",
+        ],
+        env={"PATH": os.environ["PATH"]},
+    )
+
+    assert result.stdout.strip() == "None"
+
+
+NOISY = "for i in range(200): print(f'line-{i}:' + 'x' * 1000)"
+
+
+async def test_output_past_the_cap_keeps_the_tail_and_the_process_still_finishes() -> (
+    None
+):
+    # Killing the process on the cap would lose a result the code had already
+    # computed, and simply not reading would deadlock it against a full pipe.
+    # Keep draining, keep the most recent bytes, let it exit.
+    backend = LocalBackend(output_limit=2000)
+
+    result = await backend.exec([sys.executable, "-c", NOISY])
+
+    assert result.returncode == 0
+    assert len(result.stdout) <= 2000
+    assert result.stdout.rstrip().endswith("x" * 100)
+    assert "line-199:" in result.stdout
+    assert "line-0:" not in result.stdout
+    assert result.stdout_truncated
+
+
+def _sleeper(sentinel: object, *, ignore_sigterm: bool = False) -> str:
+    """Code that outlives its timeout and records the fact if it is allowed to."""
+    guard = (
+        "import signal; signal.signal(signal.SIGTERM, signal.SIG_IGN)\n"
+        if ignore_sigterm
+        else ""
+    )
+    return f"{guard}import time; time.sleep(0.6); open({str(sentinel)!r}, 'w').close()"
+
+
+async def test_a_call_past_the_timeout_raises_and_the_process_does_not_survive(
+    tmp_path,
+) -> None:
+    sentinel = tmp_path / "survived"
+    backend = LocalBackend()
+
+    with pytest.raises(ExecTimeoutError):
+        await backend.exec([sys.executable, "-c", _sleeper(sentinel)], timeout=0.15)
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()
+
+
+async def test_a_process_that_handles_sigterm_gets_to_clean_up_first(tmp_path) -> None:
+    # SIGKILL first would strand whatever the worker was in the middle of.
+    # Ask politely, then insist.
+    marker = tmp_path / "cleaned-up"
+    code = (
+        "import signal, sys, time\n"
+        f"signal.signal(signal.SIGTERM, lambda *a: (open({str(marker)!r}, 'w').close(), sys.exit(0)))\n"
+        "time.sleep(5)\n"
+    )
+    backend = LocalBackend()
+
+    with pytest.raises(ExecTimeoutError):
+        await backend.exec([sys.executable, "-c", code], timeout=0.15)
+
+    assert marker.exists()
+
+
+async def test_a_process_that_ignores_sigterm_is_killed_anyway(tmp_path) -> None:
+    sentinel = tmp_path / "survived"
+    backend = LocalBackend(terminate_grace=0.1)
+
+    with pytest.raises(ExecTimeoutError):
+        await backend.exec(
+            [sys.executable, "-c", _sleeper(sentinel, ignore_sigterm=True)],
+            timeout=0.15,
+        )
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()
+
+
+class _NeverReaped:
+    """A process that takes its signals but whose exit is never observed.
+
+    Stands in for the race where the child watcher misses the exit. There is
+    no way to provoke that on demand, so this is the one place the suite
+    substitutes a stand-in for a real process.
+    """
+
+    returncode: int | None = None
+
+    def __init__(self) -> None:
+        self.signals: list[str] = []
+
+    def terminate(self) -> None:
+        self.signals.append("term")
+
+    def kill(self) -> None:
+        self.signals.append("kill")
+
+    async def wait(self) -> int:
+        await asyncio.sleep(3600)
+        return 0
+
+
+async def test_terminate_gives_up_when_the_exit_is_never_reaped() -> None:
+    process = _NeverReaped()
+
+    await asyncio.wait_for(_terminate(cast(Any, process), 0.05), timeout=2)
+
+    assert process.signals == ["term", "kill"]
+
+
+async def test_input_to_a_process_that_never_reads_it_is_not_an_error() -> None:
+    # A worker that dies during startup leaves nobody on the other end of the
+    # pipe. That is a failed call to report, not an exception from the plumbing.
+    backend = LocalBackend()
+
+    result = await backend.exec(
+        [sys.executable, "-c", "raise SystemExit(3)"],
+        input="x" * (4 * 1024 * 1024),
+    )
+
+    assert result.returncode == 3
+
+
+async def test_input_larger_than_the_pipe_buffer_arrives_in_full() -> None:
+    # Handles cross this boundary, so delivery cannot quietly stop at whatever
+    # the operating system's pipe buffer happens to be.
+    backend = LocalBackend()
+    payload = "y" * (4 * 1024 * 1024)
+
+    result = await backend.exec(
+        [sys.executable, "-c", "import sys; print(len(sys.stdin.read()))"],
+        input=payload,
+    )
+
+    assert result.stdout.strip() == str(len(payload))
+
+
+def test_the_local_backend_satisfies_the_backend_interface() -> None:
+    # The annotation is the real assertion: pyrefly rejects an implementation
+    # whose signature has drifted from the interface a container-hosted
+    # backend would also have to meet.
+    backend: ExecBackend = LocalBackend()
+
+    assert isinstance(backend, ExecBackend)

From 77a2fc43eac593c68472832a16b4d7b0bf74bcc0 Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Tue, 1 Sep 2026 18:07:31 -0600
Subject: [PATCH 2/6] fix(py): apply the exec timeout to process exit, not just
 output

Draining stdout and stderr ends at end-of-file, which a process can reach
while still running: closing both streams and carrying on defeated the
deadline entirely, and the call then waited on process.wait() with no
bound at all.

Both halves now sit inside the caller's deadline, so neither can outlast
it. Found by review of 48934b2.
---
 pkg-py/src/commons/_execution/_backend.py | 23 +++++++++++++++++------
 pkg-py/tests/test_execution_backend.py    | 20 ++++++++++++++++++++
 2 files changed, 37 insertions(+), 6 deletions(-)

diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py
index 74e627f1..f4402337 100644
--- a/pkg-py/src/commons/_execution/_backend.py
+++ b/pkg-py/src/commons/_execution/_backend.py
@@ -95,6 +95,22 @@ def __init__(
         self._output_limit = output_limit
         self._terminate_grace = terminate_grace
 
+    async def _collect(
+        self, process: asyncio.subprocess.Process
+    ) -> tuple[tuple[bytes, bool], tuple[bytes, bool]]:
+        """Drain both streams, then wait for the process to actually exit.
+
+        Reaching end-of-output is not the same as being finished: code can
+        close its streams and keep running. Both halves sit inside the
+        caller's deadline so that neither can outlast it.
+        """
+        streams = await asyncio.gather(
+            _read_tail(process.stdout, self._output_limit),
+            _read_tail(process.stderr, self._output_limit),
+        )
+        await process.wait()
+        return streams[0], streams[1]
+
     async def exec(
         self,
         cmd: Sequence[str],
@@ -116,18 +132,13 @@ async def exec(
             if input is not None:
                 process.stdin.write(input.encode())
             process.stdin.close()
-        reading = asyncio.gather(
-            _read_tail(process.stdout, self._output_limit),
-            _read_tail(process.stderr, self._output_limit),
-        )
         try:
-            stdout, stderr = await asyncio.wait_for(reading, timeout)
+            stdout, stderr = await asyncio.wait_for(self._collect(process), timeout)
         except TimeoutError:
             await _terminate(process, self._terminate_grace)
             raise ExecTimeoutError(
                 f"the command exceeded its {timeout}-second time limit"
             ) from None
-        await process.wait()
         return ExecResult(
             returncode=process.returncode or 0,
             stdout=stdout[0].decode(errors="replace"),
diff --git a/pkg-py/tests/test_execution_backend.py b/pkg-py/tests/test_execution_backend.py
index a33be288..682f3043 100644
--- a/pkg-py/tests/test_execution_backend.py
+++ b/pkg-py/tests/test_execution_backend.py
@@ -218,3 +218,23 @@ def test_the_local_backend_satisfies_the_backend_interface() -> None:
     backend: ExecBackend = LocalBackend()
 
     assert isinstance(backend, ExecBackend)
+
+
+async def test_the_timeout_still_applies_after_the_output_streams_close(
+    tmp_path,
+) -> None:
+    # Reaching end-of-output is not the same as being finished. Code that
+    # closes its streams and keeps running must still hit the deadline.
+    sentinel = tmp_path / "survived"
+    code = (
+        "import os, time\n"
+        "os.close(1); os.close(2)\n"
+        f"time.sleep(0.6); open({str(sentinel)!r}, 'w').close()\n"
+    )
+    backend = LocalBackend()
+
+    with pytest.raises(ExecTimeoutError):
+        await backend.exec([sys.executable, "-c", code], timeout=0.15)
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()

From 650302f8a2b1e4f694af47bcd78810f187fbb9b6 Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Tue, 1 Sep 2026 18:10:36 -0600
Subject: [PATCH 3/6] fix(py): kill the child when an exec call is cancelled

The driver cancels calls when a conversation goes away or the agent shuts
down, and only the timeout path was ending the process. A cancelled call
left the worker running: still holding the parent's file descriptors,
still burning CPU, with nobody waiting on the result.

Cancellation now goes through the same shutdown escalation as a timeout.
The waiting happens inside an except block, where an await can be cut
short, so there is a test covering a child that ignores SIGTERM to pin
that SIGKILL still lands there.
---
 pkg-py/src/commons/_execution/_backend.py |  6 ++++
 pkg-py/tests/test_execution_backend.py    | 36 +++++++++++++++++++++++
 2 files changed, 42 insertions(+)

diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py
index f4402337..470a1a23 100644
--- a/pkg-py/src/commons/_execution/_backend.py
+++ b/pkg-py/src/commons/_execution/_backend.py
@@ -139,6 +139,12 @@ async def exec(
             raise ExecTimeoutError(
                 f"the command exceeded its {timeout}-second time limit"
             ) from None
+        except asyncio.CancelledError:
+            # Whoever started the process ends it. A cancelled call that left
+            # the worker running would keep holding the parent's file
+            # descriptors and go on burning CPU with nobody waiting on it.
+            await _terminate(process, self._terminate_grace)
+            raise
         return ExecResult(
             returncode=process.returncode or 0,
             stdout=stdout[0].decode(errors="replace"),
diff --git a/pkg-py/tests/test_execution_backend.py b/pkg-py/tests/test_execution_backend.py
index 682f3043..b4ef9a31 100644
--- a/pkg-py/tests/test_execution_backend.py
+++ b/pkg-py/tests/test_execution_backend.py
@@ -238,3 +238,39 @@ async def test_the_timeout_still_applies_after_the_output_streams_close(
 
     await asyncio.sleep(0.8)
     assert not sentinel.exists()
+
+
+async def test_cancelling_a_call_does_not_leave_the_process_running(tmp_path) -> None:
+    # The driver cancels calls when a conversation goes away or the agent
+    # shuts down. Whoever started the process has to be the one to end it.
+    sentinel = tmp_path / "survived"
+    backend = LocalBackend(terminate_grace=0.1)
+    call = asyncio.create_task(backend.exec([sys.executable, "-c", _sleeper(sentinel)]))
+    await asyncio.sleep(0.1)
+
+    call.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await call
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()
+
+
+async def test_cancellation_still_escalates_for_a_process_ignoring_sigterm(
+    tmp_path,
+) -> None:
+    # The cancellation path does its waiting inside an except block, where an
+    # await can be cut short. SIGKILL still has to land.
+    sentinel = tmp_path / "survived"
+    backend = LocalBackend(terminate_grace=0.1)
+    call = asyncio.create_task(
+        backend.exec([sys.executable, "-c", _sleeper(sentinel, ignore_sigterm=True)])
+    )
+    await asyncio.sleep(0.1)
+
+    call.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await call
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()

From 7c82d5503c791c7c0525eaf041bc3e1af7d869e0 Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Tue, 1 Sep 2026 18:12:51 -0600
Subject: [PATCH 4/6] fix(py): shutdown escalation survives a second
 cancellation

The cleanup added in 650302f did its waiting inline in the except block,
so a cancel landing during the SIGTERM grace period cut it short before
SIGKILL and a child ignoring SIGTERM survived. One cancel was covered;
two were not.

Shutdown now runs as its own task, awaited through a shield, with the
backend holding a reference so it cannot be collected mid-escalation. A
second cancel stops us waiting on it, not the escalation itself.

Found by review of 650302f.
---
 pkg-py/src/commons/_execution/_backend.py | 15 ++++++++++++++-
 pkg-py/tests/test_execution_backend.py    | 21 +++++++++++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py
index 470a1a23..966bd076 100644
--- a/pkg-py/src/commons/_execution/_backend.py
+++ b/pkg-py/src/commons/_execution/_backend.py
@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from typing import Protocol, runtime_checkable
@@ -94,6 +95,9 @@ def __init__(
     ) -> None:
         self._output_limit = output_limit
         self._terminate_grace = terminate_grace
+        # Shutdowns outlive the call that started them, so they need an owner
+        # that keeps them from being garbage-collected mid-escalation.
+        self._shutdowns: set[asyncio.Task[None]] = set()
 
     async def _collect(
         self, process: asyncio.subprocess.Process
@@ -143,7 +147,16 @@ async def exec(
             # Whoever started the process ends it. A cancelled call that left
             # the worker running would keep holding the parent's file
             # descriptors and go on burning CPU with nobody waiting on it.
-            await _terminate(process, self._terminate_grace)
+            #
+            # Shutdown runs in its own task so that a second cancellation
+            # stops us waiting on it without stopping the escalation itself;
+            # a caller cancelling twice must not be able to leave a
+            # SIGTERM-ignoring child alive.
+            shutdown = asyncio.ensure_future(_terminate(process, self._terminate_grace))
+            self._shutdowns.add(shutdown)
+            shutdown.add_done_callback(self._shutdowns.discard)
+            with contextlib.suppress(asyncio.CancelledError):
+                await asyncio.shield(shutdown)
             raise
         return ExecResult(
             returncode=process.returncode or 0,
diff --git a/pkg-py/tests/test_execution_backend.py b/pkg-py/tests/test_execution_backend.py
index b4ef9a31..d7946105 100644
--- a/pkg-py/tests/test_execution_backend.py
+++ b/pkg-py/tests/test_execution_backend.py
@@ -274,3 +274,24 @@ async def test_cancellation_still_escalates_for_a_process_ignoring_sigterm(
 
     await asyncio.sleep(0.8)
     assert not sentinel.exists()
+
+
+async def test_a_second_cancellation_cannot_abort_the_shutdown(tmp_path) -> None:
+    # Shutdown is not the caller's to interrupt. A cancel landing while the
+    # grace period is being awaited would otherwise skip SIGKILL and leave a
+    # SIGTERM-ignoring child running.
+    sentinel = tmp_path / "survived"
+    backend = LocalBackend(terminate_grace=0.3)
+    call = asyncio.create_task(
+        backend.exec([sys.executable, "-c", _sleeper(sentinel, ignore_sigterm=True)])
+    )
+    await asyncio.sleep(0.1)
+
+    call.cancel()
+    await asyncio.sleep(0.05)
+    call.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await call
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()

From 4106447b5e29674f6d4d650eb49c138e954e3bd4 Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Mon, 7 Sep 2026 16:57:19 -0600
Subject: [PATCH 5/6] fix(py): terminate the worker on every abnormal exit from
 exec

Review of the exec-backend seam found the shutdown guarantee covered
only TimeoutError and CancelledError:

- Input that fails to encode (e.g. unpaired surrogates) raised after
  spawn, and any unexpected error from _collect (a pipe failing
  mid-read) propagated, each leaving the child running unsupervised.
  The stdin write now sits inside the try, and a catch-all routes every
  abnormal exit through shutdown.
- The post-timeout shutdown awaited _terminate directly, so a
  cancellation landing mid-escalation could skip SIGKILL. Both branches
  now use the detached, shielded shutdown task the cancellation path
  already had.
- env=None inherited the whole parent environment, credentials and all;
  with the payload being model-written code the default now fails
  closed with an empty environment.

Also pins the _terminate grace bound in its test (the old 2s outer
limit could not catch the waits ignoring grace) and covers stderr's
truncation flag independently of stdout's.
---
 pkg-py/src/commons/_execution/_backend.py |  54 +++++++----
 pkg-py/tests/test_execution_backend.py    | 108 +++++++++++++++++++++-
 2 files changed, 141 insertions(+), 21 deletions(-)

diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py
index 966bd076..f812e408 100644
--- a/pkg-py/src/commons/_execution/_backend.py
+++ b/pkg-py/src/commons/_execution/_backend.py
@@ -56,6 +56,11 @@ async def exec(
     ) -> ExecResult:
         """Run ``cmd``, feeding ``input`` on stdin, and collect its output.
 
+        ``env`` replaces the parent's environment outright rather than
+        extending it, and omitting it gives the child an empty one: the
+        parent holds credentials the child has no business seeing, so the
+        default fails closed.
+
         Raises ``ExecTimeoutError`` if ``timeout`` passes before the command
         finishes, having first made sure the process is gone.
         """
@@ -130,33 +135,26 @@ async def exec(
             stdout=asyncio.subprocess.PIPE,
             stderr=asyncio.subprocess.PIPE,
             cwd=cwd,
-            env=None if env is None else dict(env),
+            env={} if env is None else dict(env),
         )
-        if process.stdin is not None:
-            if input is not None:
-                process.stdin.write(input.encode())
-            process.stdin.close()
         try:
+            if process.stdin is not None:
+                if input is not None:
+                    process.stdin.write(input.encode())
+                process.stdin.close()
             stdout, stderr = await asyncio.wait_for(self._collect(process), timeout)
         except TimeoutError:
-            await _terminate(process, self._terminate_grace)
+            await self._shutdown(process)
             raise ExecTimeoutError(
                 f"the command exceeded its {timeout}-second time limit"
             ) from None
-        except asyncio.CancelledError:
-            # Whoever started the process ends it. A cancelled call that left
-            # the worker running would keep holding the parent's file
-            # descriptors and go on burning CPU with nobody waiting on it.
-            #
-            # Shutdown runs in its own task so that a second cancellation
-            # stops us waiting on it without stopping the escalation itself;
-            # a caller cancelling twice must not be able to leave a
-            # SIGTERM-ignoring child alive.
-            shutdown = asyncio.ensure_future(_terminate(process, self._terminate_grace))
-            self._shutdowns.add(shutdown)
-            shutdown.add_done_callback(self._shutdowns.discard)
-            with contextlib.suppress(asyncio.CancelledError):
-                await asyncio.shield(shutdown)
+        except BaseException:
+            # Whoever started the process ends it. Timeout and cancellation
+            # are not the only ways out of a call: input can fail to encode,
+            # a pipe can fail mid-read. Any exit that left the worker running
+            # would keep holding the parent's file descriptors and go on
+            # burning CPU with nobody waiting on it.
+            await self._shutdown(process)
             raise
         return ExecResult(
             returncode=process.returncode or 0,
@@ -166,6 +164,22 @@ async def exec(
             stderr_truncated=stderr[1],
         )
 
+    async def _shutdown(self, process: asyncio.subprocess.Process) -> None:
+        """Terminate ``process``, outliving cancellation of the caller.
+
+        Shutdown runs in its own task so that a cancellation landing while it
+        is in flight — a caller cancelling twice, or cancelling while the
+        post-timeout escalation is still waiting — stops us waiting on it
+        without stopping the escalation itself. It must not be possible to
+        leave a SIGTERM-ignoring child alive by cancelling at the wrong
+        moment.
+        """
+        shutdown = asyncio.ensure_future(_terminate(process, self._terminate_grace))
+        self._shutdowns.add(shutdown)
+        shutdown.add_done_callback(self._shutdowns.discard)
+        with contextlib.suppress(asyncio.CancelledError):
+            await asyncio.shield(shutdown)
+
 
 async def _terminate(process: asyncio.subprocess.Process, grace: float) -> None:
     """Ask the process to exit, then insist."""
diff --git a/pkg-py/tests/test_execution_backend.py b/pkg-py/tests/test_execution_backend.py
index d7946105..5f9746ac 100644
--- a/pkg-py/tests/test_execution_backend.py
+++ b/pkg-py/tests/test_execution_backend.py
@@ -14,6 +14,7 @@
 
 import pytest
 
+from commons._execution import _backend as backend_module
 from commons._execution._backend import (
     ExecBackend,
     ExecTimeoutError,
@@ -179,7 +180,10 @@ async def wait(self) -> int:
 async def test_terminate_gives_up_when_the_exit_is_never_reaped() -> None:
     process = _NeverReaped()
 
-    await asyncio.wait_for(_terminate(cast(Any, process), 0.05), timeout=2)
+    # The bound is two grace periods (SIGTERM, then SIGKILL), so 0.5s leaves
+    # generous margin over the real 0.1s while still failing if the waits
+    # stop honouring the grace they were given.
+    await asyncio.wait_for(_terminate(cast(Any, process), 0.05), timeout=0.5)
 
     assert process.signals == ["term", "kill"]
 
@@ -295,3 +299,105 @@ async def test_a_second_cancellation_cannot_abort_the_shutdown(tmp_path) -> None
 
     await asyncio.sleep(0.8)
     assert not sentinel.exists()
+
+
+async def test_omitting_the_environment_gives_the_child_an_empty_one(
+    monkeypatch,
+) -> None:
+    # The parent holds credentials a child has no business seeing, so with no
+    # allowlist in hand the default has to fail closed: no environment at
+    # all, rather than the whole of the parent's.
+    monkeypatch.setenv("COMMONS_TEST_SECRET", "sk-not-a-real-key")
+    backend = LocalBackend()
+
+    result = await backend.exec(
+        [sys.executable, "-c", "import os; print(sorted(os.environ))"]
+    )
+
+    assert "COMMONS_TEST_SECRET" not in result.stdout
+
+
+async def test_stderr_past_the_cap_keeps_the_tail_and_sets_its_own_flag() -> None:
+    # The two streams are capped and flagged independently; a flag wired to
+    # the wrong stream is invisible if only stdout is ever exercised.
+    backend = LocalBackend(output_limit=2000)
+    code = (
+        "import sys\n"
+        "for i in range(200):\n"
+        "    sys.stderr.write(f'line-{i}:' + 'x' * 1000 + '\\n')\n"
+    )
+
+    result = await backend.exec([sys.executable, "-c", code])
+
+    assert result.returncode == 0
+    assert result.stdout == ""
+    assert not result.stdout_truncated
+    assert result.stderr_truncated
+    assert len(result.stderr) <= 2000
+    assert "line-199:" in result.stderr
+    assert "line-0:" not in result.stderr
+
+
+async def test_an_unexpected_error_mid_call_still_kills_the_process(
+    tmp_path, monkeypatch
+) -> None:
+    # Timeout and cancellation are not the only ways out of a call. A pipe
+    # failing mid-read (or any other surprise) must not leave the worker
+    # running with nobody waiting on it.
+    async def fail_read(
+        stream: asyncio.StreamReader | None, limit: int
+    ) -> tuple[bytes, bool]:
+        raise RuntimeError("pipe failed mid-read")
+
+    monkeypatch.setattr(backend_module, "_read_tail", fail_read)
+    sentinel = tmp_path / "survived"
+    backend = LocalBackend(terminate_grace=0.1)
+
+    with pytest.raises(RuntimeError, match="pipe failed"):
+        await backend.exec([sys.executable, "-c", _sleeper(sentinel)])
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()
+
+
+async def test_input_that_cannot_be_encoded_still_kills_the_process(
+    tmp_path,
+) -> None:
+    # Model-written text can contain unpaired surrogates, which fail at
+    # encode time — after the child has already been spawned.
+    sentinel = tmp_path / "survived"
+    backend = LocalBackend(terminate_grace=0.1)
+
+    with pytest.raises(UnicodeEncodeError):
+        await backend.exec(
+            [sys.executable, "-c", _sleeper(sentinel)], input="\ud800"
+        )
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()
+
+
+async def test_cancellation_during_the_timeout_shutdown_cannot_abort_it(
+    tmp_path,
+) -> None:
+    # A cancel landing while the post-timeout shutdown is in flight must not
+    # skip SIGKILL. The shutdown runs in its own task precisely so that it
+    # outlives the call that started it.
+    sentinel = tmp_path / "survived"
+    backend = LocalBackend(terminate_grace=0.3)
+    call = asyncio.create_task(
+        backend.exec(
+            [sys.executable, "-c", _sleeper(sentinel, ignore_sigterm=True)],
+            timeout=0.1,
+        )
+    )
+    # The timeout fires at ~0.1s and the SIGTERM grace then runs for 0.3s,
+    # so a cancel at 0.3s lands in the middle of the shutdown.
+    await asyncio.sleep(0.3)
+
+    call.cancel()
+    with pytest.raises((asyncio.CancelledError, ExecTimeoutError)):
+        await call
+
+    await asyncio.sleep(0.8)
+    assert not sentinel.exists()

From 7a72c0c12d93c5f9ac658be0ec1e36e370e5b6af Mon Sep 17 00:00:00 2001
From: Josh Taillon <joshua.taillon@posit.co>
Date: Mon, 7 Sep 2026 16:59:54 -0600
Subject: [PATCH 6/6] refactor(py): tighten the exec backend's contracts and
 test margins

Follow-up to review, all low-severity:

- Assert the return code is known rather than coalescing None to 0 with
  `or 0`, so a broken _collect invariant fails loudly instead of
  reporting success.
- Make ExecResult keyword-only and LocalBackend's settings
  keyword-only, before anything consumes either.
- Write the encoding contract (UTF-8 in, lossy decode out) and the
  spawn-failure behaviour (the underlying OSError propagates) into the
  ExecBackend docstring, where a second backend implementation will
  look for them; pin the spawn failure with a test.
- Document LocalBackend's output_limit and terminate_grace parameters.
- Widen the sentinel-test sleeper from 0.6s to 1.5s (and the post-check
  waits to match) so a loaded machine delays the kill into slack rather
  than into a false failure.
---
 pkg-py/src/commons/_execution/_backend.py | 24 ++++++++++++--
 pkg-py/tests/test_execution_backend.py    | 38 ++++++++++++++++-------
 2 files changed, 47 insertions(+), 15 deletions(-)

diff --git a/pkg-py/src/commons/_execution/_backend.py b/pkg-py/src/commons/_execution/_backend.py
index f812e408..047b8f6e 100644
--- a/pkg-py/src/commons/_execution/_backend.py
+++ b/pkg-py/src/commons/_execution/_backend.py
@@ -28,7 +28,7 @@ class ExecTimeoutError(TimeoutError):
     """The command ran past its deadline and was killed."""
 
 
-@dataclass(frozen=True)
+@dataclass(frozen=True, kw_only=True)
 class ExecResult:
     returncode: int
     stdout: str
@@ -61,8 +61,14 @@ async def exec(
         parent holds credentials the child has no business seeing, so the
         default fails closed.
 
+        ``input`` is encoded as UTF-8, and output is decoded as UTF-8 with
+        invalid bytes replaced. Both are the contract every backend
+        implements, not a local choice.
+
         Raises ``ExecTimeoutError`` if ``timeout`` passes before the command
-        finishes, having first made sure the process is gone.
+        finishes, having first made sure the process is gone. A command that
+        cannot be started at all raises the underlying ``OSError`` (usually
+        ``FileNotFoundError``) instead.
         """
         ...
 
@@ -95,9 +101,18 @@ class LocalBackend:
 
     def __init__(
         self,
+        *,
         output_limit: int = DEFAULT_OUTPUT_LIMIT,
         terminate_grace: float = TERMINATE_GRACE,
     ) -> None:
+        """Configure output retention and shutdown patience.
+
+        ``output_limit`` caps how many bytes are kept from each of stdout
+        and stderr; past the cap the oldest bytes are dropped, keeping the
+        tail. ``terminate_grace`` is how long to wait for SIGTERM to be
+        honoured before escalating to SIGKILL, and again for the exit to be
+        observed afterwards.
+        """
         self._output_limit = output_limit
         self._terminate_grace = terminate_grace
         # Shutdowns outlive the call that started them, so they need an owner
@@ -156,8 +171,11 @@ async def exec(
             # burning CPU with nobody waiting on it.
             await self._shutdown(process)
             raise
+        # _collect awaited wait(), so the return code is known here; if that
+        # invariant ever breaks, fail loudly rather than report "succeeded".
+        assert process.returncode is not None
         return ExecResult(
-            returncode=process.returncode or 0,
+            returncode=process.returncode,
             stdout=stdout[0].decode(errors="replace"),
             stderr=stderr[0].decode(errors="replace"),
             stdout_truncated=stdout[1],
diff --git a/pkg-py/tests/test_execution_backend.py b/pkg-py/tests/test_execution_backend.py
index 5f9746ac..bef9a1c2 100644
--- a/pkg-py/tests/test_execution_backend.py
+++ b/pkg-py/tests/test_execution_backend.py
@@ -100,13 +100,18 @@ async def test_output_past_the_cap_keeps_the_tail_and_the_process_still_finishes
 
 
 def _sleeper(sentinel: object, *, ignore_sigterm: bool = False) -> str:
-    """Code that outlives its timeout and records the fact if it is allowed to."""
+    """Code that outlives its timeout and records the fact if it is allowed to.
+
+    The sleep sits well past the point where a working shutdown has killed
+    the process, so a slow or loaded machine delays the kill into slack
+    rather than into a false failure.
+    """
     guard = (
         "import signal; signal.signal(signal.SIGTERM, signal.SIG_IGN)\n"
         if ignore_sigterm
         else ""
     )
-    return f"{guard}import time; time.sleep(0.6); open({str(sentinel)!r}, 'w').close()"
+    return f"{guard}import time; time.sleep(1.5); open({str(sentinel)!r}, 'w').close()"
 
 
 async def test_a_call_past_the_timeout_raises_and_the_process_does_not_survive(
@@ -118,7 +123,7 @@ async def test_a_call_past_the_timeout_raises_and_the_process_does_not_survive(
     with pytest.raises(ExecTimeoutError):
         await backend.exec([sys.executable, "-c", _sleeper(sentinel)], timeout=0.15)
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()
 
 
@@ -149,7 +154,7 @@ async def test_a_process_that_ignores_sigterm_is_killed_anyway(tmp_path) -> None
             timeout=0.15,
         )
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()
 
 
@@ -233,14 +238,14 @@ async def test_the_timeout_still_applies_after_the_output_streams_close(
     code = (
         "import os, time\n"
         "os.close(1); os.close(2)\n"
-        f"time.sleep(0.6); open({str(sentinel)!r}, 'w').close()\n"
+        f"time.sleep(1.5); open({str(sentinel)!r}, 'w').close()\n"
     )
     backend = LocalBackend()
 
     with pytest.raises(ExecTimeoutError):
         await backend.exec([sys.executable, "-c", code], timeout=0.15)
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()
 
 
@@ -256,10 +261,19 @@ async def test_cancelling_a_call_does_not_leave_the_process_running(tmp_path) ->
     with pytest.raises(asyncio.CancelledError):
         await call
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()
 
 
+async def test_a_command_that_cannot_be_started_raises_os_error() -> None:
+    # Nothing was spawned, so there is nothing to clean up: the failure
+    # propagates as-is rather than being dressed up as an exec result.
+    backend = LocalBackend()
+
+    with pytest.raises(FileNotFoundError):
+        await backend.exec(["/no/such/binary"])
+
+
 async def test_cancellation_still_escalates_for_a_process_ignoring_sigterm(
     tmp_path,
 ) -> None:
@@ -276,7 +290,7 @@ async def test_cancellation_still_escalates_for_a_process_ignoring_sigterm(
     with pytest.raises(asyncio.CancelledError):
         await call
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()
 
 
@@ -297,7 +311,7 @@ async def test_a_second_cancellation_cannot_abort_the_shutdown(tmp_path) -> None
     with pytest.raises(asyncio.CancelledError):
         await call
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()
 
 
@@ -356,7 +370,7 @@ async def fail_read(
     with pytest.raises(RuntimeError, match="pipe failed"):
         await backend.exec([sys.executable, "-c", _sleeper(sentinel)])
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()
 
 
@@ -373,7 +387,7 @@ async def test_input_that_cannot_be_encoded_still_kills_the_process(
             [sys.executable, "-c", _sleeper(sentinel)], input="\ud800"
         )
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()
 
 
@@ -399,5 +413,5 @@ async def test_cancellation_during_the_timeout_shutdown_cannot_abort_it(
     with pytest.raises((asyncio.CancelledError, ExecTimeoutError)):
         await call
 
-    await asyncio.sleep(0.8)
+    await asyncio.sleep(1.8)
     assert not sentinel.exists()