<!-- task=python_impl plan=0 tag=API -->

# Python 구현체 추가 (toki_socket)

## 이 파일을 읽는 구현 에이전트에게

각 항목의 체크리스트를 완료하면서 `[ ]`를 `[x]`로 표시하라.
중간 검증 명령은 해당 항목 구현 직후 실행하고, 출력을 `CODE_REVIEW.md`의 검증 결과 섹션에 붙여넣어라.
최종 검증도 마찬가지로 실행 후 출력을 기록하라.
계획과 다르게 구현한 부분은 이유와 함께 `CODE_REVIEW.md`의 "계획 대비 변경 사항"에 기록하라.

---

## 배경

`PROTOCOL.md`의 Multi-language Implementations 표에 Python이 `Planned`로 등록되어 있다.
Go 구현체(`go/`)가 레퍼런스이며, `PORTING_GUIDE.md`에 Python 매핑 표와 주의사항이 명시되어 있다.
Python 패키지를 `python/` 디렉터리에 생성하고, Go×Python 양방향 크로스 테스트를 추가하여 프로토콜 호환성을 검증한다.
TLS/WSS는 이번 범위에서 제외하고 README에 명시한다.

---

## 의존 관계 및 구현 순서

```
[API-1] scaffold → [API-2] Communicator → [API-3] BaseClient
                                         → [API-4] TCP Transport
                                         → [API-5] WebSocket Transport
[API-2~5] 완료 → [API-6] 단위 테스트 → [API-7] 크로스 테스트
```

---

## [API-1] Python 패키지 scaffold

### 문제

`python/` 디렉터리가 없다. 패키지 매니저, proto binding, 최소 패키지 구조가 필요하다.

### 해결 방법

- `python/pyproject.toml` 생성 (setuptools, Python 3.11+)
- `python/toki_socket/__init__.py` 생성
- `dart/lib/src/packets/message_common.proto`를 `python/toki_socket/packets/message_common.proto`로 복사 (schema만, 언어 option 추가 불필요)
- `protoc`로 `message_common_pb2.py`, `message_common_pb2.pyi` 생성

**pyproject.toml 핵심**:
```toml
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.backends.legacy:build"

[project]
name = "toki-socket"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "protobuf>=4.25",
    "websockets>=12.0",
]

[project.optional-dependencies]
dev = ["pytest>=8.0", "pytest-asyncio>=0.23"]
```

**디렉터리 구조**:
```
python/
  toki_socket/
    __init__.py
    communicator.py
    base_client.py
    tcp_client.py
    tcp_server.py
    ws_client.py
    ws_server.py
    packets/
      __init__.py
      message_common.proto
      message_common_pb2.py      (generated)
      message_common_pb2.pyi     (generated)
  test/
    __init__.py
    test_communicator.py
    test_tcp.py
  crosstest/
    go_python_client.py
    python_go.py
  pyproject.toml
  README.md
```

### 수정 파일 및 체크리스트

- [ ] `python/pyproject.toml` 생성
- [ ] `python/toki_socket/__init__.py` 생성 (public export 비워둠)
- [ ] `python/toki_socket/packets/__init__.py` 생성
- [ ] `python/toki_socket/packets/message_common.proto` 복사
- [ ] `python/toki_socket/packets/message_common_pb2.py` 생성 (protoc 실행 또는 직접 작성)
- [ ] `python/toki_socket/packets/message_common_pb2.pyi` 생성
- [ ] `python/README.md` 생성 (TLS 미지원 명시 포함)

### 테스트 작성

skip — scaffold 단계, 코드 로직 없음.

### 중간 검증

```bash
cd python && pip install -e ".[dev]" --quiet && python -c "from toki_socket.packets import message_common_pb2; print(message_common_pb2.TestData.DESCRIPTOR.name)"
```

예상 출력: `TestData`

---

## [API-2] Communicator 구현

### 문제

`go/communicator.go`에 해당하는 Python 비동기 구현이 없다.
`Transport` Protocol, `Communicator` 클래스, write 큐, nonce 관리, listener/requestListener 라우팅을 구현해야 한다.

### 해결 방법

`python/toki_socket/communicator.py`에 아래 구조로 구현한다.

**Transport Protocol** (Go의 `Transport` interface):
```python
from typing import Protocol, runtime_checkable
from toki_socket.packets.message_common_pb2 import PacketBase

@runtime_checkable
class Transport(Protocol):
    async def write_packet(self, base: PacketBase) -> None: ...
    async def close(self) -> None: ...
```

**ParserMap**: `dict[str, Callable[[bytes], Message]]`

**Communicator 핵심**:
```python
class Communicator:
    def __init__(self) -> None:
        self._nonce: int = 0
        self._is_alive: bool = False
        self._parser_map: dict[str, Callable] = {}
        self._handlers: dict[str, list[Callable]] = {}
        self._req_handlers: dict[str, Callable] = {}
        self._pending_requests: dict[int, asyncio.Future] = {}
        self._write_queue: asyncio.Queue = asyncio.Queue(maxsize=64)
        self._closed: asyncio.Event = asyncio.Event()
        self._transport: Transport | None = None
        self._write_loop_task: asyncio.Task | None = None
        self._write_error_handler: Callable[[Exception], None] | None = None

    def initialize(self, transport: Transport, parser_map: dict) -> None:
        ...  # 하트비트 파서 추가, write_loop Task 시작

    def is_alive(self) -> bool: ...
    def _next_nonce(self) -> int: ...
    def _shutdown(self) -> None: ...
    async def close(self) -> None: ...
    async def _write_loop(self) -> None: ...
    async def queue_packet(self, base: PacketBase) -> None: ...
    async def send(self, message: Message) -> None: ...
    async def send_request(self, req: Message, res_type: type[T], timeout: float = 30.0) -> T: ...
    def add_listener(self, type_name: str, fn: Callable) -> None: ...
    def remove_listeners(self, type_name: str) -> None: ...
    def add_request_listener(self, type_name: str, fn: Callable) -> None: ...
    def on_received_data(self, type_name: str, data: bytes, nonce: int, response_nonce: int) -> None: ...
    def _handle_response(self, type_name: str, data: bytes, response_nonce: int) -> None: ...
    def _parse(self, type_name: str, data: bytes) -> Message: ...

def type_name_of(message_class) -> str:
    return message_class.DESCRIPTOR.name
```

**write_loop 패턴** — asyncio 단일 스레드이므로 sentinel 방식 사용:
```python
_STOP = object()

async def _write_loop(self) -> None:
    while True:
        item = await self._write_queue.get()
        if item is _STOP:
            return
        base, fut = item
        try:
            await self._transport.write_packet(base)
            if not fut.done():
                fut.set_result(None)
        except Exception as e:
            if not fut.done():
                fut.set_exception(e)
            if self._write_error_handler:
                self._write_error_handler(e)
```

`_shutdown()`에서 `_write_queue.put_nowait(_STOP)` 호출.

**send_request** — `asyncio.get_event_loop().create_future()`로 pending 등록, `asyncio.wait_for(fut, timeout)`으로 대기.

**addListener / addRequestListener 상호 배타** — 같은 typeName 중복 시 `ValueError` raise.

**HeartBeat 자동 처리** — `initialize()` 내에서 HeartBeat 파서를 `_parser_map`에 추가; BaseClient에서 listener 등록.

**on_received_data** — asyncio 단일 스레드이므로 요청 핸들러는 `asyncio.create_task()`로 실행 (Go의 `go reqHandler(...)`와 동일).

### 수정 파일 및 체크리스트

- [ ] `python/toki_socket/communicator.py` 생성
  - [ ] `Transport` Protocol 정의
  - [ ] `type_name_of(cls)` 함수 — `cls.DESCRIPTOR.name` 반환
  - [ ] `Communicator.initialize()` — parser_map + HeartBeat 파서 등록, write_loop Task 시작
  - [ ] `Communicator._write_loop()` — sentinel 패턴
  - [ ] `Communicator.queue_packet()` — closed 체크, Future 등록, put, await
  - [ ] `Communicator.send()` — marshal → queue_packet
  - [ ] `Communicator.send_request()` — pending 등록, marshal, queue, asyncio.wait_for
  - [ ] `Communicator.add_listener()` / `add_request_listener()` — 상호 배타 검사
  - [ ] `Communicator.on_received_data()` — response_nonce > 0이면 handleResponse, 아니면 listener/reqHandler
  - [ ] `Communicator._shutdown()` — is_alive=False, closed.set(), write_queue에 STOP 전달

### 테스트 작성

`python/test/test_communicator.py`에 작성:
- `test_add_listener_exclusivity` — 같은 typeName에 add_listener 후 add_request_listener 시 ValueError 확인
- `test_send_and_receive` — FakeTransport (write_packet 캡처)로 send() 후 패킷 내용 검증
- `test_send_request_response` — FakeTransport로 send_request 후 수동으로 on_received_data 호출하여 Future resolve 검증
- `test_shutdown` — close() 후 is_alive() == False 검증

### 중간 검증

```bash
cd python && python -m pytest test/test_communicator.py -v
```

예상: 4개 테스트 PASS

---

## [API-3] BaseClient 구현

### 문제

`go/base_client.go`에 해당하는 heartbeat, disconnect listener, connCloseOnce 패턴이 없다.

### 해결 방법

`python/toki_socket/base_client.py`에 구현.

Go `sync.Once` → `asyncio.Lock` + `bool` 플래그:
```python
self._close_called: bool = False
self._close_lock: asyncio.Lock = asyncio.Lock()
```

```python
class BaseClient:
    def __init__(self, interval_sec: int, wait_sec: int, do_close: Callable[[], Awaitable[None]]) -> None:
        self.communicator: Communicator = Communicator()
        self._interval_sec = interval_sec
        self._wait_sec = wait_sec
        self._do_close = do_close
        self._close_called = False
        self._close_lock: asyncio.Lock  # initialize()에서 생성
        self._disconnect_listeners: list[Callable] = []
        self._hb_task: asyncio.Task | None = None
        self._waiting_hb_response: bool = False

    def _init_base(self) -> None:
        self._close_lock = asyncio.Lock()

    async def close(self) -> None:
        async with self._close_lock:
            if self._close_called:
                return
            self._close_called = True
        self.communicator._shutdown()
        self._stop_heartbeat()
        if self._do_close:
            await self._do_close()
        await self._notify_disconnected()

    def add_disconnect_listener(self, fn: Callable) -> None: ...
    def remove_disconnect_listeners(self) -> None: ...
    async def _notify_disconnected(self) -> None: ...
    async def _on_disconnected(self) -> None:
        await self.close()
    async def _send_heartbeat(self) -> None: ...  # Go sendHeartBeat() 패턴
    async def _on_heartbeat(self) -> None: ...    # Go onHeartBeat() 패턴
    def _stop_heartbeat(self) -> None: ...
```

Heartbeat 타이머: `asyncio.get_event_loop().call_later()` 또는 `asyncio.create_task(asyncio.sleep(n))`으로 구현.
Go의 `HeartbeatTimer`에 해당하는 `_HbTimer` 헬퍼 클래스를 동일 파일에 작성.

### 수정 파일 및 체크리스트

- [ ] `python/toki_socket/base_client.py` 생성
  - [ ] `_HbTimer` — `asyncio.create_task(sleep+callback)`, `cancel()` 메서드
  - [ ] `BaseClient._init_base()` — close_lock 초기화 (asyncio.Lock은 루프 바인딩 필요)
  - [ ] `BaseClient.close()` — connCloseOnce 패턴, shutdown → stop_heartbeat → do_close → notify
  - [ ] `BaseClient._send_heartbeat()` — interval > 0이면 타이머 설정
  - [ ] `BaseClient._on_heartbeat()` — waiting 상태이면 False로 리셋, 아니면 HeartBeat 송신
  - [ ] `BaseClient._notify_disconnected()` — listener 복사 후 asyncio.gather로 호출

