feat(presegment): G2 후보 A — 자식 합성 file_path + bundle_source_path 실파일 해석

uq_documents_file_path 충돌 해소: 자식 file_path = unique 합성값 '{부모}#p{s}-{e}'
(UNIQUE 통과), 실파일은 bundle_source_path() 로 부모경로 복원(접미사 strip, 결정적).
- presegment_worker: bundle_source_path() 헬퍼 + 자식 합성 file_path
- extract_worker 자식분기: bundle_source_path + NFC/NFD resolve 로 실파일 range 추출
- marker_worker: container_path = bundle_source_path(file_path) (일반 doc 무변)
인제스트는 아직 extract(검증 후 재활성). 일반 doc = bundle_source_path no-op = 무회귀.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

app/workers/extract_worker.py

@@ -340,12 +340,18 @@ async def process(document_id: int, session: AsyncSession) -> None:
     # 밖(자식은 ToC 존재 = digital text layer 전제 → 대개 OCR 불필요). PyMuPDF 텍스트가 빈약해도
     # 그대로 보존하고 사유를 남긴다.
     if fmt == "pdf" and doc.bundle_page_start is not None and doc.bundle_page_end is not None:
-        if not full_path.exists():
-            raise FileNotFoundError(f"파일 없음: {full_path}")
+        # 후보 A: 자식 file_path 는 합성값(`{부모}#p{s}-{e}`) → 실파일 = bundle_source_path 로 부모경로
+        # 복원 + NFC/NFD resolve. (자식 file_path 는 디스크에 없음.)
+        from workers.presegment_worker import _resolve_path as _resolve_bundle_path
+        from workers.presegment_worker import bundle_source_path
+        real_rel = bundle_source_path(doc.file_path)
+        src = _resolve_bundle_path(str(Path(settings.nas_mount_path) / real_rel))
+        if src is None:
+            raise FileNotFoundError(f"번들 원본 파일 없음: {real_rel}")
         start, end = doc.bundle_page_start, doc.bundle_page_end
         try:
-            pymupdf_text = _extract_pdf_pymupdf(full_path, start, end)
-            page_count = _get_pdf_page_count(full_path, start, end)
+            pymupdf_text = _extract_pdf_pymupdf(src, start, end)
+            page_count = _get_pdf_page_count(src, start, end)
         except Exception as e:
             logger.error(f"[pymupdf:child] {doc.file_path} pages={start}-{end} 실패: {e}")
             raise

app/workers/marker_worker.py

@@ -185,7 +185,10 @@ async def process(document_id: int, session: AsyncSession) -> None:
         await _fail(session, document_id, "no file_path")
         return
 
-    container_path = _to_marker_path(doc.file_path)
+    # 후보 A: 자식(bundle cols)은 합성 file_path(`{부모}#p{s}-{e}`) → 실파일 = bundle_source_path
+    # 로 부모경로 복원. 일반 doc 은 그대로(접미사 없음). marker/mineru 는 실파일 + page 범위로 변환.
+    from workers.presegment_worker import bundle_source_path
+    container_path = _to_marker_path(bundle_source_path(doc.file_path))
     suffix = Path(container_path).suffix.lower()
 
     # ---- (3) office/hwp → md (C-2): PDF 외 지원 포맷은 office_md 하이브리드 변환 ----

app/workers/presegment_worker.py

@@ -9,26 +9,26 @@
   - page_count >= MIN_BUNDLE_PAGES AND level-1 ToC 항목 >= 2 AND 모든 자식 >= MIN_CHILD_PAGES
     AND 단조 증가·비중첩 AND [1, page_count] 전 범위 커버 AND 2 <= N <= MAX_CHILDREN.
 
