diff --git a/ckanext/datapusher_plus/qsv_utils.py b/ckanext/datapusher_plus/qsv_utils.py
index 8029b32..55ef853 100644
--- a/ckanext/datapusher_plus/qsv_utils.py
+++ b/ckanext/datapusher_plus/qsv_utils.py
@@ -14,6 +14,9 @@
 import ckanext.datapusher_plus.utils as utils
 from typing import Optional, Dict, Any, List, Union
 
+from qsv_client import Qsv, QsvError, QsvNotFound, QsvTimeout
+from qsv_client.errors import error_from_run
+
 from ckanext.datapusher_plus.logging_utils import TRACE
 
 logger = logging.getLogger(__name__)
@@ -40,6 +43,10 @@ def __init__(self, logger: Optional[logging.Logger] = None) -> None:
         # Verify qsv binary exists
         if not Path(self.qsv_bin).is_file():
             raise utils.JobError(f"{self.qsv_bin} not found.")
+        try:
+            self._qsv = Qsv(binary=self.qsv_bin, timeout=conf.QSV_COMMAND_TIMEOUT)
+        except QsvNotFound as e:
+            raise utils.JobError(f"{self.qsv_bin} not found.") from e
 
         # Check qsv version
         try:
@@ -58,97 +65,96 @@ def _run_command(
         timeout: Optional[float] = None,
     ) -> Union[subprocess.CompletedProcess, Dict[str, Any], str]:
         """
-        Run a qsv command with the given arguments.
+        Run a qsv command with the given arguments, via ``qsv-client``.
+
+        ``qsv-client`` runs qsv in its own process group, so a timeout kills
+        qsv *and* any children it started, and it asks qsv for structured
+        errors (``QSV_ERROR_FORMAT=json``), whose kind and message go into
+        the ``JobError``. Any non-zero exit code is a failure, as with
+        ``subprocess.run(check=True)``.
 
         Args:
             args: List of arguments for the qsv command
             check: Whether to raise an exception if the command fails
-            capture_output: Whether to capture stdout and stderr
+            capture_output: Ignored: stdout and stderr are always captured.
             text: Whether to return output as text
-            env: Optional environment variables
+            env: Optional environment variables, added to the worker's own
+                environment (not replacing it).
+            uses_stdio: On failure with ``check``, return
+                ``{"stdout": ..., "stderr": ...}`` instead of raising.
             timeout: Per-call override (seconds). Defaults to
                 ``conf.QSV_COMMAND_TIMEOUT`` so a hung qsv invocation never
                 pins an RQ worker indefinitely.
 
         Returns:
-            The result of subprocess.run
+            A ``subprocess.CompletedProcess``.
 
         Raises:
             utils.JobError: If the command fails and check is True, or if the
                 command times out.
         """
-
-        args = [self.qsv_bin] + args
-
-        # Convert all args to str to avoid TypeError with Path objects
         str_args = [str(arg) for arg in args]
-
+        command, rest = (str_args[0], str_args[1:]) if str_args else ("", [])
+        cmdline = " ".join([str(self.qsv_bin)] + str_args)
         effective_timeout = timeout if timeout is not None else conf.QSV_COMMAND_TIMEOUT
 
+        self.logger.trace(f"Running qsv command: {cmdline}")
         try:
