Claw Code 깊이 읽기 #6 — Python 포팅 워크스페이스

조현상·2026년 4월 2일
AIAgentClaudeClawCodeclaudecode

ClaudeCode

목록 보기
6/17
post-thumbnail

Claude Code 클론 프로젝트 Claw Code의 소스를 한 줄씩 읽으며, 아키텍처 결정의 "왜"를 추적하는 시리즈입니다.

들어가며: 왜 Python 워크스페이스가 따로 있는가?

5편까지 우리는 Rust 크레이트들의 세계에 머물렀다. trait 기반 추상화, 제네릭 런타임, SSE 파서, 샌드박스까지. 그런데 Claw Code 저장소를 살펴보면 src/ 아래에 순수 Python 모듈이 24개나 존재한다. Rust가 프로덕션 런타임이라면, 이 Python 코드는 무엇을 위한 것일까?

답은 참조 구현(reference implementation)이다. TypeScript로 작성된 원본 Claude Code의 207개 커맨드와 184개 도구를 Python으로 먼저 미러링하고, 이 "목록"과 "동작 시뮬레이션"을 통해 Rust 포팅의 방향을 잡는다. 말하자면 이 Python 워크스페이스는 포팅의 나침반 역할을 한다.

이번 편에서는 이 Python 워크스페이스의 전체 아키텍처를 분석한다. frozen 데이터클래스로 불변성을 보장하는 모델 설계, JSON 스냅샷 기반 메타데이터 관리, 토큰 스코어링 라우팅 알고리즘, 그리고 턴 기반 쿼리 엔진까지 — 규모는 작지만 설계 원칙은 놀라울 만큼 정교하다.

1. 5계층 아키텍처와 모듈 맵

전체 24개 모듈은 다음과 같은 계층으로 분리된다:

CLI Layer          → main.py (27개 서브커맨드)
Runtime Layer      → runtime.py, query_engine.py
Mirroring Layer    → commands.py, tools.py
Data/Context Layer → models.py, context.py, session_store.py
Infrastructure     → setup.py, permissions.py, transcript.py, prefetch.py

여기서 주목할 점은 의존 방향이 항상 위에서 아래로만 흐른다는 것이다. main.py는 runtime.py를 호출하고, runtime.py는 commands.py와 tools.py를 사용하지만, 그 반대는 없다. 순환 의존을 원천 차단하는 이 구조는 Rust 크레이트 워크스페이스의 DAG 의존 그래프와 동일한 철학이다.

2. 불변 데이터 모델: frozen=True의 철학

models.py의 핵심 데이터클래스들을 보자:

@dataclass(frozen=True)
class PortingModule:
    name: str
    responsibility: str
    source_hint: str
    status: str = 'planned'

@dataclass(frozen=True)
class UsageSummary:
    input_tokens: int = 0
    output_tokens: int = 0

    def add_turn(self, prompt: str, output: str) -> 'UsageSummary':
        return UsageSummary(
            input_tokens=self.input_tokens + len(prompt.split()),
            output_tokens=self.output_tokens + len(output.split()),
        )

전체 데이터클래스의 약 85%가 frozen=True다. 이것이 의미하는 바는 명확하다 — 상태 변경은 새 인스턴스 반환으로만 이루어진다.

UsageSummary.add_turn()이 대표적인 예시다. 토큰 수를 누적할 때 기존 객체를 수정하지 않고, 새로운 UsageSummary를 만들어 반환한다. Rust의 소유권 모델에서 Clone + 변환이 관용적이듯, 이 Python 코드도 함수형 불변 패턴을 따른다.

예외적으로 PortingBacklog만 frozen=True가 아닌데, 이는 모듈 목록을 동적으로 추가해야 하기 때문이다. 불변과 가변의 경계를 의도적으로 구분하고 있다는 신호다.

한 가지 흥미로운 디테일: 토큰 카운팅에 len(prompt.split())을 사용한다. Rust 쪽이 chars/4 + 1 휴리스틱을 쓰는 것과 다른 전략이다. 단어 단위 분할은 영어에서 BPE 토큰과 대략 1:1 대응하므로, 참조 구현으로서는 충분히 합리적인 근사다.

3. 스냅샷 기반 메타데이터: 207개 커맨드와 184개 도구

이 워크스페이스의 가장 독특한 설계는 JSON 스냅샷이다:

