feat(safety): B-4 PR①② — licensed_restricted 차단 술어 + watch 폴더 license 주입

PR① licensed_restricted 단일 술어(_license_sql) — retrieval 3-leg(text/vec-doc/
vec-chunk) + digest loader 공유. a안(U-2①): 색인 허용·구매자료 verbatim 을 RAG 증거/
digest 발행에서 구조적 제외. 술어=COALESCE(extract_meta->'license'->>'restricted',
'false')<>'true' (restricted 부재/false 미제외 → 기존 코퍼스 결과 불변). 개인 파일
열람 미차단. chunk leg 는 outer 의 documents JOIN(항상) 활용 post-rank(restricted 소수).
PR② file_watcher _TARGET_AXIS 확장 — Books/Papers_Purchased=restricted / Manuals=
non-restricted(사용자 결정) / KGS=law·KR·kogl. ingest 시 extract_meta.license
deterministic 주입(classify material IS NULL 일 때만 제안·meta 미기록=보존).
PR③(KGS 버전 flip)=별 슬라이스 deferred(파일 포맷 조사 선행).

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

app/services/digest/loader.py

@@ -41,6 +41,9 @@ _NEWS_WINDOW_SQL = text("""
       AND d.created_at < :window_end
       AND d.embedding IS NOT NULL
       AND d.ai_summary IS NOT NULL
+      -- 안전 자료실 B-4: licensed_restricted 발행 차단 (retrieval_service._license_sql 와 동일 술어).
+      -- news 채널엔 현재 restricted 부재 = 방어적 게이트(미래 유료 news 소스 대비, 경로 누락 방지).
+      AND COALESCE(d.extract_meta -> 'license' ->> 'restricted', 'false') <> 'true'
 """)
 
 

app/services/search/retrieval_service.py

@@ -139,6 +139,22 @@ def _axis_sql(alias: str, af: "AxisFilter | None", params: dict) -> str:
     return " AND " + " AND ".join(cl)
 
 
+# ─── 안전 자료실 B-4: licensed_restricted 단일 술어 (a안 U-2① — 항상 적용) ──────
+def _license_sql(alias: str) -> str:
+    """licensed_restricted(extract_meta.license.restricted=true) 문서를 retrieval 에서 제외.
+
+    a안: 색인은 허용하되, 구매 전자책/유료자료의 verbatim span 이 RAG 증거·digest 발행에
+    들어가는 경로를 구조적으로 차단. 이 단일 술어를 모든 retrieval leg + digest loader 가
+    공유 — 경로별 가드 누락 방지([[feedback_structural_integrity_over_path_discipline]]).
+    개인 파일 열람(GET /documents/{id}?download)은 a안상 허용이라 미적용.
+
+    axis 필터(조건부)와 달리 항상 적용. restricted 부재/false = COALESCE 로 미제외 →
+    기존 코퍼스(restricted=true 0건)에서 결과 불변. 반환 ' AND ...' (alias='' = 컬럼 직접).
+    """
+    p = (alias + ".") if alias else ""
+    return f" AND COALESCE({p}extract_meta -> 'license' ->> 'restricted', 'false') <> 'true'"
+
+
 # 2단계 gate (R2-B1) — SQL string interpolation 직전 final allowlist.
 _VALID_DOCS_TABLE = re.compile(r"^(documents|documents_cand_[a-z0-9_]+)$")
 # corpus_chunks = document_chunks WHERE in_corpus=true 뷰 (Hier-Decomp-1 c2 choke point).
