diff --git a/agent/secret_sources/base.py b/agent/secret_sources/base.py index 070051e046b..886a02950c5 100644 --- a/agent/secret_sources/base.py +++ b/agent/secret_sources/base.py @@ -239,6 +239,10 @@ _ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") # ANSI CSI/OSC escape sequences — helper-CLI stderr often carries color # codes that must not reach Hermes' own startup output. +# NOTE: intentionally NOT migrated to tools.ansi_strip.strip_ansi — the +# optional terminator here (``(?:\x07|\x1b\\)?``) also strips *unterminated* +# OSC sequences (common when a CLI is killed mid-write), which strip_ansi +# leaves untouched. strip_ansi is not a superset of this regex. _ANSI_RE = re.compile(r"\x1b(?:\[[0-9;?]*[ -/]*[@-~]|\][^\x07\x1b]*(?:\x07|\x1b\\)?)") diff --git a/agent/secret_sources/onepassword.py b/agent/secret_sources/onepassword.py index 71b06492612..d8216dfcbdf 100644 --- a/agent/secret_sources/onepassword.py +++ b/agent/secret_sources/onepassword.py @@ -42,7 +42,6 @@ from __future__ import annotations import hashlib import logging import os -import re import shutil import subprocess import time @@ -73,10 +72,10 @@ _OP_RUN_TIMEOUT = 30 # looks for. _DEFAULT_TOKEN_ENV = "OP_SERVICE_ACCOUNT_TOKEN" -# Strip whole ANSI CSI sequences (colour, cursor moves, line erases) from any -# `op` diagnostic we surface — not just the lone ESC byte — so a control -# sequence can't reposition the cursor or hide text after a redaction marker. -_ANSI_CSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") +# ANSI stripping for `op` diagnostics we surface uses the shared +# tools.ansi_strip.strip_ansi (full ECMA-48: CSI, OSC, DCS/SOS/PM/APC, +# C1) so a control sequence can't reposition the cursor or hide text +# after a redaction marker. # Env vars the `op` child actually needs. We build a minimal allowlisted env # rather than copying all of os.environ (which, post-dotenv, holds every @@ -231,7 +230,10 @@ def find_op(binary_path: str = "") -> Optional[Path]: def _scrub(text: str) -> str: """Remove ANSI control sequences and trim, for safe message surfacing.""" - return _ANSI_CSI_RE.sub("", text).replace("\x1b", "").strip() + from tools.ansi_strip import strip_ansi + + # strip_ansi removes well-formed sequences; drop any stray lone ESC too. + return strip_ansi(text).replace("\x1b", "").strip() def _op_child_env(token_value: str) -> Dict[str, str]: diff --git a/gateway/run.py b/gateway/run.py index 71f1ab0a8eb..c0ea379bf40 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -17528,7 +17528,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew return def _strip_ansi(text: str) -> str: - return re.sub(r'\x1b\[[0-9;]*[A-Za-z]', '', text) + from tools.ansi_strip import strip_ansi + return strip_ansi(text) bytes_sent = 0 last_stream_time = loop.time() @@ -17770,7 +17771,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew adapter=adapter, ) # Strip ANSI escape codes for clean display - output = re.sub(r'\x1b\[[0-9;]*m', '', output).strip() + from tools.ansi_strip import strip_ansi + output = strip_ansi(output).strip() if output: if len(output) > 3500: output = "…" + output[-3500:]