# commands.py
SNAPSHOT_PATH = Path(__file__).resolve().parent / 'reference_data' / 'commands_snapshot.json'

@lru_cache(maxsize=1)
def load_command_snapshot() -> tuple[PortingModule, ...]:
    raw_entries = json.loads(SNAPSHOT_PATH.read_text())
    return tuple(
        PortingModule(
            name=entry['name'],
            responsibility=entry['responsibility'],
            source_hint=entry['source_hint'],
            status='mirrored',
        )
        for entry in raw_entries
    )

PORTED_COMMANDS = load_command_snapshot()  # 모듈 로드 시점에 싱글턴 생성

commands_snapshot.json에는 원본 TypeScript 아카이브에서 추출한 207개 커맨드의 메타데이터가 들어있고, tools_snapshot.json에는 184개 도구가 있다. 각 항목의 구조는 이렇다:

{
  "name": "AgentTool",
  "source_hint": "tools/AgentTool/AgentTool.tsx",
  "responsibility": "Tool module mirrored from archived TypeScript path"
}

왜 코드가 아니라 스냅샷인가? 이것이 핵심 설계 결정이다. 원본 Claude Code의 207개 커맨드를 전부 Python으로 재구현하는 것은 현실적이지 않다. 대신 메타데이터만 추출해서 "이런 커맨드가 존재한다"는 카탈로그를 만들고, 이를 기반으로 라우팅, 검색, 패리티 추적을 수행한다.

@lru_cache(maxsize=1) + tuple 반환의 조합도 의미 깊다. LRU 캐시는 디스크 I/O를 최초 1회로 제한하고, tuple은 반환된 데이터의 불변성을 보장한다. list로 반환했다면 외부에서 append()나 pop()으로 전역 상태를 오염시킬 수 있다. tuple은 그 가능성을 원천 차단한다.

4. 토큰 스코어링 라우터: O(n×m) 매칭

runtime.py의 PortRuntime.route_prompt()는 사용자 입력을 커맨드/도구에 매칭하는 라우터다:

def route_prompt(self, prompt: str, limit: int = 5) -> list[RoutedMatch]:
    tokens = {token.lower() for token in prompt.replace('/', ' ').replace('-', ' ').split() if token}

    by_kind = {
        'command': self._collect_matches(tokens, PORTED_COMMANDS, 'command'),
        'tool':    self._collect_matches(tokens, PORTED_TOOLS, 'tool'),
    }

    # 각 종류에서 최소 1개씩 선택
    selected: list[RoutedMatch] = []
    for kind in ('command', 'tool'):
        if by_kind[kind]:
            selected.append(by_kind[kind].pop(0))

    # 나머지 슬롯을 스코어 순으로 채움
    leftovers = sorted(
        [match for matches in by_kind.values() for match in matches],
        key=lambda item: (-item.score, item.kind, item.name),
    )
    selected.extend(leftovers[: max(0, limit - len(selected))])
    return selected[:limit]

스코어링 함수는 단순하지만 효과적이다:

@staticmethod
def _score(tokens: set[str], module: PortingModule) -> int:
    haystacks = [module.name.lower(), module.source_hint.lower(), module.responsibility.lower()]
    score = 0
    for token in tokens:
        if any(token in haystack for haystack in haystacks):
            score += 1
    return score

예를 들어 "find bash command"라는 프롬프트가 들어오면:

  • 토큰: {'find', 'bash', 'command'}
  • BashTool (name='bash', source_hint='tools/BashTool/...') → 'bash' 매치 + 'command' 불일치 → 스코어 1
  • 하지만 bash 커맨드(source_hint에 'command' 포함) → 스코어 2

설계의 영리함은 "보장된 다양성"에 있다. 커맨드에서 1개, 도구에서 1개를 먼저 뽑고, 나머지를 스코어 순으로 채운다. limit=5일 때 모든 결과가 커맨드로만 채워지는 것을 방지한다. 이는 검색 엔진의 diversification 전략과 동일한 패턴이다.

시간 복잡도는 O(n×m)으로, n=391(207+184)개 모듈 × m=토큰 수다. 퍼지 매칭이 아닌 부분 문자열 매칭이므로 오탈자에는 취약하지만, 참조 구현으로서는 충분하다.

5. QueryEnginePort: 턴 기반 상태 머신

query_engine.py는 이 워크스페이스의 두뇌다. 대화 턴을 관리하고, 토큰 예산을 추적하며, 메시지를 컴팩트하는 상태 머신이다.

