docs(realestate): naver 워커 BE 구현 plan

realestate-lab/docs/superpowers/plans/2026-07-09-naver-listing-worker-be.md

@@ -0,0 +1,467 @@
+# 네이버 매물 fetch 워커 — BE 구현 Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** naver-fetch 워커가 네이버 호가 매물을 push할 수 있도록 realestate-lab에 내부 계약(targets 조회 + listings ingest), 관측 등재, cortarNo 매핑을 구축한다. 워커(web-ai)·FE(/infra)는 별도 세션.
+
+**Architecture:** 워커는 `GET /api/internal/realestate/targets`로 대상 동+10자리 cortarNo를 받고, 네이버 raw 매물을 `POST /api/internal/realestate/listings-ingest`로 push. NAS는 기존 `_parse_naver_article`로 파싱→`upsert_listing`→`run_listing_matching`→`notify_new_listings`. 인증은 render 워커와 동형(X-Internal-Key + nginx IP화이트리스트). 관측은 heartbeat + node_monitor `fetcher` kind.
+
+**Tech Stack:** Python 3.12, FastAPI(APIRouter+Depends), SQLite, pytest. 외부 호출 테스트는 전부 mock.
+
+**스펙:** `realestate-lab/docs/superpowers/specs/2026-07-09-naver-listing-worker-design.md`
+
+## Global Constraints
+
+- **기존 매물/청약 파이프라인 불변** — 신규 내부 라우터·매핑·등재만 추가. 안전마진 판정 로직 불변.
+- **인증 패턴 준수**: `verify_internal_key(x_internal_key: str = Header(...))`, env **`INTERNAL_API_KEY`**, 실패 시 `HTTPException(401, "Invalid X-Internal-Key")`. (스펙의 `REALESTATE_INTERNAL_KEY` 대신 기존 서비스 관례 `INTERNAL_API_KEY` 사용 — image/video/insta-lab 일관성.)
+- **파싱은 한 곳**: 네이버 article 파싱은 기존 `listing_collector._parse_naver_article` 재사용(중복 구현 금지).
+- **멱등**: `listings.article_no` UNIQUE upsert + `listing_matches.notified_at` → 재push해도 재알림 없음.
+- **테스트**: `cd realestate-lab && PYTHONPATH="C:/Users/jaeoh/Desktop/workspace/web-backend" python -m pytest tests/<file> -q` (main import은 `_shared` 때문에 PYTHONPATH에 web-backend 루트 필요). agent-office 테스트는 `cd agent-office && python -m pytest`.
+- **cortarNo는 10자리 법정동코드** (기존 `LAWD_CODES` 5자리 시군구는 MOLIT 전용 유지).
+- **nginx/deploy는 locked 리소스**: nginx 변경·push는 `nginx-conf`·`nas-deploy` 락 획득 후(Task 6).
+
+## File Structure
+
+| 파일 | 신규/수정 | 책임 |
+|---|---|---|
+| `realestate-lab/app/lawd_codes.py` | Modify | `NAVER_CORTAR`(동→10자리) + `naver_cortar()` 추가 |
+| `realestate-lab/app/auth.py` | Create | `verify_internal_key` (X-Internal-Key) |
+| `realestate-lab/app/internal_router.py` | Create | `GET targets` + `POST listings-ingest` |
+| `realestate-lab/app/main.py` | Modify | `include_router(internal_router)` |
+| `nginx/default.conf` | Modify | `/api/internal/realestate/` 블록 |
+| `realestate-lab/app/listing_collector.py` | Modify | `collect_listings` MOLIT 전용화(naver deprecate) |
+| `agent-office/app/node_monitor.py` | Modify | `WORKER_REGISTRY` fetcher 등재 + link 분기 |
+
+---
+
+### Task 1: NAVER_CORTAR 매핑 (10자리 법정동코드)
+
+**Files:**
+- Modify: `realestate-lab/app/lawd_codes.py`
+- Test: `realestate-lab/tests/test_lawd_codes.py` (기존 파일에 append)
+
+**Interfaces:**
+- Produces: `NAVER_CORTAR: dict[str,str]`, `naver_cortar(dong: str) -> str | None`
+
+- [ ] **Step 1: 실패 테스트 append** — `tests/test_lawd_codes.py`
+
+```python
+def test_naver_cortar_10digit():
+    from app.lawd_codes import naver_cortar, NAVER_CORTAR
+    assert naver_cortar("신대방동") == "1159010100"
+    assert naver_cortar("봉천동") == "1162010100"
+    assert naver_cortar("신길동") == "1156013200"
+    assert naver_cortar("없는동") is None
+    # 6개 초기 대상 동 전부 10자리
+    for d in ["신대방동","대방동","상도동","봉천동","대림동","신길동"]:
+        assert len(NAVER_CORTAR[d]) == 10
+```
+
+- [ ] **Step 2: 실패 확인** — `cd realestate-lab && PYTHONPATH="C:/Users/jaeoh/Desktop/workspace/web-backend" python -m pytest tests/test_lawd_codes.py -q` → FAIL(ImportError naver_cortar)
+
+- [ ] **Step 3: 구현** — `app/lawd_codes.py` 하단에 추가
+
+```python
+# 네이버부동산 cortarNo = 10자리 법정동코드 (LAWD_CODES 5자리 시군구와 별개).
+# 출처: 행정표준코드관리시스템 법정동코드. 대상 동 추가 시 여기에.
+NAVER_CORTAR = {
+    "신대방동": "1159010100",
+    "대방동":   "1159010200",
+    "상도동":   "1159010600",
+    "봉천동":   "1162010100",
+    "대림동":   "1156013600",
+    "신길동":   "1156013200",
+}
+
+
+def naver_cortar(dong: str) -> str | None:
+    return NAVER_CORTAR.get((dong or "").strip())
+```
+
+- [ ] **Step 4: 통과 확인** — 같은 pytest → PASS
+- [ ] **Step 5: 커밋**
+
+```bash
+git add realestate-lab/app/lawd_codes.py realestate-lab/tests/test_lawd_codes.py
+git commit -m "feat(realestate): 네이버 cortarNo 10자리 법정동 매핑(NAVER_CORTAR)"
+```
+
+---
+
+### Task 2: 내부 인증 + targets 엔드포인트 + 라우터 배선
+
+**Files:**
+- Create: `realestate-lab/app/auth.py`, `realestate-lab/app/internal_router.py`
+- Modify: `realestate-lab/app/main.py`
+- Test: `realestate-lab/tests/test_internal_targets.py`
+
+**Interfaces:**
+- Consumes: Task1 `naver_cortar`; 기존 `db.get_listing_criteria`.
+- Produces: `verify_internal_key` dependency; `GET /api/internal/realestate/targets` → `{"dongs":[{"dong","cortar_no"}],"deal_types":[...],"page_limit":int}`; `router` (APIRouter, main에 include).
+
+- [ ] **Step 1: 실패 테스트** — `tests/test_internal_targets.py`
+
+```python
+import os
+from fastapi.testclient import TestClient
+
+def _client(monkeypatch):
+    monkeypatch.setenv("INTERNAL_API_KEY", "SECRET")
+    import importlib
+    from app import auth as _a
+    importlib.reload(_a)  # env 반영
+    from app.main import app
+    return TestClient(app)
+
+def test_targets_requires_key(monkeypatch):
+    c = _client(monkeypatch)
+    assert c.get("/api/internal/realestate/targets").status_code == 401           # 헤더 없음
+    assert c.get("/api/internal/realestate/targets",
+                 headers={"X-Internal-Key": "WRONG"}).status_code == 401
+
+def test_targets_returns_dongs_with_cortar(monkeypatch):
+    c = _client(monkeypatch)
+    r = c.get("/api/internal/realestate/targets", headers={"X-Internal-Key": "SECRET"})
+    assert r.status_code == 200
+    body = r.json()
+    dongs = {d["dong"]: d["cortar_no"] for d in body["dongs"]}
+    assert dongs.get("신대방동") == "1159010100"     # criteria seed 6동 + cortar join
+    assert "전세" in body["deal_types"]
+    assert isinstance(body["page_limit"], int)
+```
+
+- [ ] **Step 2: 실패 확인** — `PYTHONPATH=... python -m pytest tests/test_internal_targets.py -q` → FAIL(404/ImportError)
+
+- [ ] **Step 3: 구현**
+
+`app/auth.py`:
+```python
+"""Windows naver-fetch 워커 → NAS realestate-lab 내부 엔드포인트 인증."""
+import os
+from fastapi import Header, HTTPException
+
+
+def verify_internal_key(x_internal_key: str = Header(...)):
+    expected = os.getenv("INTERNAL_API_KEY")
+    if not expected:
+        raise HTTPException(401, "INTERNAL_API_KEY not configured on server")
+    if x_internal_key != expected:
+        raise HTTPException(401, "Invalid X-Internal-Key")
+```
+
+`app/internal_router.py`:
+```python
+"""naver-fetch 워커 내부 계약: targets 조회 + listings ingest."""
+import os
+import logging
+from fastapi import APIRouter, Depends
+
+from .auth import verify_internal_key
+from .db import get_listing_criteria
+from .lawd_codes import naver_cortar
+
+logger = logging.getLogger("realestate-lab")
+router = APIRouter()
+
+NAVER_PAGE_LIMIT = int(os.getenv("NAVER_PAGE_LIMIT", "2"))
+
+
+@router.get("/api/internal/realestate/targets", dependencies=[Depends(verify_internal_key)])
+def listing_targets():
+    crit = get_listing_criteria()
+    dongs = []
+    for d in crit.get("dongs", []):
+        code = naver_cortar(d)
+        if not code:
+            logger.warning("naver cortarNo 매핑 없음 — 대상 제외: %s", d)
+            continue
+        dongs.append({"dong": d, "cortar_no": code})
+    return {"dongs": dongs, "deal_types": crit.get("deal_types", []),
+            "page_limit": NAVER_PAGE_LIMIT}
+```
+
+`app/main.py`: import 블록에 추가(기존 `from .models import (...)` 아래)
+```python
+from .internal_router import router as internal_router
+```
+그리고 `app = FastAPI(lifespan=lifespan)` + `install_access_log(app)` **직후**에:
+```python
+app.include_router(internal_router)
+```
+
+`nginx/default.conf`: `/api/realestate/` location 블록 **앞**에 추가(더 구체적 prefix 우선):
+```nginx
+location /api/internal/realestate/ {
+    allow 192.168.45.0/24;
+    allow 100.64.0.0/10;
+    allow 127.0.0.1;
+    deny all;
+
+    proxy_http_version 1.1;
+    proxy_set_header Host $host;
+    proxy_set_header X-Real-IP $remote_addr;
+    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+    proxy_set_header X-Internal-Key $http_x_internal_key;
+    proxy_pass http://realestate-lab:8000/api/internal/realestate/;
+}
+```
+
+- [ ] **Step 4: 통과 확인** — 같은 pytest → PASS (nginx는 로컬 테스트 무관, 배포 시 검증)
+- [ ] **Step 5: 커밋**
+
+```bash
+git add realestate-lab/app/auth.py realestate-lab/app/internal_router.py realestate-lab/app/main.py nginx/default.conf realestate-lab/tests/test_internal_targets.py
+git commit -m "feat(realestate): 내부 인증 + naver 워커 targets 엔드포인트 + nginx internal 블록"
+```
+
+---
+
+### Task 3: listings-ingest 엔드포인트
+
+**Files:**
+- Modify: `realestate-lab/app/internal_router.py`
+- Test: `realestate-lab/tests/test_internal_ingest.py`
+
+**Interfaces:**
+- Consumes: 기존 `db.upsert_listing`, `listing_collector._parse_naver_article`, `listing_matcher.run_listing_matching`, `notifier.notify_new_listings`.
+- Produces: `POST /api/internal/realestate/listings-ingest` body `{fetched_at:str, batches:[{dong:str, articles:[dict]}]}` → `{"received":int,"new":int,"matched":int}`.
+
+- [ ] **Step 1: 실패 테스트** — `tests/test_internal_ingest.py`
+
+```python
+import os, importlib
+from unittest.mock import patch
+from fastapi.testclient import TestClient
+
+def _client(monkeypatch):
+    monkeypatch.setenv("INTERNAL_API_KEY", "SECRET")
+    from app import auth as _a
+    importlib.reload(_a)
+    from app.main import app
+    return TestClient(app)
+
+_ARTICLE = {"articleNo": "N100", "tradeTypeName": "전세", "dealOrWarrantPrc": "2억 9,000",
+            "area2": "42.5", "floorInfo": "5/15", "articleName": "OO아파트",
+            "buildingName": "OO", "cortarNo": "1159010100"}
+
+def test_ingest_auth(monkeypatch):
+    c = _client(monkeypatch)
+    assert c.post("/api/internal/realestate/listings-ingest", json={"batches": []}).status_code == 401
+
+def test_ingest_upserts_and_matches(monkeypatch):
+    c = _client(monkeypatch)
+    with patch("app.internal_router.notify_new_listings", return_value={"sent": 1, "sent_ids": [1]}) as noti:
+        r = c.post("/api/internal/realestate/listings-ingest",
+                   headers={"X-Internal-Key": "SECRET"},
+                   json={"fetched_at": "2026-07-09T00:00:00Z",
+                         "batches": [{"dong": "신대방동", "articles": [_ARTICLE]}]})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["received"] == 1 and body["new"] == 1 and body["matched"] == 1
+    assert noti.call_count == 1
+    # 저장 확인
+    from app import db
+    ls = db.get_listings()
+    assert any(l["article_no"] == "N100" and l["dong"] == "신대방동" for l in ls)
+
+def test_ingest_idempotent(monkeypatch):
+    c = _client(monkeypatch)
+    payload = {"batches": [{"dong": "신대방동", "articles": [_ARTICLE]}]}
+    with patch("app.internal_router.notify_new_listings", return_value={"sent": 0}):
+        c.post("/api/internal/realestate/listings-ingest", headers={"X-Internal-Key": "SECRET"}, json=payload)
+        r2 = c.post("/api/internal/realestate/listings-ingest", headers={"X-Internal-Key": "SECRET"}, json=payload)
+    assert r2.json()["new"] == 0   # 재push → 신규 0
+```
+
+- [ ] **Step 2: 실패 확인** — `PYTHONPATH=... python -m pytest tests/test_internal_ingest.py -q` → FAIL(404)
+
+- [ ] **Step 3: 구현** — `app/internal_router.py`에 추가
+
+상단 import 확장:
+```python
+from pydantic import BaseModel
+from typing import List, Dict, Any
+from .db import get_listing_criteria, upsert_listing
+from .listing_collector import _parse_naver_article
+from .listing_matcher import run_listing_matching
+from .notifier import notify_new_listings
+```
+(기존 `from .db import get_listing_criteria`는 위 확장 라인으로 합치거나 중복 없이 정리)
+
+모델 + 핸들러:
+```python
+class _Batch(BaseModel):
+    dong: str
+    articles: List[Dict[str, Any]] = []
+
+
+class ListingsIngest(BaseModel):
+    fetched_at: str | None = None
+    batches: List[_Batch] = []
+
+
+@router.post("/api/internal/realestate/listings-ingest",
+             dependencies=[Depends(verify_internal_key)])
+def listings_ingest(body: ListingsIngest):
+    received = new = 0
+    for batch in body.batches:
+        for raw in batch.articles:
+            try:
+                d = _parse_naver_article(raw)
+                if not d.get("article_no"):
+                    continue
+                d["dong"] = batch.dong
+                _, is_new = upsert_listing(d)
+                received += 1
+                if is_new:
+                    new += 1
+            except Exception as e:
+                logger.warning("네이버 ingest 파싱 실패: %s", e)
+    matched = 0
+    if received:
+        run_listing_matching()
+        noti = notify_new_listings()
+        matched = int(noti.get("sent", 0)) if isinstance(noti, dict) else 0
+    return {"received": received, "new": new, "matched": matched}
+```
+
+- [ ] **Step 4: 통과 확인** — 같은 pytest → PASS
+- [ ] **Step 5: 커밋**
+
+```bash
+git add realestate-lab/app/internal_router.py realestate-lab/tests/test_internal_ingest.py
+git commit -m "feat(realestate): naver 워커 listings-ingest(파싱→upsert→매칭→알림)"
+```
+
+---
+
+### Task 4: node_monitor fetcher 등재
+
+**Files:**
+- Modify: `agent-office/app/node_monitor.py`
+- Test: `agent-office/tests/test_node_monitor.py` (기존 파일에 append; 없으면 `test_node_monitor_fetcher.py` 생성)
+
+**Interfaces:**
+- Produces: `WORKER_REGISTRY`에 `{"name":"naver-fetch","kind":"fetcher","queue":None}`; `collect_status()`가 fetcher에 대해 `{from:"naver-fetch", to:"nas-realestate", type:"http-pull", status:...}` link 생성.
+
+- [ ] **Step 1: 실패 테스트** — 등재 여부(트리비얼) + 링크 분기
+
+```python
+def test_naver_fetch_registered():
+    from app.node_monitor import WORKER_REGISTRY
+    entry = next((w for w in WORKER_REGISTRY if w["name"] == "naver-fetch"), None)
+    assert entry is not None
+    assert entry["kind"] == "fetcher" and entry["queue"] is None
+```
+> collect_status의 fetcher-link는 기존 `test_node_monitor.py`의 redis-mock 픽스처(trader/render 링크 검증부)를 그대로 따라, "naver-fetch heartbeat 존재 시 links에 from=naver-fetch·type=http-pull이 있다"를 추가 검증하라. 픽스처가 없으면 이 트리비얼 등재 테스트만으로 충분.
+
+- [ ] **Step 2: 실패 확인** — `cd agent-office && python -m pytest tests/test_node_monitor.py -q` (또는 신규 파일) → FAIL
+
+- [ ] **Step 3: 구현** — `agent-office/app/node_monitor.py`
+
+`WORKER_REGISTRY` 리스트에 추가:
+```python
+    {"name": "naver-fetch",  "kind": "fetcher", "queue": None},
+```
+
+`collect_status()`의 kind 분기(`if w["kind"] == "trader": ... elif w["kind"] == "render": ...`)에 추가:
+```python
+    elif w["kind"] == "fetcher":
+        out["links"].append({"from": w["name"], "to": "nas-realestate", "type": "http-pull",
+                             "status": "healthy" if w["alive"] else "down"})
+```
+
+- [ ] **Step 4: 통과 확인** — 같은 pytest → PASS + 회귀 `cd agent-office && python -m pytest -q`
+- [ ] **Step 5: 커밋**
+
+```bash
+git add agent-office/app/node_monitor.py agent-office/tests/test_node_monitor.py
+git commit -m "feat(agent-office): node_monitor에 naver-fetch(fetcher) 등재 + http-pull 링크"
+```
+
+---
+
+### Task 5: collect_listings MOLIT 전용화 (naver deprecate)
+
+**Files:**
+- Modify: `realestate-lab/app/listing_collector.py`
+- Test: `realestate-lab/tests/test_listing_collector.py` (기존 수정)
+
+**Interfaces:**
+- `collect_listings()` → `{"new_count":0,"total_count":0,"naver_ok":None,"diag":str}` (naver는 워커가 담당하므로 NAS는 MOLIT만). `_fetch_naver_all`/`_parse_naver_article`/`_won_to_manwon`는 **존치**(ingest가 `_parse_naver_article` 사용).
+
+- [ ] **Step 1: 실패 테스트** — 기존 `test_collect_naver_block_sets_naver_ok_false`를 대체
+
+```python
+def test_collect_is_molit_only(monkeypatch):
+    """collect_listings는 MOLIT만 수행(naver fetch 미호출). naver는 워커 담당."""
+    from app import listing_collector as lc
+    monkeypatch.setattr(lc, "_fetch_molit_all", lambda: (5, "calls=1 saved=5"))
+    called = {"naver": False}
+    def naver_spy():
+        called["naver"] = True
+        return (0, 0)
+    monkeypatch.setattr(lc, "_fetch_naver_all", naver_spy)
+    r = lc.collect_listings()
+    assert called["naver"] is False          # naver 미호출
+    assert "molit" in r["diag"]
+```
+
+- [ ] **Step 2: 실패 확인** — `PYTHONPATH=... python -m pytest tests/test_listing_collector.py -q` → FAIL(naver 호출됨)
+
+- [ ] **Step 3: 구현** — `collect_listings` 교체
+
+```python
+def collect_listings() -> dict:
+    """MOLIT 실거래(합법 baseline)만 수집. 네이버 호가는 naver-fetch 워커가
+    /api/internal/realestate/listings-ingest로 push (spec 2026-07-09-naver-listing-worker)."""
+    molit_diag = None
+    try:
+        _, molit_diag = _fetch_molit_all()
+    except Exception as e:
+        molit_diag = f"exc={type(e).__name__}:{str(e)[:80]}"
+        logger.error("MOLIT 수집 오류(안전마진 baseline): %s", e)
+    diag = f"molit[{molit_diag}] naver[worker]"
+    save_listing_collect_log(0, 0, None, error=diag)
+    return {"new_count": 0, "total_count": 0, "naver_ok": None, "diag": diag}
+```
+> `save_listing_collect_log`의 naver_ok 인자는 `1 if naver_ok else 0` 처리이므로 None→0 저장(무해). `_fetch_naver_all`/`_parse_naver_article`는 삭제하지 말 것(ingest 재사용).
+> 기존 `test_collect_writes_molit_diag_to_log`(계측 테스트)가 `_fetch_naver_all`을 `lambda: (0,0)`로 mock하는데 이제 미호출이므로 여전히 통과(무해). `test_molit_call_returns_diag_on_http_error`·`_parse_*`도 무관.
+
+- [ ] **Step 4: 통과 확인** — `PYTHONPATH=... python -m pytest tests/test_listing_collector.py -q` → PASS (신규 + 기존 계측/파서 테스트)
+- [ ] **Step 5: 커밋**
+
+```bash
+git add realestate-lab/app/listing_collector.py realestate-lab/tests/test_listing_collector.py
+git commit -m "refactor(realestate): collect_listings MOLIT 전용화(네이버는 워커 ingest로 이전)"
+```
+
+---
+
+### Task 6: 회귀 + 카탈로그/메모리 + 배포 + 계약 핸드오프
+
+**Files:** `CLAUDE.md`, 메모리 `service_realestate.md`/`infra_distributed_workers.md`
+
+- [ ] **Step 1: 전체 회귀** — `cd realestate-lab && PYTHONPATH="C:/Users/jaeoh/Desktop/workspace/web-backend" python -m pytest -q` (0 fail) + `cd agent-office && python -m pytest -q` (0 fail)
+- [ ] **Step 2: CLAUDE.md** — §5 nginx 표에 `/api/internal/realestate/`(IP화이트리스트) 추가, §9 realestate 표에 `GET /api/internal/realestate/targets`·`POST /listings-ingest` 추가
+- [ ] **Step 3: 메모리** — `service_realestate.md`(내부 계약·NAVER_CORTAR·collect MOLIT전용화), `infra_distributed_workers.md`(naver-fetch fetcher kind 등재, /api/internal/realestate/ 계약)
+- [ ] **Step 4: 계약 잠금·핸드오프** — co-gahusb `acquire_lock("nginx-conf","BE")`+`acquire_lock("nas-deploy","BE")` → 커밋 push(자동배포). AI(web-ai)에 §4 계약(targets/ingest 스키마·heartbeat kind=fetcher·INTERNAL_API_KEY)·병렬 spike 요청, FE(web-ui)에 /infra fetcher 노드 요청 `post_message`. 배포 후 `GET /api/internal/realestate/targets`(X-Internal-Key) 200 검증 → 락 해제.
+
+```bash
+git add CLAUDE.md
+git commit -m "docs(CLAUDE.md): naver 워커 internal 엔드포인트 + nginx 등재"
+git push origin main
+```
+
+---
+
+## Self-Review
+
+**Spec coverage:** §4.1 targets→T2, §4.2 ingest→T3, §4.3 cortarNo→T1, §4.4 heartbeat(계약, 워커측)→T4 등재로 관측, §5 관측→T4, §6 인증(nginx+X-Internal-Key)→T2, §7 멱등/에러→T3(try/except·article_no UNIQUE), §8 테스트→각 Task, §3 _fetch_naver_all deprecate→T5, §10 핸드오프→T6. ✓
+- 스펙 대비 조정: env명 `INTERNAL_API_KEY`(기존 관례, 스펙의 REALESTATE_INTERNAL_KEY 대체) — Global Constraints에 명시. matched=notify sent 수로 구체화(스펙 {received,new,matched} 유지).
+
+**Placeholder scan:** 각 Step 실제 코드/명령. T4의 collect_status fetcher-link 상세 테스트만 "기존 test_node_monitor 픽스처 따라"로 위임(기존 파일 참조, 트리비얼 등재 테스트는 완전). ✓
+
+**Type consistency:** `naver_cortar`(T1)→internal_router(T2) 사용 일치. `upsert_listing`→`(dict,is_new)`(기존)·`_parse_naver_article`(기존 반환 dict)·`notify_new_listings`→`{sent,...}`(기존)→ingest 소비 일치. targets 반환 `{dongs:[{dong,cortar_no}],deal_types,page_limit}`=스펙 §4.1. ingest 반환 `{received,new,matched}`=스펙 §4.2. WORKER_REGISTRY 항목 형식(T4)=기존 구조 일치. ✓
+
+**범위:** BE 단일 세션(realestate T1-3,5 + agent-office T4 + 통합 T6). 워커(web-ai)·FE(web-ui)는 별도. 단일 실행 계획 적정(6 tasks).