-            self.logger.trace(f"Running qsv command: {' '.join(str_args)}")
-            result = subprocess.run(
-                str_args,
-                check=check,
-                capture_output=capture_output,
-                text=text,
+            res = self._qsv.run(
+                command,
+                *rest,
                 env=env,
                 timeout=effective_timeout,
+                check=False,
+                text=text,
             )
-            return result
-        except subprocess.TimeoutExpired as e:
-            error_msg = (
-                f"qsv command timed out after {effective_timeout}s: "
-                f"{' '.join(str_args)}"
-            )
+        except QsvTimeout as e:
+            error_msg = f"qsv command timed out after {effective_timeout}s: {cmdline}"
             self.logger.error(error_msg)
             raise utils.JobError(error_msg) from e
-        except subprocess.CalledProcessError as e:
-            if uses_stdio:
-                # Callers that opt into stdio capture handle their own failure
-                # logic and just want the raw stdout/stderr back.
-                return {"stdout": e.stdout, "stderr": e.stderr}
-
-            # Always log AND always raise when check=True — previously the
-            # raise was nested inside `if e.stderr:`, so failures with empty
-            # stderr silently returned None/"" and slipped past callers that
-            # had asked for check=True. Build the message with stderr when we
-            # have it, fall back to the exception's str otherwise.
-            error_msg = f"qsv command failed: {e}"
-            if getattr(e, "stderr", None):
-                error_msg += f" - {e.stderr}"
-            self.logger.error(error_msg)
-            if check:
-                raise utils.JobError(error_msg) from e
-            return e.stderr
+
+        stdout = res.stdout if text else res.stdout_bytes
+        stderr: Union[str, bytes] = res.stderr if text else res.stderr.encode("utf-8")
+        result = subprocess.CompletedProcess(
+            args=list(res.args), returncode=res.exit_code, stdout=stdout, stderr=stderr
+        )
+        if res.exit_code == 0 or not check:
+            return result
+        if uses_stdio:
+            # Callers that opt into stdio capture handle their own failure
+            # logic and just want the raw stdout/stderr back.
+            return {"stdout": stdout, "stderr": stderr}
+
+        # Always log AND always raise when check=True, even with empty stderr.
+        err = error_from_run(
+            exit_code=res.exit_code, stderr=res.stderr, args=res.args, command=command
+        )
+        error_msg = f"qsv command failed: {err} [command: {cmdline}]"
+        if not err.structured and res.stderr.strip():
+            error_msg += f" - {res.stderr}"
+        self.logger.error(error_msg)
+        raise utils.JobError(error_msg) from err
 
     def version(self) -> str:
         """
         Get the qsv version.
 
+        Read from ``qsv --capabilities`` on qsv versions that have it, and
+        from ``qsv --version`` otherwise.
+
         Returns:
-            The qsv version string
+            The qsv version string, e.g. ``"24.0.0"``
 
         Raises:
             utils.JobError: If the version command fails
+            ValueError: If the version output cannot be parsed
         """
-        result = self._run_command(["--version"])
-        version_info = result.stdout.strip()
-
-        if not version_info:
+        try:
+            return self._qsv.version
+        except QsvError as e:
             raise utils.JobError(
-                f"We expect qsv version info to be returned. Command: {self.qsv_bin} --version. Response: {version_info}"
-            )
-
-        # Extract version number
-        version_start = version_info.find(" ")
-        version_end = version_info.find("-")
-        if version_start > 0 and version_end > version_start:
-            version = version_info[version_start:version_end].lstrip()
-        else:
-            version = version_info
-
-        return version
+                f"Cannot get qsv version info. Command: {self.qsv_bin} --version. {e}"
+            ) from e
 
     def check_version(self) -> bool:
         """
@@ -712,7 +718,11 @@ def describegpt(
             env: Optional environment overrides for the subprocess
                 (e.g. ``{"OPENAI_API_KEY": "..."}`` if the caller has a
                 key it wants to inject out-of-band).
-            api_key: Optional value for ``--api-key``. Required when
+            api_key: Optional API key, passed to qsv as ``QSV_LLM_APIKEY``
+                rather than ``--api-key``: describegpt copies its command
+                line into the attribution of what it generates, and argv is
+                visible in ``ps``. Only the ``"NONE"`` sentinel, which qsv
+                honors on the flag alone, goes on the command line. Required when
                 the LLM endpoint isn't on ``localhost`` (qsv treats
                 non-localhost base URLs as "not a local LLM" and
                 refuses to start without an API key). For unauthenticated
@@ -759,7 +769,10 @@ def describegpt(
         # is nonsensical for these flags, but matching the documented
         # contract keeps the wrapper predictable.
         if api_key is not None:
-            args.extend(["--api-key", api_key])
+            if api_key.upper() == "NONE":
+                args.extend(["--api-key", api_key])
+            else:
+                env = {**(env or {}), "QSV_LLM_APIKEY": api_key}
         if base_url is not None:
             args.extend(["--base-url", base_url])
 
diff --git a/requirements.txt b/requirements.txt
index 0ee1502..1b1354c 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -29,3 +29,7 @@ blake3>=1.0.7
 # explicitly because DP+ imports ``babel.numbers`` directly now and
 # shouldn't rely on the upstream pin staying loose.
 babel>=2.9
+# qsv-client: runs qsv for QSVCommand. Kills the whole process group on
+# timeout (qsv and anything it started) and reads qsv's structured errors
+# (QSV_ERROR_FORMAT=json) where the binary supports them. No dependencies.
+qsv-client>=0.1.0
diff --git a/tests/test_ai_suggestions.py b/tests/test_ai_suggestions.py
index 45c0ff0..225f681 100644
--- a/tests/test_ai_suggestions.py
+++ b/tests/test_ai_suggestions.py
@@ -9,7 +9,7 @@
 flow continue. These tests exercise:
 
 * ``QSVCommand.describegpt`` argument assembly (no real qsv binary
-  required — subprocess.run is mocked).
+  required — qsv-client's process layer is mocked).
 * ``AISuggestionsStage.should_skip`` honors the config flag.
 * ``AISuggestionsStage.process`` happy / failure paths.
 * ``helpers.scheming_get_ai_suggestion`` lookup logic across the
@@ -53,6 +53,20 @@ def _completed(stdout: str = "", returncode: int = 0) -> subprocess.CompletedPro
     )
 
 
+# ``QSVCommand`` runs qsv through qsv-client, whose process layer is the
+# subprocess seam: its first positional argument is the argv, and it takes
+# ``env=`` and ``timeout=`` keywords, just like ``subprocess.run``.
+RUN_SYNC = "qsv_client._process.run_sync"
+
+
+def _raw(stdout: str = "", exit_code: int = 0):
+    from qsv_client._process import RawResult
+
+    return RawResult(
+        exit_code=exit_code, stdout=stdout.encode(), stderr=b"", timed_out=False
+    )
+
+
 def test_describegpt_default_flags(qsv_command):
     """Default invocation emits --description --dictionary --tags --format JSON.
 
