refactor: use shared strip_ansi for inline ANSI regexes

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\\)?)")
 
 

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]:

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:]