### 테스트 작성

단위 테스트 skip — Transport 없이 단독 테스트가 어렵고, TCP 테스트에서 통합 검증.

### 중간 검증

```bash
cd python && python -c "from toki_socket.base_client import BaseClient; print('OK')"
```

예상: `OK`

---

## [API-4] TCP Transport 구현

### 문제

Go의 `TcpClient` / `TcpServer`에 해당하는 asyncio TCP 구현이 없다.

### 해결 방법

`python/toki_socket/tcp_client.py`, `python/toki_socket/tcp_server.py` 구현.

**framing** (PROTOCOL.md 기준):
- 송신: 4바이트 big-endian 길이 + PacketBase bytes
- 수신: 4바이트 읽기 → 길이 → 나머지 읽기
- 최대 패킷 크기: 64 MiB (Go와 동일)

```python
# tcp_client.py
class TcpClient(BaseClient):
    def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter,
                 interval_sec: int, wait_sec: int, parser_map: ParserMap) -> None:
        super().__init__(interval_sec, wait_sec, do_close=self._do_close_impl)
        self._reader = reader
        self._writer = writer
        self._init_base()
        self.communicator.initialize(self, parser_map)
        self.communicator.set_write_error_handler(lambda e: asyncio.create_task(self._on_disconnected()))
        self.communicator.add_listener(
            type_name_of(HeartBeat), lambda m: asyncio.create_task(self._on_heartbeat())
        )
        asyncio.create_task(self._read_loop())
        asyncio.create_task(self._send_heartbeat())

    async def write_packet(self, base: PacketBase) -> None:
        data = base.SerializeToString()
        header = struct.pack(">I", len(data))
        self._writer.write(header + data)
        await self._writer.drain()

    async def _do_close_impl(self) -> None:
        self._writer.close()
        await self._writer.wait_closed()

    async def _read_loop(self) -> None:
        while self.communicator.is_alive():
            try:
                header = await self._reader.readexactly(4)
                length = struct.unpack(">I", header)[0]
                if length == 0:
                    continue
                if length > MAX_PACKET_SIZE:
                    await self._on_disconnected(); return
                data = await self._reader.readexactly(length)
                base = PacketBase()
                base.ParseFromString(data)
                self.communicator.on_received_data(
                    base.typeName, base.data, base.nonce, base.responseNonce
                )
                asyncio.create_task(self._send_heartbeat())
            except (asyncio.IncompleteReadError, ConnectionError, OSError):
                await self._on_disconnected(); return

async def connect_tcp(host: str, port: int, interval_sec: int, wait_sec: int,
                      parser_map: ParserMap) -> TcpClient:
    reader, writer = await asyncio.open_connection(host, port)
    return TcpClient(reader, writer, interval_sec, wait_sec, parser_map)
```

```python
# tcp_server.py
class TcpServer:
    def __init__(self, host: str, port: int, new_client: Callable) -> None:
        self._host = host
        self._port = port
        self._new_client = new_client
        self._clients: list[TcpClient] = []
        self._server: asyncio.Server | None = None
        self.on_client_connected: Callable[[TcpClient], None] = lambda c: None

    async def start(self) -> None: ...
    async def stop(self) -> None: ...
    def clients(self) -> list[TcpClient]: ...
    async def broadcast(self, message) -> None: ...
```

### 수정 파일 및 체크리스트

- [ ] `python/toki_socket/tcp_client.py` 생성
  - [ ] `MAX_PACKET_SIZE = 64 << 20`
  - [ ] `TcpClient(BaseClient)` — reader/writer, read_loop, write_packet, do_close
  - [ ] `connect_tcp(host, port, interval_sec, wait_sec, parser_map)` async 함수
- [ ] `python/toki_socket/tcp_server.py` 생성
  - [ ] `TcpServer` — start/stop/clients/broadcast, on_client_connected 콜백
  - [ ] `_handle_connection()` — TcpClient 생성, disconnect_listener로 clients 목록 관리

