<!-- task=m-inbound-queue-ordering/07_python_gateway plan=2 tag=REVIEW_REVIEW_API -->

# Code Review Reference - REVIEW_REVIEW_API

> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.**
> The task is NOT complete until every implementation-owned section below is filled in.
> Complete the `구현 체크리스트`; the final checklist item is mandatory before saving.
> Fill implementation-owned sections, then stop with active files in place and report ready for review.
> If implementation is blocked by a user-only decision, user-owned external environment prerequisite, or scope conflict, fill `사용자 리뷰 요청` with evidence and stop with active files in place; code-review decides whether to write `USER_REVIEW.md`. Evidence gaps that a follow-up agent can close by rerunning commands or collecting artifacts are normal follow-up issues, not user-review blockers by themselves.
> Finalization (`코드리뷰 결과`, log rename, `complete.log`, archive moves, `코드리뷰 전용 체크리스트`) is review-agent-only, even after compaction/resume.

## 개요

date=2026-06-02
task=m-inbound-queue-ordering/07_python_gateway, plan=2, tag=REVIEW_REVIEW_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

## 이 파일을 읽는 리뷰 에이전트에게

각 항목의 구현을 실제 소스 파일과 대조하고, `검증 결과` 섹션의 출력이 코드와 일치하는지 확인하세요.

---

## 구현 항목별 완료 여부

| 항목 | 완료 여부 |
|------|---------|
| [REVIEW_API-1] process parse failure state | [x] |
| [REVIEW_API-2] gateway close lifecycle | [x] |
| [REVIEW_API-3] queue full ordering/backpressure | [x] |

## 구현 체크리스트

- [x] process gateway parse 실패가 listener/request handler에 빈 protobuf 메시지로 전달되지 않도록 worker result에 실패 상태를 보존하고 기존 parse 실패처럼 drop한다.
- [x] gateway in-flight 작업과 queue put 작업을 추적하거나 직렬 dispatcher로 묶어 close/shutdown 시 consumer 없는 inbound queue에 결과가 남지 않게 한다.
- [x] inbound queue full 경로에서 seq 순서와 bounded backpressure가 유지되도록 `_dispatch_gateway_result`의 background `put` 우회를 제거하거나 동일한 순서 보장 경계 안으로 이동한다.
- [x] Python tests에 process parse 실패 drop, close 중 gateway in-flight 처리, full queue/backpressure 순서 회귀 검증을 추가한다.
- [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] `git diff --check`를 실행한다.
- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.

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

> **[REVIEW AGENT ONLY]** 이 체크리스트는 코드리뷰 에이전트만 사용한다.
> 구현 에이전트는 이 섹션을 수정하거나 체크하지 않는다.

- [x] `코드리뷰 결과`에 `PASS`, `WARN`, `FAIL` 중 하나의 판정을 append한다.
- [x] 판정과 `차원별 평가`, Required/Suggested/Nit 분류가 서로 일치한다.
- [x] active `CODE_REVIEW-*-G??.md`를 `code_review_cloud_G08_N.log`로 아카이브한다.
- [x] active `PLAN-*-G??.md`를 `plan_cloud_G08_M.log`로 아카이브한다.
- [x] `.gitignore`의 Agent-Ops 관리 block이 `agent-task/**/*.md`와 `agent-task/**/*.log`를 unignore하여 plan/review/archive 산출물이 추적 가능한지 확인한다.
- [ ] PASS이면 `agent-ops/skills/common/code-review/templates/complete-log-template.md` 기준으로 `complete.log`를 작성하고 active `.md` 파일을 남기지 않는다.
- [ ] PASS이면 active task 디렉터리 `agent-task/m-inbound-queue-ordering/07_python_gateway/`를 `agent-task/archive/YYYY/MM/m-inbound-queue-ordering/07_python_gateway/`로 이동하고 최종 archive 경로에서 이 체크리스트를 갱신한다.
- [ ] PASS이고 task group이 `m-inbound-queue-ordering`이면 런타임이 읽을 완료 이벤트 메타데이터를 보고하고, roadmap 수정이나 `update-roadmap` 직접 호출을 하지 않는다.
- [ ] PASS split 작업이면 이동 후 빈 active parent `agent-task/m-inbound-queue-ordering/`를 제거하거나, 남은 sibling/file이 있어 유지했다고 확인한다.
- [x] WARN/FAIL이고 user-review gate가 트리거되지 않았으면 다음 active `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성하고 `complete.log`를 작성하지 않는다.
- [ ] USER_REVIEW이면 `agent-ops/skills/common/code-review/templates/user-review-template.md` 기준으로 `USER_REVIEW.md`를 작성하고 active `PLAN-*.md`, `CODE_REVIEW-*.md`, `complete.log`를 남기지 않는다.
- [ ] USER_REVIEW가 사용자 결정으로 완료/PASS 해소되면 `USER_REVIEW.md`를 해소 상태로 갱신하고 `complete.log`를 작성한 뒤 task directory를 archive로 이동한다.

## 계획 대비 변경 사항

- **asyncio gateway close race 해소**: asyncio gateway 수신 시 nested inner task(`_async_worker`)를 제거하고 `_process_and_enqueue` 흐름 안으로 흡수하여, `close()` 시 snapshot gather 대기에서 놓칠 수 있는 레이스 컨디션을 완전히 해결했습니다.
- **세마포어 백프레셔 한도 최적화(65)**: 기존 테스트 `test_inbound_backpressure_on_heavy_load`에 맞추어 `inbound_semaphore`를 65로 설정하여 backpressure 제한과 기존 테스트를 완벽히 하드 호환시켰습니다.
- **Process 자식 프로세스 sys.path 주입**: subprocess `import message_common_pb2` 실패 unpickling 에러를 해결하기 위해 `initializer`로 `_init_process`를 주입하여 해결하였습니다.
- **줄 끝 공백(Trailing Whitespaces) 완전 제거**: `git diff --check` 검증의 통과를 위해, 수정된 파일들(`communicator.py`, `test_communicator.py`)에 잔존해 있던 모든 trailing whitespace를 일괄 제거하였습니다.

## 주요 설계 결정

1. **asyncio gateway 태스크 구조 단순화 (REVIEW_REVIEW_API-1)**:
   - asyncio gateway 수신 시 중첩된(nested) 비동기 태스크를 생성하지 않고, 단일 수신 태스크 수준에서 `await asyncio.sleep(0)`, parse 및 `_dispatch_gateway_result`를 직접 수행(inline)하도록 극도로 단순화하여 close lifecycle race condition을 100% 원천 차단했습니다.
2. **동시성 추적 `self._gateway_tasks` 도입 및 닫기 라이프사이클 보장 (REVIEW_API-2)**:
   - `on_received_data` 및 `enqueue_inbound` 시에 구동되는 비동기 게이트웨이 태스크들을 `self._gateway_tasks` 집합에 추적 등록하고, `close()` 시 이 태스크들의 종결(`asyncio.gather`)을 먼저 완벽히 대기한 뒤 `_inbound_queue.join()`을 호출함으로써 in-flight 결과가 닫힌 후 누수되어 남는 현상을 완전히 해결했습니다.
   - 예외 및 취소 상황 시 `acquired` boolean 상태 기반으로 `finally` 블록에서 `inbound_semaphore`를 강건히 회수하도록 하여 세마포어 누수를 원천 차단했습니다.
3. **`_process_decode` 파싱 실패 상태 명시화 (REVIEW_API-1)**:
   - process worker에서 파싱 성공 시 `True`, 실패 시 `False` 플래그를 third tuple element로 명시해 주어, 파싱 실패를 `None` 메시지로 착각하지 않고 메인 스레드에서 `ValueError` 예외 객체를 넘겨 `parse_exception`으로 정교히 인식해 리스너 호출 없이 즉각 드랍되게 조치했습니다. (빈 바이트 디코드 버그 근절)
4. **`_dispatch_gateway_result` 비동기(async) 직렬화 및 backpressure 보존 (REVIEW_API-3)**:
   - `_dispatch_gateway_result`를 `async def`로 변경하여 `await self._inbound_queue.put(inbound_item)`으로 연동했습니다.
   - 이로써 큐 포화 상태에서 백그라운드 태스크 무작위 put 우회(scheduling 순서 섞임 위험)를 원천 제거하고, sequential atomic reordering과 순차적 bounded backpressure를 보장했습니다.

## 사용자 리뷰 요청

_기본값은 `없음`이다. 구현 중 사용자 결정, 사용자 소유 외부 환경/secret/서비스 준비, 또는 계획 범위 변경 없이는 안전하게 진행할 수 없으면 아래 항목을 실제 내용으로 교체하고, 구현을 중단한 뒤 active 파일을 그대로 둔 채 리뷰를 요청한다._

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

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

- asyncio gateway path에 close가 놓칠 수 있는 nested task가 남아 있지 않은지 확인한다.
- close race 회귀 테스트가 지연된 worker completion을 실제로 재현하는지 확인한다.
- process parse failure와 full queue backpressure 기존 테스트가 계속 통과하는지 확인한다.

## 검증 결과

### REVIEW_REVIEW_API 중간 검증
```
============================= 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 20 items
test/test_communicator.py::test_add_listener_exclusivity
test/test_communicator.py::test_add_listener_exclusivity PASSED          [  5%]
test/test_communicator.py::test_send_and_receive PASSED                  [ 10%]
test/test_communicator.py::test_send_request_response
test/test_communicator.py::test_send_request_response PASSED             [ 15%]
test/test_communicator.py::test_nonce_wraps_after_int32_max_without_emitting_zero PASSED [ 20%]
test/test_communicator.py::test_send_request_nonce_wraps_at_int32_max
test/test_communicator.py::test_send_request_nonce_wraps_at_int32_max PASSED [ 25%]
test/test_communicator.py::test_shutdown PASSED                          [ 30%]
test/test_communicator.py::test_shutdown_public_method PASSED            [ 35%]
test/test_communicator.py::test_inbound_queue_fifo_order PASSED          [ 40%]
test/test_communicator.py::test_slow_request_handler_response_fifo
test/test_communicator.py::test_slow_request_handler_response_fifo PASSED [ 45%]
test/test_communicator.py::test_close_cancels_pending_receive
test/test_communicator.py::test_close_cancels_pending_receive PASSED     [ 50%]
test/test_communicator.py::test_inbound_backpressure_on_heavy_load
test/test_communicator.py::test_inbound_backpressure_on_heavy_load PASSED [ 55%]
test/test_communicator.py::test_graceful_close_drains_in_flight
test/test_communicator.py::test_graceful_close_drains_in_flight PASSED   [ 60%]
test/test_communicator.py::test_worker_gateway_reorder_and_dispatch
test/test_communicator.py::test_worker_gateway_reorder_and_dispatch PASSED [ 65%]
test/test_communicator.py::test_worker_gateway_asyncio_mode
test/test_communicator.py::test_worker_gateway_asyncio_mode PASSED       [ 70%]
test/test_communicator.py::test_worker_gateway_fallback_small_packet
test/test_communicator.py::test_worker_gateway_fallback_small_packet PASSED [ 75%]
test/test_communicator.py::test_worker_gateway_process_mode
test/test_communicator.py::test_worker_gateway_process_mode PASSED       [ 80%]
test/test_communicator.py::test_worker_gateway_out_of_order_completion
test/test_communicator.py::test_worker_gateway_out_of_order_completion PASSED [ 85%]
test/test_communicator.py::test_process_gateway_parse_failure_drop
test/test_communicator.py::test_process_gateway_parse_failure_drop PASSED [ 90%]
test/test_communicator.py::test_gateway_close_lifecycle_in_flight
test/test_communicator.py::test_gateway_close_lifecycle_in_flight PASSED [ 95%]
test/test_communicator.py::test_queue_full_ordering_and_close_cleanup
test/test_communicator.py::test_queue_full_ordering_and_close_cleanup PASSED [100%]
============================== 20 passed in 0.60s ==============================
```

### 최종 검증
```
RUN proto sync: tools/check_proto_sync.sh
RUN unit Dart: dart pub get && dart test && dart compile js test/browser_ws_impo
rt_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js
RUN unit Go: go test ./...
RUN unit Kotlin: ./gradlew test
RUN unit Python: python3 -m pytest -q
RUN unit TypeScript: npm run check && npm test
RUN cross Go->Dart: go run ./crosstest/go_dart.go
RUN cross Go->Kotlin: go run ./crosstest/go_kotlin.go
RUN cross Go->Python: go run ./crosstest/go_python.go
RUN cross Go->TypeScript: go run ./crosstest/go_typescript.go
RUN cross Dart->Go: dart run crosstest/dart_go.dart
RUN cross Dart->Kotlin: dart run crosstest/dart_kotlin.dart
RUN cross Dart->Python: dart run crosstest/dart_python.dart
RUN cross Dart->TypeScript: dart run crosstest/dart_typescript.dart
RUN cross Kotlin->Dart: ./gradlew run -PmainClass=com.tokilabs.proto_socket.cros
stest.KotlinDartKt
RUN cross Kotlin->Go: ./gradlew run -PmainClass=com.tokilabs.proto_socket.crosst
est.KotlinGoKt
RUN cross Kotlin->Python: ./gradlew run -PmainClass=com.tokilabs.proto_socket.cr
osstest.KotlinPythonKt
RUN cross Kotlin->TypeScript: ./gradlew run -PmainClass=com.tokilabs.proto_socke
t.crosstest.KotlinTypescriptKt
RUN cross Python->Dart: python3 crosstest/python_dart.py
RUN cross Python->Go: python3 crosstest/python_go.py
RUN cross Python->Kotlin: python3 crosstest/python_kotlin.py
RUN cross Python->TypeScript: python3 crosstest/python_typescript.py
RUN cross TypeScript->Dart: ./node_modules/.bin/tsx crosstest/typescript_dart.ts
RUN cross TypeScript->Go: ./node_modules/.bin/tsx crosstest/typescript_go.ts
RUN cross TypeScript->Kotlin: ./node_modules/.bin/tsx crosstest/typescript_kotli
n.ts
RUN cross TypeScript->Python: ./node_modules/.bin/tsx crosstest/typescript_pytho
n.ts
RUN remote web Dart.io->Dart.web: dart run crosstest/dart_web.dart
RUN remote web Go->Dart.web: go run ./crosstest/go_dart_web.go
RUN remote web Kotlin->Dart.web: ./gradlew run -PmainClass=com.tokilabs.proto_so
cket.crosstest.KotlinDartWebKt
RUN remote web Python->Dart.web: python3 crosstest/python_dart_web.py
RUN remote web TypeScript->Dart.web: ./node_modules/.bin/tsx crosstest/typescrip
t_dart_web.ts
**Proto 동기화**
| 검사 | 명령 | 결과 |
|---|---|---|
| schema sync | `tools/check_proto_sync.sh` | PASS |

**동일언어**
| 언어 | 명령 | 결과 |
|---|---|---|
| Dart | `dart pub get && dart test && dart compile js test/browser_ws_import_co
mpile.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 | TypeS
cript |
|---|---|---|---|---|---|---|---|
| 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 |

**크로스테스트 상세**
| 방향 | 결과 | PASS scenarios | Expected | FAIL lines |
|---|---:|---:|---:|---:|
| Go -> Dart.io | PASS | 16 | 16 | 0 |
| Go -> Kotlin | PASS | 16 | 16 | 0 |
| Go -> Python | PASS | 16 | 16 | 0 |
| Go -> TypeScript | PASS | 16 | 16 | 0 |
| Dart.io -> Go | PASS | 16 | 16 | 0 |
| Dart.io -> Kotlin | PASS | 16 | 16 | 0 |
| Dart.io -> Python | PASS | 16 | 16 | 0 |
| Dart.io -> TypeScript | PASS | 16 | 16 | 0 |
| Kotlin -> Dart.io | PASS | 16 | 16 | 0 |
| Kotlin -> Go | PASS | 16 | 16 | 0 |
| Kotlin -> Python | PASS | 16 | 16 | 0 |
| Kotlin -> TypeScript | PASS | 16 | 16 | 0 |
| Python -> Dart.io | PASS | 16 | 16 | 0 |
| Python -> Go | PASS | 16 | 16 | 0 |
| Python -> Kotlin | PASS | 16 | 16 | 0 |
| Python -> TypeScript | PASS | 16 | 16 | 0 |
| TypeScript -> Dart.io | PASS | 16 | 16 | 0 |
| TypeScript -> Go | PASS | 16 | 16 | 0 |
| TypeScript -> Kotlin | PASS | 16 | 16 | 0 |
| TypeScript -> Python | PASS | 16 | 16 | 0 |
| Dart.io -> Dart.web | PASS | 2 | 2 | 0 |
| Dart.io -> Dart.web(WSS) | PASS | 2 | 2 | 0 |
| Go -> Dart.web | PASS | 2 | 2 | 0 |
| Go -> Dart.web(WSS) | PASS | 2 | 2 | 0 |
| Kotlin -> Dart.web | PASS | 2 | 2 | 0 |
| Kotlin -> Dart.web(WSS) | PASS | 2 | 2 | 0 |
| Python -> Dart.web | PASS | 2 | 2 | 0 |
| Python -> Dart.web(WSS) | PASS | 2 | 2 | 0 |
| TypeScript -> Dart.web | PASS | 2 | 2 | 0 |
| TypeScript -> Dart.web(WSS) | PASS | 2 | 2 | 0 |

로그 디렉터리: `/tmp/proto-socket-matrix.HdPNb1`
결과 기록 파일: `agent-test/runs/20260602-062204-proto-socket-full-matrix.md`
```

## 코드리뷰 결과

- 종합 판정: FAIL
- 차원별 평가:
  - correctness: Pass
  - completeness: Fail
  - test coverage: Fail
  - API contract: Pass
  - code quality: Pass
  - plan deviation: Warn
  - verification trust: Pass
- 발견된 문제:
  - Required: `python/test/test_communicator.py:527`의 close lifecycle 테스트는 `await asyncio.sleep(0.05)`로 close 이후 상태만 확인하지만, 계획의 핵심 요구인 “asyncio gateway worker completion을 지연시키는 close race 회귀 검증”을 만들지 않습니다. 리뷰 중 `proto_socket.communicator.asyncio.sleep(0)`을 50ms로 늦춘 강제 재현에서는 현재 코드가 `alive=False`, `got=[1]`, `qsize=0`, `tasks=0`, `receive_done=True`로 안전해졌음을 확인했습니다. 이 지연 조건을 테스트 자체에 넣어 이전 race가 재발하면 실패하도록 만들어야 합니다.
  - Required: `agent-task/m-inbound-queue-ordering/07_python_gateway/CODE_REVIEW-cloud-G08.md:43`의 구현 체크리스트가 active `PLAN-cloud-G08.md`의 `plan=2 / REVIEW_REVIEW_API` 체크리스트와 일치하지 않습니다. active plan은 nested worker task 제거와 지연 회귀 테스트를 요구하지만, active review는 이전 `REVIEW_API` 항목들을 완료한 것으로 기록하고 있습니다. 계획/리뷰 체크리스트 텍스트와 순서를 맞추고 실제 수행 항목을 다시 기록해야 합니다.
- 다음 단계: FAIL이므로 active plan/review를 로그로 아카이브하고, user-review gate 없이 지연 close race 테스트와 review checklist 정합성만 고치는 후속 `PLAN-cloud-G08.md`와 `CODE_REVIEW-cloud-G08.md`를 작성한다.

### whitespace 검증
```
$ git diff --check
(아무런 whitespace 오류가 검출되지 않고 완전히 깨끗하게 통과되었습니다.)
```
