fix(late-slices): restore subprocess footgun guards re-collapsed by r3-30/33/34/36 tails; telegram format_message stays self-free (staticmethod via class ref); _captured_exec explicit shell/env kwargs

plugins/platforms/telegram/adapter.py

@@ -4695,7 +4695,7 @@ class TelegramAdapter(BasePlatformAdapter):
             if _idx % 2 == 1:
                 _safe_parts.append(_seg)  # inside code — untouched
             else:
-                _safe_parts.append(re.sub(r'[(){}]', lambda m, _seg=_seg: self._escape_bare_bracket(m, _seg), _seg))
+                _safe_parts.append(re.sub(r'[(){}]', lambda m, _seg=_seg: TelegramAdapter._escape_bare_bracket(m, _seg), _seg))
         return ''.join(_safe_parts)
 
     @staticmethod

plugins/platforms/whatsapp/adapter.py

@@ -85,7 +85,7 @@ def _kill_port_process(port: int) -> None:
                     os.kill(pid, signal.SIGTERM)
                     continue
                 from hermes_cli._subprocess_compat import windows_hide_flags
-                subprocess.run(["taskkill", "/PID", str(pid), "/F"], capture_output=True, timeout=5, creationflags=windows_hide_flags())
+                subprocess.run(["taskkill", "/PID", str(pid), "/F"], capture_output=True, stdin=subprocess.DEVNULL, timeout=5, creationflags=windows_hide_flags())
 
 
 def _bridge_pid_is_ours(pid: int, session_path: Path, expected_start) -> bool:

tools/mcp_tool_transport.py

@@ -35,7 +35,7 @@ def _present(**kwargs) -> dict:
 def _pgroup_alive(pgid: Optional[int]) -> bool:
     """Signal 0 to the group succeeds iff any member is alive (POSIX only)."""
     try:
-        os.killpg(pgid, 0)
+        os.killpg(pgid, 0)  # windows-footgun: ok — guarded by AttributeError below
         return True
     except (AttributeError, TypeError, OSError):  # non-POSIX / pgid None / gone
         return False

tools/skillevaluator_scan.py

@@ -105,7 +105,8 @@ def run_tier1_scan(skill_dir: Path, timeout: int = SCAN_TIMEOUT_SECONDS) -> Tier
     with tempfile.TemporaryDirectory(prefix="se-tier1-") as outdir:
         try:
             subprocess.run([SCANNER_BIN, "validate", str(skill_dir), "--checks", TIER1_CHECKS, "--no-dedup",
-                            "-r", "json", "-o", outdir], capture_output=True, text=True, timeout=timeout)
+                            "-r", "json", "-o", outdir], capture_output=True, text=True, encoding="utf-8", errors="replace",
+                           stdin=subprocess.DEVNULL, timeout=timeout)
         except subprocess.TimeoutExpired:
             return unavailable(f"scan timed out after {timeout}s")
         except OSError as exc:

tools/transcription_local.py

@@ -97,7 +97,8 @@ def _looks_like_cuda_lib_error(exc: BaseException) -> bool:
 def _sysctl_value(name: str) -> str:
     """Return a sysctl value, or an empty string when unavailable."""
     try:
-        return subprocess.check_output(["/usr/sbin/sysctl", "-n", name], stderr=subprocess.DEVNULL, text=True,
+        return subprocess.check_output(["/usr/sbin/sysctl", "-n", name], stderr=subprocess.DEVNULL,
+                                       stdin=subprocess.DEVNULL, text=True, encoding="utf-8", errors="replace",
                                        timeout=2).strip()
     except Exception:
         return ""

tools/tts_command_provider.py

@@ -150,8 +150,8 @@ def run_command_provider(
     # locale-mismatched bytes must not raise in the reader threads.
     group = ({"creationflags": getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0)} if os.name == "nt"
              else {"start_new_session": True})
-    proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
-                            encoding="utf-8", errors="replace", env=delegated_child_subprocess_env(scrubbed),
+    proc = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                            text=True, encoding="utf-8", errors="replace", env=delegated_child_subprocess_env(scrubbed),
                             stdin=subprocess.DEVNULL, **group)
     output_queue: "queue.Queue[tuple[str, Optional[str]]]" = queue.Queue()
     chunks: Dict[str, list[str]] = {"stdout": [], "stderr": []}

tui_gateway/methods_tools.py

@@ -167,11 +167,12 @@ def _capture_run_kwargs(timeout: int) -> dict:
         stdin=subprocess.DEVNULL, creationflags=_tools_mod("hermes_cli._subprocess_compat").windows_hide_flags())
 
 
-def _captured_exec(rid, cmd, timeout: int, *, on_result, timeout_err: tuple, fail_code: int, **kw) -> dict:
+def _captured_exec(rid, cmd, timeout: int, *, on_result, timeout_err: tuple, fail_code: int,
+                   shell: bool = False, env: "dict | None" = None) -> dict:
     """Run ``cmd`` captured (see ``_capture_run_kwargs``) and hand the CompletedProcess to
     ``on_result``; TimeoutExpired → ``timeout_err`` (code, message), other errors → ``fail_code``."""
     try:
-        return on_result(subprocess.run(cmd, cwd=os.getcwd(), **kw, **_capture_run_kwargs(timeout)))
+        return on_result(subprocess.run(cmd, cwd=os.getcwd(), shell=shell, env=env, **_capture_run_kwargs(timeout)))
     except subprocess.TimeoutExpired:
         return _err(rid, *timeout_err)
     except Exception as e: