From 119dd9a486c423d7955ee64ba2a9319984b020c5 Mon Sep 17 00:00:00 2001 From: jihoson Date: Tue, 17 Feb 2026 04:13:05 +0900 Subject: [PATCH] Reduce tmux TUI noise with pane capture --- README.md | 2 +- pyproject.toml | 1 - src/lazy_enter/pty_manager.py | 100 ++++++++++++++++++++++------------ tests/test_pty_manager.py | 85 ++++++++++++++++++----------- 4 files changed, 119 insertions(+), 69 deletions(-) diff --git a/README.md b/README.md index 2134d17..3da069e 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ 1. 로컬에서 Claude를 tmux 세션으로 미리 실행 (`tmux new -s claude claude`) 2. Slack에서 `/start-claude` 실행 -3. 브릿지가 기존 tmux 세션에 attach +3. 브릿지가 기존 tmux 세션에 연결해 pane 출력/입력을 중계 4. Slack 채널 메시지가 CLI 입력으로 전달됨 5. CLI 출력이 Slack으로 다시 전송됨 6. `/stop-claude`로 브릿지 연결 해제 (tmux 세션은 유지) diff --git a/pyproject.toml b/pyproject.toml index ae3953c..2b6f982 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,6 @@ authors = [ dependencies = [ "slack-bolt>=1.21.0", "slack-sdk>=3.33.0", - "pexpect>=4.9.0", "python-dotenv>=1.0.0", ] diff --git a/src/lazy_enter/pty_manager.py b/src/lazy_enter/pty_manager.py index 2b7853f..b592d1f 100644 --- a/src/lazy_enter/pty_manager.py +++ b/src/lazy_enter/pty_manager.py @@ -1,73 +1,105 @@ -"""기존 tmux 세션에 attach하여 CLI 입출력을 중계한다.""" +"""기존 tmux 세션의 pane을 캡처/입력하여 CLI 입출력을 중계한다.""" from __future__ import annotations import logging import subprocess -import pexpect - logger = logging.getLogger(__name__) class PtyManager: - """기존 tmux 세션에 attach하고 입출력을 제어한다.""" + """기존 tmux 세션의 pane을 읽고 입력을 보낸다.""" def __init__(self, session_name: str = "claude") -> None: self.session_name = session_name - self._process: pexpect.spawn | None = None + self._connected = False + self._last_capture: list[str] = [] @property def is_alive(self) -> bool: - return self._process is not None and self._process.isalive() + return self._connected and self._session_exists() - def _ensure_session_exists(self) -> None: - """attach 대상 tmux 세션 존재 여부를 검증한다.""" - result = subprocess.run( - ["tmux", "has-session", "-t", self.session_name], + def _run_tmux(self, args: list[str]) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["tmux", *args], check=False, capture_output=True, text=True, ) + + def _session_exists(self) -> bool: + return self._run_tmux(["has-session", "-t", self.session_name]).returncode == 0 + + def _ensure_session_exists(self) -> None: + """대상 tmux 세션 존재 여부를 검증한다.""" + result = self._run_tmux(["has-session", "-t", self.session_name]) if result.returncode != 0: raise RuntimeError( "tmux 세션이 없습니다. 먼저 실행하세요: " f"tmux new -s {self.session_name} claude" ) - def start(self) -> None: - """기존 tmux 세션에 attach한다.""" - self._ensure_session_exists() - logger.info("tmux 세션 attach: %s", self.session_name) - self._process = pexpect.spawn( - "tmux", - ["attach-session", "-t", self.session_name], - encoding="utf-8", - timeout=None, + def _capture_lines(self) -> list[str]: + """pane의 최근 출력 스냅샷을 줄 단위로 가져온다.""" + result = self._run_tmux( + ["capture-pane", "-p", "-t", self.session_name, "-S", "-200"] ) + if result.returncode != 0: + return [] + return result.stdout.splitlines() + + @staticmethod + def _diff_lines(previous: list[str], current: list[str]) -> list[str]: + """이전 스냅샷 대비 새로 추가된 줄만 계산한다.""" + if not previous: + return [] + + prefix = 0 + for prev, curr in zip(previous, current): + if prev != curr: + break + prefix += 1 + + if prefix > 0: + return current[prefix:] + + # 화면 스크롤/리프레시로 앞부분이 달라진 경우 tail overlap으로 보정. + max_overlap = min(len(previous), len(current)) + for overlap in range(max_overlap, 0, -1): + if previous[-overlap:] == current[:overlap]: + return current[overlap:] + return current + + def start(self) -> None: + """기존 tmux 세션에 연결한다.""" + self._ensure_session_exists() + logger.info("tmux 세션 연결: %s", self.session_name) + self._connected = True + self._last_capture = self._capture_lines() def send(self, text: str) -> None: - """프로세스에 텍스트 입력을 전달한다.""" + """세션에 텍스트 입력을 전달한다.""" if not self.is_alive: raise RuntimeError("프로세스가 실행 중이 아닙니다.") - assert self._process is not None logger.debug("입력 전송: %s", text) - self._process.sendline(text) + result = self._run_tmux(["send-keys", "-t", self.session_name, text, "C-m"]) + if result.returncode != 0: + raise RuntimeError("tmux 입력 전송에 실패했습니다.") def read_output(self, timeout: int = 5) -> str: - """프로세스의 출력을 읽는다.""" + """세션 pane 출력을 읽는다.""" + _ = timeout if not self.is_alive: raise RuntimeError("프로세스가 실행 중이 아닙니다.") - assert self._process is not None - try: - self._process.expect(pexpect.TIMEOUT, timeout=timeout) - except pexpect.TIMEOUT: - pass - return self._process.before or "" + current = self._capture_lines() + delta_lines = self._diff_lines(self._last_capture, current) + self._last_capture = current + return "\n".join(delta_lines).strip() def stop(self) -> None: - """attach 연결만 종료한다(원격 세션은 유지).""" - if self._process is not None: - logger.info("tmux attach 종료") - self._process.close(force=True) - self._process = None + """브릿지 연결만 종료한다(원격 세션은 유지).""" + if self._connected: + logger.info("tmux 세션 연결 해제") + self._connected = False + self._last_capture = [] diff --git a/tests/test_pty_manager.py b/tests/test_pty_manager.py index 4b596a9..f80efa2 100644 --- a/tests/test_pty_manager.py +++ b/tests/test_pty_manager.py @@ -3,29 +3,17 @@ from __future__ import annotations from collections.abc import Sequence +from dataclasses import dataclass import pytest from lazy_enter.pty_manager import PtyManager -class FakeSpawn: - def __init__(self, *_args: object, **_kwargs: object) -> None: - self.before = "" - self._alive = True - - def isalive(self) -> bool: - return self._alive - - def sendline(self, _text: str) -> None: - return None - - def expect(self, *_args: object, **_kwargs: object) -> None: - return None - - def close(self, force: bool = False) -> None: - assert force is True - self._alive = False +@dataclass +class Result: + returncode: int + stdout: str = "" def test_initial_state(): @@ -45,10 +33,7 @@ def test_start_raises_when_tmux_session_missing(monkeypatch: pytest.MonkeyPatch) assert capture_output is True assert text is True - class Result: - returncode = 1 - - return Result() + return Result(returncode=1) monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run) @@ -57,9 +42,47 @@ def test_start_raises_when_tmux_session_missing(monkeypatch: pytest.MonkeyPatch) pty.start() -def test_start_and_stop_attach(monkeypatch: pytest.MonkeyPatch): +def test_start_send_read_and_stop(monkeypatch: pytest.MonkeyPatch): + calls: list[list[str]] = [] + def fake_run( - _cmd: Sequence[str], + cmd: Sequence[str], + check: bool, + capture_output: bool, + text: bool, + ): + calls.append(list(cmd)) + assert check is False + assert capture_output is True + assert text is True + + if cmd[:4] == ["tmux", "has-session", "-t", "claude"]: + return Result(returncode=0) + if cmd[:7] == ["tmux", "capture-pane", "-p", "-t", "claude", "-S", "-200"]: + # 첫 캡처는 baseline, 두 번째 캡처에서 한 줄 증가. + if calls.count(list(cmd)) == 1: + return Result(returncode=0, stdout="line1\nline2\n") + return Result(returncode=0, stdout="line1\nline2\nline3\n") + if cmd[:6] == ["tmux", "send-keys", "-t", "claude", "hello", "C-m"]: + return Result(returncode=0) + return Result(returncode=1) + + monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run) + + pty = PtyManager("claude") + pty.start() + assert pty.is_alive + + pty.send("hello") + assert pty.read_output() == "line3" + + pty.stop() + assert not pty.is_alive + + +def test_read_output_returns_empty_when_unchanged(monkeypatch: pytest.MonkeyPatch): + def fake_run( + cmd: Sequence[str], check: bool, capture_output: bool, text: bool, @@ -67,18 +90,14 @@ def test_start_and_stop_attach(monkeypatch: pytest.MonkeyPatch): assert check is False assert capture_output is True assert text is True - - class Result: - returncode = 0 - - return Result() + if cmd[:4] == ["tmux", "has-session", "-t", "claude"]: + return Result(returncode=0) + if cmd[:7] == ["tmux", "capture-pane", "-p", "-t", "claude", "-S", "-200"]: + return Result(returncode=0, stdout="same\n") + return Result(returncode=1) monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run) - monkeypatch.setattr("lazy_enter.pty_manager.pexpect.spawn", FakeSpawn) pty = PtyManager("claude") pty.start() - assert pty.is_alive - - pty.stop() - assert not pty.is_alive + assert pty.read_output() == ""