5.1 설정 객체의 동작 파라미터화

@dataclass(frozen=True)
class QueryEngineConfig:
    max_turns: int = 8
    max_budget_tokens: int = 2000
    compact_after_turns: int = 12
    structured_output: bool = False
    structured_retry_limit: int = 2

이 4개의 필드가 엔진의 전체 동작을 결정한다. frozen=True이므로 런타임 중 변경 불가. Rust 쪽 CompactionConfig와 동일한 "설정 객체" 패턴인데, Python에서는 디폴트 값이 있어 QueryEngineConfig()만으로 합리적 기본값이 적용된다.

5.2 submit_message() — 6단계 턴 처리

def submit_message(self, prompt, matched_commands, matched_tools, denied_tools) -> TurnResult:
    # 1단계: 턴 수 한도 확인
    if len(self.mutable_messages) >= self.config.max_turns:
        return TurnResult(..., stop_reason='max_turns_reached')

    # 2단계: 출력 포맷팅
    output = self._format_output(summary_lines)

    # 3단계: 토큰 사용량 투영 (add_turn → 새 UsageSummary)
    projected_usage = self.total_usage.add_turn(prompt, output)

    # 4단계: 예산 초과 확인
    if projected_usage.input_tokens + projected_usage.output_tokens > self.config.max_budget_tokens:
        stop_reason = 'max_budget_reached'

    # 5단계: 상태 변이 (mutable 구간)
    self.mutable_messages.append(prompt)
    self.transcript_store.append(prompt)
    self.total_usage = projected_usage

    # 6단계: 필요시 컴팩트
    self.compact_messages_if_needed()

    return TurnResult(..., stop_reason=stop_reason)

주목할 패턴이 있다. 3단계에서 projected_usage를 먼저 계산하고, 4단계에서 예산 확인 후, 5단계에서야 실제 상태를 변이한다. 즉, "예산을 초과하더라도 메시지는 기록한다." 이것은 의도적 설계다 — stop_reason='max_budget_reached'를 반환하되 해당 턴의 내용 자체는 보존한다. 트랜스크립트의 연속성을 위해서다.

5.3 메시지 컴팩트: 슬라이스 할당의 마법

def compact_messages_if_needed(self) -> None:
    if len(self.mutable_messages) > self.config.compact_after_turns:
        self.mutable_messages[:] = self.mutable_messages[-self.config.compact_after_turns:]
    self.transcript_store.compact(self.config.compact_after_turns)

self.mutable_messages[:] = ... — 이 한 줄에 주목하자. self.mutable_messages = ...와 다르다. 전자는 같은 리스트 객체를 유지하면서 내용만 교체(in-place mutation)하고, 후자는 새 리스트 객체를 바인딩한다. QueryEnginePort가 mutable_messages 참조를 외부와 공유하는 경우(예: 스트리밍 중), 슬라이스 할당이 안전하다.

Rust 쪽의 drain() + extend() 패턴과 의미적으로 동일하다.

6. 스트리밍 이벤트 제너레이터

stream_submit_message()는 Python 제너레이터로 스트리밍을 구현한다:

def stream_submit_message(self, prompt, matched_commands, matched_tools, denied_tools):
    yield {'type': 'message_start', 'session_id': self.session_id, 'prompt': prompt}

    if matched_commands:
        yield {'type': 'command_match', 'commands': matched_commands}
    if matched_tools:
        yield {'type': 'tool_match', 'tools': matched_tools}
    if denied_tools:
        yield {'type': 'permission_denial', 'denials': [d.tool_name for d in denied_tools]}

    result = self.submit_message(prompt, matched_commands, matched_tools, denied_tools)

    yield {'type': 'message_delta', 'text': result.output}
    yield {'type': 'message_stop', 'usage': {...}, 'stop_reason': result.stop_reason}

이벤트 시퀀스가 message_start → match → delta → message_stop인 것에 주목하자. 이것은 Anthropic Messages API의 SSE 이벤트 구조(message_start → content_block_delta → message_stop)를 그대로 미러링한 것이다. Rust 쪽 SseParser가 파싱하는 프레임과 1:1 대응되는 설계다.

Python의 yield 기반 제너레이터는 lazy evaluation이므로, 소비자가 중간에 읽기를 멈추면 이후 이벤트는 생성되지 않는다. Rust의 Stream trait와 동일한 pull 기반 모델이다.

