jihoson/LazyEnter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: jihoson:fix/slack-input-drop-logs
Branches Tags
jihoson:main jihoson:feat/unify-start-stop-args jihoson:fix/output-throttle jihoson:fix/slack-input-drop-logs jihoson:fix/codex-tui-noise-filter jihoson:chore/record-pr-workflow-safety jihoson:feat/add-codex-bridge-support jihoson:chore/gitea-workflow-notes jihoson:feat/tmux-attach-bridge jihoson:fix/slack-ansi-filter jihoson:feat/remoteenter-hardening jihoson:docs/contributor-guide-and-readme
..
compare: jihoson:feat/tmux-attach-bridge
Branches Tags
jihoson:main jihoson:feat/unify-start-stop-args jihoson:fix/output-throttle jihoson:fix/slack-input-drop-logs jihoson:fix/codex-tui-noise-filter jihoson:chore/record-pr-workflow-safety jihoson:feat/add-codex-bridge-support jihoson:chore/gitea-workflow-notes jihoson:feat/tmux-attach-bridge jihoson:fix/slack-ansi-filter jihoson:feat/remoteenter-hardening jihoson:docs/contributor-guide-and-readme

1 Commits
fix/slack- ... feat/tmux-

Author SHA1 Message Date
jihoson
119dd9a486 Reduce tmux TUI noise with pane capture 2026-02-17 04:13:05 +09:00
14 changed files with 156 additions and 655 deletions
Expand all files Collapse all files

1
.env.example
View File

@@ -8,7 +8,6 @@ SLACK_ALLOWED_CHANNEL_ID=C0XXXXXXXXX
# PTY / tmux 연결 설정 # PTY / tmux 연결 설정
TMUX_SESSION_NAME=claude TMUX_SESSION_NAME=claude
CODEX_TMUX_SESSION_NAME=codex
PTY_READ_TIMEOUT=5 PTY_READ_TIMEOUT=5
# 출력 버퍼 설정 # 출력 버퍼 설정

39
AGENTS.md
View File

