feat: dart isolate transport path optimization - G08 task completion

- Moved archive files to 2026/06 task archive - Optimized communicator.dart for isolate transport - Updated protobuf_client.dart changes - Refactored ws_protobuf_client_io.dart and ws_protobuf_client_web.dart
2026-06-06 14:02:08 +09:00 · 2026-06-06 14:02:08 +09:00 · ce2457c0d3
commit ce2457c0d3
parent 5d3022ba46
8 changed files with 292 additions and 120 deletions
--- a/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/code_review_cloud_G08_0.log
+++ b/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/code_review_cloud_G08_0.log
@ -0,0 +1,133 @@
+<!-- task=m-performance-hotspot-optimization/11+10_dart_isolate_transport_path plan=0 tag=REFACTOR -->
+
+# Code Review Reference - REFACTOR
+
+> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.**
+> Do not implement until predecessor `10` has `complete.log`.
+
+## 개요
+
+date=2026-06-06
+task=m-performance-hotspot-optimization/11+10_dart_isolate_transport_path, plan=0, tag=REFACTOR
+
+## Roadmap Targets
+
+- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md`
+- Task ids:
+  - `dart-isolate-real-path`: 실제 TCP/WS transport 수신 경로 hardening 연결
+- Completion mode: check-on-pass
+
+## 이 파일을 읽는 리뷰 에이전트에게
+
+> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다.
+
+## 구현 항목별 완료 여부
+
+| 항목 | 완료 여부 |
+|------|---------|
+| [REFACTOR-1] TCP transport receive path | [x] |
+| [REFACTOR-2] WS IO/web receive path | [x] |
+| [REFACTOR-3] Performance row inclusion | [x] |
+
+## 구현 체크리스트
+
+- [x] predecessor `10_dart_isolate_gateway_contract` complete.log 존재를 확인한다. (archive: `agent-task/archive/2026/06/m-performance-hotspot-optimization/10_dart_isolate_gateway_contract/complete.log` 존재 + git `9082c0c`에서 archive 확인. 의존성 충족.)
+- [x] `ProtobufClient`가 decoded inline path 대신 hardened gateway-backed `onReceivedFrame`을 사용하게 한다. (`_parsing()`이 `onReceivedFrame(packetBytes)` 호출, 생성자에서 `IsolateInboundGateway` attach+start.)
+- [x] `WsProtobufClient` IO/web이 같은 receive coordinator path를 사용하고 web은 `SyncInboundGateway` fallback을 유지한다. (IO는 `inbound_gateway_io.dart`의 `IsolateInboundGateway`, web은 `inbound_gateway_web.dart`의 `SyncInboundGateway` 기반 `IsolateInboundGateway`. 둘 다 `_dispatch`가 `onReceivedFrame(bytes)`만 수행.)
+- [x] 내부 단일 수신 경로로 동작하며 public on/off 옵션을 추가하지 않는다. (gateway는 각 transport 생성자 내부에서만 attach. public API/생성자 시그니처 변경 없음.)
+- [x] TCP/WS FIFO, nonce matching, close cleanup regression tests와 focused stress 결과를 기록한다. (`dart test` 67 통과, focused stress dart 12 rows violations=0. 아래 검증 결과 참조.)
+- [x] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
+
+## 코드리뷰 전용 체크리스트
+
+- [x] 판정을 append한다.
+- [x] PASS이면 complete/archive 절차와 런타임 완료 이벤트를 처리한다.
+
+## 계획 대비 변경 사항
+
+- 계획 범위(REFACTOR-1/2)는 transport 3종을 `onReceivedFrame`으로 전환하는 것이었으나, 전환 후 기존 회귀 테스트 `TCP inbound backpressure pauses subscription on heavy load`가 실패했다. 원인은 transport swap 자체가 아니라 predecessor `10`의 gateway wiring이 **decode in-flight**(gateway capacity)로만 backpressure를 측정하는데, 실제 병목은 **dispatch**(stalled handler)라는 점이다. decode는 병목이 아니므로 handler가 막혀도 socket이 멈추지 않고 decoded frame이 unbounded로 쌓여 기존 64-bound 회귀가 발생했다.
+- 이를 plan의 "심볼 참조"가 명시한 변경 후보(`Communicator.attachInboundGateway`, `onReceivedFrame`) 범위 안에서 `dart/lib/src/communicator.dart`를 추가 수정해 해결했다. 계획의 "수정 파일 요약" 표에는 communicator.dart가 없었으나, 같은 plan 본문이 두 심볼을 변경 후보로 적시했으므로 범위 내로 판단했고 사용자 리뷰 요청으로 올리지 않았다.
+- web `IsolateInboundGateway`는 `SyncInboundGateway`를 상속한 conditional-export 구현이라, 계획 체크리스트의 "web `SyncInboundGateway` fallback 유지"를 그대로 만족한다(별도 분기 코드 불필요).
+
+## 주요 설계 결정
+
+- **gateway 경로 전용 backlog 카운터로 dispatch backpressure를 transport read loop에 전파.** `_gatewayBacklog`는 "gateway에 submit했지만 아직 dispatch 완료되지 않은 frame 수"다. `onReceivedFrame`(gateway branch)이 submit 전에 `_gatewayBacklog >= _inboundQueueCapacity`(64)면 대기하고, 통과 시 `_gatewayBacklog++` 후 submit한다. backlog는 frame이 실제로 handle된 시점(데이터/요청은 dispatch 체인 `whenComplete`, 응답은 pending 완료 직후, error는 즉시)에만 `_releaseGatewayBacklog()`로 감소한다. 따라서 stalled handler 하에서 backlog가 64까지 차면 read loop가 멈춰 inline 경로와 동일한 outstanding-frame bound를 회복한다.
+- **gateway 경로를 `onReceivedData`의 큐 회계와 분리.** `onReceivedData`(+`_inboundQueueLength`/`_awaitInboundCapacity`)는 inline 경로와 직접 호출 테스트가 계속 쓰므로 그대로 두고, gateway listener는 decoded frame을 `_onGatewayResult`에서 직접 dispatch 체인에 태운다. 이렇게 하면 큐 슬롯을 enqueue 시점이 아니라 submit 시점에 예약(backlog)하게 되어, submit→decode→enqueue 사이 비동기 지연으로 read loop가 앞서 달리던 문제를 제거한다(단일 카운터로 두 회계를 섞지 않는다).
+- **응답 immediacy 보존.** gateway 경로에서도 `responseNonce > 0` frame은 dispatch 체인을 거치지 않고 pending request를 즉시 완료한다 — inline `onReceivedData`와 동일한 out-of-band 처리.
+- **decode pipeline 깊이는 gateway capacity(1024)가, dispatch bound는 backlog(64)가 담당**하도록 두 backpressure 축을 분리해, 병렬 decode 이점과 메모리 bound를 동시에 유지한다.
+- **gateway start()는 await하지 않는다.** async 함수는 첫 await 전까지 동기 실행되어 `_ready` completer가 생성자 동기 구간에 설정되므로, isolate spawn 완료 전 도착한 첫 frame도 `submit`이 readiness를 기다려 drop되지 않는다.
+- **teardown 안전.** `close()`/`shutdown()`에 `_drainGatewayBacklog()`를 추가해 backlog를 0으로 리셋하고 대기 중인 `onReceivedFrame`을 깨워, gateway 경로가 read loop를 strand하지 않게 한다.
+
+## 사용자 리뷰 요청
+
+- 상태: 없음
+- 사유 유형: 없음
+- 결정 필요: 없음
+- 차단 근거: 없음
+- 실행한 검증/명령: 없음
+- 자동 후속 불가 이유: 없음
+- 재개 조건: 없음
+
+## 리뷰어를 위한 체크포인트
+
+- `10` predecessor가 실제 PASS인지 확인한다.
+- TCP/WS/web 모두 inline decode bypass 없이 단일 receive coordinator를 쓰는지 확인한다.
+- public gateway on/off 옵션이 생기지 않았는지 확인한다.
+
+## 검증 결과
+
+### REFACTOR-1/2 중간 검증
+```text
+$ cd dart && dart analyze lib/src/protobuf_client.dart lib/src/ws_protobuf_client_io.dart lib/src/ws_protobuf_client_web.dart lib/src/communicator.dart
+No issues found!
+
+$ cd dart && dart test test/socket_test.dart test/communicator_test.dart -r expanded
+00:15 +64: ... Request-Response (WS plain) TCP inbound backpressure pauses subscription on heavy load
+00:16 +65: ... TCP fragmentation and large payload TCP receives fragmented large frame without reordering or truncation
+00:16 +66: ... TCP fragmentation and large payload TCP 1MB payload roundtrip
+00:17 +67: All tests passed!
+```
+- FIFO/fragmentation/1MB/request-response/close/broadcast 회귀 및 gateway 단위 테스트 모두 통과(67).
+- backpressure 회귀 테스트는 transport swap 직후 FAIL이었고, communicator backlog gate 수정 후 PASS로 회복(계획 대비 변경 사항 참조).
+
+### REFACTOR-3 중간 검증 (focused stress)
+```text
+$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_stress.sh --quick --lang dart --transport tcp,ws --profile roundtrip,payload
+DONE dart: PASS (rows=12 skips=0 required_skips=0 violations=0)
+| 언어 | 결과 | ROW | SKIP(전체) | SKIP(required) | stability_violations | 비고 |
+| dart | PASS | 12 | 0 | 0 | 0 |  |
+전체 결과값: PASS
+결과 기록 파일: agent-test/runs/20260606-042547-proto-socket-stress-quick.md
+```
+
+### 최종 검증
+```text
+$ cd dart && dart test
+00:17 +67: All tests passed!
+
+$ cd dart && dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js
+Compiled 10,874,863 input bytes (5,876,091 characters source) to 11,692 characters JavaScript in 0.50 seconds
+
+$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_performance.sh --quick
+| 구성 | 결과 | exit code |
+| same-language | PASS | 0 |  (Dart 18 rows, 모든 row stability_violations=0)
+| cross-language | PASS | 0 |
+| typescript-gateway | PASS | 0 |
+전체 결과값: PASS
+결과 기록 파일: agent-test/runs/20260606-042556-proto-socket-performance-quick.md
+```
+- same-language run의 Dart 실제 수신 경로(gateway-backed) rows: roundtrip(c1/16/64/256), burst, sustained, parallel, payload(1KB/64KB) × tcp/ws 전부 PASS, stability_violations 0.
+
+## 코드리뷰 결과
+
+- 종합 판정: PASS
+- 차원별 평가:
+  - correctness: Pass
+  - completeness: Pass
+  - test coverage: Pass
+  - API contract: Pass
+  - code quality: Pass
+  - plan deviation: Pass
+  - verification trust: Pass
+- 발견된 문제: 없음
+- 다음 단계: PASS이므로 active plan/review를 `.log`로 아카이브하고 `complete.log` 작성 후 task 디렉터리를 archive로 이동한다.
--- a/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/complete.log
+++ b/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/complete.log
@ -0,0 +1,44 @@
+# Complete - m-performance-hotspot-optimization/11+10_dart_isolate_transport_path
+
+## 완료 일시
+
+2026-06-06
+
+## 요약
+
+Dart TCP/WS transport 수신 경로를 gateway-backed receive coordinator로 연결한 구현을 1회 리뷰했고 최종 판정은 PASS다.
+
+## 루프 이력
+
+| Plan | Review | Verdict | 메모 |
+|------|--------|---------|------|
+| `plan_cloud_G08_0.log` | `code_review_cloud_G08_0.log` | PASS | TCP/WS IO/web 수신 경로 전환, communicator gateway backlog 보정, focused stress/performance 검증이 계획 범위를 충족했다. |
+
+## 구현/정리 내용
+
+- `ProtobufClient` TCP 수신 path를 inline `PacketBase.fromBuffer` decode 대신 `onReceivedFrame` gateway-backed coordinator로 전환했다.
+- `WsProtobufClient` IO/web 수신 path를 동일한 coordinator contract로 전환하고 web conditional export의 sync fallback을 유지했다.
+- gateway decode와 dispatch 병목을 분리해 `_gatewayBacklog`로 stalled handler 상황의 transport backpressure를 보존했다.
+
+## 최종 검증
+
+- `dart analyze lib/src/protobuf_client.dart lib/src/ws_protobuf_client_io.dart lib/src/ws_protobuf_client_web.dart lib/src/communicator.dart` - PASS; No issues found.
+- `dart test test/socket_test.dart test/communicator_test.dart -r expanded` - PASS; 67 tests passed.
+- `dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js` - PASS; JavaScript output generated.
+- `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_stress.sh --quick --lang dart --transport tcp,ws --profile roundtrip,payload` - PASS; 12 rows, required skips 0, stability violations 0; result=`agent-test/runs/20260606-045133-proto-socket-stress-quick.md`.
+- `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_performance.sh --quick` - PASS; same-language/cross-language/typescript-gateway all PASS, regression SKIPPED; result=`agent-test/runs/20260606-045148-proto-socket-performance-quick.md`.
+
+## Roadmap Completion
+
+- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md`
+- Completed task ids:
+  - `dart-isolate-real-path`: PASS; evidence=`agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/plan_cloud_G08_0.log`, `agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/code_review_cloud_G08_0.log`; verification=`dart test test/socket_test.dart test/communicator_test.dart -r expanded`, `dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js`, `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_stress.sh --quick --lang dart --transport tcp,ws --profile roundtrip,payload`, `bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_performance.sh --quick`
+- Not completed task ids: 없음
+
+## 잔여 Nit
+
+- 없음
+
+## 후속 작업
+
+- 없음
--- a/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/plan_cloud_G08_0.log
+++ b/agent-task/archive/2026/06/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/plan_cloud_G08_0.log
--- a/agent-task/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/CODE_REVIEW-cloud-G08.md
+++ b/agent-task/m-performance-hotspot-optimization/11+10_dart_isolate_transport_path/CODE_REVIEW-cloud-G08.md
@ -1,92 +0,0 @@
-<!-- task=m-performance-hotspot-optimization/11+10_dart_isolate_transport_path plan=0 tag=REFACTOR -->
-
-# Code Review Reference - REFACTOR
-
-> **[IMPLEMENTING AGENT — READ FIRST] Filling in this file is the mandatory final step of implementation.**
-> Do not implement until predecessor `10` has `complete.log`.
-
-## 개요
-
-date=2026-06-06
-task=m-performance-hotspot-optimization/11+10_dart_isolate_transport_path, plan=0, tag=REFACTOR
-
-## Roadmap Targets
-
- Milestone: `agent-roadmap/milestones/performance-hotspot-optimization.md`
- Task ids:
-  - `dart-isolate-real-path`: 실제 TCP/WS transport 수신 경로 hardening 연결
- Completion mode: check-on-pass
-
-## 이 파일을 읽는 리뷰 에이전트에게
-
-> **[REVIEW AGENT ONLY]** 종결 절차는 코드리뷰 에이전트 전용이다.
-
-## 구현 항목별 완료 여부
-
-| 항목 | 완료 여부 |
-|------|---------|
-| [REFACTOR-1] TCP transport receive path | [ ] |
-| [REFACTOR-2] WS IO/web receive path | [ ] |
-| [REFACTOR-3] Performance row inclusion | [ ] |
-
-## 구현 체크리스트
-
- [ ] predecessor `10_dart_isolate_gateway_contract` complete.log 존재를 확인한다.
- [ ] `ProtobufClient`가 decoded inline path 대신 hardened gateway-backed `onReceivedFrame`을 사용하게 한다.
- [ ] `WsProtobufClient` IO/web이 같은 receive coordinator path를 사용하고 web은 `SyncInboundGateway` fallback을 유지한다.
- [ ] 내부 단일 수신 경로로 동작하며 public on/off 옵션을 추가하지 않는다.
- [ ] TCP/WS FIFO, nonce matching, close cleanup regression tests와 focused stress 결과를 기록한다.
- [ ] CODE_REVIEW-*-G??.md의 구현 에이전트 소유 섹션을 실제 구현 내용과 검증 출력으로 채운다. 이 항목이 완료되기 전에는 구현이 완료된 것이 아니다.
-
-## 코드리뷰 전용 체크리스트
-
- [ ] 판정을 append한다.
- [ ] PASS이면 complete/archive 절차와 런타임 완료 이벤트를 처리한다.
-
-## 계획 대비 변경 사항
-
-_구현 에이전트가 계획과 다르게 구현한 부분을 이유와 함께 기록한다._
-
-## 주요 설계 결정
-
-_구현 에이전트가 주요 설계 결정 사항을 기록한다._
-
-## 사용자 리뷰 요청
-
- 상태: 없음
- 사유 유형: 없음
- 결정 필요: 없음
- 차단 근거: 없음
- 실행한 검증/명령: 없음
- 자동 후속 불가 이유: 없음
- 재개 조건: 없음
-
-## 리뷰어를 위한 체크포인트
-
- `10` predecessor가 실제 PASS인지 확인한다.
- TCP/WS/web 모두 inline decode bypass 없이 단일 receive coordinator를 쓰는지 확인한다.
- public gateway on/off 옵션이 생기지 않았는지 확인한다.
-
-## 검증 결과
-
-### REFACTOR-1 중간 검증
-```text
-$ cd dart && dart test test/socket_test.dart -r expanded
-(output)
-```
-
-### REFACTOR-3 중간 검증
-```text
-$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_stress.sh --quick --lang dart --transport tcp,ws --profile roundtrip,payload
-(output)
-```
-
-### 최종 검증
-```text
-$ cd dart && dart test test/socket_test.dart test/communicator_test.dart -r expanded
-(output)
-$ cd dart && dart compile js test/browser_ws_import_compile.dart -o /tmp/proto_socket_browser_ws_import_compile.js
-(output)
-$ bash agent-ops/skills/project/run-proto-socket-test-matrix/scripts/run_performance.sh --quick
-(output)
-```
--- a/dart/lib/src/communicator.dart
+++ b/dart/lib/src/communicator.dart
@ -47,6 +47,16 @@ abstract class Communicator {
  StreamSubscription<DecodedFrame>? _gatewaySubscription;
  int _frameSeq = 0;

+  /// Frames handed to the gateway but not yet accepted into the dispatch queue.
+  ///
+  /// Decode is never the bottleneck, so the gateway's own in-flight capacity
+  /// cannot throttle a stalled dispatch handler. Gating [onReceivedFrame] on
+  /// this backlog makes a full inbound queue propagate to the transport read
+  /// loop and keeps decoded frames from piling up unbounded behind a stuck
+  /// handler — the same bound the inline path gets for free.
+  int _gatewayBacklog = 0;
+  Completer<void>? _gatewayBacklogWaiter;
+
  /// Monotonically increasing nonce. Shared by send / sendRequest / response.
  int _nonce = 0;
  int get nonce => _nonce;
@ -233,24 +243,56 @@ abstract class Communicator {

  /// Attaches a worker [gateway] in front of the receive coordinator.
  ///
-  /// Decoded results are forwarded to [onReceivedData] in input [seq] order, so
-  /// the coordinator keeps its FIFO dispatch and sole ownership of stateful
-  /// handling. Attaching is opt-in; without it [onReceivedFrame] decodes inline.
+  /// Decoded results are dispatched in input [seq] order, so the coordinator
+  /// keeps its FIFO dispatch and sole ownership of stateful handling. Each
+  /// result releases the backlog slot its frame reserved in [onReceivedFrame],
+  /// only once the frame is actually handled. Attaching is opt-in; without it
+  /// [onReceivedFrame] decodes inline.
  @protected
  void attachInboundGateway(InboundGateway gateway) {
    _inboundGateway = gateway;
-    _gatewaySubscription = gateway.results.listen((frame) {
-      if (frame.isError) {
-        onGatewayDecodeError(frame.seq, frame.error!);
-        return;
-      }
-      onReceivedData(
-        frame.typeName,
-        frame.data,
-        incomingNonce: frame.incomingNonce,
-        responseNonce: frame.responseNonce,
-      );
-    });
+    _gatewaySubscription = gateway.results.listen(_onGatewayResult);
+  }
+
+  /// Consumes one ordered gateway result and frees its reserved backlog slot
+  /// once handled — not merely decoded — so a stalled handler keeps the
+  /// transport read loop paused via [onReceivedFrame].
+  void _onGatewayResult(DecodedFrame frame) {
+    if (frame.isError) {
+      onGatewayDecodeError(frame.seq, frame.error!);
+      _releaseGatewayBacklog();
+      return;
+    }
+    if (frame.responseNonce > 0) {
+      // Responses complete out of band, exactly as the inline coordinator does.
+      final pending = _pendingRequests.remove(frame.responseNonce);
+      pending?.complete(frame.typeName, frame.data);
+      _releaseGatewayBacklog();
+      return;
+    }
+    // Data/request frames serialize on the dispatch chain. The reserved backlog
+    // slot is held until dispatch completes, so the chain depth — and thus the
+    // outstanding frame bound — matches the inline coordinator's queue.
+    final item = _InboundItem(
+      typeName: frame.typeName,
+      data: frame.data,
+      incomingNonce: frame.incomingNonce,
+      responseNonce: frame.responseNonce,
+    );
+    _inboundDispatch = _inboundDispatch
+        .then((_) => _dispatchInbound(item))
+        .whenComplete(_releaseGatewayBacklog);
+  }
+
+  /// Frees one backlog slot and wakes a waiting [onReceivedFrame] if the gateway
+  /// has fallen back under capacity.
+  void _releaseGatewayBacklog() {
+    if (_gatewayBacklog > 0) _gatewayBacklog--;
+    final waiter = _gatewayBacklogWaiter;
+    if (waiter != null && _gatewayBacklog < _inboundQueueCapacity) {
+      _gatewayBacklogWaiter = null;
+      waiter.complete();
+    }
  }

  /// Handles an ordered decode failure surfaced by the gateway.
@ -279,6 +321,15 @@ abstract class Communicator {
    if (!isAlive || _isClosing) return;
    final gateway = _inboundGateway;
    if (gateway != null) {
+      // Throttle on the dispatch backlog before handing the frame off. This
+      // propagates a full inbound queue to the transport read loop even though
+      // the gateway's in-flight capacity (decode) is never the bottleneck.
+      while (isAlive && !_isClosing && _gatewayBacklog >= _inboundQueueCapacity) {
+        _gatewayBacklogWaiter ??= Completer<void>();
+        await _gatewayBacklogWaiter!.future;
+      }
+      if (!isAlive || _isClosing) return;
+      _gatewayBacklog++;
      // Await so gateway backpressure propagates to the transport read loop.
      await gateway.submit(InboundFrame(seq: _nextFrameSeq(), bytes: frame));
      return;
@ -288,6 +339,15 @@ abstract class Communicator {
        incomingNonce: common.nonce, responseNonce: common.responseNonce);
  }

+  /// Blocks until the inbound dispatch queue has room. Shared by the inline and
+  /// gateway receive paths so a stalled handler throttles the transport equally.
+  Future<void> _awaitInboundCapacity() async {
+    while (_inboundQueueLength >= _inboundQueueCapacity) {
+      _queueWaiter ??= Completer<void>();
+      await _queueWaiter!.future;
+    }
+  }
+
  /// Enqueues an inbound packet for serial dispatch.
  ///
  /// The receive worker processes items one at a time, preserving FIFO order
@ -302,10 +362,7 @@ abstract class Communicator {
      }
      return;
    }
-    while (_inboundQueueLength >= _inboundQueueCapacity) {
-      _queueWaiter ??= Completer<void>();
-      await _queueWaiter!.future;
-    }
+    await _awaitInboundCapacity();
    if (!isAlive || _isClosing) return;
    _inboundQueueLength++;
    final item = _InboundItem(
@ -337,6 +394,7 @@ abstract class Communicator {
      _queueWaiter!.complete();
      _queueWaiter = null;
    }
+    _drainGatewayBacklog();
    await _closeInboundGateway();
    cancelPendingRequests();
  }
@ -351,10 +409,22 @@ abstract class Communicator {
      _queueWaiter!.complete();
      _queueWaiter = null;
    }
+    _drainGatewayBacklog();
    unawaited(_closeInboundGateway());
    cancelPendingRequests();
  }

+  /// Releases any [onReceivedFrame] blocked on backlog and clears the counter so
+  /// teardown never strands the transport read loop.
+  void _drainGatewayBacklog() {
+    _gatewayBacklog = 0;
+    final waiter = _gatewayBacklogWaiter;
+    if (waiter != null) {
+      _gatewayBacklogWaiter = null;
+      waiter.complete();
+    }
+  }
+
  Future<void> _closeInboundGateway() async {
    final subscription = _gatewaySubscription;
    final gateway = _inboundGateway;
--- a/dart/lib/src/protobuf_client.dart
+++ b/dart/lib/src/protobuf_client.dart
@ -7,6 +7,7 @@ import 'dart:collection';

 import 'package:protobuf/protobuf.dart';
 import 'base_client.dart';
+import 'inbound_gateway_io.dart';
 import 'packets/message_common.pb.dart';
 import 'transport.dart';

@ -52,6 +53,12 @@ abstract class ProtobufClient extends BaseClient<ProtobufClient> {
      (HeartBeat).toString(): HeartBeat.fromBuffer,
    });
    super.initialize(parserMap, transport: _transport);
+    // 제품 수신 경로를 단일 gateway-backed coordinator로 통일한다. start()는
+    // await하지 않아도 _ready completer를 동기 구간에서 설정하므로, spawn 완료
+    // 전에 도착한 첫 프레임도 submit에서 readiness를 기다려 drop되지 않는다.
+    final gateway = IsolateInboundGateway();
+    attachInboundGateway(gateway);
+    unawaited(gateway.start());
    _subscription = _socket.listen(onData, onError: onError);
    _subscription!.asFuture<void>().then((_) {
      close();
@ -108,9 +115,10 @@ abstract class ProtobufClient extends BaseClient<ProtobufClient> {
      _frameBuffer.takeBytes(_headerSize);
      final packetBytes = _frameBuffer.takeBytes(_length!);
      _length = null;
-      final common = PacketBase.fromBuffer(packetBytes);
-      await onReceivedData(common.typeName, common.data,
-          incomingNonce: common.nonce, responseNonce: common.responseNonce);
+      // raw frame을 coordinator로 넘긴다. gateway가 envelope decode를 off-loop로
+      // 처리하고 seq 순서로 onReceivedData에 dispatch한다. submit await는 gateway
+      // backpressure를 socket read loop로 전파한다.
+      await onReceivedFrame(packetBytes);
      sendHeartBeat();
    }
  }
--- a/dart/lib/src/ws_protobuf_client_io.dart
+++ b/dart/lib/src/ws_protobuf_client_io.dart
@ -6,6 +6,7 @@ import 'dart:typed_data';

 import 'package:protobuf/protobuf.dart';
 import 'base_client.dart';
+import 'inbound_gateway_io.dart';
 import 'packets/message_common.pb.dart';
 import 'transport.dart';

@ -44,6 +45,11 @@ abstract class WsProtobufClient extends BaseClient<WsProtobufClient> {
    isAlive = true;
    parserMap.addAll({(HeartBeat).toString(): HeartBeat.fromBuffer});
    super.initialize(parserMap, transport: _transport);
+    // TCP 경로와 동일한 gateway-backed receive coordinator를 사용한다. start()는
+    // _ready completer를 동기 구간에서 설정하므로 spawn 전 첫 프레임도 drop되지 않는다.
+    final gateway = IsolateInboundGateway();
+    attachInboundGateway(gateway);
+    unawaited(gateway.start());
    _subscription = _ws.listen(_onMessage, onError: onError, onDone: close);
    addListener(onHeartBeat);
    sendHeartBeat();
@ -65,9 +71,7 @@ abstract class WsProtobufClient extends BaseClient<WsProtobufClient> {

  Future<void> _dispatch(dynamic data) async {
    final bytes = data is List<int> ? data : (data as Uint8List).toList();
-    final common = PacketBase.fromBuffer(bytes);
-    await onReceivedData(common.typeName, common.data,
-        incomingNonce: common.nonce, responseNonce: common.responseNonce);
+    await onReceivedFrame(bytes);
    sendHeartBeat();
  }

--- a/dart/lib/src/ws_protobuf_client_web.dart
+++ b/dart/lib/src/ws_protobuf_client_web.dart
@ -6,6 +6,7 @@ import 'dart:typed_data';

 import 'package:protobuf/protobuf.dart';
 import 'base_client.dart';
+import 'inbound_gateway_web.dart';
 import 'packets/message_common.pb.dart';
 import 'transport.dart';

@ -67,6 +68,12 @@ abstract class WsProtobufClient extends BaseClient<WsProtobufClient> {
    isAlive = true;
    parserMap.addAll({(HeartBeat).toString(): HeartBeat.fromBuffer});
    super.initialize(parserMap, transport: _transport);
+    // web에는 worker isolate가 없으므로 IsolateInboundGateway는 SyncInboundGateway
+    // fallback으로 in-process decode하되 동일한 seq-ordered receive coordinator
+    // contract를 유지한다.
+    final gateway = IsolateInboundGateway();
+    attachInboundGateway(gateway);
+    unawaited(gateway.start());
    _subscription = _ws.onMessage.listen(_onMessage, onError: onError);
    _ws.onClose.listen((_) => close());
    addListener(onHeartBeat);
@ -104,9 +111,7 @@ abstract class WsProtobufClient extends BaseClient<WsProtobufClient> {
  }

  Future<void> _dispatch(List<int> bytes) async {
-    final common = PacketBase.fromBuffer(bytes);
-    await onReceivedData(common.typeName, common.data,
-        incomingNonce: common.nonce, responseNonce: common.responseNonce);
+    await onReceivedFrame(bytes);
    sendHeartBeat();
  }