python: handle invalid UTF-8 from local CLI

sdks/python/src/ctx_agent_history/transport.py

@@ -374,7 +374,6 @@ def _run(self, args: Sequence[str]) -> subprocess.CompletedProcess[str]:
                 command,
                 cwd=str(self.config.cwd) if self.config.cwd is not None else None,
                 env=env,
-                text=True,
                 capture_output=True,
                 timeout=self.config.timeout,
                 check=False,
@@ -392,8 +391,8 @@ def _run(self, args: Sequence[str]) -> subprocess.CompletedProcess[str]:
                 "ctx CLI timed out",
                 details={
                     "command": command,
-                    "stderr": exc.stderr or "",
-                    "stdout": exc.stdout or "",
+                    "stderr": _decode_process_output(exc.stderr),
+                    "stdout": _decode_process_output(exc.stdout),
                     "timeout": self.config.timeout,
                 },
                 cause=exc,
@@ -403,10 +402,26 @@ def _run(self, args: Sequence[str]) -> subprocess.CompletedProcess[str]:
                 "ctx CLI command failed",
                 command=command,
                 exit_code=completed.returncode,
-                stderr=completed.stderr,
-                stdout=completed.stdout,
+                stderr=_decode_process_output(completed.stderr),
+                stdout=_decode_process_output(completed.stdout),
             )
-        return completed
+        try:
+            stdout = _decode_process_output_strict(completed.stdout)
+            stderr = _decode_process_output_strict(completed.stderr)
+        except UnicodeDecodeError as exc:
+            raise CtxAgentHistoryProtocolError(
+                "ctx returned invalid UTF-8",
+                details={
+                    "command": command,
+                },
+                cause=exc,
+            ) from exc
+        return subprocess.CompletedProcess(
+            command,
+            completed.returncode,
+            stdout=stdout,
+            stderr=stderr,
+        )
 
     def _command(self, args: Sequence[str]) -> list[str]:
         command = [self.config.ctx_binary]
@@ -478,3 +493,23 @@ def ctx_version(self) -> Optional[str]:
 def _extend_option(args: list[str], flag: str, value: Optional[str]) -> None:
     if value is not None:
         args.extend([flag, value])
+
+
+def _decode_process_output(value: object) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    if isinstance(value, bytes):
+        return value.decode("utf-8", errors="replace")
+    return str(value)
+
+
+def _decode_process_output_strict(value: object) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    if isinstance(value, bytes):
+        return value.decode("utf-8", errors="strict")
+    return str(value)

sdks/python/tests/test_client.py

@@ -145,6 +145,37 @@ def test_invalid_json_raises_protocol_error(self) -> None:
 
         self.assertEqual(raised.exception.code, "decode_error")
 
+    def test_invalid_utf8_raises_protocol_error(self) -> None:
+        with fake_ctx(invalid_utf8=True) as cli:
+            client = AgentHistoryClient.local(ctx_binary=str(cli))
+
+            with self.assertRaises(CtxAgentHistoryProtocolError) as raised:
+                client.status()
+
+        self.assertEqual(raised.exception.code, "decode_error")
+        self.assertEqual(raised.exception.message, "ctx returned invalid UTF-8")
+        self.assertIsInstance(raised.exception.cause, UnicodeDecodeError)
+        self.assertIn("command", raised.exception.details)
+
+    def test_invalid_utf8_stderr_on_failed_cli_preserves_cli_error(self) -> None:
+        with fake_ctx(invalid_utf8_stderr=True) as cli:
+            client = AgentHistoryClient.local(ctx_binary=str(cli))
+
+            with self.assertRaises(CtxAgentHistoryCliError) as raised:
+                client.status()
+
+        self.assertEqual(raised.exception.code, "adapter_error")
+        self.assertEqual(raised.exception.exit_code, 42)
+        self.assertIn("\ufffd", raised.exception.stderr)
+
+    def test_invalid_utf8_ctx_version_returns_none(self) -> None:
+        with fake_ctx(invalid_utf8=True) as cli:
+            client = AgentHistoryClient.local(ctx_binary=str(cli))
+
+            version = client.version()
+
+        self.assertIsNone(version.ctx_version)
+
     def test_timeout_raises_contract_timeout_error(self) -> None:
         with fake_ctx(sleep=True) as cli:
             client = AgentHistoryClient.local(
@@ -225,18 +256,28 @@ def __init__(
         *,
         fail: bool = False,
         invalid_json: bool = False,
+        invalid_utf8: bool = False,
+        invalid_utf8_stderr: bool = False,
         sleep: bool = False,
     ) -> None:
         self.fail = fail
         self.invalid_json = invalid_json
+        self.invalid_utf8 = invalid_utf8
+        self.invalid_utf8_stderr = invalid_utf8_stderr
         self.sleep = sleep
         self._tmp: tempfile.TemporaryDirectory[str] | None = None
         self.path: Path | None = None
 
     def __enter__(self) -> Path:
         self._tmp = tempfile.TemporaryDirectory()
         self.path = Path(self._tmp.name) / "ctx"
-        script = _fake_ctx_script(fail=self.fail, invalid_json=self.invalid_json, sleep=self.sleep)
+        script = _fake_ctx_script(
+            fail=self.fail,
+            invalid_json=self.invalid_json,
+            invalid_utf8=self.invalid_utf8,
+            invalid_utf8_stderr=self.invalid_utf8_stderr,
+            sleep=self.sleep,
+        )
         self.path.write_text(script, encoding="utf-8")
         self.path.chmod(self.path.stat().st_mode | stat.S_IXUSR)
         return self.path
@@ -246,11 +287,22 @@ def __exit__(self, exc_type, exc, tb) -> None:  # type: ignore[no-untyped-def]
             self._tmp.cleanup()
 
 
-def _fake_ctx_script(*, fail: bool, invalid_json: bool, sleep: bool) -> str:
+def _fake_ctx_script(
+    *,
+    fail: bool,
+    invalid_json: bool,
+    invalid_utf8: bool,
+    invalid_utf8_stderr: bool,
+    sleep: bool,
+) -> str:
     if fail:
         return "#!/usr/bin/env python3\nimport sys\nsys.stderr.write('boom\\n')\nsys.exit(42)\n"
     if invalid_json:
         return "#!/usr/bin/env python3\nprint('not json')\n"
+    if invalid_utf8:
+        return "#!/usr/bin/env python3\nimport sys\nsys.stdout.buffer.write(b'\\xff\\xfe')\n"
+    if invalid_utf8_stderr:
+        return "#!/usr/bin/env python3\nimport sys\nsys.stderr.buffer.write(b'\\xff\\xfe')\nsys.exit(42)\n"
     if sleep:
         return "#!/usr/bin/env python3\nimport time\ntime.sleep(1)\nprint('{}')\n"