diff --git a/API_COVERAGE.md b/API_COVERAGE.md
index 6b20cfe..960176e 100644
--- a/API_COVERAGE.md
+++ b/API_COVERAGE.md
@@ -1,6 +1,6 @@
 # API coverage
 
-Snapshot date: 2026-06-11
+상태 기준일: 2026-09-19 (`traffic.flow` 검증 갱신)
 
 이 문서는 섞이기 쉬운 세 가지 상태를 분리한다.
 
@@ -8,6 +8,13 @@ Snapshot date: 2026-06-11
 - **구현됨**: `KrexClient` method로 노출됨.
 - **Live 검증됨**: `tests/test_live_ex.py`에서 실제 provider를 호출해 확인함.
 
+0405는 2026-09-19에 인증 요청 총 3회로 조사·검증했다. 마지막 공개
+`traffic.flow(num_of_rows=20000)` 호출은 8,370행/69개 노선을 모두 파싱했다.
+기점 방향 4,188행, 종점 방향 4,182행, 음수 속도 결측 1,520행이었다.
+개수는 해당 시점의 관측값이며 고정 계약이 아니다. 인증키나 인증 URL은 저장하지 않았다.
+이후 추가한 `flow_all()`과 공개 `vds_id`는 같은 실제 응답 표본 및 가짜 세션으로
+검증했다. 추가 실호출은 하지 않았으며 새 메서드의 직접 실호출 검증 상태는 미검증이다.
+
 이 프로젝트는 `data.ex.co.kr`와 `data.go.kr`에 흩어진 한국도로공사 API 전체를 아직 완전히 감싼 wrapper가 아니다. 현재 저장소에 문서화된 endpoint set을 typed client로 제공하고, 더 넓은 공식 coverage는 backlog로 명시한다.
 
 ## 현재 저장소 coverage
@@ -16,7 +23,8 @@ Snapshot date: 2026-06-11
 |---|---|---|---:|---:|---|
 | `traffic.by_ic()` | `data.ex.co.kr` | `Page[TrafficByIc]` | Yes | Yes | 실제 응답은 top-level `trafficIc`와 `trafficAmout` 같은 field variant를 사용한다. |
 | `traffic.by_route()` | `data.ex.co.kr` | `Page[dict]` | Yes | Yes | Empty success가 `count=0`, `list=[]`일 수 있다. |
-| `traffic.flow()` | `data.ex.co.kr` | `Page[TrafficFlow]` | Yes | No | 과거 live probe에서 404가 반환되어 portal UI 확인 전까지 unverified로 둔다. |
+| `traffic.flow()` | `data.ex.co.kr` | `Page[TrafficFlow]` | Yes | Yes | 2026-09-19 공식 0405 `odtraffic/trafficAmountByRealtime`와 실제 8,370행 확인. 필터·페이지 분할은 로컬 처리. `S/E`는 기점/종점, 시각은 `stdDate/stdHour`, 음수 속도는 결측. 2행 표본으로 오프라인 재현. |
+| `traffic.flow_all()` | `data.ex.co.kr` | `Page[TrafficFlow]` | Yes | No | 동일 엔드포인트의 전체 응답을 한 번 조회하고 모든 VDS 행을 반환한다. `flow()`가 재사용한다. 실제 표본의 `vdsId` 및 1,000행 초과·같은 콘존의 여러 VDS 보존을 오프라인 검증. |
 | `traffic.incident()` | `data.ex.co.kr` | `Page[Incident]` | Yes | Yes | 구 `trafficapi/incident`는 404로 제거 확인. `burstInfo/realTimeSms`(apiId 0611)로 repoint, 2026-06-11 live 검증 (count=190). 경도는 `altitude` 키로 온다. `realTimeSMSList`와 0 이상인 `count`가 없는 HTTP 200 본문은 파싱 오류로 거부한다. |
 | `traffic.vds_raw()` | `data.ex.co.kr` | `Page[dict]` | Yes | No | High-volume raw endpoint이므로 live test는 date/time range를 좁힌다. |
 | `traffic.avc_raw()` | `data.ex.co.kr` | `Page[dict]` | Yes | No | `vds_id`, `std_date`가 필요하다. |
@@ -45,12 +53,12 @@ Snapshot date: 2026-06-11
 
 | Category | Count |
 |---|---:|
-| `endpoints.md`에 문서화된 method | 26 |
-| `KrexClient` namespace에 구현된 method | 27 |
-| Typed public model을 반환하는 method | 10 |
+| `endpoints.md`에 문서화된 method | 27 |
+| `KrexClient` namespace에 구현된 method | 28 |
+| Typed public model을 반환하는 method | 11 |
 | Raw `dict` record를 반환하는 method | 13 |
 | Local reference helper | 3 |
-| Provider live 검증 완료 method | 6 |
+| Provider live 검증 완료 method | 7 |
 
 ## 더 넓은 공식 API backlog
 
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ab054b8..2c35b8c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,21 @@
 # 변경 기록
 
+## 2026-09-19 — 실시간 소통 조회 복구
+
+- 인자 없는 `traffic.flow_all()`이 한 응답의 모든 행을 `Page[TrafficFlow]`로 반환한다.
+  `flow()`는 이를 재사용한 뒤 기존 로컬 필터·페이지 분할을 적용한다.
+  `TrafficFlow.vds_id: str | None = None`에 공식 `vdsId`를 보존한다.
+  실제 표본으로 필드명을 검증했고, 같은 콘존의 여러 VDS를 합치지 않는다.
+- `traffic.flow()`를 공식 0405 `/openapi/odtraffic/trafficAmountByRealtime`로 수정했다.
+  서버가 제공하지 않는 필터·페이지 인자는 로컬 처리하며 기존 이름·기본값을 유지한다.
+- `stdDate/stdHour`를 기존 `updated_at` 문자열로 조합하고 음수 속도는 결측 처리한다.
+  `grade=3`은 `CongestionLevel.STOP`으로 매핑하며 `timeAvg`를 자유속도로 사용하지 않는다.
+- `FlowDirection.START/END`를 공개했다. 현행 `S/E`는 기점/종점이므로
+  `flow(direction=...)`에서 기존 `Direction`과 `0/1`을 거부한다.
+  `TrafficFlow.direction`은 현행 `FlowDirection`과 과거 응답의 `Direction`을 지원한다.
+- 목록·건수 및 모든 행을 검증하고 잘못된 응답을 빈 성공이나 일부 성공으로 숨기지 않는다.
+  실제 응답에서 추출한 인증정보 없는 2행 표본과 회귀 테스트를 추가했다.
+
 ## 2026-09-14 — native async와 공통 TPS
 
 KrexClient와 서비스·debug를 native async로 전환하고 동기 bridge/Async 접두사/aio를 제거했다. 공통 AsyncTokenBucket과 max_rps/rate_limiter를 공개한다. 기존 기본 5 TPS·capacity=1과 재시도 과금을 보존하고 리다이렉트·동시 디버그·주입 세션 수명을 함께 처리한다.
diff --git a/README.md b/README.md
index b78a01c..8dc7051 100644
--- a/README.md
+++ b/README.md
@@ -28,6 +28,18 @@
 ## 핵심 특징
 
 - **네임스페이스형 클라이언트**: `client.traffic.flow()`, `client.tollfee.between_tollgates()`처럼 문서의 API 범주와 같은 구조로 호출합니다.
+
+`traffic.flow()`는 공식 0405 실시간 소통 데이터를 한 번 조회한 뒤 노선·콘존·방향
+필터와 페이지 분할을 로컬에서 적용합니다. 기본 페이지는 최대 1,000행이며
+`total_count`는 필터 후 전체 건수입니다. 전체 수집에는
+`page = await client.traffic.flow_all()`을 사용하세요. 인자 없이 한 응답의 모든 행을
+`Page[TrafficFlow]`로 반환하며 페이지 크기를 추측할 필요가 없습니다.
+`flow()`도 내부에서 `flow_all()`을 호출한 뒤 로컬 필터·페이지 처리를 적용합니다.
+방향은 `FlowDirection.START`(`S`, 기점)/`END`(`E`, 종점)를 사용합니다.
+기존 `Direction.UP/DOWN/EAST/SOUTH`와는 의미가 달라 `flow()` 필터에 사용할 수 없습니다.
+`updated_at`은 KST로 취급하는 `YYYYMMDDHHMM` 문자열이며, 음수 속도는 `None`입니다.
+같은 콘존의 여러 VDS 행은 그대로 유지하며 공식 `vdsId`는 `item.vds_id`로 제공합니다.
+과거 응답에 이 필드가 없으면 `None`입니다. [상세 계약](endpoints.md#실시간-소통)을 참고하세요.
 - **두 포털 동시 지원**: `data.ex.co.kr` 키(`KEX_EX_API_KEY`)와 `data.go.kr` 키(`DATA_GO_KR_SERVICE_KEY`)를 분리해 사용합니다.
 - **로컬 `.env` 기본 로딩**: 환경변수가 없으면 현재 작업 디렉터리부터 부모 디렉터리의 `.env`를 찾아 키를 읽고, 복붙 과정에서 섞인 공백 문자를 제거합니다.
 - **API 카탈로그**: 구현된 API의 함수명, 데이터셋명, 포털, 엔드포인트, 서비스키 발급/활용신청 링크를 `api_catalog()` 또는 `get_api_catalog()`로 조회할 수 있습니다.
@@ -177,6 +189,7 @@ client = KrexClient(
 | 교통 | `traffic.by_ic()` | `data.ex.co.kr` | `Page[TrafficByIc]` |
 | 교통 | `traffic.by_route()` | `data.ex.co.kr` | `Page[dict]` |
 | 교통 | `traffic.flow()` | `data.ex.co.kr` | `Page[TrafficFlow]` |
+| 교통 | `traffic.flow_all()` | `data.ex.co.kr` | `Page[TrafficFlow]` |
 | 교통 | `traffic.incident()` | `data.ex.co.kr` | `Page[Incident]` |
 | 교통 | `traffic.vds_raw()`, `traffic.avc_raw()` | `data.ex.co.kr` | `Page[dict]` |
 | 통행료 | `tollfee.between_tollgates()` | `data.ex.co.kr` | `Page[TollFee]` |
diff --git a/codes.md b/codes.md
index 8252fa8..6b3e1c7 100644
--- a/codes.md
+++ b/codes.md
@@ -16,6 +16,7 @@
 - [진출입 (IOType)](#진출입-iotype)
 - [시간 단위 (TimeUnit)](#시간-단위-timeunit)
 - [방향 (Direction)](#방향-direction)
+- [실시간 소통 방향 (FlowDirection)](#실시간-소통-방향-flowdirection)
 - [혼잡도 (CongestionLevel)](#혼잡도-congestionlevel)
 - [할인 유형 (DiscountType)](#할인-유형-discounttype)
 - [노선 코드](#노선-코드)
@@ -122,8 +123,20 @@ CarType.from_label("1종")     # CarType.LIGHT
 
 ---
 
+## 실시간 소통 방향 (FlowDirection)
+
+0405 `traffic.flow()`는 별도 `FlowDirection`을 사용한다.
+`FlowDirection.START="S"`는 기점 방향, `FlowDirection.END="E"`는 종점 방향이다.
+이 코드들을 기존 `Direction.SOUTH/EAST`나 `UP/DOWN`으로 해석하면 안 된다.
+과거 응답의 `dirType`은 기존 `Direction`으로 파싱하는 호환성을 유지한다.
+
 ## 혼잡도 (CongestionLevel)
 
+0405의 원본 `grade`는 `0`=판정불가, `1`=원활(80km/h 이상),
+`2`=서행(40 이상 80 미만), `3`=정체(0 이상 40 미만)다.
+각각 `None`, `SMOOTH`, `SLOW`, `STOP`으로 정규화한다. 아래 공통 enum의
+기존 값은 보존하므로 `STOP.value="4"`와 원본 `grade="3"`을 혼동하지 않는다.
+
 실시간 소통 정보의 혼잡 수준.
 
 | 코드 | Enum | 색상 | 통행속도 기준 (대략) |
diff --git a/endpoints.md b/endpoints.md
index bfdf4ac..7745334 100644
--- a/endpoints.md
+++ b/endpoints.md
@@ -158,34 +158,71 @@ asyncio.run(main())
 
 ### 실시간 소통
 
-도로별 통행속도, 혼잡도(원활/지체/정체) 정보.
+VDS별 통행속도와 콘존 정보(apiId `0405`). 같은 콘존에 여러 VDS 관측 행이 있으며
+라이브러리는 이들을 집계하거나 중복 제거하지 않는다.
 
 - **포털**: `data.ex.co.kr`
-- **경로**: `/openapi/trafficapi/realFlow`
-- **공공데이터포털 대응**: 데이터셋 ID `15076684`
-- **메서드**: `client.traffic.flow()`
+- **경로**: `/openapi/odtraffic/trafficAmountByRealtime`
+- **공식 명세**: [실시간 소통 데이터](https://data.ex.co.kr/openapi/basicinfo/openApiInfoM?apiId=0405)
+- **메서드**: `client.traffic.flow()`, `client.traffic.flow_all()`
+
+전체 수집은 인자 없는 `page = await client.traffic.flow_all()`을 사용한다.
+한 번의 논리적 조회에서 받은 전체 응답을 검증하고 모든 행을 `Page[TrafficFlow]`로
+반환한다. 필터·페이지 분할·건수 상한·중복 제거는 없으며 `len(page.items)`와
+`page.total_count`는 일치한다. `page_no`와 `num_of_rows`는 `None`, `raw`는
+원본 전체 응답이다. 공통 HTTP 재시도 정책은 유지한다.
+이는 여러 응답을 페이지별로 합치지 않는다는 보장이며, 공급자가 모든 VDS의
+수집시각을 같게 보장한다는 뜻은 아니다.
 
 **파라미터**
 
 | 이름 | 필수 | 타입 | 설명 |
 |------|------|------|------|
-| `routeNo` | O | `code` | 노선번호 (생략 시 전체) |
-| `conzoneId` | O | `str` | 콘존 ID (구간 ID) |
-| `dirType` | O | `code` | 방향 |
+| `key` | 필수 | `str` | 인증키, 클라이언트가 자동 제공 |
+| `type` | 필수 | `str` | `json`, 클라이언트가 자동 제공 |
+
+2026-09-19 공식 동적 메타데이터(`openApiInfoDetail`, `openApiInfoPop`)에는
+필터·페이지 입력이 없다. 실제 `routeNo=0010`, `numOfRows=2`, `pageNo=2` 요청도
+69개 노선 8,370행 전체를 반환했다. 따라서 이 인자들을 서버로 보내지 않는다.
+
+기존 Python 인자 중 `route_no`, `conzone_id`, `num_of_rows=1000`, `page_no=1`을
+유지하며 `flow_all()`로 전체 응답 검증 → 로컬 필터 → 로컬 페이지 분할 순으로 적용한다.
+`direction`은 `FlowDirection.START`/`END` 또는 `"S"`/`"E"`만 받는다.
+의미가 다른 `Direction` 인스턴스와 `"0"`/`"1"`은 요청 전에 거부한다.
+`Page.total_count`는 필터 후 건수, `Page.raw["count"]`는 서버 전체 건수다.
+페이지 크기·번호는 양의 정수여야 한다. 필터 결과가 없거나 마지막 페이지를 넘으면 빈 페이지다.
+각 호출은 전체 응답을 다시 가져오므로 다른 페이지 호출 사이에 수집 시각이 바뀔 수 있다.
+전체 자료가 필요하면 `flow_all()`을 한 번 호출한다. 자동 반복 호출은 하지 않는다.
 
 **응답 필드**
 
 | 필드 | 설명 |
 |------|------|
-| `conzoneId` | 구간 식별자 |
-| `conzoneName` | 구간명 (예: "서울→수원") |
-| `routeNo` / `routeName` | 노선 번호/명 |
-| `speed` | 평균 통행속도 (km/h) |
-| `tmFreeFlow` | 자유속도 |
-| `congestionLevel` | `1`=원활, `2`=서행, `3`=지체, `4`=정체 |
-| `updTime` | 갱신 시각 |
-
-> **갱신 주기**: 5분. 더 빠른 폴링은 같은 데이터를 반환하므로 호출 한도만 소모.
+| `conzoneId` / `conzoneName` | 구간 식별자/명 → `conzone_id` / `conzone_name` |
+| `routeNo` / `routeName` | 노선 번호/명. 선행 0 보존 |
+| `vdsId` | VDS 식별자 → `vds_id: str \| None = None`. 선행 0과 원문 `raw` 보존. 실제 표본은 `0010VDE00100`, `0010VDS00100`. 과거 응답의 필드 누락은 `None` |
+| `stdDate` / `stdHour` | 수집일자 `YYYYMMDD` / 수집시각 `HHMM` → `updated_at=YYYYMMDDHHMM` 문자열. 한국 현지 시각(KST)으로 취급 |
+| `updownTypeCode` | `S`=기점 방향 → `FlowDirection.START`, `E`=종점 방향 → `FlowDirection.END`. 동서남북 또는 상하행으로 추정하지 않음 |
+| `speed` | 속도(km/h) → `speed`. 실제 음수 결측값은 `None`, 0은 정지 속도로 보존 |
+| `grade` | `0`=판정불가 → `None`, `1`=원활(80 이상) → `SMOOTH`, `2`=서행(40 이상 80 미만) → `SLOW`, `3`=정체(0 이상 40 미만) → `STOP` |
+| `timeAvg` | 통행시간, `raw`에 보존. 공식 메타데이터에 단위 없음. 자유속도로 사용하지 않음 |
+| `trafficAmout` / `shareRatio` | 교통량(대) / 점유율, `raw`에 보존 |
+
+`TrafficFlow`의 기존 필드명과 `updTime`, `avgSpeed`, `dirType` 등 과거 응답 별칭은
+유지한다. 현행 응답은 자유속도를 주지 않으므로 `free_flow_speed=None`이다.
+기존 공통 `CongestionLevel.STOP`의 값은 `"4"`이며, 0405 원본 `grade="3"`과 다르다.
+원본 등급·방향·음수 속도는 항상 `raw`에서 확인할 수 있다.
+
+두 메서드 모두 응답에 `list`와 정확한 전체 `count`가 있어야 한다. 누락·잘못된 타입·건수 불일치·
+일부 행 파싱 실패는 `KrexParseError`다. 명시적인 `count=0, list=[]`는 정상 빈 응답이다.
+`NO_DATA` 오류는 기존 `strict_no_data` 설정을 따르지만, 잘못된 응답은 이 설정과
+무관하게 거부한다. 검증을 통과한 단일 객체 `list`도 1행으로 정규화한다.
+
+근거는 공식 [JavaScript](https://data.ex.co.kr/js/sub/openApiInfo.js)가 호출하는
+`POST /openapi/basicinfo/openApiInfo`, `openApiInfoDetail`, `openApiInfoPop`
+(`apiId=0405`)와 실제 응답이다. 명세에는 시간대·갱신 주기·통행시간 단위가 명시되지
+않았다. KST 해석은 한국 현지 수집시각과 검증 시각이 일치한 관측에 근거하며,
+갱신 주기를 5분으로 단정하지 않는다.
 
 ---
 
diff --git a/error-codes.md b/error-codes.md
index 4770250..ebb7345 100644
--- a/error-codes.md
+++ b/error-codes.md
@@ -96,6 +96,9 @@ asyncio.run(main())
 
 기본 `strict_no_data=True`는 공급자의 NO_DATA를 예외로 전달한다.
 빈 Page를 받으려면 생성자에 `strict_no_data=False`를 명시한다.
+`traffic.flow()`와 `traffic.flow_all()`의 명시적인 `count=0, list=[]`는 정상 빈 응답이다.
+목록·건수 누락, 건수 불일치, 일부 행 파싱 실패는 `strict_no_data=False`에서도
+`KrexParseError`로 거부한다. 빈 필터 결과나 범위 밖 로컬 페이지는 빈 `Page`다.
 `latest_weather`는 과거 시간대를 찾는 목적에 맞게 NO_DATA인 시간대를 건너뛴다.
 
 ```python
diff --git a/src/krex/__init__.py b/src/krex/__init__.py
index 37849e4..9362676 100644
--- a/src/krex/__init__.py
+++ b/src/krex/__init__.py
@@ -9,6 +9,7 @@
     CoordinateSystem,
     Direction,
     DiscountType,
+    FlowDirection,
     IOType,
     KrexCode,
     RoadOperator,
@@ -64,6 +65,7 @@
     "DiscountType",
     "DebugRun",
     "FoodPrice",
+    "FlowDirection",
     "IOType",
     "Incident",
     "KrexCode",
diff --git a/src/krex/catalog.py b/src/krex/catalog.py
index 3dbc3f0..19c92ea 100644
--- a/src/krex/catalog.py
+++ b/src/krex/catalog.py
@@ -75,10 +75,20 @@ def _go(
     _ex(
         "traffic.flow",
         "한국도로공사_실시간 소통정보",
-        "/openapi/trafficapi/realFlow",
+        "/openapi/odtraffic/trafficAmountByRealtime",
         "Page[TrafficFlow]",
-        "콘존별 실시간 속도와 정체 상태입니다.",
+        "VDS별 실시간 속도와 콘존 정보입니다. 필터와 페이지 분할은 로컬에서 처리합니다.",
+        live_verified=True,
+        fixture_supported=True,
+    ),
+    _ex(
+        "traffic.flow_all",
+        "한국도로공사_실시간 소통정보 전체",
+        "/openapi/odtraffic/trafficAmountByRealtime",
+        "Page[TrafficFlow]",
+        "하나의 전체 응답에 포함된 모든 VDS 관측값을 필터·페이지 분할 없이 반환합니다.",
         live_verified=False,
+        fixture_supported=True,
     ),
     _ex(
         "traffic.incident",
diff --git a/src/krex/client.py b/src/krex/client.py
index e3ed5c0..0b190b1 100644
--- a/src/krex/client.py
+++ b/src/krex/client.py
@@ -26,6 +26,7 @@
     CoordinateSystem,
     Direction,
     DiscountType,
+    FlowDirection,
     IOType,
     KrexCode,
     RoadOperator,
@@ -282,26 +283,65 @@ async def flow(
         *,
         route_no: str | None = None,
         conzone_id: str | None = None,
-        direction: Direction | str | None = None,
+        direction: FlowDirection | str | None = None,
         num_of_rows: int = 1000,
         page_no: int = 1,
     ) -> Page[TrafficFlow]:
+        """0405 전체 응답을 한 번 조회한 뒤 로컬 필터와 페이지 분할을 적용한다."""
         try:
-            return await self._client._page_ex(
-                "/openapi/trafficapi/realFlow",
-                {
-                    "routeNo": route_no,
-                    "conzoneId": conzone_id,
-                    "dirType": _optional_code(Direction, direction, "direction"),
-                    "numOfRows": num_of_rows,
-                    "pageNo": page_no,
-                },
-                _traffic_flow,
+            if isinstance(direction, Direction):
+                raise KrexInvalidParameterError(
+                    "flow direction requires FlowDirection, not Direction"
+                )
+            direction_code = _optional_code(FlowDirection, direction, "direction")
+            for name, value in (("num_of_rows", num_of_rows), ("page_no", page_no)):
+                if isinstance(value, bool) or not isinstance(value, int) or value < 1:
+                    raise KrexInvalidParameterError(f"{name} must be a positive integer")
+            for name, filter_value in (("route_no", route_no), ("conzone_id", conzone_id)):
+                if filter_value is not None and (
+                    not isinstance(filter_value, str) or not filter_value.strip()
+                ):
+                    raise KrexInvalidParameterError(f"{name} must be a non-empty string")
+            page = await self.flow_all()
+            items = tuple(
+                item for item in page.items
+                if (route_no is None or item.route_no == route_no)
+                and (conzone_id is None or item.conzone_id == conzone_id)
+                and (direction_code is None or item.direction == direction_code)
+            )
+            start = (page_no - 1) * num_of_rows
+            return Page(
+                items=items[start:start + num_of_rows],
+                page_no=page_no,
+                num_of_rows=num_of_rows,
+                total_count=len(items),
+                raw=page.raw,
             )
         except KrexError as exc:
             self._client._http.protect_error(exc)
             raise exc from None
 
+    async def flow_all(self) -> Page[TrafficFlow]:
+        """0405 전체 응답 한 개를 검증해 반환한다. 필터·페이지 분할·집계는 하지 않는다.
+
+        각 VDS 행을 보존하며 공통 HTTP 재시도 정책은 그대로 적용한다.
+        공급자가 모든 행의 수집시각을 같게 보장한다는 뜻은 아니다.
+        """
+        try:
+            try:
+                payload = await self._client._http.get_ex(
+                    "/openapi/odtraffic/trafficAmountByRealtime"
+                )
+            except KrexNotFoundError:
+                if self._client.strict_no_data:
+                    raise
+                return Page(items=(), total_count=0)
+            page = _parse_traffic_flow_page(payload)
+            return Page(items=page.items, total_count=page.total_count, raw=page.raw)
+        except KrexError as exc:
+            self._client._http.protect_error(exc)
+            raise exc from None
+
     async def incident(
         self,
         *,
@@ -756,20 +796,67 @@ def _traffic_by_ic(row: dict[str, Any]) -> TrafficByIc:
     )
 
 
+def _parse_traffic_flow_page(payload: NormalizedPayload) -> Page[TrafficFlow]:
+    """0405는 전체 목록을 반환하므로 누락된 목록·건수·부분 파싱을 거부한다."""
+    raw = payload.raw
+    raw_items = raw.get("list")
+    count = raw.get("count")
+    if not isinstance(raw_items, (dict, list)) or raw_items == {}:
+        raise KrexParseError("traffic flow response must contain a list or record", response=raw)
+    if (
+        isinstance(count, bool)
+        or not isinstance(count, (int, str))
+        or not str(count).isascii()
+        or not str(count).isdigit()
+        or int(count) != len(payload.items)
+    ):
+        raise KrexParseError("traffic flow count must match the complete list", response=raw)
+    page = _parse_page(payload, _traffic_flow)
+    if len(page.items) != len(payload.items):
+        raise KrexParseError("traffic flow response contains invalid records", response=raw)
+    return page
+
+
 def _traffic_flow(row: dict[str, Any]) -> TrafficFlow:
+    conzone_id = strip_or_none(_required(row, "conzoneId", "conzoneID"))
+    direction: FlowDirection | Direction | None = _enum_or_none(
+        Direction, _get(row, "dirType", "directionCode")
+    )
+    if "updownTypeCode" in row:
+        # 0405의 S/E는 South/East가 아니라 기점/종점 방향이다.
+        direction = FlowDirection(str(row["updownTypeCode"]).strip())
+    congestion_level = _enum_or_none(
+        CongestionLevel, _get(row, "congestionLevel", "conzoneGrade")
+    )
+    if "grade" in row:
+        congestion_level = {
+            "0": None, "1": CongestionLevel.SMOOTH,
+            "2": CongestionLevel.SLOW, "3": CongestionLevel.STOP,
+        }[str(row["grade"]).strip()]
+    updated_at = strip_or_none(_get(row, "updTime", "updateTime", "updatedAt"))
+    if "stdDate" in row or "stdHour" in row:
+        day = str(_required(row, "stdDate")).strip()
+        hour = str(_required(row, "stdHour")).strip()
+        if (
+            len(day) != 8 or len(hour) != 4
+            or not (day + hour).isascii() or not (day + hour).isdigit()
+        ):
+            raise ValueError("traffic flow stdDate/stdHour must be YYYYMMDD/HHMM")
+        updated_at = datetime.strptime(day + hour, "%Y%m%d%H%M").strftime("%Y%m%d%H%M")
+    speed = to_float_or_none(_get(row, "speed", "avgSpeed"))
+    if speed is not None and speed < 0:
+        speed = None
     return TrafficFlow(
-        conzone_id=strip_or_none(_get(row, "conzoneId", "conzoneID")),
+        vds_id=strip_or_none(_get(row, "vdsId")),
+        conzone_id=conzone_id,
         conzone_name=strip_or_none(_get(row, "conzoneName")),
         route_no=strip_or_none(_get(row, "routeNo")),
         route_name=strip_or_none(_get(row, "routeName")),
-        direction=_enum_or_none(Direction, _get(row, "dirType", "directionCode")),
-        speed=to_float_or_none(_get(row, "speed", "avgSpeed")),
+        direction=direction,
+        speed=speed,
         free_flow_speed=to_float_or_none(_get(row, "tmFreeFlow", "freeFlowSpeed")),
-        congestion_level=_enum_or_none(
-            CongestionLevel,
-            _get(row, "congestionLevel", "conzoneGrade"),
-        ),
-        updated_at=strip_or_none(_get(row, "updTime", "updateTime", "updatedAt")),
+        congestion_level=congestion_level,
+        updated_at=updated_at,
         raw=row,
     )
 
diff --git a/src/krex/codes.py b/src/krex/codes.py
index 1a27849..2bb6236 100644
--- a/src/krex/codes.py
+++ b/src/krex/codes.py
@@ -132,6 +132,17 @@ def _label_map(cls) -> dict[str, str]:
         return _DIRECTION_LABELS
 
 
+class FlowDirection(KrexCode):
+    """0405 소통 데이터의 기점/종점 방향. 방위나 상하행을 추측하지 않는다."""
+
+    START = "S"
+    END = "E"
+
+    @classmethod
+    def _label_map(cls) -> dict[str, str]:
+        return {"S": "기점 방향", "E": "종점 방향"}
+
+
 class CongestionLevel(KrexCode):
     SMOOTH = "1"
     SLOW = "2"
diff --git a/src/krex/models.py b/src/krex/models.py
index 43bb544..ba81b65 100644
--- a/src/krex/models.py
+++ b/src/krex/models.py
@@ -14,6 +14,7 @@
     CoordinateSystem,
     Direction,
     DiscountType,
+    FlowDirection,
     IOType,
     TCSType,
 )
@@ -93,16 +94,23 @@ class TrafficByIc(KrexModel):
 
 
 class TrafficFlow(KrexModel):
+    """0405 VDS 관측값. 시각은 KST YYYYMMDDHHMM 문자열이며 음수 속도는 None이다.
+
+    방향은 FlowDirection.START(S)/END(E)로 제공한다. 같은 콘존의 여러
+    VDS 행을 합치지 않으며 VDS ID는 vds_id와 raw에, 원본 등급·방향은 raw에 보존한다.
+    """
+
     conzone_id: str | None
     conzone_name: str | None
     route_no: str | None
     route_name: str | None
-    direction: Direction | None
+    direction: FlowDirection | Direction | None
     speed: float | None
     free_flow_speed: float | None
     congestion_level: CongestionLevel | None
     updated_at: str | None
     raw: dict[str, Any]
+    vds_id: str | None = None
 
 
 class Incident(KrexModel):
diff --git a/tests/fixtures/traffic-flow/live-sample.json b/tests/fixtures/traffic-flow/live-sample.json
new file mode 100644
index 0000000..e3c52cf
--- /dev/null
+++ b/tests/fixtures/traffic-flow/live-sample.json
@@ -0,0 +1,17 @@
+{
+  "name": "live-sample",
+  "function": "traffic.flow",
+  "description": "2026-09-19 실제 응답에서 양방향 2행을 추출했다. 인증정보·요청 URL은 제외하고 count만 표본 크기로 변경했다.",
+  "response": {
+    "body": {
+      "code": "SUCCESS",
+      "count": 2,
+      "list": [
+        {"stdDate":"20260919","stdHour":"1828","vdsId":"0010VDE00100","conzoneId":"0010CZE010","conzoneName":"구서IC-영락IC","routeNo":"0010","routeName":"경부선","updownTypeCode":"E","speed":"90","trafficAmout":"13","shareRatio":"0","timeAvg":"72","grade":"1"},
+        {"stdDate":"20260919","stdHour":"1828","vdsId":"0010VDS00100","conzoneId":"0010CZS010","conzoneName":"영락IC-구서IC","routeNo":"0010","routeName":"경부선","updownTypeCode":"S","speed":"84","trafficAmout":"33","shareRatio":"3","timeAvg":"77","grade":"1"}
+      ]
+    }
+  },
+  "processed": {},
+  "assertion": {"mode":"required_fields", "required_fields":["items", "total_count", "raw"]}
+}
diff --git a/tests/runners.py b/tests/runners.py
index 3875e09..eca9d72 100644
--- a/tests/runners.py
+++ b/tests/runners.py
@@ -4,7 +4,7 @@
 from typing import Any
 
 from krex._http import _normalize_ex_payload
-from krex.client import _parse_page, _rest_area_weather
+from krex.client import _parse_page, _parse_traffic_flow_page, _rest_area_weather
 
 
 def _parse_restarea_weather(raw: dict[str, Any]) -> Any:
@@ -16,7 +16,13 @@ def _identity(value: Any) -> Any:
     return value
 
 
+def _parse_traffic_flow(raw: dict[str, Any]) -> Any:
+    return _parse_traffic_flow_page(_normalize_ex_payload(raw, params={}))
+
+
 RUNNERS: dict[str, dict[str, Callable[[Any], Any]]] = {
+    "traffic.flow": {"parse": _parse_traffic_flow, "process": _identity},
+    "traffic.flow_all": {"parse": _parse_traffic_flow, "process": _identity},
     "restarea.weather": {
         "parse": _parse_restarea_weather,
         "process": _identity,
diff --git a/tests/test_catalog.py b/tests/test_catalog.py
index cef5de0..add2b1a 100644
--- a/tests/test_catalog.py
+++ b/tests/test_catalog.py
@@ -19,7 +19,15 @@ def test_api_catalog_exposes_human_readable_names_and_key_links() -> None:
     assert isinstance(traffic, ApiCatalogItem)
     assert traffic.dataset_name == "한국도로공사_실시간 소통정보"
     assert traffic.service_key_url == "https://data.ex.co.kr/openapi/apikey/requestKey"
-    assert traffic.endpoint == "/openapi/trafficapi/realFlow"
+    assert traffic.endpoint == "/openapi/odtraffic/trafficAmountByRealtime"
+    assert traffic.live_verified is True
+    assert traffic.fixture_supported is True
+    all_traffic = get_api_catalog_item("traffic.flow_all")
+    assert all_traffic is not None
+    assert all_traffic.endpoint == traffic.endpoint
+    assert all_traffic.return_type == "Page[TrafficFlow]"
+    assert all_traffic.fixture_supported is True
+    assert all_traffic.live_verified is False
     assert all(item.dataset_name and item.dataset_name != item.function for item in items)
     assert all(item.service_key_url for item in items if item.provider != "local")
 
diff --git a/tests/test_client.py b/tests/test_client.py
index e67ff8d..a0c16cb 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -94,14 +94,16 @@ def test_explicit_client_keys_are_normalized_before_env_fallback(tmp_path: Path)
 
 
 async def test_aio_client_matches_sync_service_shape() -> None:
-    session = FakeSession(ex_payload([{"conzoneId": "0010CZE010", "speed": "87.5"}]))
+    session = FakeSession(
+        ex_payload([{"conzoneId": "0010CZE010", "routeNo": "0010", "speed": "87.5"}])
+    )
     client = KrexClient(ex_api_key="ex-key", retry_backoff=0, session=session)
 
     page = await client.traffic.flow(route_no="0010")
 
     assert isinstance(client, KrexClient)
     assert page.items[0].conzone_id == "0010CZE010"
-    assert session.last_url.endswith("/openapi/trafficapi/realFlow")
+    assert session.last_url.endswith("/openapi/odtraffic/trafficAmountByRealtime")
 
 
 def rest_weather_row(**overrides: Any) -> dict[str, Any]:
@@ -195,11 +197,11 @@ async def test_traffic_flow_accepts_single_dict_response() -> None:
     )
     client = KrexClient(ex_api_key="ex-key", retry_backoff=0, session=session)
 
-    page = await client.traffic.flow(route_no="0010", direction=Direction.UP)
+    page = await client.traffic.flow(route_no="0010")
 
-    assert session.last_params["routeNo"] == "0010"
-    assert session.last_params["dirType"] == "0"
+    assert session.last_params == {"key": "ex-key", "type": "json"}
     flow = page.items[0]
+    assert flow.direction is Direction.UP
     assert flow.congestion_level is CongestionLevel.SMOOTH
     assert flow.speed == pytest.approx(87.5)
     assert flow.free_flow_speed == pytest.approx(100.0)
diff --git a/tests/test_codes.py b/tests/test_codes.py
index f94bda2..f97f163 100644
--- a/tests/test_codes.py
+++ b/tests/test_codes.py
@@ -1,6 +1,6 @@
 import pytest
 
-from krex import CarType, CongestionLevel, CoordinateSystem, Direction, TCSType
+from krex import CarType, CongestionLevel, CoordinateSystem, Direction, FlowDirection, TCSType
 from krex.codes import coerce_code
 from krex.exceptions import KrexInvalidParameterError
 
@@ -10,6 +10,8 @@ def test_enum_labels_are_stable() -> None:
     assert TCSType.HIPASS.label == "하이패스"
     assert Direction.UP.label == "상행"
     assert CongestionLevel.STOP.label == "정체"
+    assert FlowDirection.START.label == "기점 방향"
+    assert FlowDirection.END.label == "종점 방향"
 
 
 def test_car_type_from_label_accepts_label_or_code() -> None:
diff --git a/tests/test_debug.py b/tests/test_debug.py
index 3358c3b..aba9788 100644
--- a/tests/test_debug.py
+++ b/tests/test_debug.py
@@ -56,7 +56,7 @@ async def test_debug_call_returns_request_response_and_typed_result() -> None:
     assert isinstance(run, DebugRun)
     assert run.error is None
     assert run.request["method"] == "GET"
-    assert run.request["url"].endswith("/openapi/trafficapi/realFlow")
+    assert run.request["url"].endswith("/openapi/odtraffic/trafficAmountByRealtime")
     assert run.request["query"]["key"] == "<REDACTED>"
     assert run.response["status_code"] == 200
     assert jsonable(run.processed)["items"][0]["speed"] == 87.5
diff --git a/tests/test_live_ex.py b/tests/test_live_ex.py
index 6b756f6..e62e975 100644
--- a/tests/test_live_ex.py
+++ b/tests/test_live_ex.py
@@ -5,7 +5,7 @@
 
 import pytest
 
-from krex import CarType, KrexClient, RoadOperator, TCSType, TimeUnit
+from krex import CarType, FlowDirection, KrexClient, RoadOperator, TCSType, TimeUnit
 
 
 def _load_local_env() -> None:
@@ -19,13 +19,36 @@ def _load_local_env() -> None:
         os.environ.setdefault(key.strip(), value.strip())
 
 
-def _live_client() -> KrexClient:
+def _live_client(*, max_retries: int = 1) -> KrexClient:
     _load_local_env()
     if os.getenv("KEX_LIVE") != "1":
         pytest.skip("set KEX_LIVE=1 to run live data.ex.co.kr tests")
     if not os.getenv("KEX_EX_API_KEY"):
         pytest.skip("KEX_EX_API_KEY is not set")
-    return KrexClient.from_env(timeout=10, max_retries=1, retry_backoff=0)
+    return KrexClient.from_env(timeout=10, max_retries=max_retries, retry_backoff=0)
+
+
+@pytest.mark.live
+async def test_live_traffic_flow_parses_complete_snapshot() -> None:
+    async with _live_client(max_retries=0) as client:
+        try:
+            page = await client.traffic.flow(num_of_rows=20000)
+        except Exception as exc:
+            # 공급자/네트워크 예외의 원문이나 인증 URL을 테스트 로그에 출력하지 않는다.
+            raise AssertionError(f"live traffic flow failed: {type(exc).__name__}") from None
+    assert page.items
+    assert page.total_count == len(page.items)
+    assert page.raw is not None and page.raw["count"] == page.total_count
+    assert all(item.direction in {FlowDirection.START, FlowDirection.END} for item in page)
+    assert all(item.updated_at and len(item.updated_at) == 12 for item in page)
+    assert all(item.speed is None or item.speed >= 0 for item in page)
+    assert all(item.free_flow_speed is None for item in page)
+    print(
+        f"parsed={len(page.items)}, routes={len({item.route_no for item in page})}, "
+        f"missing_speed={sum(item.speed is None for item in page)}, "
+        f"start={sum(item.direction is FlowDirection.START for item in page)}, "
+        f"end={sum(item.direction is FlowDirection.END for item in page)}"
+    )
 
 
 @pytest.mark.live
diff --git a/tests/test_traffic_flow.py b/tests/test_traffic_flow.py
new file mode 100644
index 0000000..15876ee
--- /dev/null
+++ b/tests/test_traffic_flow.py
@@ -0,0 +1,245 @@
+from __future__ import annotations
+
+import copy
+import json
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from krex import (
+    CongestionLevel,
+    Direction,
+    FlowDirection,
+    KrexAuthError,
+    KrexClient,
+    KrexInvalidParameterError,
+    KrexNotFoundError,
+    KrexParseError,
+    TrafficFlow,
+)
+from tests.test_client import FakeSession
+
+
+def sample() -> dict[str, Any]:
+    path = Path(__file__).parent / "fixtures/traffic-flow/live-sample.json"
+    return json.loads(path.read_text(encoding="utf-8"))["response"]["body"]  # type: ignore[no-any-return]
+
+
+async def test_live_fixture_semantics_and_request() -> None:
+    raw = sample()
+    session = FakeSession(raw)
+    async with KrexClient(ex_api_key="test-key", session=session) as client:
+        page = await client.traffic.flow()
+    assert len(session.calls) == 1
+    assert session.last_url.endswith("/openapi/odtraffic/trafficAmountByRealtime")
+    assert session.last_params == {"key": "test-key", "type": "json"}
+    assert page.total_count == 2
+    assert page.raw == raw
+    assert [item.direction for item in page] == [FlowDirection.END, FlowDirection.START]
+    assert page.items[0].direction is FlowDirection.END
+    assert page.items[1].direction is FlowDirection.START
+    assert [item.speed for item in page] == [90, 84]
+    assert [item.vds_id for item in page] == ["0010VDE00100", "0010VDS00100"]
+    assert all(item.updated_at == "202609191828" for item in page)
+    assert all(item.route_no == "0010" for item in page)
+    assert all(item.free_flow_speed is None for item in page)
+    assert all(item.congestion_level is CongestionLevel.SMOOTH for item in page)
+
+
+@pytest.mark.parametrize("grade, expected", [
+    ("0", None), ("1", CongestionLevel.SMOOTH),
+    ("2", CongestionLevel.SLOW), ("3", CongestionLevel.STOP),
+])
+async def test_official_grades_and_missing_speed(grade: str, expected: Any) -> None:
+    raw = sample()
+    raw["list"][0].update(grade=grade, speed="-1")
+    async with KrexClient(ex_api_key="test", session=FakeSession(raw)) as client:
+        item = (await client.traffic.flow()).items[0]
+    assert item.congestion_level is expected
+    assert item.speed is None
+    assert item.raw["speed"] == "-1"
+
+
+@pytest.mark.parametrize("params, expected", [
+    ({"route_no": "0010"}, 2),
+    ({"route_no": "1"}, 0),
+    ({"route_no": "9999"}, 0),
+    ({"direction": FlowDirection.START}, 1),
+    ({"direction": "E"}, 1),
+    ({"conzone_id": "0010CZE010"}, 1),
+    ({"conzone_id": "0010CZE010", "direction": FlowDirection.START}, 0),
+])
+async def test_local_filters(params: dict[str, Any], expected: int) -> None:
+    session = FakeSession(sample())
+    async with KrexClient(ex_api_key="test", session=session) as client:
+        page = await client.traffic.flow(**params)
+    assert len(page) == page.total_count == expected
+    assert set(session.last_params) == {"key", "type"}
+
+
+async def test_local_paging_preserves_vds_rows_and_raw_count() -> None:
+    raw = sample()
+    second_vds = copy.deepcopy(raw["list"][0])
+    second_vds["vdsId"] = "another-vds"
+    raw["list"].append(second_vds)
+    raw["count"] = 3
+    async with KrexClient(ex_api_key="test", session=FakeSession(raw)) as client:
+        page = await client.traffic.flow(direction="E", num_of_rows=1, page_no=2)
+        beyond = await client.traffic.flow(direction="E", num_of_rows=1, page_no=3)
+    assert len(page) == 1
+    assert page.first is not None and page.first.raw["vdsId"] == "another-vds"
+    assert page.first.vds_id == "another-vds"
+    assert page.page_no == 2 and page.num_of_rows == 1 and page.total_count == 2
+    assert page.raw is not None and page.raw["count"] == 3
+    assert beyond.is_empty and beyond.total_count == 2
+
+
+@pytest.mark.parametrize("raw", [
+    {}, {"message": "upstream failure"}, {"code": "SUCCESS"},
+    {"count": 0}, {"list": []}, {"count": 1, "list": []},
+    {"count": 0, "list": None}, {"count": 0, "list": ""},
+    {"count": 0, "list": {}}, {"count": 1, "list": [None]},
+    {"count": 1, "list": [{}]}, {"count": -1, "list": []},
+    {"count": True, "list": []}, {"count": 0.0, "list": []},
+    {"count": "0 rows", "list": []}, {"count": "invalid", "list": []},
+    {"count": 0, "data": []},
+])
+@pytest.mark.parametrize("strict", [True, False])
+@pytest.mark.parametrize("method", ["flow", "flow_all"])
+async def test_malformed_envelopes_never_become_empty_success(
+    raw: Any, strict: bool, method: str
+) -> None:
+    async with KrexClient(
+        ex_api_key="test", strict_no_data=strict, session=FakeSession(raw)
+    ) as client:
+        with pytest.raises(KrexParseError):
+            await getattr(client.traffic, method)()
+
+
+async def test_explicit_empty_snapshot_and_single_record() -> None:
+    raw = sample()
+    raw["list"] = raw["list"][0]
+    raw["count"] = "1"
+    async with KrexClient(ex_api_key="test", session=FakeSession(raw)) as client:
+        assert len(await client.traffic.flow()) == 1
+    async with KrexClient(
+        ex_api_key="test", session=FakeSession({"code": "SUCCESS", "count": 0, "list": []})
+    ) as client:
+        page = await client.traffic.flow()
+    assert page.is_empty and page.total_count == 0
+
+
+@pytest.mark.parametrize("changes", [
+    {"conzoneId": " "}, {"speed": "broken"}, {"updownTypeCode": "X"},
+    {"grade": "4"}, {"stdHour": "2500"}, {"stdHour": "12"}, {"stdHour": None},
+    {"stdDate": "20260230"}, {"stdDate": None},
+])
+async def test_one_bad_record_rejects_whole_snapshot_before_filtering(changes: Any) -> None:
+    raw = sample()
+    raw["list"][0].update(changes)
+    async with KrexClient(ex_api_key="test", session=FakeSession(raw)) as client:
+        with pytest.raises(KrexParseError):
+            await client.traffic.flow(route_no="9999")
+
+
+@pytest.mark.parametrize("params", [
+    {"direction": "0"}, {"direction": "1"}, {"direction": "invalid"},
+    {"direction": Direction.EAST}, {"direction": Direction.UP},
+    {"num_of_rows": 0}, {"page_no": -1}, {"page_no": True},
+    {"num_of_rows": 1.5}, {"page_no": "1"},
+    {"route_no": ""}, {"route_no": 10}, {"conzone_id": " "},
+])
+async def test_invalid_filters_fail_before_network(params: Any) -> None:
+    session = FakeSession(sample())
+    async with KrexClient(ex_api_key="test", session=session) as client:
+        with pytest.raises(KrexInvalidParameterError):
+            await client.traffic.flow(**params)
+    assert not session.calls
+
+
+async def test_provider_error_is_not_hidden_by_snapshot_validation() -> None:
+    async with KrexClient(
+        ex_api_key="test", session=FakeSession({"code": "INVALID_KEY", "message": "invalid"})
+    ) as client:
+        with pytest.raises(KrexAuthError):
+            await client.traffic.flow()
+
+
+async def test_zero_speed_is_not_missing_and_grade_three_means_stop() -> None:
+    raw = sample()
+    raw["list"][0].update(speed="0", grade="3")
+    async with KrexClient(ex_api_key="test", session=FakeSession(raw)) as client:
+        item = (await client.traffic.flow()).items[0]
+    assert item.speed == 0
+    assert item.congestion_level is CongestionLevel.STOP
+    assert item.raw["grade"] == "3"
+
+
+async def test_flow_all_returns_over_one_thousand_vds_rows_from_one_response() -> None:
+    raw = sample()
+    row = raw["list"][0]
+    raw["list"] = [dict(row, vdsId=f"0010VDE{index:05}") for index in range(1002)]
+    raw["count"] = 1002
+    session = FakeSession(raw)
+    async with KrexClient(ex_api_key="test", session=session) as client:
+        page = await client.traffic.flow_all()
+    assert len(session.calls) == 1
+    assert session.last_params == {"key": "test", "type": "json"}
+    assert len(page) == page.total_count == 1002
+    assert page.page_no is None and page.num_of_rows is None
+    assert len({item.conzone_id for item in page}) == 1
+    assert len({item.vds_id for item in page}) == 1002
+    assert [item.vds_id for item in page] == [item["vdsId"] for item in raw["list"]]
+    assert page.raw == raw
+
+
+async def test_flow_keeps_default_local_limit_with_one_request() -> None:
+    raw = sample()
+    raw["list"] = [dict(raw["list"][0], vdsId=str(index)) for index in range(1002)]
+    raw["count"] = 1002
+    session = FakeSession(raw)
+    async with KrexClient(ex_api_key="test", session=session) as client:
+        page = await client.traffic.flow()
+    assert len(session.calls) == 1
+    assert len(page) == page.num_of_rows == 1000
+    assert page.total_count == 1002 and page.page_no == 1
+
+
+async def test_flow_all_debug_and_real_fixture_vds_fields() -> None:
+    raw = sample()
+    async with KrexClient(ex_api_key="test", session=FakeSession(raw)) as client:
+        run = await client.debug_call("traffic.flow_all")
+    assert run.error is None
+    assert run.request["query"]["key"] == "<REDACTED>"
+    assert [item.vds_id for item in run.parsed] == ["0010VDE00100", "0010VDS00100"]
+    assert all(item.vds_id == item.raw["vdsId"] for item in run.parsed)
+
+
+async def test_flow_all_preserves_legacy_missing_vds_id_and_model_default() -> None:
+    raw = sample()
+    for item in raw["list"]:
+        del item["vdsId"]
+    async with KrexClient(ex_api_key="test", session=FakeSession(raw)) as client:
+        page = await client.traffic.flow_all()
+    assert all(item.vds_id is None for item in page)
+    fields = page.items[0].model_dump(exclude={"vds_id"})
+    assert TrafficFlow(**fields).vds_id is None
+
+
+@pytest.mark.parametrize("strict", [True, False])
+async def test_flow_all_explicit_empty_and_no_data(strict: bool) -> None:
+    session = FakeSession([
+        {"code": "SUCCESS", "count": 0, "list": []},
+        {"code": "NO_DATA", "message": "empty"},
+    ])
+    async with KrexClient(ex_api_key="test", strict_no_data=strict, session=session) as client:
+        page = await client.traffic.flow_all()
+        assert page.is_empty and page.total_count == 0
+        assert page.page_no is None and page.num_of_rows is None
+        if strict:
+            with pytest.raises(KrexNotFoundError):
+                await client.traffic.flow_all()
+        else:
+            page = await client.traffic.flow_all()
+            assert page.is_empty and page.total_count == 0 and page.raw is None