diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py index 13d1b6d4a3..9afec38fcf 100644 --- a/plugins/platforms/telegram/adapter.py +++ b/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 diff --git a/plugins/platforms/whatsapp/adapter.py b/plugins/platforms/whatsapp/adapter.py index 1fb24d3315..b5126136e1 100644 --- a/plugins/platforms/whatsapp/adapter.py +++ b/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: diff --git a/tools/mcp_tool_transport.py b/tools/mcp_tool_transport.py index 8cda945d47..0816e70063 100644 --- a/tools/mcp_tool_transport.py +++ b/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 diff --git a/tools/skillevaluator_scan.py b/tools/skillevaluator_scan.py index 2a516713b1..73b20440f5 100644 --- a/tools/skillevaluator_scan.py +++ b/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: diff --git a/tools/transcription_local.py b/tools/transcription_local.py index 515d391f28..49ec0054b8 100644 --- a/tools/transcription_local.py +++ b/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 "" diff --git a/tools/tts_command_provider.py b/tools/tts_command_provider.py index c5c6f412d3..cdd0c319c1 100644 --- a/tools/tts_command_provider.py +++ b/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": []} diff --git a/tui_gateway/methods_tools.py b/tui_gateway/methods_tools.py index b31804a3fc..fb81597aab 100644 --- a/tui_gateway/methods_tools.py +++ b/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: