fix: bounded timeouts and reliable async teardown, plus three client bugs

.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

.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.

CHANGELOG.md

@@ -12,6 +12,21 @@ 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 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.
+- 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
+  `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))

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/`)

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
@@ -15,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:
@@ -78,7 +84,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)
@@ -96,15 +110,89 @@ 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
     _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.
+    """
+    _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()
+
+
+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:
+    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``."""
+    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
+
+
+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],
     *,
@@ -129,14 +217,11 @@ 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 proc.stderr.read()
+        # 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(_terminate_async(proc, kill_grace))
+        await _finish_despite_cancel(_reap_async(proc, kill_grace))
         raise
     assert proc.returncode is not None
     return RawResult(

src/qsv_client/client.py

@@ -43,12 +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=. Not
+            # normalized: collapsing `..` lexically would be wrong after a symlinked dir.
             if Path(cand).is_file():
-                return str(Path(cand))
+                return str(Path(cand).absolute())
             continue
         found = shutil.which(cand)
         if found:
-            return 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."
@@ -122,6 +124,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 +191,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 +209,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 +353,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 +478,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]]: