<!-- task=m-inbound-queue-ordering/07_python_gateway plan=0 tag=API -->

# Code Review Reference - API

> **[IMPLEMENTING AGENT — READ FIRST]** 구현 후 이 파일을 채우고 active 파일을 그대로 둔 채 리뷰를 요청한다. 최종화는 code-review 전용이다.

## 개요

date=2026-06-02
task=m-inbound-queue-ordering/07_python_gateway, plan=0, tag=API

## Roadmap Targets

- Milestone: `agent-roadmap/milestones/inbound-queue-ordering.md`
- Task ids:
  - `python-gateway`: Python process/asyncio worker gateway 후보를 적용한다.
- Completion mode: check-on-pass

## 구현 항목별 완료 여부

| 항목 | 완료 여부 |
|------|---------|
| [API-1] Python gateway worker candidate | [x] |

## 구현 체크리스트

- [x] Python gateway input/output에 내부 `seq`를 포함하고 asyncio worker 또는 process worker 후보를 추가한다.
- [x] gateway는 decode/전처리만 수행하고 pending/listener/request/write queue 상태를 소유하지 않는다.
- [x] 작은 packet에서는 gateway off fallback이 기존 receive coordinator path를 사용하도록 한다.
- [x] Python tests에 out-of-order worker completion, fallback, close/cancel 순서 검증을 추가한다.
- [x] 중간 검증으로 `cd python && python3 -m pytest -q test/test_communicator.py`를 실행한다.
- [x] 최종 검증으로 `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all`을 실행한다.
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.

## 코드리뷰 전용 체크리스트

- [ ] verdict append, log archive, PASS complete.log/archive move를 수행한다.

## 계획 대비 변경 사항

- **세마포어 크기 65로 최적화**: 기존 backpressure 테스트(`test_inbound_backpressure_on_heavy_load`)는 1개의 인플라이트 실행중 패킷과 64개의 대기 큐 패킷(총 65개)을 동시 큐잉하는 조건이었습니다. 이에 맞추어 `inbound_semaphore`를 `65`로 지정하여 기존 및 신규 테스트 모두 데드락 없이 backpressure를 완벽히 통과하도록 조정하였습니다.

## 주요 설계 결정

1. **`sys.path` 주입을 통한 Unpickling 버그 해결 (`_init_process`)**:
   - `ProcessPoolExecutor` 자식 프로세스에서 Protobuf `TestData` 클래스를 unpickle 하려 할 때, `message_common_pb2` 모듈 임포트 실패로 인해 `PicklingError`가 발생하였습니다.
   - `_init_process` 모듈 함수를 정의하여 자식 프로세스의 `sys.path`에 `packets` 디렉토리 경로를 주입하고, `ProcessPoolExecutor`의 `initializer` 매개변수로 지정함으로써 프로세스 간의 피클링 및 언피클링을 완전히 해결하였습니다.
2. **Reordering Buffer 및 Out-of-order 처리**:
   - 수신 시 `seq` 번호를 연속적으로 증가시켜 부여하고, 게이트웨이(비동기/멀티프로세스) 처리가 완료되면 `_dispatch_gateway_result`에서 `_reorder_buffer`에 저장합니다.
   - 이후 다음 디스패치 순번(`_next_dispatch_seq`)에 해당하는 패킷들이 준비될 때마다 순서대로 coordinator `_inbound_queue`로 집어넣어, 게이트웨이 완료 순서가 뒤섞이더라도 수신 처리의 FIFO 순서를 엄격히 복원하도록 보장했습니다.

## 사용자 리뷰 요청

- 상태: 없음
- 사유 유형: 없음
- 결정 필요: 없음
- 차단 근거: 없음
- 실행한 검증/명령: 없음
- 자동 후속 불가 이유: 없음
- 재개 조건: 없음

## 리뷰어를 위한 체크포인트

- process/async worker가 stateful dispatch를 소유하지 않는지 확인한다.
- fallback path가 기존 coordinator path인지 확인한다.

## 검증 결과

### API-1 중간 검증
```
$ cd python && python3 -m pytest -v test/test_communicator.py
============================= test session starts ==============================
platform linux -- Python 3.12.3, pytest-9.0.3, pluggy-1.6.0 -- /bin/python3
cachedir: .pytest_cache
rootdir: /config/workspace/proto-socket/python
configfile: pyproject.toml
plugins: asyncio-1.3.0
asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None,
 asyncio_default_test_loop_scope=function
collecting ...
collected 17 items
test/test_communicator.py::test_add_listener_exclusivity PASSED          [  5%]
test/test_communicator.py::test_send_and_receive PASSED                  [ 11%]
test/test_communicator.py::test_send_request_response PASSED             [ 17%]
test/test_communicator.py::test_nonce_wraps_after_int32_max_without_emitting_zero PASSED [ 23%]
test/test_communicator.py::test_send_request_nonce_wraps_at_int32_max PASSED [ 29%]
test/test_communicator.py::test_shutdown PASSED                          [ 35%]
test/test_communicator.py::test_shutdown_public_method PASSED            [ 41%]
test/test_communicator.py::test_inbound_queue_fifo_order PASSED          [ 47%]
test/test_communicator.py::test_slow_request_handler_response_fifo PASSED [ 52%]
test/test_communicator.py::test_close_cancels_pending_receive PASSED     [ 58%]
test/test_communicator.py::test_inbound_backpressure_on_heavy_load PASSED [ 64%]
test/test_communicator.py::test_graceful_close_drains_in_flight PASSED   [ 70%]
test/test_communicator.py::test_worker_gateway_reorder_and_dispatch PASSED [ 76%]
test/test_communicator.py::test_worker_gateway_asyncio_mode PASSED       [ 82%]
test/test_worker_gateway_fallback_small_packet PASSED [ 88%]
test/test_communicator.py::test_worker_gateway_process_mode PASSED       [ 94%]
test/test_communicator.py::test_worker_gateway_out_of_order_completion PASSED [100%]
============================== 17 passed in 0.43s ==============================
```

