From 42c81dbd0467c672120a6e9457b1b514041da35d Mon Sep 17 00:00:00 2001
From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com>
Date: Fri, 9 Oct 2026 10:49:14 -0400
Subject: [PATCH 1/8] fix: bound post-kill output drain, typed errors on
 undecodable stdout, and more

- Timeout: after killing the process group, collect output for at most
  kill_grace seconds. A descendant that left the group (setsid) could
  otherwise hold the pipes open and stall the timeout indefinitely.
- _finish: raise the typed QsvError before decoding stdout, so a failed
  run with non-UTF-8 stdout no longer surfaces as UnicodeDecodeError.
- count()/headers(): read output via QsvResult._text(), so they work
  with stdout_path= and text=False instead of raising IndexError.
- find_qsv: return absolute paths, so a relative binary works with cwd=.
- Document kill_grace in the client docstring.

Each fix has a regression test that fails on the previous code.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md               | 11 +++++
 src/qsv_client/_process.py | 16 ++++++-
 src/qsv_client/client.py   | 33 +++++++-------
 tests/test_process.py      | 91 +++++++++++++++++++++++++++++++++++++-
 4 files changed, 133 insertions(+), 18 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 527f632..aa43532 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,17 @@ All notable changes to this project are documented here. The format follows
   included repository tooling files (`CLAUDE.md`, `.serena/`, `uv.lock`, `.github/`). The wheel
   is unchanged. ([#2](https://github.com/dathere/qsv-client/pull/2))
 
+### Fixed
+
+- A timeout now raises within about `timeout + 2 * kill_grace` even when a process qsv started
+  has left the process group and still holds its output pipes. Previously the run waited for
+  that process to exit. Output it still holds is dropped.
+- A failed run whose stdout is not valid UTF-8 raises the typed `QsvError` instead of
+  `UnicodeDecodeError`.
+- `count()` and `headers()` work with `stdout_path=` and `text=False` instead of raising
+  `IndexError`.
+- A relative `binary` path now works together with `cwd=`. `find_qsv` returns absolute paths.
+
 ## [0.1.0] - 2026-10-08
 
 First release. ([#1](https://github.com/dathere/qsv-client/pull/1))
diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py
index d6a3bd3..6a84390 100644
--- a/src/qsv_client/_process.py
+++ b/src/qsv_client/_process.py
@@ -5,6 +5,10 @@
 run gets its own session (process group) and the whole group is signalled: SIGTERM first,
 then SIGKILL after a grace period. On Windows the process is started in a new process group
 and killed directly; grandchildren are not reaped there.
+
+A descendant that leaves the group (e.g. by calling ``setsid``) survives the kill and may keep
+the output pipes open; after a timeout, output is collected for at most ``kill_grace`` seconds
+and anything that descendant still holds is dropped.
 """
 
 from __future__ import annotations
@@ -78,7 +82,15 @@ def run_sync(
     except subprocess.TimeoutExpired:
         timed_out = True
         _terminate_sync(proc, kill_grace)
-        out, err = proc.communicate()
+        try:
+            out, err = proc.communicate(timeout=kill_grace)
+        except subprocess.TimeoutExpired:
+            # a descendant that left the group still holds the pipes: drop its output
+            for pipe in (proc.stdout, proc.stderr):
+                if pipe is not None:
+                    pipe.close()
+            proc.wait()
+            out, err = b"", b""
     except BaseException:
         # KeyboardInterrupt etc.: never leave a qsv process group running behind us
         _terminate_sync(proc, kill_grace)
@@ -133,7 +145,7 @@ async def run_async(
         out, err = b"", b""
         if proc.stderr is not None:
             with contextlib.suppress(Exception):
-                err = await proc.stderr.read()
+                err = await asyncio.wait_for(proc.stderr.read(), kill_grace)
     except BaseException:
         # task cancelled (or the loop is shutting down): reap the group, then propagate
         await asyncio.shield(_terminate_async(proc, kill_grace))
diff --git a/src/qsv_client/client.py b/src/qsv_client/client.py
index 79f2991..a085d45 100644
--- a/src/qsv_client/client.py
+++ b/src/qsv_client/client.py
@@ -43,12 +43,13 @@ def find_qsv(binary: str | os.PathLike[str] | None = None) -> str:
         candidates.extend(DEFAULT_BINARIES)
     for cand in candidates:
         if os.sep in cand or (os.altsep and os.altsep in cand):
+            # absolute, so a relative path still resolves when the run sets cwd=
             if Path(cand).is_file():
-                return str(Path(cand))
+                return os.path.abspath(cand)
             continue
         found = shutil.which(cand)
         if found:
-            return found
+            return os.path.abspath(found)
     raise QsvNotFound(
         f"no qsv binary found (tried {', '.join(candidates)}). Install qsv "
         "(https://github.com/dathere/qsv/releases), put it on PATH, or pass binary= / set QSV_BIN."
@@ -122,6 +123,8 @@ def __init__(
                 ``QSV_LLM_APIKEY`` / ``QSV_LLM_BASE_URL`` / ``QSV_LLM_MODEL`` environment
                 variables, so the key never appears in the process table.
             min_version: raise :class:`QsvVersionError` on first use if the binary is older.
+            kill_grace: seconds to wait after SIGTERM before SIGKILL on timeout or
+                cancellation, and the most spent collecting output after the kill.
         """
         self.binary = find_qsv(binary)
         self.timeout = timeout
@@ -187,8 +190,16 @@ def _finish(
                 args=argv,
                 stderr=stderr,
             )
+        # before decoding stdout, so undecodable output can't mask the typed error
+        if check and raw.exit_code not in SUCCESS_EXIT_CODES:
+            raise error_from_run(
+                exit_code=raw.exit_code,
+                stderr=stderr,
+                args=argv,
+                command=command or None,
+            )
         captured = stdout_path is None
-        result = QsvResult(
+        return QsvResult(
             args=tuple(argv),
             exit_code=raw.exit_code,
             stdout=raw.stdout.decode("utf-8") if captured and text else None,
@@ -197,14 +208,6 @@ def _finish(
             duration=duration,
             stdout_path=stdout_path,
         )
-        if check and not result.ok:
-            raise error_from_run(
-                exit_code=raw.exit_code,
-                stderr=stderr,
-                args=argv,
-                command=command or None,
-            )
-        return result
 
 
 def _open_stdout(
@@ -349,12 +352,12 @@ def _check_version_once(self) -> None:
 
     def count(self, path: Arg, *args: Arg, **kw: Any) -> int:
         """Number of records (``qsv count``)."""
-        out = self.run("count", path, *args, **kw).stdout or ""
+        out = self.run("count", path, *args, **kw)._text()
         return int(out.strip().split()[0].split(";")[0].replace(",", ""))
 
     def headers(self, path: Arg, *args: Arg, **kw: Any) -> list[str]:
         """Column names (``qsv headers --just-names``)."""
-        out = self.run("headers", "--just-names", path, *args, **kw).stdout or ""
+        out = self.run("headers", "--just-names", path, *args, **kw)._text()
         return [line for line in out.splitlines() if line]
 
     def stats(self, path: Arg, *args: Arg, **kw: Any) -> list[dict[str, str]]:
@@ -474,11 +477,11 @@ async def _check_version_once(self) -> None:
             await self.capabilities()
 
     async def count(self, path: Arg, *args: Arg, **kw: Any) -> int:
-        out = (await self.run("count", path, *args, **kw)).stdout or ""
+        out = (await self.run("count", path, *args, **kw))._text()
         return int(out.strip().split()[0].split(";")[0].replace(",", ""))
 
     async def headers(self, path: Arg, *args: Arg, **kw: Any) -> list[str]:
-        out = (await self.run("headers", "--just-names", path, *args, **kw)).stdout or ""
+        out = (await self.run("headers", "--just-names", path, *args, **kw))._text()
         return [line for line in out.splitlines() if line]
 
     async def stats(self, path: Arg, *args: Arg, **kw: Any) -> list[dict[str, str]]:
diff --git a/tests/test_process.py b/tests/test_process.py
index dab3f15..5ffccde 100644
--- a/tests/test_process.py
+++ b/tests/test_process.py
@@ -3,14 +3,17 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import os
+import signal
 import time
 from collections.abc import Callable
 from pathlib import Path
+from typing import Any
 
 import pytest
 
-from qsv_client import AsyncQsv, Qsv, QsvTimeout
+from qsv_client import AsyncQsv, Qsv, QsvTimeout, QsvUsageError
 
 # The fake starts a grandchild that inherits stdout and records its pid, then hangs. If only
 # the fake were killed, the grandchild would keep the stdout pipe open and the run would hang
@@ -142,3 +145,89 @@ def test_warning_exit_code_is_success(fake_qsv: Callable[[str], str]) -> None:
     res = Qsv(fake_qsv("import sys; print('done'); sys.exit(255)")).run("x")
     assert res.ok
     assert res.exit_code == 255
+
+
+# Like HANG_WITH_GRANDCHILD, but the grandchild leaves the process group, so the group kill
+# misses it and it keeps the output pipes open.
+HANG_WITH_ESCAPED_GRANDCHILD = """
+import subprocess, sys, time
+child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"],
+                         start_new_session=True)
+open(sys.argv[-1], "w").write(str(child.pid))
+time.sleep(60)
+"""
+
+
+def _kill(pid: int) -> None:
+    with contextlib.suppress(ProcessLookupError):
+        os.kill(pid, signal.SIGKILL)
+
+
+def test_timeout_bounded_when_grandchild_escapes_group(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    pidfile = tmp_path / "child.pid"
+    qsv = Qsv(fake_qsv(HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.5)
+    started = time.monotonic()
+    try:
+        with pytest.raises(QsvTimeout):
+            qsv.run("stats", pidfile, timeout=1)
+        assert time.monotonic() - started < 10
+    finally:
+        _kill(_read_pid(pidfile))
+
+
+async def test_async_timeout_bounded_when_grandchild_escapes_group(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    pidfile = tmp_path / "child.pid"
+    qsv = AsyncQsv(fake_qsv(HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.5)
+    started = time.monotonic()
+    try:
+        with pytest.raises(QsvTimeout):
+            await qsv.run("stats", pidfile, timeout=1)
+        assert time.monotonic() - started < 10
+    finally:
+        _kill(_read_pid(pidfile))
+
+
+def test_undecodable_stdout_still_raises_typed_error(fake_qsv: Callable[[str], str]) -> None:
+    qsv = Qsv(fake_qsv("import sys; sys.stdout.buffer.write(b'\\xff'); sys.exit(2)"))
+    with pytest.raises(QsvUsageError):
+        qsv.run("x")
+
+
+COUNT_AND_HEADERS = """
+import sys
+print("3" if sys.argv[1] == "count" else "name\\nn")
+"""
+
+
+def test_count_and_headers_without_captured_text(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    qsv = Qsv(fake_qsv(COUNT_AND_HEADERS))
+    kws: list[dict[str, Any]] = [{"stdout_path": tmp_path / "out"}, {"text": False}]
+    for kw in kws:
+        assert qsv.count("in.csv", **kw) == 3
+        assert qsv.headers("in.csv", **kw) == ["name", "n"]
+
+
+async def test_async_count_and_headers_without_captured_text(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    qsv = AsyncQsv(fake_qsv(COUNT_AND_HEADERS))
+    kws: list[dict[str, Any]] = [{"stdout_path": tmp_path / "out"}, {"text": False}]
+    for kw in kws:
+        assert await qsv.count("in.csv", **kw) == 3
+        assert await qsv.headers("in.csv", **kw) == ["name", "n"]
+
+
+def test_relative_binary_with_cwd(
+    fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    fake = Path(fake_qsv("print('ok')"))
+    other = tmp_path / "other"
+    other.mkdir()
+    monkeypatch.chdir(fake.parent)
+    assert Qsv(f"./{fake.name}", cwd=other).run("x").stdout == "ok\n"

From 88b7b2322cc0c3479b371a4d4de173f3df15b0d6 Mon Sep 17 00:00:00 2001
From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com>
Date: Fri, 9 Oct 2026 10:56:49 -0400
Subject: [PATCH 2/8] fix: bound async teardown waits, release held pipes, keep
 `..` in binary paths

Addresses roborev review 5154:

- AsyncQsv: proc.wait() also waits for the output pipes to close, so with
  a SIGTERM-ignoring leader and a descendant that left the group, the
  post-SIGKILL wait blocked until that descendant exited. Every teardown
  wait is now bounded by kill_grace.
- AsyncQsv: after a timeout or cancellation, close the subprocess
  transport so pipes held by an escaped descendant no longer keep two fds
  open per run until it exits, then reap the killed leader.
- find_qsv: use Path.absolute() instead of os.path.abspath, which
  collapsed `..` lexically and pointed at the wrong file after a
  symlinked directory.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md               |  7 +++---
 src/qsv_client/_process.py | 23 ++++++++++++++++++-
 src/qsv_client/client.py   |  7 +++---
 tests/test_process.py      | 45 ++++++++++++++++++++++++++++++++++++++
 4 files changed, 75 insertions(+), 7 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index aa43532..76450f1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,9 +14,10 @@ All notable changes to this project are documented here. The format follows
 
 ### Fixed
 
-- A timeout now raises within about `timeout + 2 * kill_grace` even when a process qsv started
-  has left the process group and still holds its output pipes. Previously the run waited for
-  that process to exit. Output it still holds is dropped.
+- A timeout now raises within a few `kill_grace` periods even when a process qsv started has
+  left the process group and still holds its output pipes, including when qsv ignores SIGTERM.
+  Previously the run waited for that process to exit. Output it still holds is dropped, and
+  `AsyncQsv` closes its pipe file descriptors instead of keeping them open until it exits.
 - A failed run whose stdout is not valid UTF-8 raises the typed `QsvError` instead of
   `UnicodeDecodeError`.
 - `count()` and `headers()` work with `stdout_path=` and `text=False` instead of raising
diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py
index 6a84390..fdd6453 100644
--- a/src/qsv_client/_process.py
+++ b/src/qsv_client/_process.py
@@ -110,13 +110,33 @@ async def _terminate_async(proc: asyncio.subprocess.Process, grace: float) -> No
             proc.kill()
         await proc.wait()
         return
+    # proc.wait() also waits for the output pipes to close, which a descendant that left the
+    # group can hold open indefinitely, so every wait here is bounded
     _signal_group(proc.pid, signal.SIGTERM)
     with contextlib.suppress(asyncio.TimeoutError):
         await asyncio.wait_for(proc.wait(), grace)
     _signal_group(proc.pid, signal.SIGKILL)  # type: ignore[attr-defined,unused-ignore]
+    with contextlib.suppress(asyncio.TimeoutError):
+        await asyncio.wait_for(proc.wait(), grace)
+
+
+async def _release_async(proc: asyncio.subprocess.Process) -> None:
+    """Close our ends of the pipes, then reap the (already killed) leader.
+
+    Without the close, pipes held by an escaped descendant would keep their fds open and
+    ``wait()`` pending until it exits. asyncio has no public API for this.
+    """
+    transport = getattr(proc, "_transport", None)
+    if transport is not None:
+        transport.close()
     await proc.wait()
 
 
+async def _reap_async(proc: asyncio.subprocess.Process, grace: float) -> None:
+    await _terminate_async(proc, grace)
+    await _release_async(proc)
+
+
 async def run_async(
     argv: Sequence[str],
     *,
@@ -146,9 +166,10 @@ async def run_async(
         if proc.stderr is not None:
             with contextlib.suppress(Exception):
                 err = await asyncio.wait_for(proc.stderr.read(), kill_grace)
+        await _release_async(proc)
     except BaseException:
         # task cancelled (or the loop is shutting down): reap the group, then propagate
-        await asyncio.shield(_terminate_async(proc, kill_grace))
+        await asyncio.shield(_reap_async(proc, kill_grace))
         raise
     assert proc.returncode is not None
     return RawResult(
diff --git a/src/qsv_client/client.py b/src/qsv_client/client.py
index a085d45..9a8d64d 100644
--- a/src/qsv_client/client.py
+++ b/src/qsv_client/client.py
@@ -43,13 +43,14 @@ def find_qsv(binary: str | os.PathLike[str] | None = None) -> str:
         candidates.extend(DEFAULT_BINARIES)
     for cand in candidates:
         if os.sep in cand or (os.altsep and os.altsep in cand):
-            # absolute, so a relative path still resolves when the run sets cwd=
+            # absolute, so a relative path still resolves when the run sets cwd=. Not
+            # normalized: collapsing `..` lexically would be wrong after a symlinked dir.
             if Path(cand).is_file():
-                return os.path.abspath(cand)
+                return str(Path(cand).absolute())
             continue
         found = shutil.which(cand)
         if found:
-            return os.path.abspath(found)
+            return str(Path(found).absolute())
     raise QsvNotFound(
         f"no qsv binary found (tried {', '.join(candidates)}). Install qsv "
         "(https://github.com/dathere/qsv/releases), put it on PATH, or pass binary= / set QSV_BIN."
diff --git a/tests/test_process.py b/tests/test_process.py
index 5ffccde..74ce018 100644
--- a/tests/test_process.py
+++ b/tests/test_process.py
@@ -231,3 +231,48 @@ def test_relative_binary_with_cwd(
     other.mkdir()
     monkeypatch.chdir(fake.parent)
     assert Qsv(f"./{fake.name}", cwd=other).run("x").stdout == "ok\n"
+
+
+def test_relative_binary_through_symlinked_dir(
+    fake_qsv: Callable[[str], str], tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # work/link -> bin, so work/link/../fake-qsv is tmp_path/fake-qsv, not work/fake-qsv
+    fake_qsv("print('ok')")
+    (tmp_path / "bin").mkdir()
+    work = tmp_path / "work"
+    work.mkdir()
+    (work / "link").symlink_to(tmp_path / "bin")
+    monkeypatch.chdir(work)
+    assert Qsv("link/../fake-qsv").run("x").stdout == "ok\n"
+
+
+async def test_async_timeout_bounded_when_leader_ignores_sigterm_and_grandchild_escapes(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    pidfile = tmp_path / "child.pid"
+    body = "import signal; signal.signal(signal.SIGTERM, signal.SIG_IGN)\n"
+    qsv = AsyncQsv(fake_qsv(body + HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.5)
+    started = time.monotonic()
+    try:
+        with pytest.raises(QsvTimeout):
+            await qsv.run("stats", pidfile, timeout=1)
+        assert time.monotonic() - started < 10
+    finally:
+        _kill(_read_pid(pidfile))
+
+
+async def test_async_timeout_releases_pipes_held_by_escaped_grandchild(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    qsv = AsyncQsv(fake_qsv(HANG_WITH_ESCAPED_GRANDCHILD), kill_grace=0.3)
+    pidfiles = [tmp_path / f"child{i}.pid" for i in range(3)]
+    open_fds = len(os.listdir("/dev/fd"))
+    try:
+        for pidfile in pidfiles:
+            with pytest.raises(QsvTimeout):
+                await qsv.run("stats", pidfile, timeout=0.5)
+        await asyncio.sleep(0)  # asyncio closes pipe fds in a call_soon callback
+        assert len(os.listdir("/dev/fd")) <= open_fds
+    finally:
+        for pidfile in pidfiles:
+            _kill(_read_pid(pidfile))

From 86fadbee556a0fbb1cf2f8dd0e4d3c60339f633b Mon Sep 17 00:00:00 2001
From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com>
Date: Fri, 9 Oct 2026 11:01:47 -0400
Subject: [PATCH 3/8] ci: test on Python 3.14

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .github/workflows/ci.yml       | 2 +-
 .serena/memories/tech_stack.md | 2 +-
 CLAUDE.md                      | 2 +-
 3 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index d68abdf..ed1dec9 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -24,7 +24,7 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest, macos-latest]
-        python: ["3.10", "3.11", "3.12", "3.13"]
+        python: ["3.10", "3.11", "3.12", "3.13", "3.14"]
     runs-on: ${{ matrix.os }}
     steps:
       - uses: actions/checkout@v4
diff --git a/.serena/memories/tech_stack.md b/.serena/memories/tech_stack.md
index fb13e4c..9d1836e 100644
--- a/.serena/memories/tech_stack.md
+++ b/.serena/memories/tech_stack.md
@@ -1,6 +1,6 @@
 # Tech stack
 
-- Python >=3.10 (ruff `target-version = py310`; CI matrix 3.10–3.13 on ubuntu + macos). Use `from __future__ import annotations`; no 3.11+ only stdlib features.
+- Python >=3.10 (ruff `target-version = py310`; CI matrix 3.10–3.14 on ubuntu + macos). Use `from __future__ import annotations`; no 3.11+ only stdlib features.
 - No runtime dependencies — keep it that way (stdlib `subprocess`/`asyncio` only).
 - Package/env manager: uv (`uv.lock` committed); dev deps in `[dependency-groups].dev`: pytest, pytest-asyncio, mypy, ruff.
 - Build: hatchling, src layout (`src/qsv_client`). License AGPL-3.0-or-later.
diff --git a/CLAUDE.md b/CLAUDE.md
index 04b8b74..67ac574 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -13,7 +13,7 @@ QSV_BIN=/path/to/qsvlite uv run pytest -q tests/test_integration.py   # other bi
 uv run ruff format . && uv run ruff check . && uv run mypy            # must be clean before done
 ```
 
-CI runs lint, format check, and mypy, plus tests on ubuntu/macos x Python 3.10–3.13 against
+CI runs lint, format check, and mypy, plus tests on ubuntu/macos x Python 3.10–3.14 against
 qsv, qsvlite, and qsvdp (Linux only).
 
 ## Layout (`src/qsv_client/`)

From 5835da2f09d227b2197375c55ded24d26f115511 Mon Sep 17 00:00:00 2001
From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com>
Date: Fri, 9 Oct 2026 11:04:15 -0400
Subject: [PATCH 4/8] fix: shield async timeout cleanup from cancellation;
 bound Windows wait

Addresses roborev review 5155:

- A cancellation arriving while the timeout handler was cleaning up (e.g.
  an outer asyncio deadline during the SIGTERM grace wait) escaped the
  handler, skipping the SIGKILL and the pipe release; a SIGTERM-ignoring
  qsv kept running. The timeout cleanup now runs shielded, as the
  cancellation path already did.
- Windows: bound the post-kill wait by kill_grace, since proc.wait()
  also waits for pipes a surviving child may hold. Untested (no Windows
  CI).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/qsv_client/_process.py | 22 +++++++++++++++-------
 tests/test_process.py      | 24 ++++++++++++++++++++++++
 2 files changed, 39 insertions(+), 7 deletions(-)

diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py
index fdd6453..528c183 100644
--- a/src/qsv_client/_process.py
+++ b/src/qsv_client/_process.py
@@ -108,7 +108,8 @@ async def _terminate_async(proc: asyncio.subprocess.Process, grace: float) -> No
     if IS_WINDOWS:
         with contextlib.suppress(ProcessLookupError):
             proc.kill()
-        await proc.wait()
+        with contextlib.suppress(asyncio.TimeoutError):
+            await asyncio.wait_for(proc.wait(), grace)
         return
     # proc.wait() also waits for the output pipes to close, which a descendant that left the
     # group can hold open indefinitely, so every wait here is bounded
@@ -137,6 +138,17 @@ async def _reap_async(proc: asyncio.subprocess.Process, grace: float) -> None:
     await _release_async(proc)
 
 
+async def _timeout_cleanup(proc: asyncio.subprocess.Process, grace: float) -> bytes:
+    """Kill and release a timed-out run; return whatever stderr arrives within ``grace``."""
+    await _terminate_async(proc, grace)
+    err = b""
+    if proc.stderr is not None:
+        with contextlib.suppress(Exception):
+            err = await asyncio.wait_for(proc.stderr.read(), grace)
+    await _release_async(proc)
+    return err
+
+
 async def run_async(
     argv: Sequence[str],
     *,
@@ -161,12 +173,8 @@ async def run_async(
         out, err = await asyncio.wait_for(proc.communicate(input=stdin), timeout)
     except asyncio.TimeoutError:
         timed_out = True
-        await _terminate_async(proc, kill_grace)
-        out, err = b"", b""
-        if proc.stderr is not None:
-            with contextlib.suppress(Exception):
-                err = await asyncio.wait_for(proc.stderr.read(), kill_grace)
-        await _release_async(proc)
+        # shielded, so a cancellation arriving mid-cleanup can't skip the SIGKILL or the release
+        out, err = b"", await asyncio.shield(_timeout_cleanup(proc, kill_grace))
     except BaseException:
         # task cancelled (or the loop is shutting down): reap the group, then propagate
         await asyncio.shield(_reap_async(proc, kill_grace))
diff --git a/tests/test_process.py b/tests/test_process.py
index 74ce018..3f9446e 100644
--- a/tests/test_process.py
+++ b/tests/test_process.py
@@ -276,3 +276,27 @@ async def test_async_timeout_releases_pipes_held_by_escaped_grandchild(
     finally:
         for pidfile in pidfiles:
             _kill(_read_pid(pidfile))
+
+
+IGNORE_SIGTERM_AND_HANG = """
+import os, signal, sys, time
+signal.signal(signal.SIGTERM, signal.SIG_IGN)
+open(sys.argv[-1], "w").write(str(os.getpid()))
+time.sleep(60)
+"""
+
+
+async def test_async_cancel_during_timeout_cleanup_still_kills(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    # the run times out at 0.5s and starts its 2s SIGTERM grace wait; the outer deadline at
+    # 1s cancels it mid-cleanup, which must not skip the SIGKILL
+    pidfile = tmp_path / "leader.pid"
+    qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2)
+    try:
+        with pytest.raises(asyncio.TimeoutError):
+            await asyncio.wait_for(qsv.run("stats", pidfile, timeout=0.5), 1)
+        pid = _read_pid(pidfile)
+        assert await asyncio.to_thread(_wait_dead, pid, 6), "leader survived cancellation"
+    finally:
+        _kill(_read_pid(pidfile))

From e74a51c6df2124ba79a896c2da029538b2a7c6be Mon Sep 17 00:00:00 2001
From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com>
Date: Fri, 9 Oct 2026 11:08:14 -0400
Subject: [PATCH 5/8] fix: finish async teardown before propagating a
 cancellation

Addresses roborev review 5157: asyncio.shield() re-raised the
cancellation at once and left the cleanup in a background task. When
asyncio.run() returned right after an outer deadline, the runner's
shutdown cancelled that task mid-SIGTERM-wait, skipping the SIGKILL and
leaving a SIGTERM-ignoring qsv running. Both the timeout cleanup and the
cancellation path now run the teardown to completion, then re-raise.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/qsv_client/_process.py | 31 ++++++++++++++++++++++++++-----
 tests/test_process.py      | 19 +++++++++++++++++++
 2 files changed, 45 insertions(+), 5 deletions(-)

diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py
index 528c183..0789528 100644
--- a/src/qsv_client/_process.py
+++ b/src/qsv_client/_process.py
@@ -19,12 +19,14 @@
 import signal
 import subprocess
 import sys
-from collections.abc import Mapping, Sequence
+from collections.abc import Coroutine, Mapping, Sequence
 from dataclasses import dataclass
-from typing import IO, Any
+from typing import IO, Any, TypeVar
 
 IS_WINDOWS = sys.platform == "win32"
 
+_T = TypeVar("_T")
+
 
 @dataclass(frozen=True)
 class RawResult:
@@ -149,6 +151,25 @@ async def _timeout_cleanup(proc: asyncio.subprocess.Process, grace: float) -> by
     return err
 
 
+async def _finish_despite_cancel(coro: Coroutine[Any, Any, _T]) -> _T:
+    """Run ``coro`` to completion even if we are cancelled meanwhile, then re-raise the cancel.
+
+    ``asyncio.shield`` alone would propagate the cancellation at once, leaving the cleanup in
+    a background task that an ``asyncio.run()`` returning right after would cancel, skipping
+    the SIGKILL.
+    """
+    task = asyncio.ensure_future(coro)
+    cancel: asyncio.CancelledError | None = None
+    while not task.done():
+        try:
+            await asyncio.wait({task})
+        except asyncio.CancelledError as exc:
+            cancel = exc
+    if cancel is not None:
+        raise cancel
+    return task.result()
+
+
 async def run_async(
     argv: Sequence[str],
     *,
@@ -173,11 +194,11 @@ async def run_async(
         out, err = await asyncio.wait_for(proc.communicate(input=stdin), timeout)
     except asyncio.TimeoutError:
         timed_out = True
-        # shielded, so a cancellation arriving mid-cleanup can't skip the SIGKILL or the release
-        out, err = b"", await asyncio.shield(_timeout_cleanup(proc, kill_grace))
+        # a cancellation arriving mid-cleanup must not skip the SIGKILL or the release
+        out, err = b"", await _finish_despite_cancel(_timeout_cleanup(proc, kill_grace))
     except BaseException:
         # task cancelled (or the loop is shutting down): reap the group, then propagate
-        await asyncio.shield(_reap_async(proc, kill_grace))
+        await _finish_despite_cancel(_reap_async(proc, kill_grace))
         raise
     assert proc.returncode is not None
     return RawResult(
diff --git a/tests/test_process.py b/tests/test_process.py
index 3f9446e..4fb66be 100644
--- a/tests/test_process.py
+++ b/tests/test_process.py
@@ -300,3 +300,22 @@ async def test_async_cancel_during_timeout_cleanup_still_kills(
         assert await asyncio.to_thread(_wait_dead, pid, 6), "leader survived cancellation"
     finally:
         _kill(_read_pid(pidfile))
+
+
+def test_asyncio_run_returning_mid_cleanup_still_kills(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    # as above, but asyncio.run() returns right after the outer deadline, so any cleanup still
+    # running in a background task would be cancelled by the runner's shutdown
+    pidfile = tmp_path / "leader.pid"
+    qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2)
+
+    async def main() -> None:
+        await asyncio.wait_for(qsv.run("stats", pidfile, timeout=0.5), 1)
+
+    try:
+        with pytest.raises(asyncio.TimeoutError):
+            asyncio.run(main())
+        assert _wait_dead(_read_pid(pidfile), 6), "leader survived the runner's shutdown"
+    finally:
+        _kill(_read_pid(pidfile))

From b8a971b02a0ce4c417f586a35a4db7b14a67935b Mon Sep 17 00:00:00 2001
From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com>
Date: Fri, 9 Oct 2026 11:12:07 -0400
Subject: [PATCH 6/8] fix: kill synchronously when async teardown is itself
 cancelled

Addresses roborev review 5158: when asyncio.run() returns while a
background run is still in timeout cleanup, the runner cancels the
cleanup task directly, mid-SIGTERM-wait, so the SIGKILL and pipe close
never ran. Teardown now catches its own cancellation, skips the
remaining grace, sends SIGKILL to the group and closes the pipes
synchronously, then re-raises.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/qsv_client/_process.py | 41 +++++++++++++++++++++++++++++---------
 tests/test_process.py      | 19 ++++++++++++++++++
 2 files changed, 51 insertions(+), 9 deletions(-)

diff --git a/src/qsv_client/_process.py b/src/qsv_client/_process.py
index 0789528..8cec986 100644
--- a/src/qsv_client/_process.py
+++ b/src/qsv_client/_process.py
@@ -129,25 +129,48 @@ async def _release_async(proc: asyncio.subprocess.Process) -> None:
     Without the close, pipes held by an escaped descendant would keep their fds open and
     ``wait()`` pending until it exits. asyncio has no public API for this.
     """
+    _close_pipes(proc)
+    await proc.wait()
+
+
+def _close_pipes(proc: asyncio.subprocess.Process) -> None:
     transport = getattr(proc, "_transport", None)
     if transport is not None:
         transport.close()
-    await proc.wait()
+
+
+def _kill_now(proc: asyncio.subprocess.Process) -> None:
+    """Teardown was itself cancelled (e.g. by ``asyncio.run()`` shutting down): skip the
+    remaining grace and do the parts that matter synchronously."""
+    if IS_WINDOWS:
+        with contextlib.suppress(ProcessLookupError):
+            proc.kill()
+    else:
+        _signal_group(proc.pid, signal.SIGKILL)  # type: ignore[attr-defined,unused-ignore]
+    _close_pipes(proc)
 
 
 async def _reap_async(proc: asyncio.subprocess.Process, grace: float) -> None:
-    await _terminate_async(proc, grace)
-    await _release_async(proc)
+    try:
+        await _terminate_async(proc, grace)
+        await _release_async(proc)
+    except asyncio.CancelledError:
+        _kill_now(proc)
+        raise
 
 
 async def _timeout_cleanup(proc: asyncio.subprocess.Process, grace: float) -> bytes:
     """Kill and release a timed-out run; return whatever stderr arrives within ``grace``."""
-    await _terminate_async(proc, grace)
-    err = b""
-    if proc.stderr is not None:
-        with contextlib.suppress(Exception):
-            err = await asyncio.wait_for(proc.stderr.read(), grace)
-    await _release_async(proc)
+    try:
+        await _terminate_async(proc, grace)
+        err = b""
+        if proc.stderr is not None:
+            with contextlib.suppress(Exception):
+                err = await asyncio.wait_for(proc.stderr.read(), grace)
+        await _release_async(proc)
+    except asyncio.CancelledError:
+        _kill_now(proc)
+        raise
     return err
 
 
diff --git a/tests/test_process.py b/tests/test_process.py
index 4fb66be..35d243c 100644
--- a/tests/test_process.py
+++ b/tests/test_process.py
@@ -319,3 +319,22 @@ async def main() -> None:
         assert _wait_dead(_read_pid(pidfile), 6), "leader survived the runner's shutdown"
     finally:
         _kill(_read_pid(pidfile))
+
+
+def test_asyncio_run_cancelling_background_run_mid_cleanup_still_kills(
+    fake_qsv: Callable[[str], str], tmp_path: Path
+) -> None:
+    # the runner's shutdown cancels the run *and* its cleanup task directly, mid-grace-wait
+    pidfile = tmp_path / "leader.pid"
+    qsv = AsyncQsv(fake_qsv(IGNORE_SIGTERM_AND_HANG), kill_grace=2)
+
+    async def main() -> None:
+        background = asyncio.create_task(qsv.run("stats", pidfile, timeout=0.5))
+        await asyncio.sleep(1)
+        assert not background.done()  # still in its SIGTERM grace wait
+
+    try:
+        asyncio.run(main())
+        assert _wait_dead(_read_pid(pidfile), 6), "leader survived the runner's shutdown"
+    finally:
+        _kill(_read_pid(pidfile))

From 0a58bf57a68cc22093617b16868124b88e3c023a Mon Sep 17 00:00:00 2001
From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com>
Date: Fri, 9 Oct 2026 11:14:36 -0400
Subject: [PATCH 7/8] docs: note the async cancellation fixes in the changelog

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 76450f1..bba0f8f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -18,6 +18,9 @@ All notable changes to this project are documented here. The format follows
   left the process group and still holds its output pipes, including when qsv ignores SIGTERM.
   Previously the run waited for that process to exit. Output it still holds is dropped, and
   `AsyncQsv` closes its pipe file descriptors instead of keeping them open until it exits.
+- Cancelling an `AsyncQsv` run now always kills its process group, even when the cancellation
+  arrives during timeout cleanup or comes from `asyncio.run()` shutting down. A cancelled run
+  finishes its teardown before the cancellation propagates.
 - A failed run whose stdout is not valid UTF-8 raises the typed `QsvError` instead of
   `UnicodeDecodeError`.
 - `count()` and `headers()` work with `stdout_path=` and `text=False` instead of raising

From f098068cdc424d7b322c0d0fb11c3edaf547a279 Mon Sep 17 00:00:00 2001
From: Joel Natividad <1980690+jqnatividad@users.noreply.github.com>
Date: Fri, 9 Oct 2026 11:18:07 -0400
Subject: [PATCH 8/8] test: count a zombie as dead in _wait_dead

On Linux with Python 3.12+, asyncio reaps children through a pidfd
reader on the event loop. A leader SIGKILLed while asyncio.run() shuts
down is therefore never reaped by it, and os.kill(pid, 0) keeps
succeeding on the zombie, failing
test_asyncio_run_cancelling_background_run_mid_cleanup_still_kills on
ubuntu 3.12-3.14 although the process was killed. Check /proc/<pid>/stat
for state Z as well.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/test_process.py | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/tests/test_process.py b/tests/test_process.py
index 35d243c..9f1532d 100644
--- a/tests/test_process.py
+++ b/tests/test_process.py
@@ -26,6 +26,17 @@
 """
 
 
+def _is_zombie(pid: int) -> bool:
+    """Linux only: dead but not yet reaped. On Python 3.12+ asyncio reaps children from the
+    event loop (a pidfd reader), so a child killed as ``asyncio.run()`` shuts down is never
+    reaped by it and ``os.kill(pid, 0)`` keeps succeeding."""
+    try:
+        stat = Path(f"/proc/{pid}/stat").read_text()
+    except OSError:
+        return False
+    return stat.rsplit(")", 1)[1].split()[0] == "Z"
+
+
 def _wait_dead(pid: int, within: float = 5.0) -> bool:
     deadline = time.monotonic() + within
     while time.monotonic() < deadline:
@@ -33,6 +44,8 @@ def _wait_dead(pid: int, within: float = 5.0) -> bool:
             os.kill(pid, 0)
         except ProcessLookupError:
             return True
+        if _is_zombie(pid):
+            return True
         time.sleep(0.05)
     return False