@@ -61,7 +75,7 @@ def test_describegpt_default_flags(qsv_command):
     invocations to fail with ``Unknown flag: '--json'``). Output
     format is controlled by ``--format <Markdown|TSV|JSON|TOON>``.
     """
-    with mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(input_file="/tmp/sample.csv")
 
     args = run.call_args[0][0]
@@ -79,7 +93,7 @@ def test_describegpt_default_flags(qsv_command):
 
 def test_describegpt_prompt_file_threaded_through(qsv_command):
     """``prompt_file`` is passed as ``--prompt-file <path>``."""
-    with mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(
             input_file="/tmp/sample.csv",
             prompt_file="/etc/qsv/describegpt.toml",
@@ -93,7 +107,7 @@ def test_describegpt_prompt_file_threaded_through(qsv_command):
 
 def test_describegpt_output_file_threaded_through(qsv_command):
     """``output_file`` is passed as ``--output <path>``."""
-    with mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(
             input_file="/tmp/sample.csv", output_file="/tmp/out.json"
         )
@@ -106,7 +120,7 @@ def test_describegpt_output_file_threaded_through(qsv_command):
 
 def test_describegpt_skips_disabled_flags(qsv_command):
     """Falsy flags omit their CLI counterpart entirely."""
-    with mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(
             input_file="/tmp/sample.csv",
             description=False,
@@ -129,7 +143,7 @@ def test_describegpt_default_timeout_from_config(qsv_command):
     from ckanext.datapusher_plus import config as conf
 
     with mock.patch.object(conf, "DESCRIBEGPT_TIMEOUT_SECONDS", 42), \
-         mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+         mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(input_file="/tmp/sample.csv")
 
     assert run.call_args.kwargs["timeout"] == 42
@@ -140,7 +154,7 @@ def test_describegpt_explicit_timeout_overrides_config(qsv_command):
     from ckanext.datapusher_plus import config as conf
 
     with mock.patch.object(conf, "DESCRIBEGPT_TIMEOUT_SECONDS", 42), \
-         mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+         mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(input_file="/tmp/sample.csv", timeout=7)
 
     assert run.call_args.kwargs["timeout"] == 7
@@ -156,7 +170,7 @@ def test_describegpt_api_key_threaded_through(qsv_command):
     container-host hostnames (``host.docker.internal`` etc.) the
     documented incantation is ``--api-key NONE``.
     """
