Filter ANSI output before Slack forwarding

src/lazy_enter/bridge.py

@@ -7,6 +7,7 @@ import threading
 import time
 
 from lazy_enter.config import Config
+from lazy_enter.output_filter import clean_terminal_output
 from lazy_enter.pty_manager import PtyManager
 from lazy_enter.slack_handler import SlackHandler
 
@@ -23,6 +24,7 @@ class Bridge:
         self._output_thread: threading.Thread | None = None
         self._running = False
         self._channel: str = self.config.allowed_channel_id
+        self._last_sent_output: str = ""
 
         self.slack.on_message(self._handle_message)
         self.slack.on_command(self._handle_command)
@@ -51,6 +53,7 @@ class Bridge:
             return
 
         self._channel = channel
+        self._last_sent_output = ""
         self.pty = PtyManager(self.config.default_shell)
         self.pty.start()
         self._running = True
@@ -65,6 +68,7 @@ class Bridge:
         if self.pty:
             self.pty.stop()
             self.pty = None
+        self._last_sent_output = ""
         self.slack.send_message(channel, ":stop_sign: 세션이 종료되었습니다.")
 
     def _poll_output(self) -> None:
@@ -73,14 +77,18 @@ class Bridge:
         while self._running and self.pty and self.pty.is_alive:
             output = self.pty.read_output(timeout=self.config.pty_read_timeout)
             if output:
-                buffer += output
+                cleaned = clean_terminal_output(output)
+                if cleaned:
+                    buffer += f"{cleaned}\n"
 
             if buffer:
                 # 메시지 길이 제한 적용
-                message = buffer[: self.config.max_message_length]
+                message = buffer[: self.config.max_message_length].rstrip()
                 if len(buffer) > self.config.max_message_length:
                     message += "\n... (truncated)"
+                if message and message != self._last_sent_output:
                     self.slack.send_message(self._channel, f"```\n{message}\n```")
+                    self._last_sent_output = message
                 buffer = ""
 
             time.sleep(self.config.output_buffer_interval)

src/lazy_enter/output_filter.py

@@ -0,0 +1,32 @@
+"""PTY 출력에서 터미널 제어 시퀀스를 제거한다."""
+
+from __future__ import annotations
+
+import re
+
+# ANSI CSI / OSC / DCS / 단일 ESC 시퀀스를 폭넓게 제거한다.
+_ANSI_ESCAPE_RE = re.compile(
+    r"(?:\x1B\[[0-?]*[ -/]*[@-~])"
+    r"|(?:\x1B\][^\x1B\x07]*(?:\x07|\x1B\\))"
+    r"|(?:\x1BP[^\x1B]*(?:\x1B\\))"
+    r"|(?:\x1B[@-_])"
+)
+
+# ESC 문자가 유실된 상태로 남는 CSI 토큰(예: [38;2;153;153;153m) 제거.
+_BROKEN_CSI_RE = re.compile(r"\[(?:[0-9;<=>?]|(?:\d+;))*\d*[A-Za-z]")
+
+# C0 제어문자 중 개행/탭/캐리지리턴을 제외하고 제거.
+_CONTROL_RE = re.compile(r"[\x00-\x08\x0B-\x1F\x7F]")
+
+
+def clean_terminal_output(text: str) -> str:
+    """Slack 전송용 텍스트를 정리한다."""
+    cleaned = _ANSI_ESCAPE_RE.sub("", text)
+    cleaned = _BROKEN_CSI_RE.sub("", cleaned)
+    cleaned = cleaned.replace("\r", "")
+    cleaned = _CONTROL_RE.sub("", cleaned)
+
+    # 제어 코드 제거 후 남는 빈 줄을 축소한다.
+    lines = [line.rstrip() for line in cleaned.splitlines()]
+    compact = "\n".join(lines).strip()
+    return compact

tests/test_output_filter.py

@@ -0,0 +1,18 @@
+"""output_filter 모듈 테스트."""
+
+from lazy_enter.output_filter import clean_terminal_output
+
+
+def test_clean_terminal_output_removes_ansi_sequences():
+    raw = "\x1b[?2004h\x1b[38;2;153;153;153mClaude Code\x1b[39m"
+    assert clean_terminal_output(raw) == "Claude Code"
+
+
+def test_clean_terminal_output_removes_broken_csi_tokens():
+    raw = "[?2004h[38;2;153;153;153mhello[39m"
+    assert clean_terminal_output(raw) == "hello"
+
+
+def test_clean_terminal_output_preserves_plain_text():
+    raw = "line1\nline2"
+    assert clean_terminal_output(raw) == "line1\nline2"