diff --git a/docs/weekly-log/2026-05-16-batch-ingest.md b/docs/weekly-log/2026-05-16-batch-ingest.md
new file mode 100644
index 0000000..f23e7cb
--- /dev/null
+++ b/docs/weekly-log/2026-05-16-batch-ingest.md
@@ -0,0 +1,375 @@
+# 2026-05-16 — W3 batch ingest (#17)
+
+> Status: **🎉🎉🎉🎉 8/8 100% 적재 + Aura 통계 5종 + NED 3차 진단 확정 → 발표 핵심 발견 박제 완료.**
+> 발표 메시지 확정: **선택지 3 — 디버깅 자체를 trade-off 인사이트로**
+> ⭐⭐⭐ **진짜 보석 발견**: Entity 추출 라벨 품질 문제 (Company 라벨이 실은 금융 metric)
+
+## 🔥🔥🔥 5/16 진단 3종 — 발표 핵심 발견 (17:39~17:43 측정)
+
+### 진단 1: 미래에셋 4Q 문서의 Entity top 20
+
+```cypher
+MATCH (d:Document {filename: "미래에셋증권_4분기_실적보고서.pdf"})
+      -[:HAS_SECTION]->(:Section)-[:CONTAINS_CHUNK]->(c:Chunk)
+      -[:MENTIONS]->(e:Entity)
+RETURN labels(e)[1] AS type, e.name, count(c) AS chunk_count
+ORDER BY chunk_count DESC LIMIT 20;
+```
+
+결과 (top 17):
+
+| Type    | e.name                                  | chunk_count |
+|---------|-----------------------------------------|-------------|
+| Company | "공모발행액 23조 7,050억원 (2025년 10월)"  | **19** |
+| Company | "유상증자 41,186"                       | 4 |
+| Company | "유상증자 77,258"                       | 4 |
+| Company | "금융지주채 발행규모 및 비중..."         | 3 |
+| Company | "AA등급 이상 회사채 발행 비중 73.0%"     | 3 |
+| Metric  | "유상증자 47,034 (기준: 증권신고서..)"   | 3 |
+| Company | "시설 목적 회사채 발행 비중 10.7%"       | 3 |
+| Company | "기업공개 전년동기대비 증감액..."         | 2 |
+| ... (대부분 Company 인데 금융 수치)        | ... |
+
+**🚨 충격적 발견**:
+- 미래에셋 4Q 보고서 68청크에서 **회사명이 단 하나도 Company entity 로 추출 안 됨**
+- "Company" 로 분류된 entity 들이 실제로는 **금융 metric / 발행 규모 / 비중** 등 수치
+- "공모발행액 23조 7,050억원" 이 Company 라벨 + 19 청크 언급 (가장 많이!)
+- LLM (Kimi) 이 **entity 추출 시 라벨을 잘못 부여**하고 있음
+
+### 진단 2: 영문/대명사 패턴 검색
+
+```cypher
+MATCH (e:Entity:Company)
+WHERE e.name CONTAINS "Mirae" OR e.name CONTAINS "당사"
+   OR e.name CONTAINS "회사" OR e.name CONTAINS "그룹"
+RETURN e.name, e.aliases, e.member_count ORDER BY e.member_count DESC LIMIT 20;
+```
+
+결과: **회사명 0건**. 모두 "회사채 23조..." 같은 수치 데이터.
+
+→ "당사", "본 회사", "그룹" 같은 대명사로 entity 추출 안 됨 (Kimi 가 무시하거나 다른 카테고리)
+→ 영문 "Mirae" 표기로도 추출 안 됨
+
+### 진단 3: 전체 Company entity top 20 (member_count 순)
+
+```cypher
+MATCH (e:Entity:Company)
+RETURN e.name, e.member_count, e.aliases
+ORDER BY e.member_count DESC LIMIT 20;
+```
+
+결과 (top 14):
+
+| Rank | e.name (Company 라벨)                              | member_count |
+|------|---------------------------------------------------|--------------|
+| 1    | "총계 금액 전년동기대비 증감액..." (NED 4 표기)      | **6** |
+| 2    | "금융사"                                            | 3 |
+| 3    | "일반기업"                                          | 3 |
+| 4    | "회사채 발행 규모 23조 6,111억원..."                  | 2 |
+| 5    | "단기채무 12,060만 원 (2024년 1∼10월)"               | 2 |
+| 6    | "회사채 23조 6,111억원 (전월 대비 16.6%↓...)"        | 2 |
+| 7    | "한국예탁결제원"                                     | 2 |
+| 8    | "금액 전월대비 증감액(증감률) △135 (△41.8%)..."     | 2 |
+| ...  | 모두 금융 metric 또는 일반 카테고리                | ... |
+
+**🚨 더 충격적 발견**:
+- **186개 Company entity 중 진짜 회사명은 거의 없음**
+- 1~3위가 "총계 금액", "금융사", "일반기업" — 모두 **범주명 또는 metric**
+- "한국예탁결제원" 이 7위 — **유일하게 진짜 기관명**
+- 회사명이 entity 로 거의 추출 안 됨 → 미래에셋증권 빈 결과의 진짜 원인
+
+### 🎯 진단 종합 — 진짜 발견
+
+> **"GraphRAG 파이프라인의 정직한 한계 발견:
+>
+> 미래에셋증권 자기 회사 보고서 4개 (1Q~4Q, 150 청크) 적재했지만 'Company' entity 로 단 하나도 잡히지 않음.
+>
+> 진짜 원인 — Kimi 가 entity 추출 시 'Company' 라벨을 잘못 부여:
+> - 금융 metric ('공모발행액 23조 7,050억원') → Company ❌
+> - 범주명 ('금융사', '일반기업') → Company ❌
+> - 회사명 → 거의 못 잡음 ❌
+>
+> 즉 GraphRAG 의 가치 (그래프 구조, MENTIONS 1,553) 는 검증됐으나,
+> **Entity 추출 품질** 이 별도의 production challenge.
+> LLM 프롬프트 엔지니어링 + 후처리 검증이 필수."**
+
+## 🎤 발표 슬라이드 메시지 — 진짜 보석 ⭐⭐⭐
+
+### Before (5/16 17:37 측정 직후 추정)
+> "NED 의 한국어 회사명 표기 변형 challenge"
+
+### After (5/16 17:43 진단 3종 확정)
+> **"Entity 추출 라벨 품질 challenge — Kimi LLM 이 'Company' 라벨에 회사명이 아닌 금융 metric 을 부여하는 패턴 발견. 186개 Company entity 중 진짜 기관명은 1개. 자기 회사 보고서에 자기 회사명이 entity 로 안 잡히는 정직한 한계."**
+
+### VectorRAG ↔ GraphRAG 보완 관계 증명
+
+| 질의 유형                            | VectorRAG | GraphRAG |
+|--------------------------------------|-----------|----------|
+| "당사 영업이익은?" (대명사 해소)       | ◎ 청크 유사도로 답변 | ❌ entity 없음 |
+| "회사 X 와 Y 의 공통 리스크는?"        | △ LLM 후처리 필요 | ◎ FACES_RISK 그래프 |
+| "금융 지표 평균 추이는?"              | △ 청크 모음 | ◎ HAS_METRIC 그래프 |
+| "이 답변의 출처는?"                  | △ 청크 ID | ◎ Layer A 추적 |
+
+→ **둘 다 필요한 게 정량 증거로 증명됨**. 발표의 진짜 결론.
+
+## 🏆🏆🏆🏆 5/16 8/8 100% 적재 완료 (16:05 ~ 17:33, 약 76분)
+
+### 1차 적재: 8문서 (16:05~17:19, 71분) — 7/8 성공
+
+| 파일 | 포맷 | 타입 | sec | chunks(T/B) | ent(raw→grp) | rel | mentions | parse | extract | link | neo4j | total | ok |
+|---|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---|
+| DS투자증권_시황분석_리포트.pdf | pdf | report | 6 | 52/0 | 103→60 | 32 | 102 | 8.4 | 143.5 | 28.5 | 39.0 | 219.4 | ✅ |
+| 금융감독원_..._직접금융_조달실적.doc | doc | (unknown) | 0 | 0/0 | 0→0 | 0 | 0 | 0.0 | 0.0 | 0.0 | 0.0 | 0.0 | ❌ |
+| 농협_2022년_9월말_기준_사업보고서.hwp | **hwp** | filing | 1 | 11/0 | 56→56 | 0 | 56 | 2.5 | 26.4 | 12.4 | 15.3 | 56.6 | ✅ |
+| 미래에셋증권_1분기_실적보고서.pdf | pdf | ir | 1 | 28/0 | 131→120 | 17 | 131 | 0.2 | 47.0 | 36.1 | 36.9 | 120.3 | ✅ |
+| 미래에셋증권_2분기_실적보고서.pdf | pdf | ir | 1 | 27/0 | 121→116 | 52 | 120 | 0.2 | 53.8 | 29.6 | 38.4 | 121.9 | ✅ |
+| 미래에셋증권_3분기_실적보고서.pdf | pdf | ir | 1 | 27/0 | 149→145 | 42 | 149 | **1514.8** | 47.2 | 25.9 | 43.8 | 1631.7 | ✅ |
+| 미래에셋증권_4분기_실적보고서.pdf | pdf | ir | 30 | **68/6** | **335→261** | 55 | 329 | **1694.7** | 109.7 | 109.4 | 102.3 | 2016.0 | ✅ |
+| 한화투자증권_두산밥캣_기업분석_리포트.pdf | pdf | report | 12 | 15/5 | 42→33 | 31 | 42 | 12.9 | 40.7 | 9.2 | 20.3 | 83.1 | ✅ |
+
+### 2차 적재: DOC → DOCX 변환 후 (17:28~17:33, 5분) — 1/1 성공
+
+| 파일 | 포맷 | 타입 | sec | chunks(T/B) | ent(raw→grp) | rel | mentions | parse | extract | link | neo4j | total | ok |
+|---|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---|
+| 금융감독원_..._직접금융_조달실적.docx | **docx** | **disclosure** | 1 | **49/0** | **275→251** | 26 | **260** | 1.7 | 106.0 | 116.6 | 73.3 | 297.7 | ✅ |
+
+### 🎉 진짜 최종 8/8 100% 총합
+
+| 지표 | 1차 (7문서) | 2차 (1문서) | **총합** |
+|---|---|---|---|
+| 성공률 | 7/8 (87.5%) | 1/1 (100%) | **8/8 (100%)** 🎉 |
+| 소요 시간 | 71분 | 5분 | **76분** |
+| 청크 (text) | 228 | 49 | **277** |
+| Raw entity | 937 | 275 | **1,212** |
+| Grouped entity | 791 | 251 | **1,042** |
+| NED compression | 0.84 | 0.91 | **0.86** |
+| Relation | 229 | 26 | **255** |
+| MENTIONS | 929 | 260 | **1,189** |
+
+## 🎯 5/16 8문서 Aura Cypher 통계 (17:37 측정)
+
+### Q1: 노드 라벨별 (총 610 노드)
+```
+Chunk 277 / Entity 261 / Section 53 / Table 11 / Document 8
+```
+
+### Q2: 관계 타입별 (총 2,451 관계)
+```
+MENTIONS 1553 / CONTAINS_CHUNK 277 / HAS_METRIC 270 / NEXT 235
+HAS_SECTION 53 / FACES_RISK 34 / HAS_OUTLOOK 12 / CONTAINS_TABLE 11
+RECOMMENDED_FOR 6 (새 관계 타입)
+```
+
+### Q3: Entity 타입별 분포
+```
+Company 186 (1,220 mentions, but 대부분 metric! → 진단 3 참고)
+Metric  69 (297 mentions)
+Risk    24 (29 mentions)
+Outlook  7 (7 mentions)
+```
+
+⚠️ **Company 186개 중 진짜 회사명은 1개 ("한국예탁결제원")** — 진단 3에서 확정.
+
+### Q4: 미래에셋증권 NED — 빈 결과
+```
+MATCH (e:Entity:Company {name: "미래에셋증권"}) ... → No records
+```
+→ 진단 3종으로 원인 확정: **Kimi 라벨 부여 문제**.
+
+### Q5: doc_type 분포
+```
+pdf+ir 4 (미래에셋 1Q~4Q) / pdf+report 2 (DS투자증권 + 한화) / hwp+filing 1 (농협) / docx+disclosure 1 (금감원)
+```
+**3 포맷 × 4 doc_type** 다양성.
+
+## 🌟 발표 자료 보석 종합 (5/23 슬라이드 시드)
+
+### A. DOC → DOCX 변환 + 재적재 성공 ⭐⭐⭐
+LibreOffice headless 변환 → 49 청크 + 251 entity + 260 MENTIONS. 엔지니어링 사이클 완성.
+
+### B. HWP 파서 정상 동작 ⭐
+농협 사업보고서 11 청크 적재.
+
+### C. 미래에셋 4분기 풍부 ⭐
+68 청크 + 6 표 (Table 첫 적재) + 335→261 entity + 329 MENTIONS.
+
+### D. ⭐⭐⭐ Entity 추출 라벨 품질 challenge (진단 3종 확정!)
+- 186개 Company 중 진짜 회사명 1개
+- "공모발행액 23조 7,050억원" 같은 metric 이 Company 라벨
+- LLM (Kimi) 의 라벨 부여 한계 → 프롬프트 엔지니어링 + 후처리 필요
+
+### E. VectorRAG ↔ GraphRAG 보완 관계 정량 증명
+- 자기 회사 보고서에서 회사명 entity 없음 → VectorRAG 의 대명사 해소 강점
+- 그래프 traversal 은 GraphRAG 의 강점 (HAS_METRIC 270, FACES_RISK 34)
+- **둘 다 필요한 게 정량 증거로 증명됨**
+
+### F. LLM 비결정성
+- DS투자증권 2회 추출: 113→55 vs 103→60
+
+### G. doc_type 자동 분류 4가지
+report / ir / filing / disclosure
+
+### H. SemanticChunker 본질적 비효율
+820자 / 31.3초 → USE_SIMPLE_CHUNKER fallback
+
+## ✅ 5/16 토요일 검증 완료 사항
+
+- ✅ **단위 테스트 11/11 PASS**
+- ✅ **PR #45 생성** — 6 commits
+- ✅ **8/8 적재 완료** — 76분, 100%
+- ✅ **Aura Cypher 5종 측정** — 17:37
+- ✅ **진단 3종 완료** — 17:39~17:43, Entity 라벨 품질 확정
+- ✅ **새 Aura `9b57188f`** + .env 갱신
+- ✅ **LibreOffice 미설치 → 수동 변환 검증**
+
+## 🎯 5/17(일) 시작 가이드 — 진짜 마무리
+
+### Step 1 — PR #45 머지 + 이슈 #17 close (1분)
+
+https://github.com/TaskerJang/doc-graph-agent/pull/45 → **Merge pull request** 버튼.
+
+머지 후:
+```cmd
+cd C:\Users\taske\doc-graph-agent
+git checkout dev
+git pull origin dev
+git branch -d feat/17-batch-ingest
+```
+
+### Step 2 — #18 W4 Text2Cypher 시작
+
+```cmd
+git checkout -b feat/18-text2cypher
+```
+
+이슈 #18 요구사항:
+- `retrieval/text2cypher.py` — LLM 기반 자연어 → Cypher 변환
+- 스키마 프롬프트 — 노드/관계 정의 LLM 주입
+- 안전장치 — read-only 강제 + LIMIT 100
+- 결과 파싱 → 자연어 답변 (LLM 2단계)
+- 평가 셋 5개 질문 정성 검증
+
+**5/17 평가 셋 질문 후보** (8문서 그래프 위에서):
+1. "DS투자증권 시황분석 리포트에서 언급된 entity 들은? (회사+metric 혼재 검증)"
+2. "미래에셋증권 4분기 보고서의 Table 은 몇 개인가?"
+3. "보도자료(disclosure) 유형 문서의 entity 들은?"
+4. "두산밥캣과 함께 언급된 리스크가 있는가?"
+5. "전체 그래프에서 가장 많이 언급된 Company 라벨 entity 5개는? (실제 회사 vs metric 비율 확인)"
+
+5번이 진짜 흥미로움 — Text2Cypher 가 "Company" 라벨 결과를 사용자에게 보여줄 때, 실은 metric 이 섞인 결과라는 challenge.
+
+---
+
+## 작업 범위
+
+PR #44 (#24 Opik 1단계) 머지 직후 진행. 두산밥캣 1청크 파이프라인 → 8문서 전체 확장.
+
+### 변경 사항
+
+**Layer A 적재 추가** (`kg/builder.py`):
+- `build_layer_a()` — Document/Section/Chunk/Table MERGE + 4관계
+- `link_chunks_to_entities()` — Chunk → Entity `[:MENTIONS]`
+
+**옵션 3 chunk_id global prefix** (`kg/extractor.py`):
+- `{chunk_id}__ent_001` 형식 + `make_global_id` / `parse_global_id`
+
+**일괄 처리 스크립트** (`scripts/run_w3_batch.py`):
+- 8문서 순회 + stat + stdout 표
+
+**chunker.py USE_SIMPLE_CHUNKER fallback** (commit `8f0bcb1`):
+- `RecursiveCharacterTextSplitter` 분기, 기본 OFF
+
+## 🎤 발표 메시지 확정 (5/23) — 선택지 3
+
+> **"동일 chunker 로 비교하려 했으나, CPU 환경에서 SemanticChunker 가 비현실적임을 발견. 운영 비용의 trade-off 를 정량화함. 추가로 LLM Entity 추출의 라벨 품질 challenge 도 발견."**
+
+### 발표 구조 (12슬라이드)
+
+1. 문제 정의
+2. SemanticChunker 발견 (820자/31초)
+3. 분석 (LangChain sequential)
+4. trade-off 표
+5. fallback 구현
+6. **운영 데이터 (8/8 100%, 76분)**
+7. **포맷 다양성 (PDF + HWP + DOCX, 4 doc_type)**
+8. **production 의존성 (LibreOffice)** ⭐
+9. **그래프 통계 (610 / 2,451 / 1,553)** ⭐
+10. **⭐⭐⭐ Entity 라벨 품질 challenge — 186 Company 중 진짜 회사명 1개!**
+11. **VectorRAG ↔ GraphRAG 보완 관계 정량 증명** ⭐ NEW!
+12. 인사이트 메시지
+
+### "그래서 결국 성능 비교는?" 질문 대응
+
+답변 준비:
+- SemanticChunker 원본은 production 환경 후 (Issue #46)
+- chunk 정책 동일 유지, Layer 구조 효과 정량화: 1,189 MENTIONS, 0.86 compression
+- 4가지 doc_type 자동 분류
+- **Entity 라벨 품질** — Kimi 한계 발견, 프롬프트 엔지니어링 + 후처리 필요
+- **VectorRAG ↔ GraphRAG 가 보완 관계** 라는 게 정량 증명됨
+
+## 시행착오 박제
+
+5/16 토요일 디버깅 자산 (= 발표 슬라이드 시드):
+- ✅ **bge-m3 정상** (108ms/문장)
+- ✅ **PDF / OCR / Neo4j / SemanticChunker 초기화 모두 정상**
+- ❌ **SemanticChunker.split_text 본질적 비효율** — 820자/31초
+- ✅ **USE_SIMPLE_CHUNKER fallback** — commit `8f0bcb1`
+- ✅ **8/8 적재 100%** + **Aura 통계 5종** + **진단 3종**
+- ✅ **HWP / DOCX 파서 정상**
+- ✅ **Table 노드 첫 적재** (11개)
+- ✅ **doc_type 4가지 자동 분류**
+- ⚠️⚠️⚠️ **Entity 라벨 품질 challenge** — Kimi 가 Company 라벨에 metric 부여
+- ⚠️ **NED 한국어 회사명 변형** — 자기 회사 보고서에 자기 회사명 entity 없음
+- **Aura Free trial** — 새 인스턴스 `9b57188f`
+- **LibreOffice 환경 의존성** — production Docker 가이드
+- **MS Store python stub** — `uv run` 영향 없음
+
+### 미래 발견 (#46+)
+
+- **LibreOffice 자동화** — production Docker 이미지 추가
+- **OCR 시간 폭증** — 3Q/4Q parse 1500초+ → 캐시 또는 라이브러리 교체
+- **OpenAI 마이그레이션** (#46) — SemanticChunker 원본 복원
+- **Entity 추출 프롬프트 엔지니어링** ⭐ NEW! — Kimi 의 Company 라벨 품질 개선
+- **NED 한국어 후처리** — 회사명 정규화 추가 단계
+
+### local_id 충돌 — 옵션 3 으로 해결
+
+```
+adapter.parse_document() → Document.sections[].chunks[].id ("doc1:c0001")
+                              ↓
+chunks = [{doc_id, chunk_id, section, text}]
+                              ↓ extract() — chunk_id prefix
+ExtractedEntity.local_id = "doc1:c0001__ent_001"
+                              ↓ link_entities() — 청크 간 NED
+EntityGroup.members[*].local_id = "doc1:c0001__ent_001"
+                              ↓ link_chunks_to_entities()
+parse_global_id("doc1:c0001__ent_001") → ("doc1:c0001", "ent_001")
+                              ↓
+(:Chunk {id:"doc1:c0001"}) -[:MENTIONS]-> (:Entity {group_id:"grp_001"})
+```
+
+## 발표 자료 시드 (5/23 용)
+
+| 자료 | 상태 | 출처 |
+|---|---|---|
+| 1문서 → 8/8 적재 운영 데이터 표 | ✅ 완료 | 본 문서 |
+| **🎯 Aura 통계 5종 (8문서, 17:37)** | ✅ 완료 | 본 문서 |
+| **🎯 진단 3종 — Entity 라벨 품질** ⭐⭐⭐ | ✅ 완료 | 본 문서 17:43 |
+| **🎯 Company 186 중 진짜 회사명 1개 (한국예탁결제원)** | ✅ 완료 | 진단 3 |
+| **🎯 자기 회사 보고서에 자기 회사명 없음** | ✅ 완료 | 진단 1+2 |
+| **🎯 VectorRAG ↔ GraphRAG 보완 관계 정량 증명** ⭐ NEW! | ✅ 완료 | 본 문서 |
+| **🎯 doc_type × format 4×3 다양성** | ✅ 완료 | Q5 |
+| Layer A + Layer B + MENTIONS 그래프 시각화 | ⏸️ Aura console 캡처 | 일요일 |
+| **🎯 LibreOffice 의존성 발견 + 해소 사이클** | ✅ 완료 | 본 문서 |
+| `@track` 으로 자동 수집된 trace | ✅ Opik UI 확인 가능 | Opik UI |
+| **🎯 미래에셋 4분기 — 68 청크 + 6 표 + 261 entity** | ✅ 완료 | 본 문서 |
+| **🎯 LLM 비결정성 — DS투자증권 2회 추출 차이** | ✅ 완료 | 본 문서 |
+| **🎯 SemanticChunker 본질적 비효율 (820자/31초)** | ✅ 완료 | 본 문서 |
+| **🎯 USE_SIMPLE_CHUNKER fallback 코드** | ✅ 완료 | commit `8f0bcb1` |
+
+## 관련
+
+- PR: #45 (5/17 일요일 머지 예정)
+- 의존: ✅ PR #44 (#24 Opik 1단계)
+- 후속: #18 Text2Cypher (Layer A 적재된 풍성한 그래프 위에서)
+- 미래: #46 OpenAI 마이그레이션, #47 회사 발표 (6/1), [신규] Entity 추출 프롬프트 개선
diff --git a/ingestion/chunker.py b/ingestion/chunker.py
index 01ecd8f..e1733bb 100644
--- a/ingestion/chunker.py
+++ b/ingestion/chunker.py
@@ -1,14 +1,24 @@
 # Ported from doc-summary-agent/chunker/chunker.py (#11).
 # Source: TaskerJang/doc-summary-agent@dev as of 2026-05-03.
-# 변경: 없음 (원본 그대로). 회사 레포의 인제스트 결과를 동일하게 재현하기 위한
-#   의도적 복붙이며, GraphRAG vs VectorRAG 비교의 정직성을 위해 chunk 정책
-#   (size=700, overlap=200, min=50)을 회사 레포 dev와 동일하게 유지한다.
+#
+# 변경 이력:
+# - 5/16, #17: USE_SIMPLE_CHUNKER 환경변수 분기 추가 (CPU 환경 fallback).
+#   기본값 OFF — 환경변수 안 켜면 회사 레포와 100% 동일 동작.
+#   ON 시 RecursiveCharacterTextSplitter 사용 (SemanticChunker 우회).
+#
+# 정직성 원칙:
+#   chunk 정책 (size=700, overlap=200, min=50) 은 회사 레포 dev와 동일 유지 —
+#   GraphRAG vs VectorRAG 비교 시 chunk 경계 변수를 통제하기 위해.
+#   USE_SIMPLE_CHUNKER=1 fallback 은 CPU 환경에서 SemanticChunker 가 비현실적인
+#   경우 (820자/31초, 5/16 발견) 의 escape hatch — 발표 자료의 trade-off 슬라이드
+#   증거로 사용.
 import logging
+import os
 import re
 from collections import Counter
 from typing import Literal, TypedDict
 
-from langchain_text_splitters import MarkdownTextSplitter
+from langchain_text_splitters import MarkdownTextSplitter, RecursiveCharacterTextSplitter
 
 logger = logging.getLogger(__name__)
 
@@ -46,6 +56,22 @@
 # S-02: SemanticChunker 싱글턴
 _semantic_splitter = None
 
+# ── #17 SIMPLE_CHUNKER fallback ────────────────────────────
+#
+# USE_SIMPLE_CHUNKER=1 환경변수 ON 시 SemanticChunker 우회하고
+# RecursiveCharacterTextSplitter 사용.
+#
+# 배경 (5/16, #17):
+# - LangChain SemanticChunker 가 임베딩을 문장 단위 sequential 호출 (batch 미사용)
+# - bge-m3 (568M params) + CPU 환경: 820자 / 31.3초 측정 (5/16 단독 벤치)
+# - 8문서 풀 실행 예상: 분할만 40분~1시간 → production 비현실적
+# - SemanticChunker 는 OpenAI API 같은 빠른 임베딩과만 호환 (GPU/API 환경 전제)
+#
+# 기본값 OFF — 회사 레포 동일 동작 유지.
+# ON 으로 키면 chunk 경계 알고리즘이 회사 레포와 달라지므로
+# GraphRAG vs VectorRAG 정밀 비교 시에는 끄고 사용.
+_SIMPLE_CHUNKER_ENV = "USE_SIMPLE_CHUNKER"
+
 # ── #75 메타데이터 enrichment 상수 ──────────────────────────
 
 # M-01: 연도 추출 정규식 — 2010~2039 범위 (금융 보고서 실용 범위)
@@ -153,6 +179,21 @@ def _get_semantic_splitter():
 
 
 def _semantic_split(text: str) -> list[str]:
+    # #17 (5/16) — USE_SIMPLE_CHUNKER=1 켜져 있으면 SemanticChunker 우회.
+    # 배경: bge-m3 + CPU + SemanticChunker = 820자 / 31.3초 (비현실적).
+    # 발표 자료의 trade-off 슬라이드 증거 + production fallback.
+    if os.getenv(_SIMPLE_CHUNKER_ENV) == "1":
+        logger.info(
+            "%s=1 → SemanticChunker 우회, RecursiveCharacterTextSplitter 사용",
+            _SIMPLE_CHUNKER_ENV,
+        )
+        simple_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=DEFAULT_CHUNK_SIZE,
+            chunk_overlap=DEFAULT_CHUNK_OVERLAP,
+            separators=["\n\n", "\n", ". ", "。", "? ", "! ", " ", ""],
+        )
+        return simple_splitter.split_text(text)
+
     sentences = [s for s in re.split(KOREAN_SENTENCE_SPLIT_REGEX, text) if s.strip()]
     if len(sentences) < 3:
         logger.debug("문장 수 부족(%d) → SemanticChunker 우회", len(sentences))
diff --git a/kg/builder.py b/kg/builder.py
index 754229b..29d3be6 100644
--- a/kg/builder.py
+++ b/kg/builder.py
@@ -1,23 +1,27 @@
-"""Layer B 적재 — LinkingResult + Relations → Neo4j MERGE.
+"""Layer A/B 적재 — Neo4j MERGE 기반 idempotent ingest.
 
-#13 (extractor) → #14 (linking) 의 출력을 받아 Neo4j Aura 에 Layer B 를
-적재한다. Layer A (Document/Section/Chunk) 는 5/14 보너스에서.
+#13 (extractor) → #14 (linking) 의 출력을 받아 Neo4j Aura 에 Layer B 를,
+#17 의 ingestion.Document 객체를 받아 Layer A 를 적재한다.
 
 설계 원칙:
 - **Idempotent**: MERGE 기반 — 재실행해도 중복 노드 안 생김
-- **멀티 라벨**: 모든 Entity 노드에 `:Entity` (전역 검색용) + 구체 타입
-  (`:Company` `:Metric` `:Recommendation` `:Risk` `:Outlook`) 동시 부착
-- **노드 키**: `group_id` (NED 가 부여한 전역 ID) — 같은 그룹이면 같은 노드
-- **관계**: ExtractedRelation.source/target (청크 내 local_id) 는 NED 단계에
-  group_id 로 매핑하는 보조 테이블 필요 — 본 모듈이 책임
+- **멀티 라벨**: Layer B Entity 노드는 `:Entity` (전역 검색) + 구체 타입 동시 부착
+- **노드 키**: Layer B 는 `group_id` (NED), Layer A 는 `id` (file hash + 청크 idx)
+- **Layer 분리**: Layer A 적재 시 Layer B 노드 안 만지고, 그 반대도 동일.
+  연결은 별도 함수 `link_chunks_to_entities` 가 책임 — AGENTS.md 원칙
+  "Layer 책임 섞지 않기" 준수.
 
 #24 (Opik 트레이싱):
-- 공개 진입점 `build_layer_b` 에 `@track` — span 입출력은 노드/관계 적재 수.
-- 개별 MERGE 호출 (_merge_group / _merge_relation) 은 부착 안 함 — 8문서
-  적재 시 span 폭증 방지 (1문서 = 수십 개 MERGE).
+- 공개 진입점 3개 모두 `@track`: build_layer_a / build_layer_b / link_chunks_to_entities
+- 개별 MERGE 호출은 부착 안 함 — 8문서 적재 시 span 폭증 방지
 
-관련 이슈: #15 (본), #13/#14 (입력 공급), #16 (후속 — 시각화),
-          #24 (Opik 트레이싱).
+#17 (5/16 8문서 일괄):
+- Layer A 적재 — doc-ontology.md §3-§4 의 Cypher 스키마 그대로 구현
+- 청크 노드의 id 가 ingestion.adapter._chunk_id() 의 `{doc_id}:c{idx:04d}` 형식
+- entity 의 local_id 는 extractor 옵션 3 으로 `{chunk_id}__ent_001` 형식 →
+  parse_global_id 로 청크 역추적해 MENTIONS 관계 생성
+
+관련 이슈: #15 (Layer B), #17 (Layer A + 8문서), #13/#14 (입력), #24 (Opik).
 """
 
 from __future__ import annotations
@@ -26,15 +30,22 @@
 import time
 from dataclasses import dataclass
 
+from kg.extractor import parse_global_id
 from kg.linking import EntityGroup, LinkingResult
 from kg.neo4j_client import Neo4jClient
 from kg.ontology import EntityType, ExtractedEntity, ExtractedRelation, RelationType
 from observability.tracing import track
 
+# Layer A 모델 — ingestion 모듈에서 import. circular 방지를 위해 함수 안에서
+# import 하지 않고 여기서 한 번만.
+from ingestion.models import Chunk, ChunkType, Document, Section, Table
+
 logger = logging.getLogger(__name__)
 
 
-# ── 결과 타입 ───────────────────────────────────────────
+# ═══════════════════════════════════════════════════════════════════
+# 결과 타입
+# ═══════════════════════════════════════════════════════════════════
 @dataclass
 class IngestResult:
     """build_layer_b() 의 적재 결과 수치."""
@@ -45,13 +56,44 @@ class IngestResult:
     elapsed_sec:      float
 
 
-# ── 인덱스 / 제약 조건 ───────────────────────────────────
+@dataclass
+class LayerAIngestResult:
+    """build_layer_a() 의 적재 결과 수치."""
+
+    documents_merged: int
+    sections_merged:  int
+    chunks_merged:    int
+    tables_merged:    int
+    next_relations:   int  # Chunk -[:NEXT]-> Chunk 개수
+    elapsed_sec:      float
+
+
+@dataclass
+class MentionsResult:
+    """link_chunks_to_entities() 의 적재 결과 수치."""
+
+    mentions_merged:  int
+    mentions_dropped: int   # chunk_id parse 실패 또는 entity 없음
+    elapsed_sec:      float
+
+
+# ═══════════════════════════════════════════════════════════════════
+# 인덱스 / 제약 조건
+# ═══════════════════════════════════════════════════════════════════
 INDEX_QUERIES = [
-    # group_id 유일 제약 — 동시에 B-tree 인덱스 역할
+    # Layer B
     "CREATE CONSTRAINT entity_group_id IF NOT EXISTS "
     "FOR (e:Entity) REQUIRE e.group_id IS UNIQUE",
-    # name 검색 가속 (유일 제약 아님 — 동명 회사·메트릭 가능)
     "CREATE INDEX entity_name IF NOT EXISTS FOR (e:Entity) ON (e.name)",
+    # Layer A — #17 추가. doc-ontology.md §3 의 4 노드 모두 id 유일.
+    "CREATE CONSTRAINT document_id IF NOT EXISTS "
+    "FOR (d:Document) REQUIRE d.id IS UNIQUE",
+    "CREATE CONSTRAINT section_id IF NOT EXISTS "
+    "FOR (s:Section) REQUIRE s.id IS UNIQUE",
+    "CREATE CONSTRAINT chunk_id IF NOT EXISTS "
+    "FOR (c:Chunk) REQUIRE c.id IS UNIQUE",
+    "CREATE CONSTRAINT table_id IF NOT EXISTS "
+    "FOR (t:Table) REQUIRE t.id IS UNIQUE",
 ]
 
 
@@ -62,7 +104,9 @@ def ensure_indexes(client: Neo4jClient) -> None:
         logger.info("Index/constraint applied: %s", q.split("FOR")[0].strip())
 
 
-# ── 노드 MERGE ──────────────────────────────────────────
+# ═══════════════════════════════════════════════════════════════════
+# Layer B — Entity 노드 / 관계
+# ═══════════════════════════════════════════════════════════════════
 # 멀티 라벨 — EntityType 값을 동적 라벨로 부착하면서도 공통 :Entity 도 부여.
 # Cypher 는 라벨을 파라미터로 받을 수 없으므로 f-string 으로 조립 —
 # 입력 EntityType.value 는 enum 에서 온 화이트리스트 (injection 없음).
@@ -91,7 +135,6 @@ def _merge_group(client: Neo4jClient, group: EntityGroup) -> None:
     )
 
 
-# ── 관계 MERGE ─────────────────────────────────────────
 RELATION_MERGE_TEMPLATE = """
 MATCH (s:Entity {{group_id: $source_gid}})
 MATCH (t:Entity {{group_id: $target_gid}})
@@ -118,16 +161,11 @@ def _merge_relation(
     )
 
 
-# ── local_id → group_id 매핑 ──────────────────────────────────
 def _build_local_to_group(linking: LinkingResult) -> dict[str, str]:
     """ExtractedEntity.local_id → 그게 속한 EntityGroup.group_id.
 
-    ExtractedRelation 은 source/target 으로 local_id 를 싨는데 (추출 당시의 ID),
-    적재 시점에는 NED 그룹 기준으로 연결해야 하므로 매핑 필요.
-
-    주의: local_id 는 청크 간 중복 가능 (각 청크가 ent_001 부터 시작).
-    실제 운영에서는 (chunk_idx, local_id) 튜플이 key 여야 안전.
-    본 PR 은 1청크 시나리오만 취급 — #17 에서 다중 청크 시 결합.
+    옵션 3 (#17) 이후 local_id 는 `{chunk_id}__ent_001` 형식 — global 유일.
+    따라서 청크 간 충돌 없음.
     """
     mapping: dict[str, str] = {}
     for group in linking.groups:
@@ -136,7 +174,6 @@ def _build_local_to_group(linking: LinkingResult) -> dict[str, str]:
     return mapping
 
 
-# ── 공개 인터페이스 ───────────────────────────────────────
 @track
 def build_layer_b(
     linking: LinkingResult,
@@ -151,38 +188,22 @@ def build_layer_b(
       1. (옵션) 인덱스/제약조건 멱등 생성
       2. 그룹별로 :Entity:{Type} 노드 MERGE
       3. local_id → group_id 매핑 구축
-      4. 관계별로 MATCH-MATCH-MERGE (양쪽 그룹 존재 필수)
-
-    Args:
-        linking:        #14 출력 (그룹화된 Entity 들)
-        relations:      #13 출력 (청크 내 local_id 기반 관계)
-        client:         Neo4jClient 를 주입 (테스트·CLI 재활용). None 이면 from_env.
-        create_indexes: 첨 적재 시 인덱스 생성 시도 (멱등).
-
-    Returns:
-        IngestResult — 적재 수치.
-
-    #24 Opik:
-    - 본 함수에 `@track` — span 입출력은 노드/관계 적재 수·소요.
-    - 환경 미설정 시 자동 no-op.
+      4. 관계별로 MATCH-MATCH-MERGE
     """
     started = time.perf_counter()
 
     if client is None:
-        client = Neo4jClient()  # from_env
+        client = Neo4jClient()
 
     if create_indexes:
         ensure_indexes(client)
 
-    # 2. 노드 MERGE
     for group in linking.groups:
         _merge_group(client, group)
     nodes_merged = len(linking.groups)
 
-    # 3. local_id → group_id
     local_to_group = _build_local_to_group(linking)
 
-    # 4. 관계 MERGE
     relations_merged  = 0
     relations_dropped = 0
     for rel in relations:
@@ -210,3 +231,323 @@ def build_layer_b(
         result.nodes_merged, result.relations_merged, result.relations_dropped, result.elapsed_sec,
     )
     return result
+
+
+# ═══════════════════════════════════════════════════════════════════
+# Layer A — Document / Section / Chunk / Table (#17)
+# ═══════════════════════════════════════════════════════════════════
+# 노드 MERGE 템플릿. doc-ontology.md §3.1-3.4 스키마 그대로.
+
+DOCUMENT_MERGE = """
+MERGE (d:Document {id: $id})
+SET d.filename       = $filename,
+    d.doc_type       = $doc_type,
+    d.source_format  = $source_format,
+    d.total_pages    = $total_pages,
+    d.publisher      = $publisher,
+    d.subject        = $subject,
+    d.fiscal_year    = $fiscal_year,
+    d.ingested_at    = timestamp()
+RETURN d.id AS id
+""".strip()
+
+
+SECTION_MERGE = """
+MERGE (s:Section {id: $id})
+SET s.label             = $label,
+    s.doc_type          = $doc_type,
+    s.original_section  = $original_section,
+    s.heading_level     = $heading_level,
+    s.order_index       = $order_index,
+    s.page_start        = $page_start,
+    s.page_end          = $page_end,
+    s.updated_at        = timestamp()
+RETURN s.id AS id
+""".strip()
+
+
+CHUNK_MERGE = """
+MERGE (c:Chunk {id: $id})
+SET c.text         = $text,
+    c.chunk_type   = $chunk_type,
+    c.char_count   = $char_count,
+    c.page         = $page,
+    c.order_index  = $order_index,
+    c.updated_at   = timestamp()
+RETURN c.id AS id
+""".strip()
+
+
+TABLE_MERGE = """
+MERGE (t:Table {id: $id})
+SET t.raw_markdown  = $raw_markdown,
+    t.caption       = $caption,
+    t.row_count     = $row_count,
+    t.column_count  = $column_count,
+    t.page          = $page,
+    t.order_index   = $order_index,
+    t.updated_at    = timestamp()
+RETURN t.id AS id
+""".strip()
+
+
+# 관계 MERGE — Layer A 의 4 관계 (doc-ontology.md §4)
+HAS_SECTION_MERGE = """
+MATCH (d:Document {id: $doc_id})
+MATCH (s:Section  {id: $section_id})
+MERGE (d)-[:HAS_SECTION]->(s)
+""".strip()
+
+
+CONTAINS_CHUNK_MERGE = """
+MATCH (s:Section {id: $section_id})
+MATCH (c:Chunk   {id: $chunk_id})
+MERGE (s)-[:CONTAINS_CHUNK]->(c)
+""".strip()
+
+
+CONTAINS_TABLE_MERGE = """
+MATCH (s:Section {id: $section_id})
+MATCH (t:Table   {id: $table_id})
+MERGE (s)-[:CONTAINS_TABLE]->(t)
+""".strip()
+
+
+NEXT_CHUNK_MERGE = """
+MATCH (a:Chunk {id: $prev_id})
+MATCH (b:Chunk {id: $next_id})
+MERGE (a)-[:NEXT]->(b)
+""".strip()
+
+
+def _merge_document(client: Neo4jClient, doc: Document) -> None:
+    """Document 노드 MERGE."""
+    client.write(
+        DOCUMENT_MERGE,
+        id=doc.id,
+        filename=doc.filename,
+        doc_type=doc.doc_type.value,
+        source_format=doc.source_format.value,
+        total_pages=doc.total_pages,
+        publisher=doc.publisher,
+        subject=doc.subject,
+        fiscal_year=doc.fiscal_year,
+    )
+
+
+def _merge_section(client: Neo4jClient, section: Section) -> None:
+    """Section 노드 MERGE — label 은 W3 LLM 채움 전이라 None 가능."""
+    client.write(
+        SECTION_MERGE,
+        id=section.id,
+        label=section.label.value if section.label else None,
+        doc_type=section.doc_type.value,
+        original_section=section.original_section,
+        heading_level=section.heading_level,
+        order_index=section.order_index,
+        page_start=section.page_start,
+        page_end=section.page_end,
+    )
+
+
+def _merge_chunk(client: Neo4jClient, chunk: Chunk) -> None:
+    """Chunk 노드 MERGE."""
+    client.write(
+        CHUNK_MERGE,
+        id=chunk.id,
+        text=chunk.text,
+        chunk_type=chunk.chunk_type.value,
+        char_count=chunk.char_count,
+        page=chunk.page,
+        order_index=chunk.order_index,
+    )
+
+
+def _merge_table(client: Neo4jClient, table: Table) -> None:
+    """Table 노드 MERGE."""
+    client.write(
+        TABLE_MERGE,
+        id=table.id,
+        raw_markdown=table.raw_markdown,
+        caption=table.caption,
+        row_count=table.row_count,
+        column_count=table.column_count,
+        page=table.page,
+        order_index=table.order_index,
+    )
+
+
+@track(ignore_arguments=["client"])
+def build_layer_a(
+    document: Document,
+    *,
+    client: Neo4jClient | None = None,
+    create_indexes: bool = True,
+) -> LayerAIngestResult:
+    """Layer A (Document Structure) 를 Neo4j 에 적재.
+
+    파이프라인 (doc-ontology.md §3-§4):
+      1. (옵션) Layer A 인덱스/제약조건 멱등 생성
+      2. Document 노드 MERGE
+      3. Section 노드들 MERGE + HAS_SECTION 관계
+      4. Chunk / Table 노드들 MERGE + CONTAINS_CHUNK / CONTAINS_TABLE 관계
+      5. Section 내 Chunk 순서대로 NEXT 관계
+
+    Args:
+        document:       ingestion.adapter.parse_document() 출력
+        client:         Neo4jClient 주입 (테스트 재활용). None 이면 from_env.
+        create_indexes: 첫 적재 시 인덱스 생성 시도 (멱등).
+
+    Returns:
+        LayerAIngestResult — 노드/관계 적재 수치.
+
+    #24 Opik:
+    - 본 함수에 `@track(ignore_arguments=["client"])` — client 객체가
+      `<...object at 0x...>` 로 박혀 노이즈가 되는 것 회피 (5/16 박제).
+    """
+    started = time.perf_counter()
+
+    if client is None:
+        client = Neo4jClient()
+
+    if create_indexes:
+        ensure_indexes(client)
+
+    # 2. Document
+    _merge_document(client, document)
+    documents_merged = 1
+
+    sections_merged = 0
+    chunks_merged   = 0
+    tables_merged   = 0
+    next_relations  = 0
+
+    # 3-5. Section / Chunk / Table / 관계
+    for section in document.sections:
+        _merge_section(client, section)
+        client.write(HAS_SECTION_MERGE, doc_id=document.id, section_id=section.id)
+        sections_merged += 1
+
+        prev_chunk_id: str | None = None
+        for chunk in section.chunks:
+            _merge_chunk(client, chunk)
+            client.write(
+                CONTAINS_CHUNK_MERGE,
+                section_id=section.id,
+                chunk_id=chunk.id,
+            )
+            chunks_merged += 1
+
+            # Section 내 Chunk NEXT 관계
+            if prev_chunk_id is not None:
+                client.write(
+                    NEXT_CHUNK_MERGE,
+                    prev_id=prev_chunk_id,
+                    next_id=chunk.id,
+                )
+                next_relations += 1
+            prev_chunk_id = chunk.id
+
+        for table in section.tables:
+            _merge_table(client, table)
+            client.write(
+                CONTAINS_TABLE_MERGE,
+                section_id=section.id,
+                table_id=table.id,
+            )
+            tables_merged += 1
+
+    elapsed = time.perf_counter() - started
+    result = LayerAIngestResult(
+        documents_merged=documents_merged,
+        sections_merged=sections_merged,
+        chunks_merged=chunks_merged,
+        tables_merged=tables_merged,
+        next_relations=next_relations,
+        elapsed_sec=elapsed,
+    )
+    logger.info(
+        "Layer A 적재 완료 doc=%s sections=%d chunks=%d tables=%d next=%d (%.2fs)",
+        document.filename,
+        result.sections_merged, result.chunks_merged,
+        result.tables_merged, result.next_relations,
+        result.elapsed_sec,
+    )
+    return result
+
+
+# ═══════════════════════════════════════════════════════════════════
+# MENTIONS — Layer A Chunk → Layer B Entity (#17)
+# ═══════════════════════════════════════════════════════════════════
+MENTIONS_MERGE = """
+MATCH (c:Chunk  {id: $chunk_id})
+MATCH (e:Entity {group_id: $group_id})
+MERGE (c)-[:MENTIONS]->(e)
+""".strip()
+
+
+@track(ignore_arguments=["client"])
+def link_chunks_to_entities(
+    linking: LinkingResult,
+    *,
+    client: Neo4jClient | None = None,
+) -> MentionsResult:
+    """Layer A Chunk 와 Layer B Entity 를 MENTIONS 관계로 연결.
+
+    동작:
+      - linking.groups 의 각 EntityGroup.members 를 순회
+      - 각 member 의 local_id 에서 `parse_global_id` 로 chunk_id 추출
+      - `(:Chunk {id: chunk_id}) -[:MENTIONS]-> (:Entity {group_id: group.group_id})`
+
+    chunk_id 가 없는 (구식, 5/10 sanity) entity 는 graceful drop — Layer A
+    가 적재 안 됐을 가능성 높으므로 매핑 불가.
+
+    Args:
+        linking: #14 출력 (그룹화된 Entity 들). 각 member 의 local_id 가
+                 `{chunk_id}__ent_001` 형식이어야 매핑 가능 (옵션 3).
+        client:  Neo4jClient 주입. None 이면 from_env.
+
+    Returns:
+        MentionsResult — 적재 / drop 수치.
+    """
+    started = time.perf_counter()
+
+    if client is None:
+        client = Neo4jClient()
+
+    mentions_merged  = 0
+    mentions_dropped = 0
+
+    for group in linking.groups:
+        # 같은 그룹 안에서 동일 청크가 여러 member 로 등장할 수 있음 (드물게).
+        # MERGE 가 idempotent 라 여러 번 호출해도 안전하지만 통계는 정확히
+        # 청크별 1회로 집계.
+        seen_chunks: set[str] = set()
+        for member in group.members:
+            chunk_id, _ = parse_global_id(member.local_id)
+            if chunk_id is None:
+                # 구식 local_id — Layer A 매핑 불가. 5/10 sanity 호환.
+                mentions_dropped += 1
+                continue
+            if chunk_id in seen_chunks:
+                continue
+            seen_chunks.add(chunk_id)
+
+            client.write(
+                MENTIONS_MERGE,
+                chunk_id=chunk_id,
+                group_id=group.group_id,
+            )
+            mentions_merged += 1
+
+    elapsed = time.perf_counter() - started
+    result = MentionsResult(
+        mentions_merged=mentions_merged,
+        mentions_dropped=mentions_dropped,
+        elapsed_sec=elapsed,
+    )
+    logger.info(
+        "MENTIONS 적재 완료 merged=%d dropped=%d (%.2fs)",
+        result.mentions_merged, result.mentions_dropped, result.elapsed_sec,
+    )
+    return result
diff --git a/kg/extractor.py b/kg/extractor.py
index 39747e6..4b9e098 100644
--- a/kg/extractor.py
+++ b/kg/extractor.py
@@ -18,8 +18,15 @@
 - 청크 단위 (_extract_chunk) 는 의도적으로 부착 안 함 — 8문서 × N청크 일괄
   적재 시 trace 가 폭발하지 않도록. 필요하면 W5 에서 추가.
 
+#17 (5/16 8문서 일괄 적재) — local_id 충돌 해결:
+- 청크 N 개에서 각각 `ent_001` 부터 부여 → 다중 청크 시 local_id 충돌
+- 해결 (옵션 3): chunk dict 에 옵셔널 `chunk_id` 키를 추가하면 본 모듈이
+  자동으로 모든 entity local_id 와 relation source/target 에
+  `{chunk_id}__{local_id}` prefix 부여 — global 유일성 확보.
+- chunk_id 없는 호출 (5/10 두산밥캣 1청크 sanity) 은 기존 동작 그대로.
+
 관련 이슈: #13 (본 작업), #8 (Spike), #14 (NED — 본 모듈의 출력이 입력),
-          #24 (Opik 트레이싱).
+          #17 (8문서 일괄), #24 (Opik 트레이싱).
 """
 
 from __future__ import annotations
@@ -68,6 +75,11 @@
 SYSTEM_PROMPT_PATH       = PROMPTS_DIR / "entity_system_v1.md"
 EXTRACT_PROMPT_PATH      = PROMPTS_DIR / "entity_extract_v1.md"
 
+# local_id global prefix 구분자.
+# `{chunk_id}__{local_id}` 의 `__` — '_' 단일은 chunk_id 안에 흔히 등장 (예:
+# `doc_abc:c0001`) 하므로 충돌 회피 위해 이중 underscore.
+_GLOBAL_ID_SEP = "__"
+
 
 # ── 모듈 레벨 Semaphore (이벤트 루프 생명주기 공유) ──────────
 _sem: asyncio.Semaphore | None = None
@@ -118,6 +130,66 @@ def _parse_extraction_json(raw: str) -> dict[str, Any]:
         return json.loads(fixed)  # 두 번째도 실패하면 호출자가 잡음
 
 
+# ── chunk_id prefix 부여 (옵션 3 — #17) ──────────────────────
+def make_global_id(chunk_id: str, local_id: str) -> str:
+    """`{chunk_id}__{local_id}` 형식의 global id 생성.
+
+    역방향 파싱 (`parse_global_id`) 으로 어느 청크에서 나온 entity 인지
+    추적 가능 — Layer A 의 `(:Chunk) -[:MENTIONS]-> (:Entity)` 매핑에 사용.
+    """
+    return f"{chunk_id}{_GLOBAL_ID_SEP}{local_id}"
+
+
+def parse_global_id(global_id: str) -> tuple[str | None, str]:
+    """global id 를 (chunk_id, local_id) 로 분해.
+
+    prefix 가 없는 (구식) id 는 (None, id) 로 반환 — 5/10 두산밥캣 1청크
+    sanity 와 호환.
+    """
+    if _GLOBAL_ID_SEP in global_id:
+        chunk_id, local_id = global_id.split(_GLOBAL_ID_SEP, 1)
+        return chunk_id, local_id
+    return None, global_id
+
+
+def _apply_chunk_id_prefix(
+    result: ExtractionResult, chunk_id: str
+) -> ExtractionResult:
+    """ExtractionResult 의 모든 local_id / source / target 에 chunk_id prefix.
+
+    `_coerce_to_result` 가 source/target 의 local_id 유효성 검증을 마친 후
+    호출되므로, 본 함수는 매핑만 안전하게 수행.
+    """
+    if not chunk_id:
+        return result
+
+    # entity: local_id 만 prefix 부여
+    new_entities = [
+        ExtractedEntity(
+            local_id=make_global_id(chunk_id, e.local_id),
+            type=e.type,
+            canonical=e.canonical,
+            source_span=e.source_span,
+            section=e.section,
+        )
+        for e in result.entities
+    ]
+
+    # relation: source / target 둘 다 prefix (같은 청크 내 관계만 추출되므로
+    # 동일 chunk_id 로 prefix).
+    new_relations = [
+        ExtractedRelation(
+            source=make_global_id(chunk_id, r.source),
+            type=r.type,
+            target=make_global_id(chunk_id, r.target),
+            evidence=r.evidence,
+        )
+        for r in result.relations
+    ]
+
+    return ExtractionResult(entities=new_entities, relations=new_relations)
+
+
 # ── 응답 → Pydantic 변환 ─────────────────────────────────────
 def _coerce_to_result(data: dict[str, Any], section: str | None) -> ExtractionResult:
     """JSON dict → ExtractionResult.
@@ -214,12 +286,14 @@ async def _extract_chunk(
 ) -> ExtractionResult:
     """Chunk → ExtractionResult.
 
-    chunk dict 예상 키: doc_id, section, text. 결측 시 placeholder 로 graceful.
-    Semaphore 로 동시성 제한 + asyncio.to_thread 로 sync LLMClient 감쌈.
+    chunk dict 예상 키: doc_id, section, text, (optional) chunk_id.
+    결측 시 placeholder 로 graceful.
+    chunk_id 가 있으면 결과의 local_id 들에 자동으로 prefix 부여 (#17).
     """
-    doc_id  = chunk.get("doc_id", "(unknown)")
-    section = chunk.get("section") or "(no section)"
-    text    = (chunk.get("text") or "").strip()
+    doc_id   = chunk.get("doc_id", "(unknown)")
+    section  = chunk.get("section") or "(no section)"
+    text     = (chunk.get("text") or "").strip()
+    chunk_id = chunk.get("chunk_id")  # #17 — global id prefix 용 (없으면 prefix 안 함)
 
     if not text:
         logger.info("빈 청크 skip (doc=%s section=%s)", doc_id, section)
@@ -251,9 +325,15 @@ async def _extract_chunk(
         return ExtractionResult()
 
     result = _coerce_to_result(data, section=section)
+
+    # #17 — chunk_id 가 주어지면 global id prefix. 없으면 (5/10 sanity) 원본 그대로.
+    if chunk_id:
+        result = _apply_chunk_id_prefix(result, chunk_id)
+
     logger.info(
-        "추출 OK doc=%s section=%s entities=%d relations=%d (%.2fs)",
-        doc_id, section, len(result.entities), len(result.relations), elapsed,
+        "추출 OK doc=%s chunk=%s section=%s entities=%d relations=%d (%.2fs)",
+        doc_id, chunk_id or "(no-id)", section,
+        len(result.entities), len(result.relations), elapsed,
     )
     return result
 
@@ -268,9 +348,10 @@ async def extract(
 
     - 청크별로 병렬 추출 (Semaphore 제한)
     - 결과는 entities/relations 모두 평탄화하여 합침
-    - local_id 는 청크 내에서만 유효 → 호출자(NED #14)가 청크 경계 알아야 함
-      대안: prefix 부여 (`{chunk_idx}__{local_id}`) — 본 작업은 단순 합침으로,
-      NED 단계에서 청크 경계 처리.
+    - **chunk dict 에 옵셔널 `chunk_id` 키가 있으면** 본 모듈이 자동으로
+      모든 entity local_id 와 relation source/target 에
+      `{chunk_id}__{local_id}` prefix 부여 → 청크 간 global 유일성 (#17).
+      없으면 (5/10 sanity 호환) 원본 local_id 유지.
 
     DoD (#13):
     - 빈 청크 / 인식 실패 청크 graceful (KeyError 등 X)
@@ -288,9 +369,11 @@ async def extract(
     extract_tmpl   = _load_prompt(EXTRACT_PROMPT_PATH)
 
     started = time.perf_counter()
+    has_chunk_id = any(c.get("chunk_id") for c in chunks)
     logger.info(
-        "Entity 추출 시작 — 청크 %d개  MAX_CONCURRENT=%d  model=%s",
+        "Entity 추출 시작 — 청크 %d개  MAX_CONCURRENT=%d  model=%s  global_id=%s",
         len(chunks), MAX_CONCURRENT, llm.config.model,
+        "ON" if has_chunk_id else "off",
     )
 
     results: list[ExtractionResult] = await asyncio.gather(
diff --git a/scripts/run_w3_batch.py b/scripts/run_w3_batch.py
new file mode 100644
index 0000000..cc52d45
--- /dev/null
+++ b/scripts/run_w3_batch.py
@@ -0,0 +1,258 @@
+"""W3 batch — 평가 셋 8문서 일괄 적재.
+
+`eval/dataset/documents/` 의 모든 파일을 순회하며 풀 파이프라인:
+  parse_document (Layer A 구조화) → extract (Kimi) → link_entities (bge-m3)
+  → build_layer_a → build_layer_b → link_chunks_to_entities (MENTIONS)
+
+문서별 통계 (청크 / Entity / Relation / 소요시간 / 실패 여부) 수집 후
+마지막에 표로 출력. weekly-log 박제 + 5/23 발표 자료 § 운영 데이터 시드.
+
+사용:
+    uv run python -m scripts.run_w3_batch
+    uv run python -m scripts.run_w3_batch --limit 2  # 첫 2문서만 (sanity)
+    uv run python -m scripts.run_w3_batch --pattern "*.pdf"  # PDF 만
+
+전제조건:
+    .env 에 KIMI_API_KEY / NEO4J_URI / NEO4J_PASSWORD / OPIK_*
+    eval/dataset/documents/ 에 평가 셋 8문서 (gitignore — 로컬만)
+    bge-m3 최초 1회 다운로드 완료 상태
+
+#17 시행착오 / 운영 데이터:
+- 본 스크립트 출력 표를 docs/weekly-log/2026-05-16-batch-ingest.md 에 박제
+- Opik UI 에서 trace 페이지 캡처 → 발표 § How 슬라이드 시드
+
+관련 이슈: #17 (본 작업), #13 #14 #15 (1청크 검증 끝), #24 (Opik on).
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import sys
+import time
+import traceback
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from ingestion.adapter import parse_document
+from ingestion.models import ChunkType, Document
+from kg.builder import build_layer_a, build_layer_b, link_chunks_to_entities
+from kg.extractor import extract
+from kg.linking import link_entities
+from kg.neo4j_client import Neo4jClient
+from kg.ontology import ExtractionResult
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+logger = logging.getLogger("w3-batch")
+
+DOCS_DIR = Path("eval/dataset/documents")
+
+
+@dataclass
+class DocStat:
+    """문서별 처리 통계 — 마지막에 표로 출력."""
+
+    filename:           str
+    source_format:      str
+    doc_type:           str
+    success:            bool             = False
+    error:              str | None       = None
+
+    sections:           int              = 0
+    chunks_text:        int              = 0
+    chunks_table:       int              = 0
+    entities_raw:       int              = 0
+    entities_grouped:   int              = 0
+    relations:          int              = 0
+    mentions:           int              = 0
+
+    sec_parse:          float            = 0.0
+    sec_extract:        float            = 0.0
+    sec_linking:        float            = 0.0
+    sec_neo4j:          float            = 0.0
+    sec_total:          float            = 0.0
+
+
+def _doc_to_extract_chunks(doc: Document) -> list[dict]:
+    """Layer A Document → extract() 가 받는 chunk dict 리스트.
+
+    옵션 3 (#17): chunk_id 키를 함께 박아 보내 entity local_id 에 global
+    prefix 가 자동 부여되도록.
+    """
+    out: list[dict] = []
+    for section in doc.sections:
+        for chunk in section.chunks:
+            # Table 은 별도 Layer A 노드. extractor 는 text 청크만 받는 게 단순.
+            if chunk.chunk_type != ChunkType.TEXT:
+                continue
+            text = (chunk.text or "").strip()
+            if not text:
+                continue
+            out.append({
+                "doc_id":   doc.id,
+                "chunk_id": chunk.id,           # ← #17 옵션 3 핵심 키
+                "section":  section.original_section,
+                "text":     text,
+            })
+    return out
+
+
+async def _process_one(
+    path: Path,
+    client: Neo4jClient,
+    *,
+    create_indexes: bool,
+) -> DocStat:
+    """1 문서 풀 파이프라인. 실패해도 예외 안 던지고 stat 에 박제."""
+    started_total = time.perf_counter()
+
+    stat = DocStat(
+        filename=path.name,
+        source_format=path.suffix.lstrip(".").lower(),
+        doc_type="(unknown)",
+    )
+
+    try:
+        # 1. Parse — Layer A 구조화
+        t0 = time.perf_counter()
+        document = parse_document(path)
+        stat.sec_parse = time.perf_counter() - t0
+        stat.doc_type  = document.doc_type.value
+        stat.sections  = len(document.sections)
+        for sec in document.sections:
+            stat.chunks_text  += len(sec.chunks)
+            stat.chunks_table += len(sec.tables)
+
+        logger.info(
+            "[%s] Parse OK sections=%d chunks(text)=%d tables=%d (%.2fs)",
+            path.name, stat.sections, stat.chunks_text, stat.chunks_table, stat.sec_parse,
+        )
+
+        # 2. Extract — Kimi
+        chunks = _doc_to_extract_chunks(document)
+        if not chunks:
+            logger.warning("[%s] text 청크 없음 — extract skip", path.name)
+            ext_result = ExtractionResult()
+        else:
+            t0 = time.perf_counter()
+            ext_result = await extract(chunks)
+            stat.sec_extract = time.perf_counter() - t0
+        stat.entities_raw = len(ext_result.entities)
+        stat.relations    = len(ext_result.relations)
+
+        # 3. Linking — bge-m3 NED
+        if ext_result.entities:
+            t0 = time.perf_counter()
+            link_result = link_entities(ext_result.entities)
+            stat.sec_linking = time.perf_counter() - t0
+            stat.entities_grouped = link_result.grouped_count
+        else:
+            from kg.linking import LinkingResult
+            link_result = LinkingResult(groups=[], raw_count=0, grouped_count=0)
+
+        # 4. Neo4j 적재 — Layer A + Layer B + MENTIONS
+        t0 = time.perf_counter()
+        build_layer_a(document, client=client, create_indexes=create_indexes)
+        build_layer_b(link_result, ext_result.relations, client=client, create_indexes=False)
+        mentions_result = link_chunks_to_entities(link_result, client=client)
+        stat.sec_neo4j = time.perf_counter() - t0
+        stat.mentions = mentions_result.mentions_merged
+
+        stat.success = True
+
+    except Exception as exc:
+        stat.error = f"{type(exc).__name__}: {exc}"
+        logger.error("[%s] 실패: %s", path.name, stat.error)
+        logger.debug(traceback.format_exc())
+
+    stat.sec_total = time.perf_counter() - started_total
+    return stat
+
+
+def _print_table(stats: list[DocStat]) -> None:
+    """결과 표를 stdout 에 출력. weekly-log 에 그대로 복사 가능한 markdown."""
+    print()
+    print("## 8문서 일괄 적재 결과")
+    print()
+    print("| 파일 | 포맷 | 타입 | sec | chunks(T/B) | ent(raw→grp) | rel | mentions | parse | extract | link | neo4j | total | ok |")
+    print("|---|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---|")
+    for s in stats:
+        ok = "✅" if s.success else "❌"
+        rel_str = str(s.relations) if s.success else (s.error or "")[:30]
+        print(
+            f"| {s.filename} | {s.source_format} | {s.doc_type} | "
+            f"{s.sections} | {s.chunks_text}/{s.chunks_table} | "
+            f"{s.entities_raw}→{s.entities_grouped} | "
+            f"{s.relations} | {s.mentions} | "
+            f"{s.sec_parse:.1f} | {s.sec_extract:.1f} | "
+            f"{s.sec_linking:.1f} | {s.sec_neo4j:.1f} | "
+            f"{s.sec_total:.1f} | {ok} |"
+        )
+    print()
+
+    ok_count = sum(1 for s in stats if s.success)
+    print(f"총 {len(stats)}문서 · 성공 {ok_count} · 실패 {len(stats) - ok_count}")
+    print(f"전체 소요: {sum(s.sec_total for s in stats):.1f}s")
+    print(f"총 청크 (text): {sum(s.chunks_text for s in stats)}")
+    print(f"총 entity raw: {sum(s.entities_raw for s in stats)}")
+    print(f"총 entity grouped: {sum(s.entities_grouped for s in stats)}")
+    print(f"총 relation: {sum(s.relations for s in stats)}")
+    print(f"총 MENTIONS: {sum(s.mentions for s in stats)}")
+
+
+async def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--limit", type=int, default=None,
+        help="처음 N 문서만 처리 (sanity 용). 기본: 전부.",
+    )
+    parser.add_argument(
+        "--pattern", type=str, default="*",
+        help="파일명 glob 패턴 (예: '*.pdf'). 기본: 전체.",
+    )
+    parser.add_argument(
+        "--docs-dir", type=Path, default=DOCS_DIR,
+        help="문서 디렉토리. 기본: eval/dataset/documents/",
+    )
+    args = parser.parse_args()
+
+    if not args.docs_dir.exists():
+        logger.error("문서 디렉토리 없음: %s", args.docs_dir)
+        return 1
+
+    files = sorted(args.docs_dir.glob(args.pattern))
+    files = [p for p in files if p.is_file()]
+    if args.limit:
+        files = files[: args.limit]
+
+    if not files:
+        logger.error("처리할 파일 없음 (dir=%s pattern=%s)", args.docs_dir, args.pattern)
+        return 1
+
+    logger.info("=== 8문서 일괄 적재 시작 — %d 파일 ===", len(files))
+    for p in files:
+        logger.info("  - %s", p.name)
+
+    stats: list[DocStat] = []
+
+    with Neo4jClient() as client:
+        client.verify_connectivity()
+
+        for idx, path in enumerate(files):
+            logger.info("\n=== [%d/%d] %s ===", idx + 1, len(files), path.name)
+            # 첫 문서만 인덱스 생성 시도 (멱등이라 매번 호출해도 OK 지만 절약)
+            stat = await _process_one(
+                path, client=client, create_indexes=(idx == 0),
+            )
+            stats.append(stat)
+
+    _print_table(stats)
+    return 0 if all(s.success for s in stats) else 1
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))
diff --git a/tests/kg/test_builder_layer_a.py b/tests/kg/test_builder_layer_a.py
new file mode 100644
index 0000000..5074a78
--- /dev/null
+++ b/tests/kg/test_builder_layer_a.py
@@ -0,0 +1,389 @@
+"""kg/builder.py 의 Layer A 함수 + MENTIONS 매핑 단위 테스트.
+
+검증 포인트 (#17):
+- build_layer_a: Document/Section/Chunk/Table MERGE 호출 + 관계 4종
+- 청크 NEXT 관계: Section 내 순서대로만 (Section 경계 넘지 않음)
+- link_chunks_to_entities: parse_global_id 로 chunk_id 추출 후 MENTIONS
+- chunk_id 없는 구식 local_id 는 graceful drop (5/10 호환)
+
+테스트 방식: mock Neo4jClient — 실제 DB 없이 client.write() 호출 인자 검증.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+import pytest
+
+from ingestion.models import (
+    Chunk as LayerAChunk,
+    ChunkType,
+    DocType,
+    Document,
+    Section,
+    SectionLabel,
+    SourceFormat,
+    Table,
+)
+from kg.builder import build_layer_a, link_chunks_to_entities
+from kg.linking import EntityGroup, LinkingResult
+from kg.ontology import EntityType, ExtractedEntity
+
+
+# ── Mock Neo4jClient ──────────────────────────────────────
+@dataclass
+class _MockClient:
+    """Neo4jClient 인터페이스 흉내 — write() 호출을 기록만 함."""
+
+    writes: list[tuple[str, dict]] = field(default_factory=list)
+
+    def write(self, query: str, **params) -> list:
+        self.writes.append((query, params))
+        return []
+
+    def read(self, query: str, **params) -> list:
+        return []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return False
+
+
+def _writes_with_keyword(client: _MockClient, keyword: str) -> list[tuple[str, dict]]:
+    """write 호출 중 query 에 keyword 포함된 것만."""
+    return [(q, p) for q, p in client.writes if keyword in q]
+
+
+# ── 픽스처: 두산밥캣 mini Document ──────────────────────────
+@pytest.fixture
+def doosan_doc() -> Document:
+    """1 Document + 2 Section + 3 Chunk + 1 Table 의 미니 문서."""
+    section1 = Section(
+        id="doc1:s000",
+        original_section="3분기 실적",
+        doc_type=DocType.REPORT,
+        label=SectionLabel.PERFORMANCE,
+        heading_level=2,
+        order_index=0,
+        page_start=3,
+        page_end=4,
+        chunks=[
+            LayerAChunk(
+                id="doc1:c0000",
+                text="두산밥캣 영업이익 35% 감소.",
+                chunk_type=ChunkType.TEXT,
+                char_count=20,
+                page=3,
+                order_index=0,
+            ),
+            LayerAChunk(
+                id="doc1:c0001",
+                text="북미 고금리 장기화가 원인.",
+                chunk_type=ChunkType.TEXT,
+                char_count=18,
+                page=3,
+                order_index=1,
+            ),
+        ],
+        tables=[
+            Table(
+                id="doc1:t000",
+                raw_markdown="| 항목 | 값 |\n|---|---|\n| 영업이익 | 1.2조 |",
+                row_count=3,
+                column_count=2,
+                page=4,
+                order_index=0,
+            ),
+        ],
+    )
+    section2 = Section(
+        id="doc1:s001",
+        original_section="전망",
+        doc_type=DocType.REPORT,
+        label=SectionLabel.OUTLOOK,
+        heading_level=2,
+        order_index=1,
+        chunks=[
+            LayerAChunk(
+                id="doc1:c0002",
+                text="2026 회복 전망.",
+                chunk_type=ChunkType.TEXT,
+                char_count=10,
+                page=5,
+                order_index=2,
+            ),
+        ],
+    )
+    return Document(
+        id="doc1",
+        filename="한화_두산밥캣.pdf",
+        source_format=SourceFormat.PDF,
+        doc_type=DocType.REPORT,
+        total_pages=5,
+        publisher="한화투자증권",
+        subject="두산밥캣",
+        fiscal_year=2024,
+        sections=[section1, section2],
+    )
+
+
+# ── build_layer_a 테스트 ──────────────────────────────────
+def test_build_layer_a_node_counts(doosan_doc: Document):
+    """Document 1 + Section 2 + Chunk 3 + Table 1 노드 MERGE 발생."""
+    client = _MockClient()
+    result = build_layer_a(doosan_doc, client=client, create_indexes=False)
+
+    assert result.documents_merged == 1
+    assert result.sections_merged == 2
+    assert result.chunks_merged == 3
+    assert result.tables_merged == 1
+
+
+def test_build_layer_a_document_params(doosan_doc: Document):
+    """Document MERGE 에 publisher / fiscal_year 등 메타 정확히 전달."""
+    client = _MockClient()
+    build_layer_a(doosan_doc, client=client, create_indexes=False)
+
+    doc_writes = _writes_with_keyword(client, "MERGE (d:Document")
+    assert len(doc_writes) == 1
+    _, params = doc_writes[0]
+    assert params["id"] == "doc1"
+    assert params["filename"] == "한화_두산밥캣.pdf"
+    assert params["doc_type"] == "report"
+    assert params["source_format"] == "pdf"
+    assert params["publisher"] == "한화투자증권"
+    assert params["fiscal_year"] == 2024
+
+
+def test_build_layer_a_section_label_serialized(doosan_doc: Document):
+    """Section.label (enum) → 문자열 값으로 직렬화."""
+    client = _MockClient()
+    build_layer_a(doosan_doc, client=client, create_indexes=False)
+
+    section_writes = _writes_with_keyword(client, "MERGE (s:Section")
+    assert len(section_writes) == 2
+    labels = {p["label"] for _, p in section_writes}
+    assert labels == {"Performance", "Outlook"}
+
+
+def test_build_layer_a_section_label_none_safe():
+    """Section.label 이 None 이어도 적재 가능 (W3 미분류 청크 대응)."""
+    doc = Document(
+        id="d", filename="t.pdf", source_format=SourceFormat.PDF,
+        doc_type=DocType.IR, total_pages=1,
+        sections=[
+            Section(
+                id="s", original_section="raw", doc_type=DocType.IR,
+                label=None,  # ← 미분류
+            ),
+        ],
+    )
+    client = _MockClient()
+    build_layer_a(doc, client=client, create_indexes=False)
+    section_writes = _writes_with_keyword(client, "MERGE (s:Section")
+    _, params = section_writes[0]
+    assert params["label"] is None
+
+
+def test_build_layer_a_next_relations_only_within_section(doosan_doc: Document):
+    """NEXT 관계는 Section 내에서만. Section 경계 넘지 않음."""
+    client = _MockClient()
+    result = build_layer_a(doosan_doc, client=client, create_indexes=False)
+
+    # section1: chunks c0000 → c0001 (1개), section2: c0002 단독 (0개)
+    assert result.next_relations == 1
+
+    next_writes = _writes_with_keyword(client, "[:NEXT]")
+    assert len(next_writes) == 1
+    _, params = next_writes[0]
+    assert params["prev_id"] == "doc1:c0000"
+    assert params["next_id"] == "doc1:c0001"
+
+
+def test_build_layer_a_table_attached_to_section(doosan_doc: Document):
+    """CONTAINS_TABLE 관계가 올바른 Section 에 연결."""
+    client = _MockClient()
+    build_layer_a(doosan_doc, client=client, create_indexes=False)
+
+    table_writes = _writes_with_keyword(client, "[:CONTAINS_TABLE]")
+    assert len(table_writes) == 1
+    _, params = table_writes[0]
+    assert params["section_id"] == "doc1:s000"
+    assert params["table_id"] == "doc1:t000"
+
+
+def test_build_layer_a_idempotent_uses_merge(doosan_doc: Document):
+    """모든 쿼리가 MERGE 기반 (idempotent). CREATE 단독 없음."""
+    client = _MockClient()
+    build_layer_a(doosan_doc, client=client, create_indexes=False)
+
+    for query, _ in client.writes:
+        # CREATE CONSTRAINT/INDEX 는 create_indexes=False 라 호출 안 됨
+        assert "MERGE" in query
+
+
+# ── link_chunks_to_entities 테스트 ────────────────────────
+def _make_group(group_id: str, type_: EntityType, members: list[ExtractedEntity]) -> EntityGroup:
+    return EntityGroup(
+        group_id=group_id,
+        type=type_,
+        representative_name=members[0].canonical if members else "",
+        members=members,
+    )
+
+
+def test_link_chunks_to_entities_basic():
+    """각 EntityGroup.member 마다 chunk_id 를 추출해 MENTIONS MERGE."""
+    linking = LinkingResult(
+        groups=[
+            _make_group(
+                "grp_001",
+                EntityType.COMPANY,
+                [
+                    ExtractedEntity(
+                        local_id="doc1:c0000__ent_001",
+                        type=EntityType.COMPANY,
+                        canonical="두산밥캣",
+                        source_span="두산밥캣",
+                        section="3분기 실적",
+                    ),
+                ],
+            ),
+            _make_group(
+                "grp_002",
+                EntityType.METRIC,
+                [
+                    ExtractedEntity(
+                        local_id="doc1:c0000__ent_002",
+                        type=EntityType.METRIC,
+                        canonical="영업이익 -35%",
+                        source_span="영업이익이 전년 동기 대비 35% 감소",
+                        section="3분기 실적",
+                    ),
+                ],
+            ),
+        ],
+        raw_count=2,
+        grouped_count=2,
+    )
+
+    client = _MockClient()
+    result = link_chunks_to_entities(linking, client=client)
+
+    assert result.mentions_merged == 2
+    assert result.mentions_dropped == 0
+
+    # 각 MENTIONS write 가 올바른 chunk_id + group_id 를 가지는지
+    mentions = _writes_with_keyword(client, "[:MENTIONS]")
+    assert len(mentions) == 2
+    pairs = {(p["chunk_id"], p["group_id"]) for _, p in mentions}
+    assert pairs == {
+        ("doc1:c0000", "grp_001"),
+        ("doc1:c0000", "grp_002"),
+    }
+
+
+def test_link_chunks_to_entities_dedupe_same_chunk():
+    """한 그룹 내 같은 청크의 member 가 여러 개여도 청크당 1회만 MENTIONS."""
+    linking = LinkingResult(
+        groups=[
+            _make_group(
+                "grp_001",
+                EntityType.COMPANY,
+                [
+                    # 같은 청크에서 같은 그룹의 alias 가 2번 등장 (드물지만 가능)
+                    ExtractedEntity(
+                        local_id="doc1:c0000__ent_001",
+                        type=EntityType.COMPANY, canonical="두산밥캣",
+                        source_span="두산밥캣", section="3분기",
+                    ),
+                    ExtractedEntity(
+                        local_id="doc1:c0000__ent_003",
+                        type=EntityType.COMPANY, canonical="Doosan Bobcat",
+                        source_span="Doosan Bobcat", section="3분기",
+                    ),
+                ],
+            ),
+        ],
+        raw_count=2,
+        grouped_count=1,
+    )
+
+    client = _MockClient()
+    result = link_chunks_to_entities(linking, client=client)
+
+    # 청크 1개, 그룹 1개 → MENTIONS 1번만
+    assert result.mentions_merged == 1
+    mentions = _writes_with_keyword(client, "[:MENTIONS]")
+    assert len(mentions) == 1
+
+
+def test_link_chunks_to_entities_multi_chunk_same_group():
+    """다른 청크의 member 들은 각각 MENTIONS — 청크당 1번씩."""
+    linking = LinkingResult(
+        groups=[
+            _make_group(
+                "grp_001",
+                EntityType.COMPANY,
+                [
+                    ExtractedEntity(
+                        local_id="doc1:c0000__ent_001",
+                        type=EntityType.COMPANY, canonical="미래에셋",
+                        source_span="미래에셋", section="1Q",
+                    ),
+                    ExtractedEntity(
+                        local_id="doc1:c0010__ent_001",
+                        type=EntityType.COMPANY, canonical="미래에셋증권",
+                        source_span="미래에셋증권", section="2Q",
+                    ),
+                    ExtractedEntity(
+                        local_id="doc1:c0020__ent_002",
+                        type=EntityType.COMPANY, canonical="미래에셋",
+                        source_span="미래에셋", section="3Q",
+                    ),
+                ],
+            ),
+        ],
+        raw_count=3,
+        grouped_count=1,
+    )
+
+    client = _MockClient()
+    result = link_chunks_to_entities(linking, client=client)
+
+    # 3개 청크 → 3개 MENTIONS
+    assert result.mentions_merged == 3
+    mentions = _writes_with_keyword(client, "[:MENTIONS]")
+    chunk_ids = sorted({p["chunk_id"] for _, p in mentions})
+    assert chunk_ids == ["doc1:c0000", "doc1:c0010", "doc1:c0020"]
+
+
+def test_link_chunks_to_entities_legacy_local_id_dropped():
+    """chunk_id prefix 없는 구식 local_id 는 graceful drop (5/10 sanity 호환)."""
+    linking = LinkingResult(
+        groups=[
+            _make_group(
+                "grp_001",
+                EntityType.COMPANY,
+                [
+                    ExtractedEntity(
+                        local_id="ent_001",  # ← prefix 없음
+                        type=EntityType.COMPANY, canonical="두산밥캣",
+                        source_span="두산밥캣", section="3분기",
+                    ),
+                ],
+            ),
+        ],
+        raw_count=1,
+        grouped_count=1,
+    )
+
+    client = _MockClient()
+    result = link_chunks_to_entities(linking, client=client)
+
+    assert result.mentions_merged == 0
+    assert result.mentions_dropped == 1
+    # 실제 write 호출 없음
+    assert _writes_with_keyword(client, "[:MENTIONS]") == []