-    with mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(input_file="/tmp/sample.csv", api_key="NONE")
 
     args = run.call_args[0][0]
@@ -166,7 +180,7 @@ def test_describegpt_api_key_threaded_through(qsv_command):
 
 def test_describegpt_base_url_threaded_through(qsv_command):
     """``--base-url`` overrides the endpoint without rewriting the prompt-file."""
-    with mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(
             input_file="/tmp/sample.csv",
             base_url="http://host.docker.internal:1234/v1",
@@ -182,7 +196,7 @@ def test_describegpt_base_url_threaded_through(qsv_command):
 
 def test_describegpt_optional_flags_omitted_by_default(qsv_command):
     """``api_key`` / ``base_url`` are None by default — let qsv discover."""
-    with mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(input_file="/tmp/sample.csv")
 
     args = run.call_args[0][0]
@@ -503,12 +517,31 @@ def test_describegpt_empty_api_key_passes_through(qsv_command):
     """``api_key=""`` is unusual but the docstring contract is
     ``None`` ⇒ omit, anything else ⇒ pass through. ``""`` should
     not be silently dropped."""
-    with mock.patch("subprocess.run", return_value=_completed("{}")) as run:
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
         qsv_command.describegpt(input_file="/tmp/sample.csv", api_key="")
 
-    args = run.call_args[0][0]
-    assert "--api-key" in args
-    assert args[args.index("--api-key") + 1] == ""
+    assert run.call_args.kwargs["env"]["QSV_LLM_APIKEY"] == ""
+    assert "--api-key" not in run.call_args[0][0]
+
+
+def test_describegpt_api_key_goes_in_the_environment_not_argv(qsv_command):
+    """describegpt copies its command line into the attribution of the
+    dictionary/description/tags it generates, and argv shows up in ``ps``,
+    so a real key must travel in ``QSV_LLM_APIKEY``. The worker's own
+    environment is kept, and caller ``env`` entries are merged in."""
+    key = "sk-or-v1-0123456789abcdef"
+    with mock.patch(RUN_SYNC, return_value=_raw("{}")) as run:
+        qsv_command.describegpt(
+            input_file="/tmp/sample.csv", api_key=key, env={"EXTRA": "1"}
+        )
+
+    argv = run.call_args[0][0]
+    env = run.call_args.kwargs["env"]
+    assert "--api-key" not in argv
+    assert not any(key in a for a in argv)
+    assert env["QSV_LLM_APIKEY"] == key
+    assert env["EXTRA"] == "1"
+    assert "PATH" in env
 
 
 def test_process_reshape_skips_unknown_envelope_keys(stage_cls, context_factory):
diff --git a/tests/test_qsv_command_client.py b/tests/test_qsv_command_client.py
new file mode 100644
index 0000000..e675055
--- /dev/null
+++ b/tests/test_qsv_command_client.py
@@ -0,0 +1,191 @@
+# -*- coding: utf-8 -*-
+"""``QSVCommand`` runs qsv through qsv-client.
+
+These tests run ``QSVCommand`` against a fake ``qsvdp`` (a Python script) and
+check the contract callers rely on: a ``subprocess.CompletedProcess`` on
+success, ``JobError`` on failure (or the ``{"stdout", "stderr"}`` dict with
+``uses_stdio``), the result itself with ``check=False``, and a timeout that
+kills qsv's children too.
+"""
+
+import json
+import os
+import subprocess
+import sys
+import time
+from unittest import mock
+
+import pytest
+
+pytestmark = pytest.mark.skipif(
+    sys.platform == "win32", reason="the fake qsv is a POSIX script"
+)
+
+FAKE_QSV = r'''
+import json, os, subprocess, sys, time
+
+args = sys.argv[1:]
+cmd = args[0] if args else ""
+if cmd == "--capabilities":
+    print(json.dumps({"binary": "qsvdp", "version": os.environ.get("FAKE_QSV_VERSION", "99.0.0"),
+                      "features": [], "commands": ["count"], "error_formats": ["text", "json"]}))
+elif cmd == "ok":
+    print("out")
+    print("progress", file=sys.stderr)
+elif cmd == "fail-json":
+    print("out-before-failure")
+    print(json.dumps({"error": {"kind": "csv", "level": "error", "message": "ragged row 3",
+                                "exit_code": 1, "command": "fail-json", "qsv_version": "99.0.0"}}),
+          file=sys.stderr)
+    sys.exit(1)
+elif cmd == "fail-text":
+    print("plain failure text", file=sys.stderr)
+    sys.exit(1)
+elif cmd == "fail-silent":
+    sys.exit(1)
+elif cmd == "warn":
+    print("broken pipe", file=sys.stderr)
+    sys.exit(255)
+elif cmd == "env":
+    print(json.dumps({k: os.environ.get(k) for k in ("FOO", "PATH", "QSV_ERROR_FORMAT")}))
+elif cmd == "hang":
+    child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(60)"])
+    with open(args[1], "w") as f:
+        f.write(str(child.pid))
+    time.sleep(60)
+'''
+
+
+@pytest.fixture
+def fake_qsv(tmp_path):
+    path = tmp_path / "qsvdp"
+    path.write_text(f"#!{sys.executable}\n{FAKE_QSV}")
+    path.chmod(0o755)
+    return path
+
+
+@pytest.fixture
+def qsv(fake_qsv):
+    pytest.importorskip("ckan")
+    from ckanext.datapusher_plus import qsv_utils
+
+    with mock.patch.object(qsv_utils.conf, "QSV_BIN", fake_qsv):
+        yield qsv_utils.QSVCommand(logger=mock.Mock())
+
+
+def _job_error():
+    from ckanext.datapusher_plus import utils
+
+    return utils.JobError
+
+
+def test_constructor_checks_the_version_via_capabilities(qsv):
+    assert qsv.version() == "99.0.0"
+
+
+def test_constructor_rejects_a_too_old_qsv(fake_qsv, monkeypatch):
+    pytest.importorskip("ckan")
+    from ckanext.datapusher_plus import qsv_utils
+
+    monkeypatch.setenv("FAKE_QSV_VERSION", "1.0.0")
+    with mock.patch.object(qsv_utils.conf, "QSV_BIN", fake_qsv), pytest.raises(
+        _job_error(), match="At least qsv version"
+    ):
+        qsv_utils.QSVCommand(logger=mock.Mock())
+
+
+def test_success_returns_a_completed_process(qsv, fake_qsv):
+    result = qsv._run_command(["ok"])
+    assert isinstance(result, subprocess.CompletedProcess)
+    assert result.returncode == 0
+    assert result.stdout == "out\n"
+    assert result.stderr == "progress\n"
+    assert result.args == [str(fake_qsv), "ok"]
+
+
+def test_failure_raises_job_error_with_the_structured_error(qsv, fake_qsv):
+    with pytest.raises(_job_error()) as exc:
+        qsv._run_command(["fail-json", "/data/in.csv", "--flag"])
+    msg = str(exc.value)
+    assert msg.startswith("qsv command failed: ")
+    assert "csv" in msg and "ragged row 3" in msg and "exit 1" in msg
+    # operators reading the job log need to see what ran
+    assert f"[command: {fake_qsv} fail-json /data/in.csv --flag]" in msg
+
+
+def test_failure_raises_job_error_with_text_stderr(qsv):
+    with pytest.raises(_job_error(), match="plain failure text"):
+        qsv._run_command(["fail-text"])
+
+
+def test_failure_with_empty_stderr_still_raises(qsv):
+    # an earlier DP+ bug: empty stderr meant check=True failures slipped through
+    with pytest.raises(_job_error(), match="exited with code 1"):
+        qsv._run_command(["fail-silent"])
+
+
+def test_exit_255_is_a_failure(qsv):
+    # qsv-client counts 255 (qsv's "warning") as success; QSVCommand keeps
+    # subprocess.run(check=True)'s rule that any non-zero exit is a failure
+    with pytest.raises(_job_error(), match="broken pipe"):
+        qsv._run_command(["warn"])
+
+
+def test_failure_with_uses_stdio_returns_the_raw_output(qsv):
+    out = qsv._run_command(["fail-json"], uses_stdio=True)
+    assert out["stdout"] == "out-before-failure\n"
+    assert '"kind": "csv"' in out["stderr"]
+
+
+def test_failure_with_check_false_returns_the_completed_process(qsv):
+    result = qsv._run_command(["fail-text"], check=False)
+    assert isinstance(result, subprocess.CompletedProcess)
+    assert result.returncode == 1
+    assert "plain failure text" in result.stderr
+
+
+def test_env_is_merged_into_the_workers_environment(qsv):
+    # subprocess.run(env=...) used to REPLACE the environment, dropping PATH
+    seen = json.loads(qsv._run_command(["env"], env={"FOO": "bar"}).stdout)
+    assert seen["FOO"] == "bar"
+    assert seen["PATH"] == os.environ["PATH"]
+    assert seen["QSV_ERROR_FORMAT"] == "json"
+
+
+def test_text_false_returns_bytes(qsv):
+    result = qsv._run_command(["ok"], text=False)
+    assert result.stdout == b"out\n"
+    assert result.stderr == b"progress\n"
+
+
+def test_timeout_raises_job_error_and_kills_qsvs_children(qsv, tmp_path):
+    pid_file = tmp_path / "child.pid"
+    started = time.monotonic()
+    with pytest.raises(_job_error(), match=r"timed out after 1s"):
+        qsv._run_command(["hang", str(pid_file)], timeout=1)
+    assert time.monotonic() - started < 30
+
+    child = int(pid_file.read_text())
+    deadline = time.monotonic() + 10
+    while _alive(child):
+        if time.monotonic() > deadline:
+            os.kill(child, 9)
+            pytest.fail("qsv's child process survived the timeout")
+        time.sleep(0.05)
+
+
+def _alive(pid):
+    """True while ``pid`` runs. A zombie counts as dead: an orphan is only
+    reaped if PID 1 reaps, and a container's PID 1 may not."""
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    stat = f"/proc/{pid}/stat"
+    if os.path.exists(stat):
+        with open(stat) as f:
+            return f.read().rsplit(")", 1)[-1].split()[0] != "Z"
+    state = subprocess.run(
+        ["ps", "-o", "stat=", "-p", str(pid)], capture_output=True, text=True
+    ).stdout.strip()
+    return bool(state) and not state.startswith("Z")