-분할 시 Option A(파일 물리분할 없음): 자식은 부모 file_path 를 그대로 공유하고
-bundle_page_start/end(1-based inclusive) 로 자기 page 범위만 가리킨다. 부모-자식 관계 자체는
-document_lineage(relation_type='segmented_from'). 부모(presegment_role='parent')는 파일 홀더라
-자체 extract/embed 안 함 — enqueue_next_stage 의 presegment→extract 전이가 'parent' 면 억제된다
-(queue_consumer.enqueue_next_stage 참조). 자식의 extract 는 이 워커가 직접 enqueue 한다.
+분할 시 ★후보 A(물리분할 없음, uq_documents_file_path 해소): 자식 file_path = unique 합성값
+`{부모경로}#p{start}-{end}` (UNIQUE 제약 통과), 실파일은 `bundle_source_path()` 로 부모 경로 복원.
+자식은 bundle_page_start/end(1-based inclusive) 로 부모 파일의 자기 page 범위만 가리킨다.
+부모-자식 관계 정본 = document_lineage(relation_type='segmented_from'). 부모(presegment_role='parent')는
+파일 홀더라 자체 extract/embed 안 함 — enqueue_next_stage 의 presegment→extract 전이가 'parent' 면
+억제된다(queue_consumer 참조). 자식의 extract 는 이 워커가 직접 enqueue. extract_worker/marker_worker
+가 자식 처리 시 bundle_source_path() 로 실파일 접근.
 
 멱등: 재실행 시 같은 부모로 이미 자식이 있으면(document_lineage segmented_from) 재생성하지 않고
 수렴(각 자식이 extract 활성/완료 상태인지만 보장)한다.
 
-★★ BLOCKER (2026-06-18 실측): Option A(자식이 부모 file_path 공유)는 `uq_documents_file_path`
-UNIQUE 제약과 충돌 → 자식 INSERT UniqueViolation 으로 실패한다. 따라서 **현재 인제스트는 presegment
-를 enqueue 하지 않음**(documents.py/file_watcher.py 가 직접 extract). 본 워커는 재설계 전까지 미사용.
-재설계 후보: 자식 file_path=unique 합성값(`{parent}#p{s}-{e}`)+실파일은 부모(lineage source)에서 해석
-/ 또는 file_path NULL + 별도 bundle_source_path 컬럼. 결정 후 인제스트 재활성.
+★해결 이력 (2026-06-18): 최초 Option A(자식이 부모 file_path 그대로 공유)는 uq_documents_file_path
+UNIQUE 위반(실번들 검증서 발견) → 합성 file_path(후보 A)로 해소. 인제스트 재활성 = 합성번들 재검증 PASS 후.
 
 plan: G2 pre-segmentation (PR-G2-2 deterministic ToC segmentation)
 """
 
 import hashlib
 import os
+import re
 import unicodedata
 from pathlib import Path
 
@@ -82,6 +82,21 @@ def _to_container_path(file_path: str) -> str:
     return f"{CONTAINER_PATH_PREFIX}/{file_path}"
 
 
+# 후보 A: 자식 합성 file_path 패턴 `{부모경로}#p{start}-{end}` (uq_documents_file_path 유일성).
+_BUNDLE_SUFFIX_RE = re.compile(r"#p\d+-\d+$")
+
+
+def bundle_source_path(file_path: str | None) -> str | None:
+    """자식 합성 file_path → 부모 실파일 경로 복원. 일반 doc(접미사 없음)은 그대로 반환.
+
+    extract_worker/marker_worker 가 자식 처리 시 실제 파일 접근에 사용 (자식 file_path 는
+    합성값이라 디스크에 없음). 결정적·세션 불필요. lineage 가 부모-자식 관계의 정본 기록.
+    """
+    if not file_path:
+        return file_path
+    return _BUNDLE_SUFFIX_RE.sub("", file_path)
+
+
 def _is_pdf(doc: Document) -> bool:
     """PDF 판정 — file_format=pdf 또는 .pdf 확장자."""
     fmt = (doc.file_format or "").lower()
@@ -300,8 +315,10 @@ async def process(document_id: int, session: AsyncSession) -> None:
     for seg in segments:
         start, end = seg["start_page"], seg["end_page"]
         child = Document(
-            # Option A: 부모 파일 그대로 공유(물리 분할 없음). 자식은 bundle_page_start/end 로 슬라이스.
-            file_path=doc.file_path,
+            # 후보 A: 자식 file_path = unique 합성값 `{부모경로}#p{s}-{e}` (uq_documents_file_path
+            # 충돌 회피). 실파일은 bundle_source_path() 로 복원(부모 경로). 물리 분할 없음 —
+            # 자식은 bundle_page_start/end 로 부모 파일을 슬라이스.
+            file_path=f"{doc.file_path}#p{start}-{end}",
             file_hash=_child_file_hash(doc.file_hash, start, end),
             file_format=doc.file_format,
             file_size=doc.file_size,