@@ -42,42 +42,3 @@ Current git history is minimal (`Initial commit`), so enforce clear conventions
## Security & Configuration Tips ## Security & Configuration Tips
Do not commit secrets. Copy `.env.example` to `.env` locally. Do not commit secrets. Copy `.env.example` to `.env` locally.
Restrict Slack access with `SLACK_ALLOWED_USER_ID` and `SLACK_ALLOWED_CHANNEL_ID` before running in shared workspaces. Restrict Slack access with `SLACK_ALLOWED_USER_ID` and `SLACK_ALLOWED_CHANNEL_ID` before running in shared workspaces.
## Git & Gitea Workflow Notes
Use `tea` (Gitea CLI), not `gh`.
Hard rule:
- Never implement changes, stage files, or commit on `main`.
- Always create/use a feature branch first, and merge via PR.
- Check login:
- `tea login ls`
- `tea whoami`
- Create/update feature branch:
- `git checkout -b <branch>` (or `git checkout <branch>`)
- `git add <files>`
- `git commit -m "<message>"`
- Push to Gitea:
- Normal: `git push -u origin <branch>`
- If HTTP username prompt fails in this environment, use the token from `~/.config/tea/config.yml`:
- `TOKEN=$(sed -n 's/^[[:space:]]*token: //p' ~/.config/tea/config.yml | head -n1)`
- `git push "http://agentson:${TOKEN}@localhost:3000/jihoson/LazyEnter.git" <branch>`
- After token-based push, ensure tracking is on `origin/<branch>` (not token URL):
- `git fetch origin <branch>:refs/remotes/origin/<branch>`
- `git branch --set-upstream-to=origin/<branch> <branch>`
- Create PR on Gitea:
- `tea pr create --base main --head <branch> --title "<title>" --description "<body>"`
- If PR body includes backticks (`` ` ``), slash commands, or markdown that can be shell-expanded, do not pass it directly in double quotes.
- Preferred safe flow:
- `cat > /tmp/pr_body.md <<'EOF'` ... `EOF`
- `tea pr create --base main --head <branch> --title "<title>" --description "$(cat /tmp/pr_body.md)"`
- If the body is malformed after creation, patch it with API:
- `tea api -X PATCH repos/{owner}/{repo}/pulls/<number> -F body=@/tmp/pr_body.md`
- Sync local main:
- `git checkout main`
- `git pull --ff-only`
- When user confirms the PR is merged, always run this sync immediately to keep local `main` up to date before any next task.
Safety:
- Do not commit or print tokens in logs/docs.
- Keep unrelated local files (for example `uv.lock`) out of PRs unless intentionally changed.

25
README.md
View File

@@ -1,15 +1,13 @@
# LazyEnter # LazyEnter
소파에서 엔터나 치자. 소파에서 엔터나 치자.
로컬 PC의 Claude/Codex CLI 세션을 Slack Socket Mode로 중계해 모바일에서도 원격으로 입력/승인을 처리하는 브릿지입니다. 로컬 PC의 Claude Code CLI 세션을 Slack Socket Mode로 중계해 모바일에서도 원격으로 입력/승인을 처리하는 브릿지입니다.
## 동작 방식 ## 동작 방식
1. 로컬에서 Claude 또는 Codex를 tmux 세션으로 미리 실행 1. 로컬에서 Claude를 tmux 세션으로 미리 실행 (`tmux new -s claude claude`)
- Claude: `tmux new -s claude claude` 2. Slack에서 `/start-claude` 실행
- Codex: `tmux new -s codex codex` 3. 브릿지가 기존 tmux 세션에 연결해 pane 출력/입력을 중계
2. Slack에서 `/start-claude` 또는 `/start-codex` 실행
3. 브릿지가 기존 tmux 세션에 attach
4. Slack 채널 메시지가 CLI 입력으로 전달됨 4. Slack 채널 메시지가 CLI 입력으로 전달됨
5. CLI 출력이 Slack으로 다시 전송됨 5. CLI 출력이 Slack으로 다시 전송됨
6. `/stop-claude`로 브릿지 연결 해제 (tmux 세션은 유지) 6. `/stop-claude`로 브릿지 연결 해제 (tmux 세션은 유지)
@@ -37,8 +35,7 @@ cp .env.example .env
- `SLACK_ALLOWED_CHANNEL_ID` - `SLACK_ALLOWED_CHANNEL_ID`
선택 환경 변수: 선택 환경 변수:
- `TMUX_SESSION_NAME` (기본: `claude`, `/start-claude` 대상) - `TMUX_SESSION_NAME` (기본: `claude`)
- `CODEX_TMUX_SESSION_NAME` (기본: `codex`, `/start-codex` 대상)
- `PTY_READ_TIMEOUT` (기본: `5`) - `PTY_READ_TIMEOUT` (기본: `5`)
- `OUTPUT_BUFFER_INTERVAL` (기본: `2.0`) - `OUTPUT_BUFFER_INTERVAL` (기본: `2.0`)
- `MAX_MESSAGE_LENGTH` (기본: `3000`) - `MAX_MESSAGE_LENGTH` (기본: `3000`)
@@ -55,18 +52,14 @@ cp .env.example .env
- Slash Commands 생성: - Slash Commands 생성:
- `/start-claude` - `/start-claude`
- `/stop-claude` - `/stop-claude`
- `/start-codex`
- `/stop-codex`
- 앱을 워크스페이스에 설치 후 토큰을 `.env`에 반영 - 앱을 워크스페이스에 설치 후 토큰을 `.env`에 반영
## 실행 ## 실행
먼저 로컬에서 원하는 tmux 세션을 실행: 먼저 로컬에서 Claude tmux 세션을 실행:
```bash ```bash
tmux new -s claude claude tmux new -s claude claude
# 또는
tmux new -s codex codex
``` ```
그 다음 브릿지 실행: 그 다음 브릿지 실행:
@@ -78,9 +71,9 @@ python -m lazy_enter
``` ```
실행 후 Slack의 허용된 채널에서: 실행 후 Slack의 허용된 채널에서:
- `/start-claude`, `/start-codex`: 기존 세션에 연결 - `/start-claude`: 기존 세션에 연결
- 일반 메시지 전송: 현재 연결된 CLI(Claude/Codex)로 입력 전달 - 일반 메시지 전송: Claude CLI로 입력 전달
- `/stop-claude`, `/stop-codex`: 브릿지 연결 해제 (세션 유지) - `/stop-claude`: 브릿지 연결 해제 (세션 유지)
## 테스트 및 품질 점검 ## 테스트 및 품질 점검

1
pyproject.toml
View File

@@ -11,7 +11,6 @@ authors = [
dependencies = [ dependencies = [
"slack-bolt>=1.21.0", "slack-bolt>=1.21.0",
"slack-sdk>=3.33.0", "slack-sdk>=3.33.0",
"pexpect>=4.9.0",
"python-dotenv>=1.0.0", "python-dotenv>=1.0.0",
] ]

2
src/lazy_enter/__init__.py
View File

@@ -1,3 +1,3 @@
"""LazyEnter - Slack을 통한 Claude/Codex 원격 제어 브릿지.""" """LazyEnter - Slack을 통한 Claude Code 원격 제어 브릿지."""
__version__ = "0.1.0" __version__ = "0.1.0"

131
src/lazy_enter/bridge.py
View File

@@ -25,9 +25,7 @@ class Bridge:
self._output_thread: threading.Thread | None = None self._output_thread: threading.Thread | None = None
self._running = False self._running = False
self._channel: str = self.config.allowed_channel_id self._channel: str = self.config.allowed_channel_id
self._active_target: str | None = None
self._last_sent_output: str = "" self._last_sent_output: str = ""
self._last_sent_fingerprint: str | None = None
self._last_input_at = time.monotonic() self._last_input_at = time.monotonic()
self._last_output_at = time.monotonic() self._last_output_at = time.monotonic()
self._input_idle_reported = False self._input_idle_reported = False
@@ -46,9 +44,6 @@ class Bridge:
if self.pty and self.pty.is_alive: if self.pty and self.pty.is_alive:
self.pty.send(text) self.pty.send(text)
# 입력 이후 출력은 동일 문자열이어도 한 번 더 전달한다.
self._last_sent_output = ""
self._last_sent_fingerprint = None
self._last_input_at = time.monotonic() self._last_input_at = time.monotonic()
self._input_idle_reported = False self._input_idle_reported = False
logger.info("입력 전달: %s", text) logger.info("입력 전달: %s", text)
@@ -70,30 +65,15 @@ class Bridge:
) )
return any(pattern in normalized for pattern in blocked_patterns) return any(pattern in normalized for pattern in blocked_patterns)
@staticmethod def _handle_command(self, command: str, channel: str) -> None:
def _display_name(target: str) -> str:
if target == "codex":
return "Codex"
return "Claude"
def _session_name_for_target(self, target: str) -> str:
if target == "codex":
return self.config.codex_tmux_session_name
return self.config.tmux_session_name
def _handle_command(self, action: str, target: str, channel: str) -> None:
"""슬래시 커맨드를 처리한다.""" """슬래시 커맨드를 처리한다."""
if target not in {"claude", "codex"}: if command == "start":
self.slack.send_message(channel, ":warning: 지원하지 않는 대상입니다.") self._start_session(channel)
return elif command == "stop":
if action == "start":
self._start_session(channel, target)
elif action == "stop":
self._stop_session(channel) self._stop_session(channel)
def _start_session(self, channel: str, target: str) -> None: def _start_session(self, channel: str) -> None:
"""지정한 대상의 tmux 세션에 연결한다.""" """기존 Claude tmux 세션에 연결한다."""
if self.pty and self.pty.is_alive: if self.pty and self.pty.is_alive:
self.slack.send_message( self.slack.send_message(
channel, ":information_source: 이미 세션에 연결되어 있습니다." channel, ":information_source: 이미 세션에 연결되어 있습니다."
@@ -101,13 +81,8 @@ class Bridge:
return return
self._channel = channel self._channel = channel
self._active_target = target
self._last_sent_output = "" self._last_sent_output = ""
self._last_sent_fingerprint = None self.pty = PtyManager(self.config.tmux_session_name)
self.pty = PtyManager(
self._session_name_for_target(target),
cli_name=target,
)
try: try:
self.pty.start() self.pty.start()
except RuntimeError as exc: except RuntimeError as exc:
@@ -123,10 +98,7 @@ class Bridge:
self._output_thread = threading.Thread(target=self._poll_output, daemon=True) self._output_thread = threading.Thread(target=self._poll_output, daemon=True)
self._output_thread.start() self._output_thread.start()
display_name = self._display_name(target) self.slack.send_message(channel, ":link: Claude 세션에 연결되었습니다.")
self.slack.send_message(
channel, f":link: {display_name} 세션에 연결되었습니다."
)
def _stop_session(self, channel: str) -> None: def _stop_session(self, channel: str) -> None:
"""브릿지 연결만 해제한다.""" """브릿지 연결만 해제한다."""
@@ -134,9 +106,7 @@ class Bridge:
if self.pty: if self.pty:
self.pty.stop() self.pty.stop()
self.pty = None self.pty = None
self._active_target = None
self._last_sent_output = "" self._last_sent_output = ""
self._last_sent_fingerprint = None
self.slack.send_message(channel, ":electric_plug: 세션 연결이 해제되었습니다.") self.slack.send_message(channel, ":electric_plug: 세션 연결이 해제되었습니다.")
def _poll_output(self) -> None: def _poll_output(self) -> None:
@@ -153,7 +123,13 @@ class Bridge:
self._output_idle_reported = False self._output_idle_reported = False
if buffer: if buffer:
self._send_output_chunks(buffer) # 메시지 길이 제한 적용
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 = "" buffer = ""
output_idle = now - self._last_output_at output_idle = now - self._last_output_at
@@ -191,83 +167,6 @@ class Bridge:
# attach 프로세스가 예기치 않게 종료된 경우 # attach 프로세스가 예기치 않게 종료된 경우
self.slack.send_message(self._channel, ":warning: 세션 연결이 종료되었습니다.") self.slack.send_message(self._channel, ":warning: 세션 연결이 종료되었습니다.")
@staticmethod
def _split_message(text: str, max_length: int) -> list[str]:
"""긴 텍스트를 메시지 길이 제한에 맞게 분할한다."""
if max_length <= 0:
return [text] if text else []
lines = text.splitlines(keepends=True)
chunks: list[str] = []
current = ""
def flush() -> None:
nonlocal current
if current:
chunks.append(current)
current = ""
for line in lines:
if len(line) > max_length:
flush()
start = 0
while start < len(line):
piece = line[start : start + max_length]
chunks.append(piece)
start += max_length
continue
if len(current) + len(line) > max_length:
flush()
current += line
flush()
return chunks
def _send_output_chunks(self, text: str) -> None:
"""출력을 잘라서 Slack으로 순차 전송한다."""
chunks = self._split_message(text, self.config.max_message_length)
snapshot = "\n\x00".join(chunks)
fingerprint = self._output_fingerprint(snapshot)
if not chunks:
return
if (
self._last_sent_fingerprint is not None
and fingerprint == self._last_sent_fingerprint
):
return
for chunk in chunks:
if chunk.endswith("\n"):
message = f"```\n{chunk}```"
else:
message = f"```\n{chunk}\n```"
self.slack.send_message(self._channel, message)
self._last_sent_output = snapshot
self._last_sent_fingerprint = fingerprint
@staticmethod
def _output_fingerprint(text: str) -> str:
"""중복 전송 억제를 위한 정규화 지문을 생성한다."""
normalized_lines: list[str] = []
for raw_line in text.splitlines():
line = raw_line.rstrip()
if not line.strip():
continue
# tmux 상태줄의 시계/날짜 라인은 프레임마다 변할 수 있어 제외한다.
if re.fullmatch(
(
r'\s*\w*odex\]\s+\d+:[^\[]*'
r'\[\d+,\d+\]\s+".+"\s+\d{2}:\d{2}\s+\d{2}-[A-Za-z]{3}-\d{2}'
),
line,
):
continue
normalized_lines.append(line)
return "\n".join(normalized_lines)
def run(self) -> None: def run(self) -> None:
"""브릿지를 시작한다.""" """브릿지를 시작한다."""
logging.basicConfig( logging.basicConfig(

1
src/lazy_enter/config.py
View File

@@ -20,7 +20,6 @@ class Config:
# PTY / tmux attach # PTY / tmux attach
tmux_session_name: str = os.getenv("TMUX_SESSION_NAME", "claude") tmux_session_name: str = os.getenv("TMUX_SESSION_NAME", "claude")
codex_tmux_session_name: str = os.getenv("CODEX_TMUX_SESSION_NAME", "codex")
pty_read_timeout: int = int(os.getenv("PTY_READ_TIMEOUT", "5")) pty_read_timeout: int = int(os.getenv("PTY_READ_TIMEOUT", "5"))
# Buffer # Buffer

53
src/lazy_enter/output_filter.py
View File

@@ -15,67 +15,18 @@ _ANSI_ESCAPE_RE = re.compile(
# ESC 문자가 유실된 상태로 남는 CSI 토큰(예: [38;2;153;153;153m) 제거. # ESC 문자가 유실된 상태로 남는 CSI 토큰(예: [38;2;153;153;153m) 제거.
_BROKEN_CSI_RE = re.compile(r"\[(?:[0-9;<=>?]|(?:\d+;))*\d*[A-Za-z]") _BROKEN_CSI_RE = re.compile(r"\[(?:[0-9;<=>?]|(?:\d+;))*\d*[A-Za-z]")
# ESC(B 같은 문자셋 지정 시퀀스가 ESC 유실 후 남긴 토큰 제거.
_BROKEN_CHARSET_RE = re.compile(r"[\(\)][B0]")
# C0 제어문자 중 개행/탭/캐리지리턴을 제외하고 제거. # C0 제어문자 중 개행/탭/캐리지리턴을 제외하고 제거.
_CONTROL_RE = re.compile(r"[\x00-\x08\x0B-\x1F\x7F]") _CONTROL_RE = re.compile(r"[\x00-\x08\x0B-\x1F\x7F]")
_CODEX_TUI_LINE_PATTERNS = (
re.compile(r"openai codex", re.IGNORECASE),
re.compile(r"/model to change", re.IGNORECASE),
re.compile(r"^\s*[|│]\s*model:", re.IGNORECASE),
re.compile(r"^\s*[|│]\s*directory:", re.IGNORECASE),
re.compile(r"^\s*tip:\s*new\s+codex", re.IGNORECASE),
re.compile(r"find and fix a bug in @filename", re.IGNORECASE),
re.compile(r"\?\s*for shortcuts", re.IGNORECASE),
re.compile(r"\bcontext left\b", re.IGNORECASE),
re.compile(r"^\s*\w*odex\]\s+\d+:", re.IGNORECASE),
)
def _is_box_border_line(line: str) -> bool:
stripped = line.strip()
if not stripped:
return False
# ESC(B 유실 잔재가 앞에 붙는 경우(예: =╭────╮)도 border로 본다.
stripped = stripped.lstrip("= ")
return all(ch in "╭╮╰╯─│┌┐└┘├┤┬┴┼" for ch in stripped)
def _is_box_empty_line(line: str) -> bool:
stripped = line.strip()
if not stripped:
return False
return bool(re.fullmatch(r"[│|]\s+[│|]", stripped))
def _is_noise_line(line: str) -> bool:
if _is_box_border_line(line):
return True
if _is_box_empty_line(line):
return True
return any(pattern.search(line) for pattern in _CODEX_TUI_LINE_PATTERNS)
def clean_terminal_output(text: str) -> str: def clean_terminal_output(text: str) -> str:
"""Slack 전송용 텍스트를 정리한다.""" """Slack 전송용 텍스트를 정리한다."""
cleaned = _ANSI_ESCAPE_RE.sub("", text) cleaned = _ANSI_ESCAPE_RE.sub("", text)
cleaned = _BROKEN_CSI_RE.sub("", cleaned) cleaned = _BROKEN_CSI_RE.sub("", cleaned)
cleaned = _BROKEN_CHARSET_RE.sub("", cleaned)
cleaned = cleaned.replace("\r", "") cleaned = cleaned.replace("\r", "")
cleaned = _CONTROL_RE.sub("", cleaned) cleaned = _CONTROL_RE.sub("", cleaned)
lines: list[str] = [] # 제어 코드 제거 후 남는 빈 줄을 축소한다.
for raw_line in cleaned.splitlines(): lines = [line.rstrip() for line in cleaned.splitlines()]
line = raw_line.rstrip()
if not line:
continue
if _is_noise_line(line):
continue
if lines and lines[-1] == line:
continue
lines.append(line)
compact = "\n".join(lines).strip() compact = "\n".join(lines).strip()
return compact return compact

105
src/lazy_enter/pty_manager.py
View File

@@ -1,74 +1,105 @@
"""기존 tmux 세션에 attach하여 CLI 입출력을 중계한다.""" """기존 tmux 세션의 pane을 캡처/입력하여 CLI 입출력을 중계한다."""
from __future__ import annotations from __future__ import annotations
import logging import logging
import subprocess import subprocess
import pexpect
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class PtyManager: class PtyManager:
"""기존 tmux 세션에 attach하고 입출력을 제어한다.""" """기존 tmux 세션의 pane을 읽고 입력을 보낸다."""
def __init__(self, session_name: str = "claude", cli_name: str = "claude") -> None: def __init__(self, session_name: str = "claude") -> None:
self.session_name = session_name self.session_name = session_name
self.cli_name = cli_name self._connected = False
self._process: pexpect.spawn | None = None self._last_capture: list[str] = []
@property @property
def is_alive(self) -> bool: 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: def _run_tmux(self, args: list[str]) -> subprocess.CompletedProcess[str]:
"""attach 대상 tmux 세션 존재 여부를 검증한다.""" return subprocess.run(
result = subprocess.run( ["tmux", *args],
["tmux", "has-session", "-t", self.session_name],
check=False, check=False,
capture_output=True, capture_output=True,
text=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: if result.returncode != 0:
raise RuntimeError( raise RuntimeError(
"tmux 세션이 없습니다. 먼저 실행하세요: " "tmux 세션이 없습니다. 먼저 실행하세요: "
f"tmux new -s {self.session_name} {self.cli_name}" f"tmux new -s {self.session_name} claude"
) )
def start(self) -> None: def _capture_lines(self) -> list[str]:
"""기존 tmux 세션에 attach한다.""" """pane의 최근 출력 스냅샷을 줄 단위로 가져온다."""
self._ensure_session_exists() result = self._run_tmux(
logger.info("tmux 세션 attach: %s", self.session_name) ["capture-pane", "-p", "-t", self.session_name, "-S", "-200"]
self._process = pexpect.spawn(
"tmux",
["attach-session", "-t", self.session_name],
encoding="utf-8",
timeout=None,
) )
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: def send(self, text: str) -> None:
"""프로세스에 텍스트 입력을 전달한다.""" """세션에 텍스트 입력을 전달한다."""
if not self.is_alive: if not self.is_alive:
raise RuntimeError("프로세스가 실행 중이 아닙니다.") raise RuntimeError("프로세스가 실행 중이 아닙니다.")
assert self._process is not None
logger.debug("입력 전송: %s", text) 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: def read_output(self, timeout: int = 5) -> str:
"""프로세스의 출력을 읽는다.""" """세션 pane 출력을 읽는다."""
_ = timeout
if not self.is_alive: if not self.is_alive:
raise RuntimeError("프로세스가 실행 중이 아닙니다.") raise RuntimeError("프로세스가 실행 중이 아닙니다.")
assert self._process is not None current = self._capture_lines()
try: delta_lines = self._diff_lines(self._last_capture, current)
self._process.expect(pexpect.TIMEOUT, timeout=timeout) self._last_capture = current
except pexpect.TIMEOUT: return "\n".join(delta_lines).strip()
pass
return self._process.before or ""
def stop(self) -> None: def stop(self) -> None:
"""attach 연결만 종료한다(원격 세션은 유지).""" """브릿지 연결만 종료한다(원격 세션은 유지)."""
if self._process is not None: if self._connected:
logger.info("tmux attach 종료") logger.info("tmux 세션 연결 해제")
self._process.close(force=True) self._connected = False
self._process = None self._last_capture = []

43
src/lazy_enter/slack_handler.py
View File

@@ -4,7 +4,6 @@ from __future__ import annotations
import logging import logging
import time import time
from collections.abc import Callable
from slack_bolt import App from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler from slack_bolt.adapter.socket_mode import SocketModeHandler
@@ -23,8 +22,8 @@ class SlackHandler:
self.app = App(token=config.slack_bot_token) self.app = App(token=config.slack_bot_token)
self._handler: SocketModeHandler | None = None self._handler: SocketModeHandler | None = None
self._stop_requested = False self._stop_requested = False
self._on_message_callback: Callable[[str, str], None] | None = None self._on_message_callback: callable | None = None
self._on_command_callback: Callable[[str, str, str], None] | None = None self._on_command_callback: callable | None = None
self._register_listeners() self._register_listeners()
@@ -47,7 +46,7 @@ class SlackHandler:
"""Slack 이벤트 리스너를 등록한다.""" """Slack 이벤트 리스너를 등록한다."""
@self.app.event("message") @self.app.event("message")
def handle_message(event: dict, say: Callable[..., None]) -> None: def handle_message(event: dict, say: callable) -> None:
user_id = event.get("user", "") user_id = event.get("user", "")
channel_id = event.get("channel", "") channel_id = event.get("channel", "")
text = event.get("text", "") text = event.get("text", "")
@@ -59,7 +58,7 @@ class SlackHandler:
self._on_message_callback(text, channel_id) self._on_message_callback(text, channel_id)
@self.app.command("/start-claude") @self.app.command("/start-claude")
def handle_start_claude(ack: Callable[..., None], body: dict) -> None: def handle_start(ack: callable, body: dict) -> None:
ack() ack()
user_id = body.get("user_id", "") user_id = body.get("user_id", "")
channel_id = body.get("channel_id", "") channel_id = body.get("channel_id", "")
@@ -68,10 +67,10 @@ class SlackHandler:
return return
if self._on_command_callback: if self._on_command_callback:
self._on_command_callback("start", "claude", channel_id) self._on_command_callback("start", channel_id)
@self.app.command("/stop-claude") @self.app.command("/stop-claude")
def handle_stop_claude(ack: Callable[..., None], body: dict) -> None: def handle_stop(ack: callable, body: dict) -> None:
ack() ack()
user_id = body.get("user_id", "") user_id = body.get("user_id", "")
channel_id = body.get("channel_id", "") channel_id = body.get("channel_id", "")
@@ -80,37 +79,13 @@ class SlackHandler:
return return
if self._on_command_callback: if self._on_command_callback:
self._on_command_callback("stop", "claude", channel_id) self._on_command_callback("stop", channel_id)
@self.app.command("/start-codex") def on_message(self, callback: callable) -> None:
def handle_start_codex(ack: Callable[..., None], body: dict) -> None:
ack()
user_id = body.get("user_id", "")
channel_id = body.get("channel_id", "")
if not self._is_authorized(user_id, channel_id):
return
if self._on_command_callback:
self._on_command_callback("start", "codex", channel_id)
@self.app.command("/stop-codex")
def handle_stop_codex(ack: Callable[..., None], body: dict) -> None:
ack()
user_id = body.get("user_id", "")
channel_id = body.get("channel_id", "")
if not self._is_authorized(user_id, channel_id):
return
if self._on_command_callback:
self._on_command_callback("stop", "codex", channel_id)
def on_message(self, callback: Callable[[str, str], None]) -> None:
"""메시지 수신 콜백을 등록한다.""" """메시지 수신 콜백을 등록한다."""
self._on_message_callback = callback self._on_message_callback = callback
def on_command(self, callback: Callable[[str, str, str], None]) -> None: def on_command(self, callback: callable) -> None:
"""슬래시 커맨드 콜백을 등록한다.""" """슬래시 커맨드 콜백을 등록한다."""
self._on_command_callback = callback self._on_command_callback = callback

299
tests/test_bridge.py
View File

@@ -1,299 +0,0 @@
"""Bridge 명령 라우팅 테스트."""
from __future__ import annotations
import sys
import types
slack_bolt = types.ModuleType("slack_bolt")
slack_bolt.App = object
socket_mode = types.ModuleType("slack_bolt.adapter.socket_mode")
socket_mode.SocketModeHandler = object
slack_sdk = types.ModuleType("slack_sdk")
slack_sdk.WebClient = object
sys.modules.setdefault("slack_bolt", slack_bolt)
sys.modules.setdefault("slack_bolt.adapter.socket_mode", socket_mode)
sys.modules.setdefault("slack_sdk", slack_sdk)
class FakeSlackHandler:
def __init__(self, _config) -> None:
self.message_callback = None
self.command_callback = None
self.sent_messages: list[tuple[str, str]] = []
def on_message(self, callback) -> None:
self.message_callback = callback
def on_command(self, callback) -> None:
self.command_callback = callback
def send_message(self, channel: str, text: str) -> None:
self.sent_messages.append((channel, text))
def start(self) -> None:
return None
def stop(self) -> None:
return None
class FakePtyManager:
instances: list[FakePtyManager] = []
def __init__(self, session_name: str, cli_name: str = "claude") -> None:
self.session_name = session_name
self.cli_name = cli_name
self._alive = False
self.sent_inputs: list[str] = []
FakePtyManager.instances.append(self)
@property
def is_alive(self) -> bool:
return self._alive
def start(self) -> None:
self._alive = True
def stop(self) -> None:
self._alive = False
def send(self, text: str) -> None:
self.sent_inputs.append(text)
def read_output(self, timeout: int = 5) -> str:
return ""
def _make_bridge(monkeypatch):
from lazy_enter.bridge import Bridge
from lazy_enter.config import Config
monkeypatch.setattr("lazy_enter.bridge.SlackHandler", FakeSlackHandler)
monkeypatch.setattr("lazy_enter.bridge.PtyManager", FakePtyManager)
config = Config()
config.tmux_session_name = "claude-room"
config.codex_tmux_session_name = "codex-room"
return Bridge(config)
def test_start_claude_routes_to_claude_session(monkeypatch) -> None:
FakePtyManager.instances.clear()
bridge = _make_bridge(monkeypatch)
bridge._handle_command("start", "claude", "C1")
assert FakePtyManager.instances[-1].session_name == "claude-room"
assert FakePtyManager.instances[-1].cli_name == "claude"
assert bridge.slack.sent_messages[-1] == (
"C1",
":link: Claude 세션에 연결되었습니다.",
)
def test_start_codex_routes_to_codex_session(monkeypatch) -> None:
FakePtyManager.instances.clear()
bridge = _make_bridge(monkeypatch)
bridge._handle_command("start", "codex", "C1")
assert FakePtyManager.instances[-1].session_name == "codex-room"
assert FakePtyManager.instances[-1].cli_name == "codex"
assert bridge.slack.sent_messages[-1] == (
"C1",
":link: Codex 세션에 연결되었습니다.",
)
def test_start_session_resets_output_dedup_state(monkeypatch) -> None:
FakePtyManager.instances.clear()
bridge = _make_bridge(monkeypatch)
bridge._last_sent_output = "stale"
bridge._last_sent_fingerprint = "stale-fp"
bridge._handle_command("start", "codex", "C1")
assert bridge._last_sent_output == ""
assert bridge._last_sent_fingerprint is None
def test_unknown_target_is_rejected(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._handle_command("start", "unknown", "C1")
assert bridge.slack.sent_messages[-1] == (
"C1",
":warning: 지원하지 않는 대상입니다.",
)
def test_handle_message_resets_last_sent_output_after_input(monkeypatch) -> None:
FakePtyManager.instances.clear()
bridge = _make_bridge(monkeypatch)
bridge._handle_command("start", "codex", "C1")
pty = FakePtyManager.instances[-1]
bridge._last_sent_output = "previous output"
bridge._handle_message("status", "C1")
assert pty.sent_inputs[-1] == "status"
assert bridge._last_sent_output == ""
assert bridge._last_sent_fingerprint is None
def test_split_message_preserves_all_content(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
chunks = bridge._split_message("line1\nline2\nline3", max_length=7)
assert "".join(chunks) == "line1\nline2\nline3"
def test_split_message_preserves_trailing_whitespace_and_blank_line(
monkeypatch,
) -> None:
bridge = _make_bridge(monkeypatch)
text = "ab \n\ncd "
chunks = bridge._split_message(text, max_length=4)
assert "".join(chunks) == text
def test_send_output_chunks_sends_multiple_messages(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._channel = "C1"
bridge._send_output_chunks("1234567890")
assert bridge.slack.sent_messages == [
("C1", "```\n1234567890\n```"),
]
bridge.config.max_message_length = 4
bridge.slack.sent_messages.clear()
bridge._send_output_chunks("abcdefghij")
assert bridge.slack.sent_messages == [
("C1", "```\nabcd\n```"),
("C1", "```\nefgh\n```"),
("C1", "```\nij\n```"),
]
def test_send_output_chunks_skips_duplicate_snapshot(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._channel = "C1"
bridge.config.max_message_length = 4
bridge._send_output_chunks("abcdefghij")
first = list(bridge.slack.sent_messages)
bridge._send_output_chunks("abcdefghij")
assert bridge.slack.sent_messages == first
def test_send_output_chunks_skips_duplicate_with_volatile_status_line(
monkeypatch,
) -> None:
bridge = _make_bridge(monkeypatch)
bridge._channel = "C1"
bridge.config.max_message_length = 1000
first_output = (
"Would you like to make the following edits?\n"
"src/main.py (+68 -1)\n"
'odex] 0:node* [0,0] "jihoson-home" 05:12 17-Feb-26\n'
"Press enter to confirm or esc to cancel\n"
)
second_output = (
"Would you like to make the following edits?\n"
"src/main.py (+68 -1)\n"
'odex] 0:node* [0,0] "jihoson-home" 05:13 17-Feb-26\n'
"Press enter to confirm or esc to cancel\n"
)
bridge._send_output_chunks(first_output)
first = list(bridge.slack.sent_messages)
bridge._send_output_chunks(second_output)
assert bridge.slack.sent_messages == first
def test_send_output_chunks_sends_first_when_fingerprint_is_empty(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._channel = "C1"
volatile_only = 'odex] 0:node* [0,0] "jihoson-home" 05:12 17-Feb-26\n'
bridge._send_output_chunks(volatile_only)
assert bridge.slack.sent_messages == [("C1", f"```\n{volatile_only.rstrip()}\n```")]
bridge._send_output_chunks(volatile_only)
assert bridge.slack.sent_messages == [("C1", f"```\n{volatile_only.rstrip()}\n```")]
def test_send_output_chunks_keeps_non_tmux_timestamp_lines(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._channel = "C1"
bridge.config.max_message_length = 1000
first_output = (
"report generated at 05:12 17-Feb-26\n"
"count=10\n"
)
second_output = (
"report generated at 05:13 17-Feb-26\n"
"count=10\n"
)
bridge._send_output_chunks(first_output)
bridge._send_output_chunks(second_output)
assert bridge.slack.sent_messages == [
("C1", f"```\n{first_output.rstrip()}\n```"),
("C1", f"```\n{second_output.rstrip()}\n```"),
]
def test_send_output_chunks_preserves_whitespace_signals(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._channel = "C1"
bridge.config.max_message_length = 1000
first_output = "def f():\n return 1\n"
second_output = "def f():\n return 1\n"
bridge._send_output_chunks(first_output)
bridge._send_output_chunks(second_output)
assert bridge.slack.sent_messages == [
("C1", f"```\n{first_output.rstrip()}\n```"),
("C1", f"```\n{second_output.rstrip()}\n```"),
]
def test_send_output_chunks_keeps_standalone_time_lines(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._channel = "C1"
bridge.config.max_message_length = 1000
bridge._send_output_chunks("[5:12 AM]\n")
bridge._send_output_chunks("[5:13 AM]\n")
assert bridge.slack.sent_messages == [
("C1", "```\n[5:12 AM]\n```"),
("C1", "```\n[5:13 AM]\n```"),
]
def test_send_output_chunks_keeps_non_tmux_status_like_lines(monkeypatch) -> None:
bridge = _make_bridge(monkeypatch)
bridge._channel = "C1"
bridge.config.max_message_length = 1000
first_output = '[2,3] "job-runner" 05:12 17-Feb-26\n'
second_output = '[2,3] "job-runner" 05:13 17-Feb-26\n'
bridge._send_output_chunks(first_output)
bridge._send_output_chunks(second_output)
assert bridge.slack.sent_messages == [
("C1", "```\n[2,3] \"job-runner\" 05:12 17-Feb-26\n```"),
("C1", "```\n[2,3] \"job-runner\" 05:13 17-Feb-26\n```"),
]