### 테스트 작성

`python/test/test_tcp.py`:
- `test_tcp_send_receive` — 루프백 TCP로 양방향 TestData 교환
- `test_tcp_request_response` — send_request / add_request_listener 검증

```python
@pytest.mark.asyncio
async def test_tcp_send_receive():
    received = asyncio.Future()
    server = TcpServer("127.0.0.1", 0, lambda r, w: TcpClient(r, w, 0, 0, parser_map()))
    server.on_client_connected = lambda c: c.communicator.add_listener(
        "TestData", lambda m: received.set_result(m)
    )
    await server.start()
    port = server.port  # 동적 포트
    client = await connect_tcp("127.0.0.1", port, 0, 0, parser_map())
    await client.communicator.send(TestData(index=1, message="hello"))
    result = await asyncio.wait_for(received, 2.0)
    assert result.index == 1 and result.message == "hello"
    await client.close()
    await server.stop()
```

### 중간 검증

```bash
cd python && python -m pytest test/test_tcp.py -v
```

예상: 2개 테스트 PASS

---

## [API-5] WebSocket Transport 구현

### 문제

Go의 `WsClient` / `WsServer`에 해당하는 asyncio WebSocket 구현이 없다.

### 해결 방법

`websockets` 라이브러리 사용. WebSocket은 메시지 경계가 보장되므로 길이 헤더 없이 바이너리 프레임으로 전송.

```python
# ws_client.py
import websockets

class WsClient(BaseClient):
    def __init__(self, ws: websockets.WebSocketClientProtocol | websockets.WebSocketServerProtocol,
                 interval_sec: int, wait_sec: int, parser_map: ParserMap) -> None:
        super().__init__(interval_sec, wait_sec, do_close=self._do_close_impl)
        self._ws = ws
        self._init_base()
        self.communicator.initialize(self, parser_map)
        self.communicator.set_write_error_handler(lambda e: asyncio.create_task(self._on_disconnected()))
        self.communicator.add_listener(
            type_name_of(HeartBeat), lambda m: asyncio.create_task(self._on_heartbeat())
        )
        asyncio.create_task(self._read_loop())
        asyncio.create_task(self._send_heartbeat())

    async def write_packet(self, base: PacketBase) -> None:
        await self._ws.send(base.SerializeToString())

    async def _do_close_impl(self) -> None:
        await self._ws.close()

    async def _read_loop(self) -> None:
        try:
            async for message in self._ws:
                if not isinstance(message, bytes):
                    continue
                base = PacketBase()
                base.ParseFromString(message)
                self.communicator.on_received_data(
                    base.typeName, base.data, base.nonce, base.responseNonce
                )
                asyncio.create_task(self._send_heartbeat())
        except Exception:
            pass
        finally:
            await self._on_disconnected()

async def connect_ws(host: str, port: int, path: str,
                     interval_sec: int, wait_sec: int, parser_map: ParserMap) -> WsClient:
    uri = f"ws://{host}:{port}{path}"
    ws = await websockets.connect(uri)
    return WsClient(ws, interval_sec, wait_sec, parser_map)
```

```python
# ws_server.py
class WsServer:
    def __init__(self, host: str, port: int, path: str, new_client: Callable) -> None: ...
    async def start(self) -> None: ...
    async def stop(self) -> None: ...
    def clients(self) -> list[WsClient]: ...
    async def broadcast(self, message) -> None: ...
```

### 수정 파일 및 체크리스트

- [ ] `python/toki_socket/ws_client.py` 생성
  - [ ] `WsClient(BaseClient)` — websocket conn, read_loop, write_packet, do_close
  - [ ] `connect_ws(host, port, path, interval_sec, wait_sec, parser_map)` async 함수
- [ ] `python/toki_socket/ws_server.py` 생성
  - [ ] `WsServer` — start/stop/clients/broadcast, path 기반 핸들러
  - [ ] `_handler(ws, path)` — WsClient 생성, disconnect_listener로 목록 관리