7. 권한 컨텍스트: 2단계 필터링

permissions.py의 ToolPermissionContext는 간결하지만 효과적인 RBAC를 구현한다:

@dataclass(frozen=True)
class ToolPermissionContext:
    deny_names: frozenset[str] = field(default_factory=frozenset)
    deny_prefixes: tuple[str, ...] = ()

    def blocks(self, tool_name: str) -> bool:
        lowered = tool_name.lower()
        return lowered in self.deny_names or any(
            lowered.startswith(prefix) for prefix in self.deny_prefixes
        )

frozenset으로 정확 매칭(O(1) 룩업), tuple로 접두사 매칭(O(k) 순회). 두 가지 차단 전략을 조합한다.

실제 사용 예시:

python -m src tools --deny-tool BashTool --deny-prefix mcp_

이 명령은 BashTool(정확 매칭)과 mcp_로 시작하는 모든 도구(접두사 매칭)를 필터링한다. Rust 쪽의 PermissionMode 3단계 모델(ReadOnly < WorkspaceWrite < DangerFullAccess)보다 단순하지만, 참조 구현에서 "차단" 개념을 빠르게 프로토타이핑하기에 적합한 설계다.

from_iterables() 클래스메서드는 빌더 패턴으로 CLI 인자를 받아 불변 컨텍스트를 생성한다. 한 번 생성되면 어디서도 수정할 수 없으므로 스레드 안전하다.

8. 세션 영속화와 턴 루프

8.1 세션 저장소

@dataclass(frozen=True)
class StoredSession:
    session_id: str
    messages: tuple[str, ...]
    input_tokens: int
    output_tokens: int

def save_session(session: StoredSession, directory: Path | None = None) -> Path:
    target_dir = directory or DEFAULT_SESSION_DIR
    target_dir.mkdir(parents=True, exist_ok=True)
    path = target_dir / f'{session.session_id}.json'
    path.write_text(json.dumps(asdict(session), indent=2))
    return path

uuid4().hex로 세션 ID를 생성하고, .port_sessions/{id}.json에 저장한다. asdict() + json.dumps()는 frozen 데이터클래스의 직렬화에 가장 관용적인 Python 패턴이다.

복원 시 tuple(data['messages'])로 리스트를 다시 튜플로 변환하는 점도 일관성 있다 — JSON에서 역직렬화하면 리스트로 돌아오기 때문이다.

8.2 턴 루프: 반복적 대화 실행

def run_turn_loop(self, prompt, limit=5, max_turns=3, structured_output=False):
    engine = QueryEnginePort.from_workspace(config=QueryEngineConfig(
        max_turns=max_turns * 2,  # 여유 있는 턴 한도
        structured_output=structured_output,
    ))
    routed = self.route_prompt(prompt, limit=limit)

    results = []
    for turn in range(max_turns):
        turn_prompt = prompt if turn == 0 else f'{prompt} [turn {turn + 1}]'
        result = engine.submit_message(turn_prompt, command_names, tool_names, ())
        results.append(result)
        if result.stop_reason != 'completed':
            break
    return results

max_turns * 2로 엔진의 내부 한도를 설정하는 것이 흥미롭다. 턴 루프의 외부 제어(for turn in range(max_turns))와 엔진의 내부 제어(config.max_turns)가 이중 안전장치로 작동한다. 어느 쪽이든 먼저 한도에 도달하면 멈춘다.

턴 번호를 f'{prompt} [turn {turn + 1}]'로 프롬프트에 붙이는 것도 영리하다 — 같은 프롬프트의 반복 제출이 아니라 진행 상황이 표시된 별개의 턴임을 명시한다.

9. 부트스트랩: 10단계 세션 초기화

bootstrap_session()은 전체 흐름을 하나로 엮는 오케스트레이터다:

1. PortContext 생성 (context.py)
2. WorkspaceSetup 실행 — Python 버전, 플랫폼 감지
3. 프리페치 시작 — MDM, keychain, 프로젝트 스캔
4. HistoryLog 초기화
5. QueryEnginePort 생성
6. 프롬프트 라우팅 → RoutedMatch 리스트
7. ExecutionRegistry 빌드 (207+184 항목 등록)
8. 커맨드/도구 실행 시뮬레이션
9. 쿼리 엔진에 메시지 제출
10. 세션 영속화 → RuntimeSession 반환