1
tests/test_config.py
View File

@@ -8,7 +8,6 @@ from lazy_enter.config import Config
def test_config_defaults(): def test_config_defaults():
config = Config() config = Config()
assert config.tmux_session_name == "claude" assert config.tmux_session_name == "claude"
assert config.codex_tmux_session_name == "codex"
assert config.pty_read_timeout == 5 assert config.pty_read_timeout == 5
assert config.output_buffer_interval == 2.0 assert config.output_buffer_interval == 2.0
assert config.max_message_length == 3000 assert config.max_message_length == 3000

25
tests/test_output_filter.py
View File

@@ -16,28 +16,3 @@ def test_clean_terminal_output_removes_broken_csi_tokens():
def test_clean_terminal_output_preserves_plain_text(): def test_clean_terminal_output_preserves_plain_text():
raw = "line1\nline2" raw = "line1\nline2"
assert clean_terminal_output(raw) == "line1\nline2" assert clean_terminal_output(raw) == "line1\nline2"
def test_clean_terminal_output_removes_broken_charset_tokens():
raw = "=(Bhello(B"
assert clean_terminal_output(raw) == "=hello"
def test_clean_terminal_output_filters_codex_tui_redraw_noise():
raw = """=(B╭─────────────────────────────────────────────╮(B
│ >_ OpenAI Codex (v0.101.0) │(B
│ model: gpt-5.3-codex /model to change │(B
│ directory: ~/repos/The-Ouroboros │(B
╰─────────────────────────────────────────────╯(B
Tip: New Codex is included in your plan for free
› Find and fix a bug in @filename
? for shortcuts
100% context left
Assistant: 작업을 시작합니다.
"""
assert clean_terminal_output(raw) == "Assistant: 작업을 시작합니다."
def test_clean_terminal_output_removes_box_empty_line_noise():
raw = "│ │\n› Explain this codebase\n"
assert clean_terminal_output(raw) == "› Explain this codebase"