### 테스트 작성

skip — TCP 테스트로 핵심 로직이 검증되며, WS는 크로스 테스트에서 검증.

### 중간 검증

```bash
cd python && python -c "from toki_socket.ws_client import WsClient; from toki_socket.ws_server import WsServer; print('OK')"
```

예상: `OK`

---

## [API-6] 단위 테스트

API-2, API-4 항목에서 이미 기술. 해당 항목 체크리스트 참조.

---

## [API-7] Python×Go 크로스 테스트

### 문제

Python이 Go 서버와 통신할 수 있는지, Go가 Python 서버와 통신할 수 있는지 검증이 없다.

### 해결 방법

포트 할당:
- Go 서버 / Python 클라이언트: TCP `29390`, WS `29392`
- Python 서버 / Go 클라이언트: TCP `29490`, WS `29492`

**Python 클라이언트 헬퍼** `python/crosstest/go_python_client.py`:
- `--mode tcp|ws`, `--port N`, `--phase send-push|requests` 플래그
- `INFO typeName python=TestData` 출력
- 각 시나리오 결과를 `PASS scenario=N detail=...` / `FAIL scenario=N error=...` 출력
- send-push: 시나리오 1(fire-and-forget), 2(server push 수신)
- requests: 시나리오 3(단일 요청), 4(동시 5개 요청 nonce 검증)
- Go crosstest 클라이언트(`go/crosstest/dart_go_client/main.go`)와 동일한 구조

**Go 오케스트레이터** `go/crosstest/go_python.go`:
- Go 서버를 시작하고 `python go_python_client.py` 서브프로세스를 실행
- 기존 `go_dart.go` 패턴과 동일 (`//go:build ignore`, `runPythonClient()` 헬퍼)
- Python 패키지 디렉터리를 소스 위치 기반으로 찾는 `pythonPackageDir()` 함수

**Python 오케스트레이터** `python/crosstest/python_go.py`:
- Python 서버를 시작하고 `go run ./crosstest/python_go_client` 서브프로세스를 실행
- `go/crosstest/dart_go_client/main.go` 패턴을 참고하여 `--mode`, `--port`, `--phase` 플래그 지원
- Go 패키지 디렉터리를 소스 위치 기반으로 찾는 헬퍼

**Go 클라이언트 헬퍼** `go/crosstest/python_go_client/main.go`:
- 기존 `dart_go_client/main.go`와 동일한 구조
- Python 서버 포트(29490 TCP, 29492 WS)에 연결
- `--phase send-push`: 시나리오 1, 2
- `--phase requests`: 시나리오 3, 4
- push 메시지 검증: `index=200 && message=="push from python server"`

**시나리오별 메시지 규격**:
| 시나리오 | 클라이언트 송신 | 서버 검증 / 응답 |
|----------|----------------|-----------------|
| 1 (send) | `index=101, message="fire from python client"` | Go서버: index==101, message 검증 |
| 2 (push) | — | Go서버 → `index=200, message="push from go server"` |
| 3 (request) | `index=21, message="single request from python"` | Go서버 → `index=42, message="echo: single request from python"` |
| 4 (concurrent) | `index=30+i, message="multi request {i} from python"` | Go서버 → `index=(30+i)*2, message="echo: ..."` |

Python서버일 때는 go클라이언트 메시지 검증:
- 1: `index=101, message="fire from go client"`
- 2: Python서버 → `index=200, message="push from python server"`

### 수정 파일 및 체크리스트

- [ ] `python/crosstest/__init__.py` 생성 (빈 파일)
- [ ] `python/crosstest/go_python_client.py` 생성
  - [ ] argparse로 `--mode`, `--port`, `--phase` 파싱
  - [ ] `INFO typeName python=TestData` 출력
  - [ ] `dial_with_retry(mode, port)` — 3초 내 재시도
  - [ ] `run_send_push(client)` — 시나리오 1, 2
  - [ ] `run_requests(client)` — 시나리오 3, 4 (asyncio.gather로 동시 요청)
  - [ ] 비동기 main → `asyncio.run(main())`
