fix(prs): decode gh/git/claude output as UTF-8, not the Windows cp1252 codec

`graphify prs` reads gh, git, and claude output via subprocess.run(text=True)
with no explicit encoding=, so on Windows stdout is decoded with the cp1252
locale codec. gh emits UTF-8 JSON whose PR titles and logins routinely carry
non-Latin1 bytes (emoji, or the Persian title in PR #1281), and cp1252 has no
mapping for bytes such as 0x81. The capture reader thread raises
UnicodeDecodeError, subprocess.run then returns stdout=None, and the caller dies
one line later on json.loads(None) (TypeError) or None.splitlines()
(AttributeError). Neither is caught by _gh's except clause, so the whole `prs`
command aborts, with a spurious reader-thread traceback on stderr, against any
repo that has non-ASCII PR metadata.

Pass encoding="utf-8", errors="replace" to all five subprocess reads in prs.py
so decoding matches Linux and macOS. This is the decode-side sibling of #1505,
which applied the same change to the llm.py claude-cli subprocess. CI runs Linux
only (a UTF-8 locale), so this path is never exercised there.

Reproduced on Windows 11 / CPython 3.13 with a cp1252 locale:
prs._gh("pr", "view", "1281", ...) crashed with TypeError before and returns the
decoded title after. The added regression tests assert encoding="utf-8" on each
call, mirroring tests/test_charmap_encoding.py, and are red without the fix.
This commit is contained in:
Luke J
2026-07-18 12:03:47 +01:00
committed by safishamsi
parent a0386848be
commit a86666ae94
2 changed files with 70 additions and 5 deletions
+9 -5
View File
@@ -142,7 +142,10 @@ def _gh(*args: str) -> list | dict | None:
try:
result = subprocess.run(
["gh", *args],
capture_output=True, text=True, timeout=30
# Decode gh's output as UTF-8, not the Windows cp1252 locale codec: gh
# emits UTF-8 JSON with non-Latin1 titles/logins (emoji, فارسی), and the
# default text=True decode crashes on those (#1505 fixed the same in llm).
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30
)
if result.returncode != 0:
return None
@@ -164,7 +167,7 @@ def _detect_default_branch(repo: str | None = None) -> str:
try:
result = subprocess.run(
["git", "symbolic-ref", "refs/remotes/origin/HEAD"],
capture_output=True, text=True, timeout=5
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5
)
if result.returncode == 0:
# refs/remotes/origin/main → main
@@ -229,7 +232,7 @@ def fetch_pr_files(number: int, repo: str | None = None) -> list[str]:
if repo:
args += ["--repo", repo]
try:
result = subprocess.run(["gh", *args], capture_output=True, text=True, timeout=30)
result = subprocess.run(["gh", *args], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30)
if result.returncode != 0:
return []
return [l.strip() for l in result.stdout.splitlines() if l.strip()]
@@ -300,7 +303,7 @@ def fetch_worktrees() -> dict[str, str]:
try:
result = subprocess.run(
["git", "worktree", "list", "--porcelain"],
capture_output=True, text=True, timeout=10
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10
)
if result.returncode != 0:
return {}
@@ -655,7 +658,8 @@ def triage_with_opus(prs: list[PRInfo], base: str) -> None:
_claude = _shutil.which("claude.cmd") or _shutil.which("claude") or "claude"
proc = _sp.run(
[_claude, "-p", "--no-session-persistence"],
input=prompt, capture_output=True, text=True, timeout=120,
input=prompt, capture_output=True, text=True,
encoding="utf-8", errors="replace", timeout=120,
)
if proc.returncode != 0:
print(red(f" claude -p failed: {proc.stderr.strip()[:300]}"), file=sys.stderr)
+61
View File
@@ -1,6 +1,7 @@
"""Tests for graphify/prs.py."""
from __future__ import annotations
import json
import subprocess
from datetime import datetime, timedelta, timezone
from unittest.mock import patch, MagicMock
@@ -11,10 +12,12 @@ import pytest
from graphify.prs import (
PRInfo,
_classify,
_gh,
_parse_ci,
_path_match,
build_community_labels,
compute_pr_impact,
fetch_pr_files,
fetch_worktrees,
format_prs_text,
_detect_default_branch,
@@ -401,3 +404,61 @@ class TestBuildCommunityLabels:
def test_empty_nodes(self):
assert build_community_labels({}) == {}
assert build_community_labels({"nodes": []}) == {}
# ── Windows cp1252 subprocess-decode hardening (decode-side sibling of #1505) ──
class TestSubprocessOutputEncoding:
"""prs.py reads gh/git/claude output via subprocess.run(text=True). Without an
explicit encoding= the stdout is decoded with the locale codec (cp1252 on
Windows), so non-Latin1 output (emoji, non-ASCII PR titles/logins/paths)
fails: the reader thread raises UnicodeDecodeError, subprocess.run returns
stdout=None, and the caller then dies on json.loads(None) / None.splitlines().
Every call must pass encoding="utf-8" so decoding matches Linux/macOS.
#1505 fixed the same defect on the llm.py claude-cli subprocess.
"""
_NON_LATIN1 = "docs: add Persian (فارسی) 🏆"
def test_fixture_is_cp1252_undecodable(self):
"""Guard: the fixture's UTF-8 bytes must be undecodable as cp1252, else
these tests would prove nothing about the failure surface."""
raw = self._NON_LATIN1.encode("utf-8")
with pytest.raises(UnicodeDecodeError):
raw.decode("cp1252")
assert raw.decode("utf-8") == self._NON_LATIN1
def test_gh_decodes_output_as_utf8(self):
completed = MagicMock(
returncode=0, stdout=json.dumps([{"title": self._NON_LATIN1}]), stderr=""
)
with patch("subprocess.run", return_value=completed) as mock_run:
data = _gh("pr", "list", "--json", "title")
_args, kwargs = mock_run.call_args
assert kwargs.get("encoding") == "utf-8", (
f"_gh subprocess must use encoding='utf-8'; got {kwargs.get('encoding')!r}"
)
assert data == [{"title": self._NON_LATIN1}]
def test_fetch_pr_files_decodes_output_as_utf8(self):
completed = MagicMock(returncode=0, stdout="src/café.py\n", stderr="")
with patch("subprocess.run", return_value=completed) as mock_run:
fetch_pr_files(1)
_args, kwargs = mock_run.call_args
assert kwargs.get("encoding") == "utf-8"
def test_fetch_worktrees_decodes_output_as_utf8(self):
completed = MagicMock(returncode=0, stdout="", stderr="")
with patch("subprocess.run", return_value=completed) as mock_run:
fetch_worktrees()
_args, kwargs = mock_run.call_args
assert kwargs.get("encoding") == "utf-8"
def test_detect_default_branch_decodes_output_as_utf8(self):
# Force the git symbolic-ref fallback: gh returns None -> git subprocess runs.
completed = MagicMock(returncode=0, stdout="refs/remotes/origin/v8\n", stderr="")
with patch("graphify.prs._gh", return_value=None), \
patch("subprocess.run", return_value=completed) as mock_run:
_detect_default_branch()
_args, kwargs = mock_run.call_args
assert kwargs.get("encoding") == "utf-8"