3 Commits

5 changed files with 95 additions and 113 deletions

View File

@@ -42,3 +42,35 @@ 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>"`
- Sync local main:
- `git checkout main`
- `git pull --ff-only`
Safety:
- Do not commit or print tokens in logs/docs.
- Keep unrelated local files (for example `uv.lock`) out of PRs unless intentionally changed.

View File

@@ -7,7 +7,7 @@
1. 로컬에서 Claude를 tmux 세션으로 미리 실행 (`tmux new -s claude claude`) 1. 로컬에서 Claude를 tmux 세션으로 미리 실행 (`tmux new -s claude claude`)
2. Slack에서 `/start-claude` 실행 2. Slack에서 `/start-claude` 실행
3. 브릿지가 기존 tmux 세션에 연결해 pane 출력/입력을 중계 3. 브릿지가 기존 tmux 세션에 attach
4. Slack 채널 메시지가 CLI 입력으로 전달됨 4. Slack 채널 메시지가 CLI 입력으로 전달됨
5. CLI 출력이 Slack으로 다시 전송됨 5. CLI 출력이 Slack으로 다시 전송됨
6. `/stop-claude`로 브릿지 연결 해제 (tmux 세션은 유지) 6. `/stop-claude`로 브릿지 연결 해제 (tmux 세션은 유지)

View File

@@ -11,6 +11,7 @@ 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",
] ]

View File

@@ -1,105 +1,73 @@
"""기존 tmux 세션의 pane을 캡처/입력하여 CLI 입출력을 중계한다.""" """기존 tmux 세션에 attach하여 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 세션의 pane을 읽고 입력을 보낸다.""" """기존 tmux 세션에 attach하고 입력을 제어한다."""
def __init__(self, session_name: str = "claude") -> None: def __init__(self, session_name: str = "claude") -> None:
self.session_name = session_name self.session_name = session_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._connected and self._session_exists() return self._process is not None and self._process.isalive()
def _run_tmux(self, args: list[str]) -> subprocess.CompletedProcess[str]: def _ensure_session_exists(self) -> None:
return subprocess.run( """attach 대상 tmux 세션 존재 여부를 검증한다."""
["tmux", *args], result = subprocess.run(
["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} claude" f"tmux new -s {self.session_name} claude"
) )
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: def start(self) -> None:
"""기존 tmux 세션에 연결한다.""" """기존 tmux 세션에 attach한다."""
self._ensure_session_exists() self._ensure_session_exists()
logger.info("tmux 세션 연결: %s", self.session_name) logger.info("tmux 세션 attach: %s", self.session_name)
self._connected = True self._process = pexpect.spawn(
self._last_capture = self._capture_lines() "tmux",
["attach-session", "-t", self.session_name],
encoding="utf-8",
timeout=None,
)
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)
result = self._run_tmux(["send-keys", "-t", self.session_name, text, "C-m"]) self._process.sendline(text)
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("프로세스가 실행 중이 아닙니다.")
current = self._capture_lines() assert self._process is not None
delta_lines = self._diff_lines(self._last_capture, current) try:
self._last_capture = current self._process.expect(pexpect.TIMEOUT, timeout=timeout)
return "\n".join(delta_lines).strip() except pexpect.TIMEOUT:
pass
return self._process.before or ""
def stop(self) -> None: def stop(self) -> None:
"""브릿지 연결만 종료한다(원격 세션은 유지).""" """attach 연결만 종료한다(원격 세션은 유지)."""
if self._connected: if self._process is not None:
logger.info("tmux 세션 연결 해제") logger.info("tmux attach 종료")
self._connected = False self._process.close(force=True)
self._last_capture = [] self._process = None

View File

@@ -3,17 +3,29 @@
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
@dataclass class FakeSpawn:
class Result: def __init__(self, *_args: object, **_kwargs: object) -> None:
returncode: int self.before = ""
stdout: str = "" 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
def test_initial_state(): def test_initial_state():
@@ -33,7 +45,10 @@ 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
return Result(returncode=1) class Result:
returncode = 1
return Result()
monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run) monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run)
@@ -42,62 +57,28 @@ def test_start_raises_when_tmux_session_missing(monkeypatch: pytest.MonkeyPatch)
pty.start() pty.start()
def test_start_send_read_and_stop(monkeypatch: pytest.MonkeyPatch): def test_start_and_stop_attach(monkeypatch: pytest.MonkeyPatch):
calls: list[list[str]] = []
def fake_run( def fake_run(
cmd: Sequence[str], _cmd: Sequence[str],
check: bool, check: bool,
capture_output: bool, capture_output: bool,
text: bool, text: bool,
): ):
calls.append(list(cmd))
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"]:
# 첫 캡처는 baseline, 두 번째 캡처에서 한 줄 증가. return Result()
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) 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.is_alive
pty.send("hello")
assert pty.read_output() == "line3"
pty.stop() pty.stop()
assert not pty.is_alive 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,
):
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"]:
return Result(returncode=0, stdout="same\n")
return Result(returncode=1)
monkeypatch.setattr("lazy_enter.pty_manager.subprocess.run", fake_run)
pty = PtyManager("claude")
pty.start()
assert pty.read_output() == ""