- [ ] `go/crosstest/go_python.go` 생성
  - [ ] `//go:build ignore` 태그
  - [ ] `goPythonTCPPort = 29390`, `goPythonWSPort = 29392`
  - [ ] `runTCPSendPush/TCPRequests/WSSendPush/WSRequests()` 4개 함수
  - [ ] `runPythonClient(mode, port, phase, expected)` — `python go_python_client.py` 실행
  - [ ] `pythonPackageDir()` — 소스 위치 기반 탐색
- [ ] `python/crosstest/python_go.py` 생성
  - [ ] `pythonGoTCPPort = 29490`, `pythonGoWSPort = 29492`
  - [ ] `run_tcp_send_push/tcp_requests/ws_send_push/ws_requests()` 4개 함수
  - [ ] `run_go_client(mode, port, phase, expected)` — `go run ./crosstest/python_go_client` 실행
  - [ ] `go_package_dir()` — 소스 위치 기반 탐색
  - [ ] 비동기 main → `asyncio.run(main())`
- [ ] `go/crosstest/python_go_client/main.go` 생성
  - [ ] `dart_go_client/main.go`와 동일 구조
  - [ ] Python서버 포트(`--port N`)에 연결
  - [ ] push 메시지: `"push from python server"`

### 테스트 작성

크로스 테스트가 곧 테스트. 별도 단위 테스트 skip.

### 중간 검증

Go 서버 / Python 클라이언트:
```bash
cd go && go run ./crosstest/go_python.go
```

Python 서버 / Go 클라이언트:
```bash
cd python && python crosstest/python_go.py
```

예상: 각각 `PASS all ...` 출력

---

## 수정 파일 요약

| 파일 | 항목 |
|------|------|
| `python/pyproject.toml` | API-1 |
| `python/toki_socket/__init__.py` | API-1 |
| `python/toki_socket/packets/__init__.py` | API-1 |
| `python/toki_socket/packets/message_common.proto` | API-1 |
| `python/toki_socket/packets/message_common_pb2.py` | API-1 |
| `python/toki_socket/packets/message_common_pb2.pyi` | API-1 |
| `python/README.md` | API-1 |
| `python/toki_socket/communicator.py` | API-2 |
| `python/test/__init__.py` | API-2 |
| `python/test/test_communicator.py` | API-2 |
| `python/toki_socket/base_client.py` | API-3 |
| `python/toki_socket/tcp_client.py` | API-4 |
| `python/toki_socket/tcp_server.py` | API-4 |
| `python/test/test_tcp.py` | API-4 |
| `python/toki_socket/ws_client.py` | API-5 |
| `python/toki_socket/ws_server.py` | API-5 |
| `python/crosstest/__init__.py` | API-7 |
| `python/crosstest/go_python_client.py` | API-7 |
| `python/crosstest/python_go.py` | API-7 |
| `go/crosstest/go_python.go` | API-7 |
| `go/crosstest/python_go_client/main.go` | API-7 |

---

## 최종 검증

```bash
# 1. Python 단위 테스트
cd python && pip install -e ".[dev]" && python -m pytest test/ -v

# 2. Go 서버 / Python 클라이언트 크로스 테스트
cd go && go run ./crosstest/go_python.go

# 3. Python 서버 / Go 클라이언트 크로스 테스트
cd python && python crosstest/python_go.py

# 4. 기존 Go 크로스 테스트 회귀 확인
cd go && go run ./crosstest/go_dart.go
cd go && go run ./crosstest/go_kotlin.go
```

예상 결과:
1. `pytest`: 모든 테스트 PASS
2. `go run ./crosstest/go_python.go`: `PASS all go-server/python-client crosstests passed`
3. `python crosstest/python_go.py`: `PASS all python-server/go-client crosstests passed`
4. 기존 크로스 테스트: 회귀 없음 (PASS)