85
tests/test_pty_manager.py
View File

@@ -3,29 +3,17 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Sequence from collections.abc import Sequence
from dataclasses import dataclass
import pytest import pytest
from lazy_enter.pty_manager import PtyManager from lazy_enter.pty_manager import PtyManager
class FakeSpawn: @dataclass
def __init__(self, *_args: object, **_kwargs: object) -> None: class Result:
self.before = "" returncode: int
self._alive = True stdout: str = ""
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
def test_initial_state(): 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 capture_output is True
assert text is True assert text is True
class Result: return Result(returncode=1)
returncode = 1
return Result()
monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run) 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() 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( 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, check: bool,
capture_output: bool, capture_output: bool,
text: bool, text: bool,
@@ -67,18 +90,14 @@ def test_start_and_stop_attach(monkeypatch: pytest.MonkeyPatch):
assert check is False assert check is False
assert capture_output is True assert capture_output is True
assert text is True assert text is True
if cmd[:4] == ["tmux", "has-session", "-t", "claude"]:
class Result: return Result(returncode=0)
returncode = 0 if cmd[:7] == ["tmux", "capture-pane", "-p", "-t", "claude", "-S", "-200"]:
return Result(returncode=0, stdout="same\n")
return Result() return Result(returncode=1)
monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run) monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run)
monkeypatch.setattr("lazy_enter.pty_manager.pexpect.spawn", FakeSpawn)
pty = PtyManager("claude") pty = PtyManager("claude")
pty.start() pty.start()
assert pty.is_alive assert pty.read_output() == ""
pty.stop()
assert not pty.is_alive