@@ -308,7 +324,8 @@ async def search_text(
     await session.execute(text("SELECT set_limit(0.15)"))
 
     _params: dict[str, Any] = {"q": query, "limit": limit}
-    _where = _axis_sql("d", axis, _params)   # 미지정 시 '' (byte 불변)
+    # license(항상) + axis(조건부). license 가 항상 ' AND ...' 이라 WHERE 는 늘 존재.
+    _where = _license_sql("d") + _axis_sql("d", axis, _params)
 
     result = await session.execute(
         text(f"""
@@ -368,7 +385,7 @@ async def search_text(
                    d.material_type, d.jurisdiction, d.published_date
             FROM documents d
             JOIN candidates c ON d.id = c.id
-            {("WHERE" + _where[4:]) if _where else ""}
+            WHERE{_where[4:]}
             ORDER BY score DESC
             LIMIT :limit
         """),
@@ -514,6 +531,7 @@ async def _search_vector_docs(
             snapshot_clause = " AND id <= :snapshot_doc_id_max"
             params["snapshot_doc_id_max"] = snapshot_doc_id_max
         axis_clause = _axis_sql("", axis, params)   # alias 없음 (단일 FROM documents)
+        license_clause = _license_sql("")            # B-4: restricted 항상 제외
         sql = f"""
             SELECT id, title, ai_domain, ai_summary, file_format,
                    (1 - (embedding <=> cast(:embedding AS vector))) AS score,
@@ -522,13 +540,14 @@ async def _search_vector_docs(
                    NULL::bigint AS chunk_id, NULL::integer AS chunk_index, NULL::text AS section_title,
                    material_type, jurisdiction, published_date
             FROM documents
-            WHERE embedding IS NOT NULL AND deleted_at IS NULL{snapshot_clause}{axis_clause}
+            WHERE embedding IS NOT NULL AND deleted_at IS NULL{snapshot_clause}{axis_clause}{license_clause}
             ORDER BY embedding <=> cast(:embedding AS vector)
             LIMIT :limit
         """
     else:
         # candidate: docs_table 은 (doc_id, embed_input, embed_input_hash, embedding) 만 보유 → JOIN documents
         axis_clause = _axis_sql("d", axis, params)
+        license_clause = _license_sql("d")           # B-4: restricted 항상 제외
         sql = f"""
             SELECT d.id, d.title, d.ai_domain, d.ai_summary, d.file_format,
                    (1 - (c.embedding <=> cast(:embedding AS vector))) AS score,
@@ -538,7 +557,7 @@ async def _search_vector_docs(
                    d.material_type, d.jurisdiction, d.published_date
             FROM {docs_table} c
             JOIN documents d ON d.id = c.doc_id
-            WHERE d.deleted_at IS NULL{axis_clause}
+            WHERE d.deleted_at IS NULL{axis_clause}{license_clause}
             ORDER BY c.embedding <=> cast(:embedding AS vector)
             LIMIT :limit
         """
@@ -589,6 +608,10 @@ async def _search_vector_chunks(
         chunk_join = ""
         chunk_axis = ""
 
+    # B-4: restricted 제외 — outer 가 documents d 를 항상 JOIN 하므로 post-rank 위치.
+    # restricted 는 소수(구매자료)라 inner topk 후 제외해도 candidate collapse 없음(axis 와 상이).
+    license_clause = _license_sql("d")
+
     sql = f"""
         WITH topk AS (
             SELECT c.id AS chunk_id, c.doc_id, c.chunk_index, c.section_title, c.text,
@@ -612,7 +635,7 @@ async def _search_vector_chunks(
                d.published_date AS published_date
         FROM ranked r
         JOIN documents d ON d.id = r.doc_id
-        WHERE r.rn <= 2 AND d.deleted_at IS NULL
+        WHERE r.rn <= 2 AND d.deleted_at IS NULL{license_clause}
         ORDER BY r.dist
         LIMIT :limit
     """

app/workers/file_watcher.py

@@ -58,11 +58,21 @@ SCAN_TARGETS: list[tuple[str, str | None]] = [
     ("Videos", "video"),
 ]
 
-# 안전 자료실 A-2 — additional watch 타깃별 자료유형 (폴더 = deterministic 축).
-# 키 = 타깃 경로의 마지막 성분. KGS Code = 법정 위임 상세기준 = law/KR (plan 0-1 확정).
-# B-4 에서 Books(book)/Manuals(manual)/Papers_Purchased(paper) + license 주입 표로 확장.
-_TARGET_MATERIAL: dict[str, tuple[str, str | None]] = {
-    "KGS_Code": ("law", "KR"),
+# 안전 자료실 A-2/B-4 — watch 타깃별 (material_type, jurisdiction, license) deterministic 축.
+# 키 = 타깃 경로의 마지막 성분. license = extract_meta.license 에 그대로 주입(None=미주입).
+#   restricted=true → retrieval_service._license_sql 가 RAG 증거·digest 에서 제외(a안 U-2① —
+#   구매자료 verbatim span 차단, 색인 자체는 허용. 개인 파일 열람은 미차단).
+# 사용자 결정(2026-06-13): Books/Papers=proprietary+restricted / Manuals=proprietary·restricted=false
+#   (검색·RAG 활용) / KGS=법정 위임 상세기준 law/KR·KOGL 공공·restricted 아님.
+_TARGET_AXIS: dict[str, tuple[str, str | None, dict | None]] = {
+    "KGS_Code": ("law", "KR", {"scheme": "kogl", "redistribute": True,
+                               "restricted": False, "attribution": "한국가스안전공사(KGS)"}),
+    "Books": ("book", None, {"scheme": "proprietary", "redistribute": False,
+                             "restricted": True, "attribution": "구매 도서"}),
+    "Papers_Purchased": ("paper", None, {"scheme": "proprietary", "redistribute": False,
+                                         "restricted": True, "attribution": "구매 논문"}),
+    "Manuals": ("manual", None, {"scheme": "proprietary", "redistribute": False,
+                                 "restricted": False, "attribution": "기술 매뉴얼"}),
 }
 
 
@@ -249,9 +259,9 @@ async def watch_inbox():
             if not scan_root.exists():
                 continue
 
-            # 안전 자료실 A-2 — 타깃 폴더 기반 자료유형 (없으면 (None, None))
-            target_mt, target_jur = _TARGET_MATERIAL.get(
-                Path(sub).name, (None, None)
+            # 안전 자료실 A-2/B-4 — 타깃 폴더 기반 (material, jurisdiction, license)
+            target_mt, target_jur, target_license = _TARGET_AXIS.get(
+                Path(sub).name, (None, None, None)
             )
 
             for file_path in scan_root.rglob("*"):
@@ -287,10 +297,14 @@ async def watch_inbox():
                         source_channel="drive_sync",
                         category=category,
                         needs_conversion=needs_conversion,
-                        # 안전 자료실 A-2 — watch 타깃 매핑 (KGS=law/KR 등, 비대상=NULL)
+                        # 안전 자료실 A-2/B-4 — watch 타깃 매핑 (KGS=law/KR 등, 비대상=NULL)
                         material_type=target_mt,
                         jurisdiction=target_jur,
                     )
+                    # B-4 — 타깃 폴더 license 주입(restricted 포함, 비대상=미주입). classify 는
+                    # material_type IS NULL 일 때만 제안 + extract_meta 미기록이라 주입 보존.
+                    if target_license:
+                        doc.extract_meta = {"license": dict(target_license)}
                     session.add(doc)
                     await session.flush()