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

graphify/prs.py

@@ -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)

tests/test_prs.py

@@ -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 @@
 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 @@ def test_no_community_field_skipped(self):
     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"