이 10단계가 RuntimeSession 데이터클래스 하나로 응축된다. 각 단계의 결과물이 세션의 필드로 저장되므로, 나중에 디버깅이나 감사 추적 시 "이 세션이 어떤 경로로 생성되었는가"를 완전히 재구성할 수 있다.

Rust 쪽의 12단계 부트스트랩과 비교하면 단계 수는 비슷하지만, Python 버전은 시뮬레이션 중심이고 Rust 버전은 실제 실행 중심이다. 참조 구현과 프로덕션 구현의 역할 분담이 여기서도 드러난다.

10. Rust와 Python의 대칭: 의도된 미러링

지금까지 분석한 Python 모듈들을 Rust 크레이트와 나란히 놓으면, 설계의 대칭이 보인다:

개념PythonRust
불변 데이터@dataclass(frozen=True)#[derive(Clone)] + 소유권
누적 패턴UsageSummary.add_turn() → 새 인스턴스UsageTracker 누적
스냅샷 로딩@lru_cache(maxsize=1) + tupleonce_cell::Lazy
라우팅토큰 스코어링 _score()trait dispatch
턴 루프submit_message() 상태 머신run_turn() agentic 루프
컴팩트mutable_messages[:] = ...[-N:]summarize_messages() + LLM
스트리밍yield 제너레이터Stream trait + SseParser
세션 저장json.dumps(asdict(...))serde JSON
권한ToolPermissionContext.blocks()PermissionMode Ord 비교

Python은 "빠르게 프로토타이핑하고 구조를 검증"하는 역할, Rust는 "성능과 안전성을 보장하며 프로덕션에 배포"하는 역할. 이 듀얼 포팅 전략이 Claw Code 프로젝트의 근본 철학이다.

마무리: 참조 구현에서 배우는 것

이번 편에서 분석한 Python 워크스페이스의 핵심 설계 원칙을 정리하면:

첫째, 불변 우선 설계. frozen 데이터클래스 85%는 "변경하지 않겠다"는 의지의 표현이다. 가변 상태가 필요한 곳(QueryEnginePort의 mutable_messages, TranscriptStore의 entries)은 명시적으로 구분한다.

둘째, 스냅샷은 코드보다 오래 산다. 207개 커맨드를 전부 구현하지 않아도, 메타데이터 스냅샷만으로 라우팅, 검색, 패리티 추적이 가능하다. "구현 없는 카탈로그"의 가치를 보여주는 설계다.

셋째, 이중 안전장치. 턴 루프의 외부 제어와 엔진 내부 제어, 토큰 예산과 턴 수 한도, 슬라이스 할당과 별도 컴팩트 — 모든 제어 흐름에 두 겹의 보호막이 있다.

넷째, API 미러링은 의도적이다. 스트리밍 이벤트의 message_start → delta → message_stop 시퀀스가 Anthropic API와 동일한 것은 우연이 아니다. 참조 구현이 프로덕션 API의 동작을 시뮬레이션하도록 설계되었다.

실습 가이드

# 워크스페이스 요약 확인
python3 -m src.main summary

# "review MCP tool" 프롬프트 라우팅 테스트
python3 -m src.main route "review MCP tool" --limit 5

# 턴 루프 실행 (3턴)
python3 -m src.main turn-loop "help me find a bash command" --max-turns 3

# 세션 저장 후 복원
python3 -m src.main flush-transcript "test prompt"
ls .port_sessions/  # 생성된 세션 파일 확인

# 도구 목록에서 MCP 제외
python3 -m src.main tools --no-mcp --limit 10

시리즈 목차
1편: 프로젝트 배경과 아키텍처 오버뷰
2편: Rust 워크스페이스와 크레이트 구조
3편: API 통신과 SSE 스트리밍
4편: 대화 런타임과 세션 관리
5편: 도구 시스템과 권한 모델
6편: Python 포팅 워크스페이스 분석 ← 현재 글
7편: 테스팅 전략과 패리티 추적
8편: TUI 개선과 미래 로드맵

#ClawCode #ClaudeCode #Python #Rust #OpenSource #코드분석 #아키텍처 #포팅전략

profile
조현상
꿈꾸는 개발자
이전 포스트

# Claw Code 깊이 읽기 #5 — 도구 시스템과 권한 모델

다음 포스트

Claw Code 깊이 읽기 #7 — 테스팅 전략과 패리티 추적

0개의 댓글