fix: decode SSH subprocess output tolerantly

tests/tools/test_ssh_environment_decoding.py

@@ -0,0 +1,38 @@
+from types import SimpleNamespace
+
+from tools.environments import ssh as ssh_env
+
+
+def test_ensure_remote_dirs_decodes_subprocess_output_tolerantly(monkeypatch):
+    calls = []
+
+    def fake_run(cmd, **kwargs):
+        calls.append((cmd, kwargs))
+        return SimpleNamespace(returncode=0, stdout="", stderr="")
+
+    env = object.__new__(ssh_env.SSHEnvironment)
+    env._remote_home = "/home/testuser"
+    env._build_ssh_command = lambda: ["ssh", "example.com"]
+
+    monkeypatch.setattr(ssh_env.subprocess, "run", fake_run)
+
+    env._ensure_remote_dirs()
+
+    assert calls
+    _, kwargs = calls[0]
+    assert kwargs["text"] is True
+    assert kwargs["encoding"] == "utf-8"
+    assert kwargs["errors"] == "replace"
+
+
+def test_ssh_subprocess_text_captures_all_use_tolerant_decoding():
+    source = ssh_env.Path(ssh_env.__file__).read_text(encoding="utf-8")
+    snippets = [
+        line for line in source.splitlines()
+        if "subprocess.run(" in line and "capture_output=True" in line and "text=True" in line
+    ]
+
+    assert snippets
+    for line in snippets:
+        assert 'encoding="utf-8"' in line
+        assert 'errors="replace"' in line

tools/environments/ssh.py

@@ -101,7 +101,7 @@ def _establish_connection(self):
         cmd = self._build_ssh_command()
         cmd.append("echo 'SSH connection established'")
         try:
-            result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+            result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=15)
             if result.returncode != 0:
                 error_msg = result.stderr.strip() or result.stdout.strip()
                 raise RuntimeError(f"SSH connection failed: {error_msg}")
@@ -113,7 +113,7 @@ def _detect_remote_home(self) -> str:
         try:
             cmd = self._build_ssh_command()
             cmd.append("echo $HOME")
-            result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+            result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10)
             home = result.stdout.strip()
             if home and result.returncode == 0:
                 logger.debug("SSH: remote home = %s", home)
@@ -134,7 +134,7 @@ def _ensure_remote_dirs(self) -> None:
         dirs = [base, f"{base}/skills", f"{base}/credentials", f"{base}/cache"]
         cmd = self._build_ssh_command()
         cmd.append(quoted_mkdir_command(dirs))
-        subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+        subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10)
 
     # _get_sync_files provided via iter_sync_files in FileSyncManager init
 
@@ -143,15 +143,15 @@ def _scp_upload(self, host_path: str, remote_path: str) -> None:
         parent = str(Path(remote_path).parent)
         mkdir_cmd = self._build_ssh_command()
         mkdir_cmd.append(f"mkdir -p {shlex.quote(parent)}")
-        subprocess.run(mkdir_cmd, capture_output=True, text=True, timeout=10)
+        subprocess.run(mkdir_cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10)
 
         scp_cmd = ["scp", "-o", f"ControlPath={self.control_socket}"]
         if self.port != 22:
             scp_cmd.extend(["-P", str(self.port)])
         if self.key_path:
             scp_cmd.extend(["-i", self.key_path])
         scp_cmd.extend([host_path, f"{self.user}@{self.host}:{remote_path}"])
-        result = subprocess.run(scp_cmd, capture_output=True, text=True, timeout=30)
+        result = subprocess.run(scp_cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30)
         if result.returncode != 0:
             raise RuntimeError(f"scp failed: {result.stderr.strip()}")
 
@@ -174,7 +174,7 @@ def _ssh_bulk_upload(self, files: list[tuple[str, str]]) -> None:
         if parents:
             cmd = self._build_ssh_command()
             cmd.append(quoted_mkdir_command(parents))
-            result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
+            result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=30)
             if result.returncode != 0:
                 raise RuntimeError(f"remote mkdir failed: {result.stderr.strip()}")
 
@@ -266,7 +266,7 @@ def _ssh_delete(self, remote_paths: list[str]) -> None:
         """Batch-delete remote files in one SSH call."""
         cmd = self._build_ssh_command()
         cmd.append(quoted_rm_command(remote_paths))
-        result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
+        result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10)
         if result.returncode != 0:
             raise RuntimeError(f"remote rm failed: {result.stderr.strip()}")