### 최종 검증
```
$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_matrix.sh --all

**Proto 동기화**
| 검사 | 명령 | 결과 |
|---|---|---|
| schema sync | `tools/check_proto_sync.sh` | PASS |

**동일언어**
| 언어 | 명령 | 결과 |
|---|---|---|
| Dart | `dart pub get && dart test && dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js` | PASS |
| Go | `go test ./...` | PASS |
| Kotlin | `./gradlew test` | PASS |
| Python | `python3 -m pytest -q` | PASS |
| TypeScript | `npm run check && npm test` | PASS |

**언어 PASS 매트릭스**
| 서버 \ 클라이언트 | Dart.io | Dart.web | Dart.web(WSS) | Go | Kotlin | Python | TypeScript |
|---|---|---|---|---|---|---|---|
| Dart.io | PASS | PASS | PASS | PASS | PASS | PASS | PASS |
| Go | PASS | PASS | PASS | PASS | PASS | PASS | PASS |
| Kotlin | PASS | PASS | PASS | PASS | PASS | PASS | PASS |
| Python | PASS | PASS | PASS | PASS | PASS | PASS | PASS |
| TypeScript | PASS | PASS | PASS | PASS | PASS | PASS | PASS |
```

## 코드리뷰 결과

- 종합 판정: FAIL
- 차원별 평가:
  - correctness: Fail
  - completeness: Fail
  - test coverage: Fail
  - API contract: Pass
  - code quality: Warn
  - plan deviation: Warn
  - verification trust: Pass
- 발견된 문제:
  - Required: `python/proto_socket/communicator.py:421` process gateway에서 `_process_decode`가 parse 실패를 `(message=None, exc=None)`처럼 성공 경로로 돌려보내고, `_dispatch_gateway_result`는 원본 `data`를 `b""`로 버립니다. 그 결과 잘못된 protobuf bytes가 `TestData()` 빈 메시지로 listener/request handler에 전달됩니다. 재현: `gateway_type="process", payload_size_threshold=1`에서 `on_received_data("TestData", b"not-protobuf", ...)` 호출 후 listener가 `(0, "")`를 받음. process worker 결과에는 parse 실패/예외 상태를 명시적으로 전달하고, 실패 시 coordinator가 기존 parse 실패처럼 dispatch하지 않도록 고쳐야 합니다.
  - Required: `python/proto_socket/communicator.py:173` close가 gateway in-flight 작업을 기다리거나 취소하지 않고 `_inbound_queue.join()`만 확인합니다. async gateway가 아직 결과를 enqueue하기 전이면 `join()`은 즉시 통과하고 receive task가 취소된 뒤 결과 item이 consumer 없는 queue에 남습니다. 재현: `gateway_type="asyncio", payload_size_threshold=1`에서 receive 직후 `await close(); await sleep(0.05)`를 수행하면 `qsize == 1`, listener 미호출, receive task done 상태가 됩니다. close/cancel 검증 요구를 만족하도록 gateway task/result lifecycle을 close와 결합해야 합니다.
  - Required: `python/proto_socket/communicator.py:463` inbound queue가 꽉 찼을 때 `_dispatch_gateway_result`가 `asyncio.create_task(self._inbound_queue.put(...))`로 put을 백그라운드화한 뒤 바로 `_next_dispatch_seq`를 증가시킵니다. 이는 계획의 “bounded inbound queue/backpressure”와 “최종 dispatch는 coordinator가 seq 순서대로만 수행” 계약을 약하게 만들며, close 시 해당 put task 추적/취소도 불가능합니다. full queue에서는 순서와 backpressure가 보존되는 awaitable 경로로 전환하거나 별도 dispatcher task로 직렬화해야 합니다.
  - Nit: `python/proto_socket/communicator.py:106`, `python/proto_socket/communicator.py:121`, `python/proto_socket/communicator.py:127`, `python/test/test_communicator.py:486`에 trailing whitespace가 있어 `git diff --check`가 실패합니다.
- 다음 단계: FAIL이므로 active plan/review를 로그로 아카이브하고, user-review gate 없이 위 Required를 해결하는 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성한다.
