Claude 리뷰 기반 운영 안정성 보강

.env.example

@@ -7,6 +7,12 @@ RELOAD=false
 ROOT_PATH=
 # 설정하면 /v1/health를 제외한 API 요청에 X-API-Key가 필요합니다.
 API_KEY=
+# 운영 fail-closed. true인데 API_KEY가 비면 앱이 기동에 실패합니다.
+# ⚠️ King 실제 활성화는 바이칼 측 키 전달 후 진행 (docs/AI_DETECTION.md 참조).
+REQUIRE_API_KEY=false
+# 인증 없이 열어둘 경로 제어
+PUBLIC_HEALTH=true
+PUBLIC_DOCS=true
 
 ENGINE_VERSION=o2o-plagiarism-2.1.0-kosimcse
 REFERENCE_CORPUS_DIR=./data/reference

Dockerfile

@@ -5,6 +5,7 @@ WORKDIR /app
 ENV PYTHONDONTWRITEBYTECODE=1 \
     PYTHONUNBUFFERED=1 \
     PIP_NO_CACHE_DIR=1 \
+    PYTHONPATH=/app \
     HOST=0.0.0.0 \
     PORT=8000
 

app/api/schemas.py

@@ -55,11 +55,33 @@ class DetectOptions(BaseModel):
     )
 
 
+class LegalContext(BaseModel):
+    """사람이 확인한 법적 사실. 제공하지 않으면 missing_factors 로 남는다 (#10).
+
+    엔진은 이 값들을 **추론하지 않는다.** 텍스트만으로는 알 수 없는 사실이므로,
+    검토자가 확인한 경우에만 전달받아 판례 매칭과 위험도 판단에 반영한다.
+    """
+    work_type: str = Field(
+        default="literary", description="저작물 유형 (literary/musical/visual 등)"
+    )
+    access_evidence: bool | None = Field(
+        default=None,
+        description="원저작물 접근·의거 가능성이 확인되었는지. None이면 미제공.",
+    )
+    protected_expression_reviewed: bool = Field(
+        default=False, description="보호되는 창작적 표현인지 사람이 검토했는지",
+    )
+    rights_verified: bool = Field(
+        default=False, description="저작권 귀속·이용허락·인용 요건이 확인되었는지",
+    )
+
+
 class DetectRequest(BaseModel):
     doc_id: str
     text: str = Field(..., min_length=1)
     metadata: DocumentMetadata | None = None
     options: DetectOptions = Field(default_factory=DetectOptions)
+    legal_context: LegalContext | None = None
 
 
 class EvidenceSpan(BaseModel):
@@ -117,8 +139,18 @@ class MatchResult(BaseModel):
     paragraph_number: int | None = None
     source_char_start: int | None = None
     source_char_end: int | None = None
-    matched_coverage: float = Field(default=0.0, ge=0.0, le=1.0)
+    matched_coverage: float = Field(
+        default=0.0, ge=0.0, le=1.0,
+        description="이 세그먼트 일치 구간이 질의 전체 길이에서 차지하는 비율",
+    )
     longest_span: int = Field(default=0, ge=0)
+    match_reasons: list[str] = Field(
+        default_factory=list,
+        description=(
+            "이 후보가 채택된 이유 (#9). score_threshold=결합점수가 임계 초과, "
+            "exact_span=연속 일치 길이 조건 충족, coverage=커버리지 조건 충족."
+        ),
+    )
 
 
 class ExtractedElements(BaseModel):
@@ -142,6 +174,41 @@ class LegalRiskSignal(BaseModel):
     disclaimer: str
 
 
+class ScoreSemantics(BaseModel):
+    """점수와 임계값의 의미를 명시 (#9).
+
+    ``confidence``/``similarity`` 는 hashing 어휘 점수와 lemma 겹침을 설정
+    가중치로 섞은 **검색 랭킹 점수**이며, 침해 확률도 법적 판정도 아니다.
+    임계값 역시 실데이터 캘리브레이션 전이라 provisional 이다.
+    """
+    combined_score: float = Field(..., ge=0.0, le=1.0)
+    score_kind: str = Field(
+        default="lexical_lemma_blend",
+        description="점수 구성. 확률값이 아니며 서로 다른 코퍼스 간 비교 불가.",
+    )
+    threshold_used: float = Field(..., ge=0.0, le=1.0)
+    threshold_source: Literal["request_override", "server_default"]
+    threshold_calibrated: bool = Field(
+        default=False, description="실데이터 FP 분포로 캘리브레이션되었는지",
+    )
+    provisional: bool = Field(
+        default=True, description="True면 임계값이 잠정값이라 판정 근거로 쓸 수 없음",
+    )
+    union_coverage: float = Field(
+        default=0.0, ge=0.0, le=1.0,
+        description="정밀 비교 후보 전체의 비중복 일치 구간 / 질의 길이 (#3)",
+    )
+    covered_chars: int = Field(default=0, ge=0)
+    query_chars: int = Field(default=0, ge=0)
+    evidence_truncated: bool = Field(
+        default=False,
+        description="True면 CPU 상한으로 일부 후보는 정밀 비교하지 않음",
+    )
+    note: str = (
+        "검색 랭킹 점수이며 침해 확률이 아닙니다. 임계값은 캘리브레이션 전 잠정값입니다."
+    )
+
+
 class AiSegmentSignal(BaseModel):
     index: int
     start: int
@@ -203,6 +270,7 @@ class DetectResponse(BaseModel):
     has_similarity_match: bool | None = None
     corpus_scope_note: str | None = None
     legal_risk: LegalRiskSignal | None = None
+    score_semantics: ScoreSemantics | None = None
     autobiography_mode: bool = False
     candidates_before_filter: int | None = None
     engine_version: str

app/core/config.py

@@ -13,7 +13,15 @@ class Settings(BaseSettings):
     log_level: str = "info"        # debug / info / warning / error
     reload: bool = False           # 개발용 자동 재시작
     root_path: str = ""            # 리버스 프록시 sub-path (예: /plagiarism)
-    api_key: str = ""              # 설정 시 /v1/health 외 X-API-Key 필수
+
+    # --- 인증 (#1) ---
+    # api_key 만으로는 "빈 값 = 무인증"이 조용히 성립한다. 운영에서는
+    # require_api_key=true 로 명시적 fail-closed 를 걸어, 키가 없으면 앱이
+    # 아예 뜨지 않게 한다.
+    api_key: str = ""              # 설정 시 X-API-Key 필수
+    require_api_key: bool = False  # true 인데 api_key 가 비면 기동 실패
+    public_health: bool = True     # /v1/health 를 인증 없이 공개할지
+    public_docs: bool = True       # /docs, /openapi.json, /redoc 공개 여부
 
     engine_version: str = "o2o-plagiarism-2.0.0-pdf-v1.2"
     reference_corpus_dir: str = "./data/reference"
@@ -25,6 +33,8 @@ class Settings(BaseSettings):
     persistent_similarity_threshold: float = 0.65
     persistent_min_exact_span: int = 80
     persistent_min_coverage: float = 0.30
+    # 상위 N개 후보만 SequenceMatcher 정밀 비교 + union coverage 에 참여시킨다.
+    # 이 값이 곧 요청당 O(질의길이 × 세그먼트길이) 연산의 상한이다 (#3/#5).
     persistent_rerank_top_k: int = 20
     precedents_path: str = "./data/precedents/precedents.jsonl"
     ai_detector_model_path: str = "./data/models/ai_detector.joblib"
@@ -33,6 +43,9 @@ class Settings(BaseSettings):
 
     # PDF VII-4 권장: 정밀도 우선 보수적 임계값
     similarity_threshold: float = 0.85
+    # 임계값이 실데이터로 캘리브레이션되었는지. false 면 API 응답에
+    # provisional=true 로 노출된다 (#9). 79권 FP 분포 측정 후 true 로 전환.
+    similarity_threshold_calibrated: bool = False
 
     # KoSimCSE / KoSBERT (PDF VII-3 권장) - 한국어 오픈소스 임베딩
     use_kosimcse: bool = False

app/engine/detector.py

@@ -23,13 +23,16 @@ from app.api.schemas import (
     DetectRequest,
     DetectResponse,
     DocumentMetadata,
+    ExtractedElements,
     InfringementTag,
     InfringementType,
+    LegalContext,
     LegalRiskSignal,
     MatchResult,
     PartialPlagiarismSignal,
     ReviewSummary,
     ScoreBreakdown,
+    ScoreSemantics,
 )
 from app.core.config import Settings, get_settings
 from app.engine.autobiography_filter import preprocess_for_autobiography
@@ -54,8 +57,12 @@ logger = logging.getLogger(__name__)
 
 
 class PlagiarismDetector:
+    #: 참조 특징 프로세스 캐시 상한. 초과하면 통째로 비운다(단순 LRU 대용).
+    _FEATURE_CACHE_MAX = 20_000
+
     def __init__(self, settings: Settings | None = None, extractor: Extractor | None = None):
         self.settings = settings or get_settings()
+        self._feature_cache: dict[str, tuple[list[str], "ExtractedElements"]] = {}
         self._extractor: Extractor = extractor or get_extractor(self.settings)
         self.taxonomy: Taxonomy | None = load_taxonomy(self.settings.taxonomy_path)
         self._legal_engine = LegalRiskEngine(load_precedents(self.settings.precedent_path))
@@ -174,6 +181,7 @@ class PlagiarismDetector:
         metadata: DocumentMetadata | None = None,
         options: DetectOptions | None = None,
         include_ai_segments: bool = True,
+        legal_context: LegalContext | None = None,
     ) -> DetectResponse:
         opts = options or DetectOptions()
         default_threshold = (
@@ -202,18 +210,30 @@ class PlagiarismDetector:
 
         # 영속 CPU 인덱스: 전량 행렬곱으로 후보를 구하고 상위 후보만 증거 비교한다.
         persistent_hits: list[PersistentHit] = []
+        union_coverage = 0.0
+        covered_chars = 0
+        evidence_truncated = False
+        document_coverage: dict[str, float] = {}
         if self._persistent:
             persistent_query_lemmas = extract_lemmas(text)
-            persistent_hits = self._persistent.query(
+            result = self._persistent.search(
                 text,
                 top_k=max(opts.top_k, self.settings.persistent_rerank_top_k),
+                evidence_limit=self.settings.persistent_rerank_top_k,
             )
+            persistent_hits = result.hits
+            union_coverage = result.union_coverage
+            covered_chars = result.covered_chars
+            evidence_truncated = result.evidence_truncated
+            document_coverage = result.document_coverage
+            # 정밀 특징 비교도 rerank 대상(reranked=True)에만 수행한다 (#5).
             hits = [
                 self._persistent_to_similarity_hit(h, persistent_query_lemmas, elements)
-                for h in persistent_hits
+                for h in persistent_hits if h.reranked
             ]
+            self._backfill_segment_features(persistent_hits)
             hits.sort(key=lambda h: h.score, reverse=True)
-            candidates_count = len(hits)
+            candidates_count = len(persistent_hits)
             lsh_jaccards: dict[str, float] = {}
         else:
             hits = []
@@ -240,14 +260,22 @@ class PlagiarismDetector:
         persistent_by_id = {h.segment_id: h for h in persistent_hits}
         matches = []
         for h in hits:
+            if len(matches) >= opts.top_k:
+                break
             provenance_hit = persistent_by_id.get(h.doc_id)
-            exact_partial_match = bool(
+            # 채택 이유를 명시적으로 남긴다 (#9). 임계 초과가 아니라 연속 일치나
-                provenance_hit and (
+            # 커버리지 때문에 올라온 후보를 검토자가 구분할 수 있어야 한다.
-                    provenance_hit.longest_span >= self.settings.persistent_min_exact_span
+            reasons: list[str] = []
-                    or provenance_hit.coverage >= self.settings.persistent_min_coverage
+            if h.score >= threshold:
-                )
+                reasons.append("score_threshold")
-            )
+            if provenance_hit:
-            if (h.score < threshold and not exact_partial_match) or len(matches) >= opts.top_k:
+                if provenance_hit.longest_span >= self.settings.persistent_min_exact_span:
+                    reasons.append("exact_span")
+                # 커버리지 조건은 문서 단위 union 으로 판단한다. 세그먼트 단독
+                # 비율은 긴 원고에서 구조적으로 작아 조건이 성립하지 않는다 (#3).
+                if document_coverage.get(provenance_hit.document_id, 0.0) >= self.settings.persistent_min_coverage:
+                    reasons.append("coverage")
+            if not reasons:
                 continue
             match = self._to_match(
                 h, opts.return_evidence, lsh_jaccards.get(h.doc_id),
@@ -255,20 +283,25 @@ class PlagiarismDetector:
             )
             if provenance_hit:
                 match = self._add_provenance(match, provenance_hit)
-            matches.append(match)
+            matches.append(match.model_copy(update={"match_reasons": reasons}))
         confidence = matches[0].similarity if matches else (hits[0].score if hits else 0.0)
         is_infringement = bool(matches)  # 후방호환 필드. 법적 확정이 아니라 임계 초과 매칭.
         ccl_basis = self._build_ccl_basis(matches) if is_infringement else None
 
-        top_coverage = max((m.matched_coverage for m in matches), default=0.0)
         top_longest = max((m.longest_span for m in matches), default=0)
         legal_tags = [t.tag for m in matches for t in m.tags]
+        ctx = legal_context or LegalContext()
         legal = self._legal_engine.assess(
             max_similarity=matches[0].similarity if matches else 0.0,
-            coverage=top_coverage,
+            # 문서 단위 union coverage 를 쓴다. 세그먼트 최대값을 쓰면 긴 원고에서
+            # 항상 0에 가까워 strong_copy 판정이 성립하지 않았다 (#3).
+            coverage=union_coverage,
             longest_span=top_longest,
             legal_tags=legal_tags,
-            work_type="literary",
+            work_type=ctx.work_type,
+            access_evidence=ctx.access_evidence,
+            protected_expression_reviewed=ctx.protected_expression_reviewed,
+            rights_verified=ctx.rights_verified,
         )
 
         # AI 탐지는 전처리 전 raw text에서만 실행한다.
@@ -329,6 +362,19 @@ class PlagiarismDetector:
                 precedent_ids=list(legal.precedent_ids),
                 disclaimer=legal.disclaimer,
             ),
+            score_semantics=ScoreSemantics(
+                combined_score=round(confidence, 4),
+                threshold_used=threshold,
+                threshold_source=(
+                    "request_override" if opts.threshold is not None else "server_default"
+                ),
+                threshold_calibrated=self.settings.similarity_threshold_calibrated,
+                provisional=not self.settings.similarity_threshold_calibrated,
+                union_coverage=round(union_coverage, 4),
+                covered_chars=covered_chars,
+                query_chars=len(text),
+                evidence_truncated=evidence_truncated,
+            ),
             autobiography_mode=autobio_mode,
             candidates_before_filter=candidates_count,
             engine_version=self.settings.engine_version,
@@ -336,7 +382,53 @@ class PlagiarismDetector:
         )
 
     def detect_request(self, req: DetectRequest) -> DetectResponse:
-        return self.detect(req.doc_id, req.text, req.metadata, req.options)
+        return self.detect(
+            req.doc_id, req.text, req.metadata, req.options,
+            legal_context=req.legal_context,
+        )
+
+    def reference_features(self, hit: PersistentHit) -> tuple[list[str], ExtractedElements]:
+        """참조 세그먼트의 lemma/요소. 인덱스 캐시 → 프로세스 캐시 → 계산 순 (#5).
+
+        인덱싱 때 채워둔 값이 있으면 형태소 분석을 아예 하지 않는다. v1 DB 나
+        API 업로드분처럼 캐시가 없으면 계산하되 프로세스 캐시에 담고, 호출자가
+        DB 로 백필한다.
+        """
+        from app.api.schemas import ExtractedElements as _EE
+
+        if hit.reference_lemmas is not None and hit.reference_elements is not None:
+            return hit.reference_lemmas, _EE(**hit.reference_elements)
+
+        cached = self._feature_cache.get(hit.segment_id)
+        if cached is not None:
+            return cached
+
+        lemmas = extract_lemmas(hit.reference_text)
+        elements = self._extractor.extract(hit.reference_text)
+        if len(self._feature_cache) >= self._FEATURE_CACHE_MAX:
+            self._feature_cache.clear()
+        self._feature_cache[hit.segment_id] = (lemmas, elements)
+        return lemmas, elements
+
+    def _backfill_segment_features(self, hits: list[PersistentHit]) -> None:
+        """질의 중 계산한 참조 특징을 DB 에 되돌려 다음 요청부터 재사용."""
+        if not self._persistent:
+            return
+        pending = [
+            (h.segment_id, *self._feature_cache[h.segment_id])
+            for h in hits
+            if h.reranked
+            and h.reference_lemmas is None
+            and h.segment_id in self._feature_cache
+        ]
+        if not pending:
+            return
+        try:
+            self._persistent.store.update_segment_features(
+                (sid, lemmas, elements.model_dump()) for sid, lemmas, elements in pending
+            )
+        except Exception as exc:  # 백필 실패가 탐지 응답을 막아서는 안 된다
+            logger.warning("Segment feature backfill failed: %s", exc)
 
     def _persistent_to_similarity_hit(
         self, hit: PersistentHit, query_lemmas: list[str], query_elements,
@@ -345,9 +437,9 @@ class PlagiarismDetector:
         from app.engine.similarity import _element_similarities
         from app.engine.structural import lemma_overlap_ratio
 
-        reference_elements = self._extractor.extract(hit.reference_text)
+        reference_lemmas, reference_elements = self.reference_features(hit)
         element_sim = _element_similarities(query_elements, reference_elements)
-        lemma_sim = lemma_overlap_ratio(query_lemmas, extract_lemmas(hit.reference_text))
+        lemma_sim = lemma_overlap_ratio(query_lemmas, reference_lemmas)
         s = self.settings
         combined = (
             s.weight_text_sim * hit.score

app/engine/persistent_index.py

@@ -28,6 +28,29 @@ VECTORIZER_CONFIG = {
 }
 
 
+@dataclass(frozen=True)
+class PersistentQueryResult:
+    """질의 1건의 후보와 **문서 단위 union coverage**.
+
+    coverage 정의 (#3):
+      · ``PersistentHit.coverage`` — 이 세그먼트 하나가 질의 전체 길이에서
+        차지하는 비율. 분모가 질의 전체라 긴 원고에서는 필연적으로 작다.
+      · ``PersistentQueryResult.union_coverage`` — 정밀 비교 대상 후보
+        전체(evidence_limit 개)의 일치 구간을 **질의 좌표에서 합집합**으로
+        묶어 계산한 비율. 중복 구간을 두 번 세지 않는다.
+        "이 원고의 몇 %가 등록 코퍼스와 겹치는가"에 답하는 값은 이쪽이며,
+        법적 위험도 판단에는 반드시 이 값을 쓴다.
+    """
+
+    hits: list["PersistentHit"]
+    union_coverage: float
+    covered_chars: int
+    query_chars: int
+    evidence_limit: int
+    evidence_truncated: bool
+    document_coverage: dict[str, float]
+
+
 @dataclass(frozen=True)
 class PersistentHit:
     segment_id: str
@@ -44,6 +67,12 @@ class PersistentHit:
     source_char_start: int | None
     source_char_end: int | None
     reference_text: str
+    #: 인덱싱 때 캐시된 참조 특징 (#5). None 이면 호출자가 계산·백필해야 한다.
+    reference_lemmas: list[str] | None = None
+    reference_elements: dict | None = None
+    #: SequenceMatcher 정밀 비교를 실제로 수행했는지. False 면 evidence/coverage/
+    #: longest_span 은 미계산(0) 이며 score 만 의미가 있다.
+    reranked: bool = True
 
 
 def _vectorizer(n_features: int, config: dict | None = None):
@@ -61,10 +90,17 @@ def _vectorizer(n_features: int, config: dict | None = None):
     )
 
 
-def _evidence_spans(query: str, reference: str, min_match: int = 12, limit: int = 10) -> tuple[list[dict], float, int]:
+def _evidence_spans(
-    """원문 query 좌표의 공통 연속 구간과 coverage를 반환."""
+    query: str, reference: str, min_match: int = 12, limit: int = 10
+) -> tuple[list[dict], list[tuple[int, int]], int]:
+    """원문 query 좌표의 공통 연속 구간을 반환.
+
+    반환: (표시용 상위 span, coverage 계산용 전체 구간 [start,end), 최장 일치 길이)
+    두 번째 값은 union coverage 를 위해 **잘리지 않은 전체 구간**이다. 표시용은
+    limit 개로 줄이지만 coverage 는 전량으로 계산해야 과소평가되지 않는다.
+    """
     if not query or not reference:
-        return [], 0.0, 0
+        return [], [], 0
     blocks = SequenceMatcher(None, query, reference, autojunk=False).get_matching_blocks()
     useful = [b for b in blocks if b.size >= min_match]
     useful.sort(key=lambda b: (-b.size, b.a))
@@ -79,15 +115,35 @@ def _evidence_spans(query: str, reference: str, min_match: int = 12, limit: int
         }
         for b in selected
     ]
-    covered: set[int] = set()
+    intervals = [(b.a, b.a + b.size) for b in useful]
-    for b in useful:
+    return spans, intervals, max((b.size for b in useful), default=0)
-        covered.update(range(b.a, b.a + b.size))
+
-    return spans, len(covered) / max(1, len(query)), max((b.size for b in useful), default=0)
+
+def _merge_intervals(intervals: list[tuple[int, int]]) -> list[tuple[int, int]]:
+    """겹치는 구간을 합쳐 비중복 구간 목록으로. set(range(...)) 는 긴 원고에서
+    메모리를 크게 먹으므로 구간 병합으로 처리한다."""
+    if not intervals:
+        return []
+    ordered = sorted(intervals)
+    merged = [ordered[0]]
+    for start, end in ordered[1:]:
+        last_start, last_end = merged[-1]
+        if start <= last_end:
+            merged[-1] = (last_start, max(last_end, end))
+        else:
+            merged.append((start, end))
+    return merged
+
+
+def _covered_length(intervals: list[tuple[int, int]]) -> int:
+    return sum(end - start for start, end in _merge_intervals(intervals))
 
 
 class PersistentCorpusIndex:
     MATRIX_FILE = "lexical.npz"
     META_FILE = "index.json"
+    #: 정밀 비교(SequenceMatcher) 기본 상한. 호출자가 설정값으로 덮어쓴다.
+    DEFAULT_EVIDENCE_LIMIT = 20
 
     def __init__(self, store_path: str | Path, index_dir: str | Path):
         self.store = CorpusStore(store_path)
@@ -195,11 +251,28 @@ class PersistentCorpusIndex:
         self._matrix, self._meta = matrix, meta
         return {"mode": mode, "total": len(ids), "added": len(new_ids)}
 
-    def query(self, text: str, top_k: int = 50, min_score: float = 0.0) -> list[PersistentHit]:
+    def query(self, text: str, top_k: int = 50, min_score: float = 0.0,
+              evidence_limit: int | None = None) -> list[PersistentHit]:
+        """후보만 필요할 때 쓰는 얇은 래퍼. union coverage 가 필요하면 search()."""
+        return self.search(text, top_k, min_score, evidence_limit).hits
+
+    def search(
+        self,
+        text: str,
+        top_k: int = 50,
+        min_score: float = 0.0,
+        evidence_limit: int | None = None,
+    ) -> PersistentQueryResult:
+        """후보 검색 + 상위 evidence_limit 개에 대해서만 정밀 비교.
+
+        SequenceMatcher 는 O(질의청크 × 세그먼트) 라 후보 전체에 돌리면 요청당
+        수십 초가 된다. 정밀 비교 대상을 evidence_limit 로 제한하는 것이 CPU
+        상한이며, 나머지 후보는 score 만 채워 reranked=False 로 표시한다.
+        """
         if self._matrix is None:
             self.load()
         if not text.strip() or self._matrix is None or self._matrix.shape[0] == 0:
-            return []
+            return PersistentQueryResult([], 0.0, 0, len(text), evidence_limit or 0, False, {})
         vectorizer = _vectorizer(
             int(self._meta["n_features"]), self._meta["vectorizer_config"]
         )
@@ -228,7 +301,14 @@ class PersistentCorpusIndex:
         indexes = indexes[np.argsort(scores[indexes])[::-1]]
         ids = [self._meta["segment_ids"][int(i)] for i in indexes if scores[int(i)] >= min_score]
         records = self.store.get_segments(ids)
+
+        limit = self.DEFAULT_EVIDENCE_LIMIT if evidence_limit is None else evidence_limit
+        limit = max(0, limit)
         hits: list[PersistentHit] = []
+        union_intervals: list[tuple[int, int]] = []
+        document_intervals: dict[str, list[tuple[int, int]]] = {}
+        reranked_count = 0
+
         for i in indexes:
             score = float(scores[int(i)])
             if score < min_score:
@@ -237,18 +317,42 @@ class PersistentCorpusIndex:
             record = records.get(segment_id)
             if not record:
                 continue
+
+            if reranked_count >= limit:
+                # CPU 상한. 정밀 비교 없이 score 만 채운다.
+                hits.append(self._to_hit(record, score, [], 0.0, 0, reranked=False))
+                continue
+
             chunk_start, chunk_text = query_chunks[int(best_chunk[int(i)])]
-            evidence, _, longest = _evidence_spans(chunk_text, record.text)
+            evidence, intervals, longest = _evidence_spans(chunk_text, record.text)
             for span in evidence:
                 span["start"] += chunk_start
                 span["end"] += chunk_start
-            covered = sum(span["end"] - span["start"] for span in evidence)
+            shifted = [(s + chunk_start, e + chunk_start) for s, e in intervals]
-            coverage = min(1.0, covered / max(1, len(text)))
+            union_intervals.extend(shifted)
+            document_intervals.setdefault(record.document_id, []).extend(shifted)
+            # 이 세그먼트 단독 기여분 (질의 전체 길이 대비)
+            coverage = min(1.0, _covered_length(shifted) / max(1, len(text)))
             hits.append(self._to_hit(record, score, evidence, coverage, longest))
-        return hits
+            reranked_count += 1
+
+        covered_chars = _covered_length(union_intervals)
+        return PersistentQueryResult(
+            hits=hits,
+            union_coverage=min(1.0, covered_chars / max(1, len(text))),
+            covered_chars=covered_chars,
+            query_chars=len(text),
+            evidence_limit=limit,
+            evidence_truncated=any(not h.reranked for h in hits),
+            document_coverage={
+                document_id: min(1.0, _covered_length(intervals) / max(1, len(text)))
+                for document_id, intervals in document_intervals.items()
+            },
+        )
 
     @staticmethod
-    def _to_hit(record: SegmentRecord, score: float, evidence: list[dict], coverage: float, longest: int) -> PersistentHit:
+    def _to_hit(record: SegmentRecord, score: float, evidence: list[dict],
+                coverage: float, longest: int, reranked: bool = True) -> PersistentHit:
         return PersistentHit(
             segment_id=record.segment_id,
             document_id=record.document_id,
@@ -264,4 +368,7 @@ class PersistentCorpusIndex:
             source_char_start=record.char_start,
             source_char_end=record.char_end,
             reference_text=record.text,
+            reference_lemmas=record.lemmas,
+            reference_elements=record.elements,
+            reranked=reranked,
         )

app/engine/provenance.py

@@ -15,7 +15,9 @@ from pathlib import Path
 from typing import Iterable, Iterator
 
 
-SCHEMA_VERSION = 1
+#: v2 — 참조 lemma/요소 특징 캐시 컬럼 추가 (#5). 기존 v1 DB 는 ALTER TABLE 로
+#: 자동 승격되며, 캐시가 비어 있으면 질의 시 계산 후 백필된다.
+SCHEMA_VERSION = 2
 
 
 def text_sha256(text: str) -> str:
@@ -49,6 +51,11 @@ class SegmentRecord:
     char_end: int | None = None
     source_locator: str | None = None
     metadata: dict = field(default_factory=dict)
+    #: 인덱싱 시점에 계산해 둔 참조 lemma 열 (#5). None 이면 미계산 상태이며,
+    #: 질의 경로가 계산 후 백필한다. 형태소 분석을 요청마다 반복하지 않기 위한 캐시.
+    lemmas: list[str] | None = None
+    #: 인물/모티프 등 최소 요소 특징. ExtractedElements 를 dict 로 직렬화한 형태.
+    elements: dict | None = None
 
     @property
     def text_sha256(self) -> str:
@@ -110,11 +117,20 @@ class CorpusStore:
                 CREATE INDEX IF NOT EXISTS idx_segments_hash ON segments(text_sha256);
                 """
             )
+            self._migrate(con)
             con.execute(
                 "INSERT OR REPLACE INTO corpus_meta(key, value) VALUES('schema_version', ?)",
                 (str(SCHEMA_VERSION),),
             )
 
+    @staticmethod
+    def _migrate(con: sqlite3.Connection) -> None:
+        """기존 DB 를 파괴 없이 승격. v1 → v2 는 컬럼 추가만 필요하다."""
+        existing = {row["name"] for row in con.execute("PRAGMA table_info(segments)")}
+        for column in ("lemmas_json", "elements_json"):
+            if column not in existing:
+                con.execute(f"ALTER TABLE segments ADD COLUMN {column} TEXT")
+
     def upsert_document(self, record: DocumentRecord) -> None:
         self.upsert_documents([record])
 
@@ -150,8 +166,9 @@ class CorpusStore:
                     """
                     INSERT OR IGNORE INTO segments(
                       segment_id,document_id,ordinal,text,text_sha256,coordinate_scope,
-                      page_number,paragraph_number,char_start,char_end,source_locator,metadata_json
+                      page_number,paragraph_number,char_start,char_end,source_locator,
-                    ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?)
+                      metadata_json,lemmas_json,elements_json
+                    ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                     """,
                     (
                         r.segment_id,
@@ -166,6 +183,9 @@ class CorpusStore:
                         r.char_end,
                         r.source_locator,
                         json.dumps(r.metadata, ensure_ascii=False, sort_keys=True),
+                        json.dumps(r.lemmas, ensure_ascii=False) if r.lemmas is not None else None,
+                        json.dumps(r.elements, ensure_ascii=False, sort_keys=True)
+                        if r.elements is not None else None,
                     ),
                 )
                 if con.total_changes > before:
@@ -186,21 +206,66 @@ class CorpusStore:
                 """
             )
             for row in rows:
-                metadata = json.loads(row["metadata_json"] or "{}")
+                yield self._row_to_segment(row)
-                metadata.setdefault("document_title", row["title"])
+
-                yield SegmentRecord(
+    @staticmethod
-                    segment_id=row["segment_id"],
+    def _row_to_segment(row: sqlite3.Row) -> SegmentRecord:
-                    document_id=row["document_id"],
+        metadata = json.loads(row["metadata_json"] or "{}")
-                    text=row["text"],
+        metadata.setdefault("document_title", row["title"])
-                    ordinal=row["ordinal"],
+        keys = row.keys()
-                    coordinate_scope=row["coordinate_scope"],
+        lemmas = None
-                    page_number=row["page_number"],
+        elements = None
-                    paragraph_number=row["paragraph_number"],
+        # v1 DB 를 그대로 읽는 경로에서도 죽지 않도록 컬럼 존재를 확인한다.
-                    char_start=row["char_start"],
+        if "lemmas_json" in keys and row["lemmas_json"]:
-                    char_end=row["char_end"],
+            lemmas = json.loads(row["lemmas_json"])
-                    source_locator=row["source_locator"],
+        if "elements_json" in keys and row["elements_json"]:
-                    metadata=metadata,
+            elements = json.loads(row["elements_json"])
-                )
+        return SegmentRecord(
+            segment_id=row["segment_id"],
+            document_id=row["document_id"],
+            text=row["text"],
+            ordinal=row["ordinal"],
+            coordinate_scope=row["coordinate_scope"],
+            page_number=row["page_number"],
+            paragraph_number=row["paragraph_number"],
+            char_start=row["char_start"],
+            char_end=row["char_end"],
+            source_locator=row["source_locator"],
+            metadata=metadata,
+            lemmas=lemmas,
+            elements=elements,
+        )
+
+    def update_segment_features(
+        self, features: Iterable[tuple[str, list[str], dict]]
+    ) -> int:
+        """(segment_id, lemmas, elements) 를 백필한다. 반환값은 갱신된 행 수."""
+        rows = [
+            (
+                json.dumps(lemmas, ensure_ascii=False),
+                json.dumps(elements, ensure_ascii=False, sort_keys=True),
+                segment_id,
+            )
+            for segment_id, lemmas, elements in features
+        ]
+        if not rows:
+            return 0
+        self.initialize()
+        with self._connect() as con:
+            con.executemany(
+                "UPDATE segments SET lemmas_json=?, elements_json=? WHERE segment_id=?",
+                rows,
+            )
+            return con.total_changes
+
+    def count_missing_features(self) -> int:
+        if not self.path.exists():
+            return 0
+        self.initialize()
+        with self._connect() as con:
+            return int(con.execute(
+                "SELECT COUNT(*) FROM segments WHERE lemmas_json IS NULL"
+            ).fetchone()[0])
 
     def get_segments(self, segment_ids: Iterable[str]) -> dict[str, SegmentRecord]:
         ids = list(dict.fromkeys(segment_ids))
@@ -218,16 +283,7 @@ class CorpusStore:
                     batch,
                 )
                 for row in rows:
-                    metadata = json.loads(row["metadata_json"] or "{}")
+                    result[row["segment_id"]] = self._row_to_segment(row)
-                    metadata.setdefault("document_title", row["title"])
-                    result[row["segment_id"]] = SegmentRecord(
-                        segment_id=row["segment_id"], document_id=row["document_id"],
-                        text=row["text"], ordinal=row["ordinal"],
-                        coordinate_scope=row["coordinate_scope"],
-                        page_number=row["page_number"], paragraph_number=row["paragraph_number"],
-                        char_start=row["char_start"], char_end=row["char_end"],
-                        source_locator=row["source_locator"], metadata=metadata,
-                    )
         return result
 
     def stats(self) -> dict[str, int]:

app/main.py

@@ -19,12 +19,60 @@ from app.jobs.store import JobStore
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
 
 
+class AuthConfigurationError(RuntimeError):
+    """운영 인증 설정이 모순될 때 기동을 막는다."""
+
+
+def validate_auth_settings(settings) -> None:
+    """REQUIRE_API_KEY=true 인데 키가 없으면 기동 실패 (fail-closed).
+
+    '키를 깜빡해서 무인증으로 떠 있었다'가 가능한 구성을 없애는 것이 목적이다.
+    미출간 원고를 다루는 서버라 조용한 무인증이 가장 위험하다.
+    """
+    if settings.require_api_key and not settings.api_key.strip():
+        raise AuthConfigurationError(
+            "REQUIRE_API_KEY=true 인데 API_KEY 가 비어 있습니다. "
+            "키를 설정하거나 REQUIRE_API_KEY=false 로 두십시오."
+        )
+
+
+def public_paths(settings) -> set[str]:
+    """인증 없이 접근 가능한 경로. 설정으로 좁힐 수 있다."""
+    paths = {"/"}
+    if settings.public_health:
+        paths.add("/v1/health")
+    if settings.public_docs:
+        paths.update({"/docs", "/openapi.json", "/redoc"})
+    return paths
+
+
+def is_authorized(settings, path: str, supplied: str) -> bool:
+    """요청 허용 여부. 순수 함수라 앱 기동 없이 테스트할 수 있다.
+
+    api_key 가 비어 있으면(개발 기본값) 인증을 걸지 않는다. 이 경우 기동 시
+    critical 경고가 남는다.
+    """
+    configured = settings.api_key.strip()
+    if not configured:
+        return True
+    if path in public_paths(settings):
+        return True
+    protected_docs = {"/docs", "/openapi.json", "/redoc"}
+    if not path.startswith("/v1") and path not in protected_docs:
+        return True
+    # compare_digest 는 비ASCII str 에서 TypeError 를 던진다. 한글/이모지 키를
+    # 넣으면 전 요청이 500 이 되므로 반드시 bytes 로 비교한다.
+    return hmac.compare_digest(configured.encode("utf-8"), (supplied or "").encode("utf-8"))
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     settings = get_settings()
+    validate_auth_settings(settings)
     if not settings.api_key.strip():
         logging.critical(
             "API_KEY is empty: all /v1 endpoints are unauthenticated. "
+            "Set REQUIRE_API_KEY=true with a key before exposing this server. "
             "Do not expose unpublished manuscripts publicly until client key rollout is complete."
         )
     app.state.settings = settings
@@ -67,12 +115,9 @@ app.include_router(api_router)
 @app.middleware("http")
 async def optional_api_key_auth(request: Request, call_next):
     """API_KEY가 설정된 운영 환경에서만 API 인증을 강제한다."""
-    configured = _settings.api_key.strip()
+    settings = getattr(request.app.state, "settings", None) or _settings
-    public_paths = {"/", "/v1/health", "/docs", "/openapi.json", "/redoc"}
+    if not is_authorized(settings, request.url.path, request.headers.get("x-api-key", "")):
-    if configured and request.url.path not in public_paths and request.url.path.startswith("/v1"):
+        return JSONResponse(status_code=401, content={"detail": "Invalid or missing API key"})
-        supplied = request.headers.get("x-api-key", "")
-        if not hmac.compare_digest(configured, supplied):
-            return JSONResponse(status_code=401, content={"detail": "Invalid or missing API key"})
     return await call_next(request)
 
 _STATIC_DIR = Path(__file__).resolve().parent / "static"

docs/IMPLEMENTATION_RUNBOOK.md

@@ -47,6 +47,7 @@ PERSISTENT_INDEX_DIR=/app/data/runtime/index
 PERSISTENT_SIMILARITY_THRESHOLD=0.65
 # 바이칼과 키 전달을 합의한 뒤 활성화
 API_KEY=<secret-manager-or-protected-env-value>
+REQUIRE_API_KEY=true
 ```
 
 `0.65`는 영속 문자 n-gram 후보 검색의 보수적인 시작값일 뿐 운영 확정값이 아니다.
@@ -127,3 +128,74 @@ API 인증·방화벽·키 회전을 별도 운영 작업으로 완료해야 한
 포함하지 않는다. 일반 PyPI의 최신 torch가 CUDA 런타임 수 GB를 함께 설치할 수 있기
 때문이다. 레거시 KoSimCSE가 반드시 필요한 별도 이미지에서만 해당 Python 버전에 맞는
 공식 CPU 전용 torch wheel을 먼저 설치한 뒤 sentence-transformers를 추가한다.
+
+## 인증 fail-closed (#1)
+
+| 설정 | 기본 | 의미 |
+|---|---|---|
+| `API_KEY` | 빈 값 | 비어 있으면 **인증이 걸리지 않는다**. 기동 시 critical 로그가 남는다. |
+| `REQUIRE_API_KEY` | `false` | `true` 인데 `API_KEY` 가 비면 **앱이 기동에 실패**한다(`AuthConfigurationError`). |
+| `PUBLIC_HEALTH` | `true` | `/v1/health` 를 무인증 공개할지. 모니터링이 키를 못 넣으면 `true` 유지. |
+| `PUBLIC_DOCS` | `true` | `/docs`, `/openapi.json`, `/redoc` 공개 여부. |
+
+- **King 실제 활성화는 바이칼 측 키 전달 후로 보류**한다. 그때까지 `REQUIRE_API_KEY=false`
+  로 두되, 서버를 외부에 노출하지 않는다. 키를 받으면 `API_KEY` 설정과 동시에
+  `REQUIRE_API_KEY=true` 로 올려 "키를 깜빡한 채 무인증으로 떠 있는" 상태를 원천 차단한다.
+- **운영 키는 반드시 ASCII 로 발급**한다. HTTP 헤더는 비ASCII 를 전송할 수 없어
+  한글 키는 인증 자체가 불가능하다(서버는 500 대신 401 을 반환한다).
+- `PUBLIC_DOCS=false` 이면 `/docs`, `/openapi.json`, `/redoc`에도 API 키가
+  필요하다. 외부 노출 환경에서는 리버스 프록시 차단도 함께 적용하는 편이 안전하다.
+
+## coverage 정의와 CPU 상한 (#3/#5)
+
+두 가지 coverage 를 구분한다. 혼동하면 긴 원고에서 위험도가 항상 낮게 나온다.
+
+- `matches[].matched_coverage` — **세그먼트 1건**의 일치 구간 / 질의 전체 길이.
+  분모가 원고 전체라 30만 자 원고에서는 한 세그먼트가 최대 수천분의 1에 그친다.
+  개별 후보의 기여도를 볼 때만 쓴다.
+- `score_semantics.union_coverage` — **정밀 비교한 후보 전체**의 일치 구간을 질의
+  좌표에서 **합집합**으로 묶은 비율(중복 구간 1회만 계산). "이 원고의 몇 %가 등록
+  코퍼스와 겹치는가"에 답하는 값이며, `PERSISTENT_MIN_COVERAGE` 게이트와 판례
+  위험도(`legal_risk`)는 **이 값만** 사용한다.
+
+CPU 상한은 `PERSISTENT_RERANK_TOP_K`(기본 20) 하나로 통제한다. 후보 검색은 전량
+행렬곱으로 하되, `SequenceMatcher` 정밀 비교는 상위 N건에만 돌린다. 나머지 후보는
+`reranked=false` 로 반환되며 `evidence`/`coverage`/`longest_span` 이 0 이고 score 만
+의미가 있다. 잘림이 발생하면 `score_semantics.evidence_truncated=true` 로 노출된다.
+이 값을 올리면 요청당 지연이 선형으로 증가한다.
+
+참조 lemma/요소는 `scripts/build_persistent_index.py` 가 인덱싱 때 DB 에 사전계산해
+둔다(`--skip-precompute` 로 생략 가능). 캐시가 없으면 질의 시 계산 후 자동 백필된다.
+500 세그먼트×937자 기준 사전계산 시 요청 지연 0.499s → 0.415s (약 17%).
+
+## 점수 의미와 잠정 임계값 (#9)
+
+`confidence` / `matches[].similarity` 는 hashing 어휘 점수와 lemma 겹침을 설정
+가중치로 섞은 **검색 랭킹 점수**이며 침해 확률이 아니다. 응답의 `score_semantics`
+가 이를 명시한다.
+
+- `threshold_source` — `server_default` / `request_override`
+- `threshold_calibrated` — `SIMILARITY_THRESHOLD_CALIBRATED` 설정값. 79권 상호비교
+  FP 분포를 측정하기 전까지 `false` 로 두고, `provisional=true` 로 노출된다.
+- `matches[].match_reasons` — 후보가 채택된 이유. `score_threshold`(결합점수 초과),
+  `exact_span`(연속 일치 ≥ `PERSISTENT_MIN_EXACT_SPAN`), `coverage`(union coverage
+  ≥ `PERSISTENT_MIN_COVERAGE`). 이때 후보 채택용 coverage는 해당 출처 문서의
+  세그먼트끼리만 합산한다. 서로 다른 출처의 일치를 합쳐 개별 후보를 통과시키지
+  않는다. 임계값이 아니라 연속 일치 때문에 올라온 후보도 구분할 수 있다.
+
+임계값 수치는 캘리브레이션 전까지 **임의로 바꾸지 않는다.** 기존 값을 유지하고
+provisional 플래그로만 알린다.
+
+## 법적 맥락 입력 (#10)
+
+`POST /v1/plagiarism/detect` 의 `legal_context` 는 선택 필드다. 엔진은 이 사실들을
+**추론하지 않으며**, 미제공 시 `legal_risk.missing_factors` 에 그대로 남는다.
+
+```json
+{"doc_id":"x","text":"...","legal_context":{
+  "work_type":"literary","access_evidence":true,
+  "protected_expression_reviewed":true,"rights_verified":true}}
+```
+
+판례는 등록된 것만 반환한다(`precedent_ids`). 랭킹은 ① 태그 교집합 수 ②
+`work_type` 일치 ③ 사건번호 순이며, 생성형 인용은 어떤 경로로도 발생하지 않는다.

scripts/build_persistent_index.py

@@ -12,6 +12,36 @@ if __package__ in (None, ""):
     sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
 from app.engine.persistent_index import PersistentCorpusIndex
+from app.engine.provenance import CorpusStore
+
+
+def precompute_features(store: CorpusStore, batch: int = 500) -> int:
+    """참조 lemma/요소를 미리 계산해 DB 에 저장 (#5).
+
+    이 작업을 인덱싱 때 1회 해두면, 탐지 요청마다 후보 세그먼트를 형태소
+    분석하던 비용이 사라진다. 이미 채워진 세그먼트는 건너뛴다.
+    """
+    from app.engine.extractor import get_extractor
+    from app.engine.structural import extract_lemmas
+
+    extractor = get_extractor()
+    pending: list[tuple[str, list[str], dict]] = []
+    updated = 0
+    for segment in store.iter_segments():
+        if segment.lemmas is not None and segment.elements is not None:
+            continue
+        pending.append((
+            segment.segment_id,
+            extract_lemmas(segment.text),
+            extractor.extract(segment.text).model_dump(),
+        ))
+        if len(pending) >= batch:
+            updated += store.update_segment_features(pending)
+            print(f"  특징 계산 {updated}건…", flush=True)
+            pending = []
+    if pending:
+        updated += store.update_segment_features(pending)
+    return updated
 
 
 def main() -> int:
@@ -19,10 +49,22 @@ def main() -> int:
     p.add_argument("--database", type=Path, required=True)
     p.add_argument("--index-dir", type=Path, required=True)
     p.add_argument("--features", type=int, default=2**20)
+    p.add_argument(
+        "--skip-precompute", action="store_true",
+        help="참조 lemma/요소 사전계산을 건너뛴다(질의 시 계산 후 백필됨)",
+    )
     args = p.parse_args()
     if not args.database.exists():
         p.error(f"database does not exist: {args.database}")
+
+    store = CorpusStore(args.database)
+    precomputed = 0
+    if not args.skip_precompute:
+        precomputed = precompute_features(store)
+
     result = PersistentCorpusIndex(args.database, args.index_dir).sync(args.features)
+    result["precomputed_features"] = precomputed
+    result["missing_features"] = store.count_missing_features()
     print(json.dumps(result, ensure_ascii=False, indent=2))
     return 0
 

tests/test_ai_training_split.py

@@ -0,0 +1,167 @@
+"""AI 탐지기 학습 데이터 분리 검증 (#F).
+
+핵심 불변식 두 가지:
+  1) fit=train / threshold=val / report=test 로 역할이 섞이지 않는다.
+  2) 어떤 source_group 도 두 split 에 동시에 나타나지 않는다(누출 금지).
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from scripts.build_ai_training_dataset import Record, assign_splits, summarize  # noqa: E402
+
+sklearn = pytest.importorskip("sklearn", reason="scikit-learn 미설치")
+pytest.importorskip("joblib", reason="joblib 미설치")
+
+
+# ---------------------------------------------------------------------------
+# 분할 자체의 불변식 (sklearn 불필요 부분)
+# ---------------------------------------------------------------------------
+
+def _records(n_groups_per_label: int = 6, per_group: int = 6) -> list[Record]:
+    out: list[Record] = []
+    for label in (0, 1):
+        origin = "human" if label == 0 else "ai"
+        for g in range(n_groups_per_label):
+            group = f"{origin}:group-{g}"
+            for i in range(per_group):
+                out.append(Record(
+                    text=f"{origin} 문단 {g}-{i}", label=label, origin=origin,
+                    source_group=group, book=group,
+                ))
+    return out
+
+
+def test_assign_splits_never_shares_a_group():
+    records = _records()
+    assignment = assign_splits(records, (0.7, 0.15, 0.15), seed=7)
+    for rec in records:
+        rec.split = assignment[rec.source_group]
+
+    by_split: dict[str, set[str]] = {}
+    for rec in records:
+        by_split.setdefault(rec.split, set()).add(rec.source_group)
+    splits = list(by_split)
+    for i, a in enumerate(splits):
+        for b in splits[i + 1:]:
+            assert not (by_split[a] & by_split[b]), f"{a}/{b} 그룹 중복 = 누출"
+
+    assert summarize(records)["group_overlap_between_splits"] == []
+
+
+def test_assign_splits_is_deterministic_for_same_seed():
+    a = assign_splits(_records(), (0.7, 0.15, 0.15), seed=11)
+    b = assign_splits(_records(), (0.7, 0.15, 0.15), seed=11)
+    assert a == b
+
+
+def test_assign_splits_covers_all_three_splits():
+    assignment = assign_splits(_records(), (0.7, 0.15, 0.15), seed=3)
+    assert set(assignment.values()) == {"train", "val", "test"}
+
+
+# ---------------------------------------------------------------------------
+# 학습 CLI end-to-end
+# ---------------------------------------------------------------------------
+
+def _write_dataset(path: Path) -> None:
+    human = "비가 왔다. 나는 그날 학교에 가지 않았고 대신 뒷산에 올라가 온종일 앉아 있었다. 춥지는 않았다. 형이 왔다."
+    ai = "그날의 기억은 오래도록 남아, 지금까지도 선명하게 떠오르는 장면이 되었다. 아침의 공기는 서늘했고, 발걸음은 조용히 이어졌다."
+    rows = []
+    for label, body, origin in ((0, human, "human"), (1, ai, "ai")):
+        for g in range(6):
+            split = "train" if g < 4 else ("val" if g == 4 else "test")
+            for i in range(8):
+                rows.append({
+                    "text": (body + f" 변형 {g}-{i}. ") * 3,
+                    "label": label, "origin": origin,
+                    "source_group": f"{origin}:g{g}", "book": f"{origin}-{g}",
+                    "split": split,
+                })
+    path.write_text(
+        "\n".join(json.dumps(r, ensure_ascii=False) for r in rows), encoding="utf-8"
+    )
+
+
+def _run_trainer(data: Path, out: Path, *extra: str) -> subprocess.CompletedProcess:
+    return subprocess.run(
+        [sys.executable, "scripts/train_ai_detector.py", "--data", str(data),
+         "--out", str(out), *extra],
+        cwd=ROOT, capture_output=True, text=True,
+    )
+
+
+def test_trainer_uses_train_val_test_roles(tmp_path):
+    data = tmp_path / "ds.jsonl"
+    out = tmp_path / "model.joblib"
+    _write_dataset(data)
+
+    proc = _run_trainer(data, out)
+    assert proc.returncode == 0, proc.stderr[-2000:]
+
+    metrics = json.loads((tmp_path / "model.metrics.json").read_text(encoding="utf-8"))
+    # 세 역할이 모두 기록되어야 한다
+    assert metrics["n_train"] > 0 and metrics["n_val"] > 0 and metrics["n_test"] > 0
+    assert {"train", "validation", "test"} <= set(metrics)
+    # 임계값은 val 에서 뽑혔음이 지표에 남아야 한다
+    assert "validation_low_point" in metrics["cuts"]
+    assert "validation_high_point" in metrics["cuts"]
+    # 학습에 쓰인 표본 수와 보고 표본 수가 서로 다른 집합이어야 한다
+    assert metrics["n_train"] != metrics["n_test"] or metrics["n_val"] != metrics["n_test"]
+
+
+def test_trainer_rejects_group_overlap_between_splits(tmp_path):
+    data = tmp_path / "leaky.jsonl"
+    out = tmp_path / "model.joblib"
+    _write_dataset(data)
+
+    rows = [json.loads(line) for line in data.read_text(encoding="utf-8").splitlines()]
+    for row in rows:
+        # train 그룹 하나를 test 에도 등장시켜 누출을 주입
+        if row["source_group"] == "human:g0" and row["split"] == "train":
+            row["split"] = "test"
+            break
+    data.write_text(
+        "\n".join(json.dumps(r, ensure_ascii=False) for r in rows), encoding="utf-8"
+    )
+
+    proc = _run_trainer(data, out)
+    assert proc.returncode == 2, "그룹 누출은 학습을 중단시켜야 한다"
+    assert "누출" in proc.stderr or "중복" in proc.stderr
+
+
+def test_trainer_fails_on_single_class(tmp_path):
+    data = tmp_path / "one.jsonl"
+    _write_dataset(data)
+    rows = [json.loads(line) for line in data.read_text(encoding="utf-8").splitlines()]
+    kept = [r for r in rows if r["label"] == 0]
+    data.write_text(
+        "\n".join(json.dumps(r, ensure_ascii=False) for r in kept), encoding="utf-8"
+    )
+    proc = _run_trainer(data, tmp_path / "m.joblib")
+    assert proc.returncode == 2
+    assert "단일 클래스" in proc.stderr
+
+
+def test_trainer_fails_when_a_split_is_empty(tmp_path):
+    data = tmp_path / "noval.jsonl"
+    _write_dataset(data)
+    rows = [json.loads(line) for line in data.read_text(encoding="utf-8").splitlines()]
+    for row in rows:
+        if row["split"] == "val":
+            row["split"] = "train"
+    data.write_text(
+        "\n".join(json.dumps(r, ensure_ascii=False) for r in rows), encoding="utf-8"
+    )
+    proc = _run_trainer(data, tmp_path / "m.joblib")
+    assert proc.returncode == 2
+    assert "빈 split" in proc.stderr

tests/test_api.py

@@ -56,3 +56,69 @@ def test_batch_flow():
         job_id = resp.json()["job_id"]
         status = client.get(f"/v1/plagiarism/batch/{job_id}")
         assert status.status_code == 200
+
+
+def test_detect_exposes_score_semantics():
+    """#9 — 점수/임계값의 의미가 응답에 명시되어야 한다."""
+    with TestClient(app) as client:
+        resp = client.post(
+            "/v1/plagiarism/detect",
+            json={"doc_id": "s-1", "text": "어린왕자는 작은 별에서 온 소년이다."},
+        )
+        assert resp.status_code == 200
+        sem = resp.json()["score_semantics"]
+        assert sem["threshold_source"] == "server_default"
+        assert sem["threshold_calibrated"] is False
+        assert sem["provisional"] is True, "캘리브레이션 전에는 잠정값으로 노출"
+        assert 0.0 <= sem["union_coverage"] <= 1.0
+        assert sem["query_chars"] > 0
+        assert "침해 확률이 아닙니다" in sem["note"]
+
+
+def test_detect_reports_request_threshold_override():
+    with TestClient(app) as client:
+        resp = client.post(
+            "/v1/plagiarism/detect",
+            json={"doc_id": "s-2", "text": "앤 셜리는 초록 지붕 집에 입양된 소녀다.",
+                  "options": {"threshold": 0.4}},
+        )
+        sem = resp.json()["score_semantics"]
+        assert sem["threshold_source"] == "request_override"
+        assert sem["threshold_used"] == 0.4
+
+
+def test_detect_accepts_legal_context():
+    """#10 — 사람이 확인한 사실을 전달하면 missing_factors 에서 빠진다."""
+    with TestClient(app) as client:
+        base = client.post(
+            "/v1/plagiarism/detect",
+            json={"doc_id": "l-1", "text": "홍길동은 활빈당을 만들어 재물을 나누었다."},
+        ).json()["legal_risk"]
+        assert base["access_evidence"] == "not_provided"
+        assert len(base["missing_factors"]) == 3
+
+        supplied = client.post(
+            "/v1/plagiarism/detect",
+            json={
+                "doc_id": "l-2", "text": "홍길동은 활빈당을 만들어 재물을 나누었다.",
+                "legal_context": {
+                    "work_type": "literary", "access_evidence": True,
+                    "protected_expression_reviewed": True, "rights_verified": True,
+                },
+            },
+        ).json()["legal_risk"]
+        assert supplied["access_evidence"] == "provided"
+        assert supplied["protected_expression"] == "reviewed"
+        assert supplied["missing_factors"] == []
+
+
+def test_matches_carry_match_reasons():
+    with TestClient(app) as client:
+        resp = client.post(
+            "/v1/plagiarism/detect",
+            json={"doc_id": "r-1", "text": "어린왕자는 작은 별에서 온 소년이다. 그는 여우를 만난다.",
+                  "options": {"threshold": 0.01}},
+        )
+        for match in resp.json()["matches"]:
+            assert match["match_reasons"], "채택 이유가 비어 있으면 안 된다"
+            assert set(match["match_reasons"]) <= {"score_threshold", "exact_span", "coverage"}

tests/test_auth_middleware.py

@@ -0,0 +1,144 @@
+"""API 키 인증 미들웨어 (#1).
+
+앱 전체를 띄우지 않고 순수 함수로 검증한다. 엔진 기동(코퍼스 인덱싱)을 타면
+테스트가 느려지고 인증 로직과 무관한 이유로 깨진다.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.core.config import Settings
+from app.main import (
+    AuthConfigurationError,
+    is_authorized,
+    public_paths,
+    validate_auth_settings,
+)
+
+
+def _settings(**overrides) -> Settings:
+    base = {"api_key": "", "require_api_key": False,
+            "public_health": True, "public_docs": True}
+    base.update(overrides)
+    return Settings(**base)
+
+
+# ---------------------------------------------------------------------------
+# fail-closed 기동 검증
+# ---------------------------------------------------------------------------
+
+def test_require_api_key_without_key_fails_startup():
+    with pytest.raises(AuthConfigurationError) as exc:
+        validate_auth_settings(_settings(require_api_key=True, api_key=""))
+    assert "REQUIRE_API_KEY" in str(exc.value)
+
+
+def test_require_api_key_with_whitespace_only_key_fails():
+    with pytest.raises(AuthConfigurationError):
+        validate_auth_settings(_settings(require_api_key=True, api_key="   "))
+
+
+def test_require_api_key_with_key_starts_fine():
+    validate_auth_settings(_settings(require_api_key=True, api_key="secret"))
+
+
+def test_default_settings_start_without_key():
+    """기본값(개발)에서는 기동을 막지 않는다 — 기존 동작 보존."""
+    validate_auth_settings(_settings())
+
+
+# ---------------------------------------------------------------------------
+# 기존 기본 동작 보존
+# ---------------------------------------------------------------------------
+
+def test_no_key_configured_allows_everything():
+    s = _settings(api_key="")
+    assert is_authorized(s, "/v1/plagiarism/detect", "") is True
+    assert is_authorized(s, "/v1/corpus", "") is True
+
+
+def test_key_configured_rejects_missing_and_wrong_key():
+    s = _settings(api_key="secret")
+    assert is_authorized(s, "/v1/plagiarism/detect", "") is False
+    assert is_authorized(s, "/v1/plagiarism/detect", "wrong") is False
+    assert is_authorized(s, "/v1/plagiarism/detect", "secret") is True
+
+
+def test_corpus_write_paths_are_protected():
+    s = _settings(api_key="secret")
+    for path in ("/v1/corpus", "/v1/corpus/file", "/v1/corpus/doc-1", "/v1/plagiarism/batch"):
+        assert is_authorized(s, path, "") is False, path
+
+
+# ---------------------------------------------------------------------------
+# 공개 경로 설정
+# ---------------------------------------------------------------------------
+
+def test_health_public_by_default():
+    s = _settings(api_key="secret")
+    assert "/v1/health" in public_paths(s)
+    assert is_authorized(s, "/v1/health", "") is True
+
+
+def test_health_can_be_protected():
+    s = _settings(api_key="secret", public_health=False)
+    assert "/v1/health" not in public_paths(s)
+    assert is_authorized(s, "/v1/health", "") is False
+    assert is_authorized(s, "/v1/health", "secret") is True
+
+
+def test_docs_public_by_default_and_can_be_closed():
+    s = _settings(api_key="secret")
+    assert "/openapi.json" in public_paths(s)
+    closed = _settings(api_key="secret", public_docs=False)
+    assert "/openapi.json" not in public_paths(closed)
+    for path in ("/docs", "/openapi.json", "/redoc"):
+        assert is_authorized(closed, path, "") is False
+        assert is_authorized(closed, path, "secret") is True
+
+
+def test_root_console_stays_public():
+    s = _settings(api_key="secret")
+    assert is_authorized(s, "/", "") is True
+
+
+# ---------------------------------------------------------------------------
+# 회귀: 비ASCII 키가 TypeError 로 500 을 내지 않아야 한다
+# ---------------------------------------------------------------------------
+
+def test_non_ascii_key_does_not_raise():
+    s = _settings(api_key="비밀키-한글🔑")
+    assert is_authorized(s, "/v1/plagiarism/detect", "비밀키-한글🔑") is True
+    assert is_authorized(s, "/v1/plagiarism/detect", "틀린키") is False
+    assert is_authorized(s, "/v1/plagiarism/detect", "") is False
+
+
+def test_non_ascii_configured_key_returns_401_not_500():
+    """서버에 한글 키를 설정해도 500 이 아니라 401 이어야 한다.
+
+    HTTP 헤더는 비ASCII 를 전송할 수 없으므로 이런 키는 사실상 인증 불가지만,
+    최소한 서버가 TypeError 로 터지면 안 된다. (운영 키는 ASCII 로 발급할 것)
+    """
+    from fastapi import FastAPI, Request
+    from fastapi.responses import JSONResponse
+    from fastapi.testclient import TestClient
+
+    from app.main import is_authorized as guard
+
+    settings = _settings(api_key="한글키")
+    app = FastAPI()
+
+    @app.middleware("http")
+    async def auth(request: Request, call_next):
+        if not guard(settings, request.url.path, request.headers.get("x-api-key", "")):
+            return JSONResponse(status_code=401, content={"detail": "Invalid or missing API key"})
+        return await call_next(request)
+
+    @app.get("/v1/thing")
+    async def thing():
+        return {"ok": True}
+
+    with TestClient(app) as client:
+        assert client.get("/v1/thing").status_code == 401
+        assert client.get("/v1/thing", headers={"x-api-key": "ascii-guess"}).status_code == 401

tests/test_coverage_and_features.py

@@ -0,0 +1,337 @@
+"""union coverage (#3), 참조 특징 캐시 (#5), 점수 의미 (#9), 법적 맥락 (#10).
+
+scipy/sklearn 이 없으면 인덱스 관련 테스트는 skip 된다.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from app.engine.legal_risk import LegalRiskEngine, Precedent
+from app.engine.persistent_index import _covered_length, _merge_intervals
+from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord
+
+scipy = pytest.importorskip("scipy", reason="scipy 미설치")
+
+from app.engine.persistent_index import PersistentCorpusIndex  # noqa: E402
+
+
+def _segment(segment_id: str, text: str, **kw) -> SegmentRecord:
+    return SegmentRecord(
+        segment_id=segment_id, document_id=kw.pop("document_id", "doc-1"),
+        text=text, ordinal=segment_id, char_start=0, char_end=len(text),
+        source_locator=f"book.json#{segment_id}", **kw,
+    )
+
+
+def _index_with(tmp_path, segments: list[SegmentRecord], title="원본"):
+    db = tmp_path / "corpus.sqlite3"
+    store = CorpusStore(db)
+    for doc_id in {s.document_id for s in segments}:
+        store.upsert_document(DocumentRecord(document_id=doc_id, title=title))
+    store.add_segments(segments)
+    index = PersistentCorpusIndex(db, tmp_path / "index")
+    index.sync()
+    return store, index
+
+
+# ---------------------------------------------------------------------------
+# 구간 병합 (순수 함수)
+# ---------------------------------------------------------------------------
+
+def test_merge_intervals_deduplicates_overlap():
+    assert _merge_intervals([(0, 10), (5, 20), (30, 40)]) == [(0, 20), (30, 40)]
+    assert _covered_length([(0, 10), (5, 20)]) == 20
+    assert _covered_length([(0, 10), (0, 10)]) == 10, "중복 구간을 두 번 세면 안 된다"
+    assert _covered_length([]) == 0
+
+
+def test_merge_intervals_handles_adjacent_and_nested():
+    assert _merge_intervals([(0, 10), (10, 20)]) == [(0, 20)]
+    assert _merge_intervals([(0, 100), (10, 20)]) == [(0, 100)]
+
+
+# ---------------------------------------------------------------------------
+# #3 union coverage
+# ---------------------------------------------------------------------------
+
+def test_union_coverage_sums_multiple_segments(tmp_path):
+    """서로 다른 세그먼트가 질의의 다른 부분과 일치하면 coverage 가 합산된다."""
+    part_a = "바닷가 마을에서 파도 소리를 들으며 자란 기억이 아직도 선명하게 남아 있다. " * 3
+    part_b = "군에 입대하던 날 아버지는 아무 말 없이 내 어깨를 두드려 주셨던 기억이 난다. " * 3
+    _, index = _index_with(tmp_path, [
+        _segment("seg-a", part_a), _segment("seg-b", part_b),
+    ])
+    query = part_a + "완전히 무관한 중간 문단입니다. " * 20 + part_b
+    result = index.search(query, top_k=10)
+
+    assert result.union_coverage > 0.4, "두 구간이 모두 반영되어야 한다"
+    assert result.covered_chars >= len(part_a)
+    assert result.query_chars == len(query)
+    # 개별 세그먼트 coverage 는 각자 union 보다 작다
+    per_hit = [h.coverage for h in result.hits if h.reranked]
+    assert max(per_hit) < result.union_coverage
+
+
+def test_document_coverage_does_not_leak_between_sources(tmp_path):
+    part_a = "바닷가 마을에서 파도 소리를 들으며 자랐다. " * 3
+    part_b = "군에 입대하던 날 아버지가 내 어깨를 두드렸다. " * 3
+    _, index = _index_with(tmp_path, [
+        _segment("seg-a", part_a, document_id="doc-a"),
+        _segment("seg-b", part_b, document_id="doc-b"),
+    ])
+    query = part_a + ("서로 무관한 중간 문장입니다. " * 20) + part_b
+    result = index.search(query, top_k=10)
+
+    assert result.union_coverage > result.document_coverage["doc-a"]
+    assert result.union_coverage > result.document_coverage["doc-b"]
+    assert result.document_coverage["doc-a"] < 0.30
+    assert result.document_coverage["doc-b"] < 0.30
+
+
+def test_union_coverage_low_for_long_unrelated_document(tmp_path):
+    _, index = _index_with(tmp_path, [_segment("seg-1", "바닷가 마을의 파도 소리를 기억한다.")])
+    query = "전혀 다른 주제의 글입니다. 오늘 회의에서 분기 실적을 논의했습니다. " * 100
+    result = index.search(query, top_k=5)
+    assert result.union_coverage < 0.1
+
+
+def test_short_full_copy_reaches_high_coverage(tmp_path):
+    text = "나는 어린 시절 바닷가 마을에서 살았고 매일 파도 소리를 들으며 잠들었다."
+    _, index = _index_with(tmp_path, [_segment("seg-1", text)])
+    result = index.search(text, top_k=5)
+    assert result.union_coverage > 0.8
+
+
+def test_long_document_partial_copy_is_not_structurally_zero(tmp_path):
+    """긴 원고 안의 부분 복사가 coverage 에 실제로 잡히는지 (#3 회귀)."""
+    copied = "바닷가 마을에서 파도 소리를 들으며 자란 기억이 선명하다. " * 10
+    _, index = _index_with(tmp_path, [_segment("seg-1", copied)])
+    filler = "무관한 문장입니다. " * 300
+    query = filler + copied + filler
+    result = index.search(query, top_k=5)
+    assert result.covered_chars >= len(copied) * 0.5
+    assert result.union_coverage > 0.0
+
+
+# ---------------------------------------------------------------------------
+# CPU 상한
+# ---------------------------------------------------------------------------
+
+def test_evidence_limit_caps_precise_comparison(tmp_path):
+    segments = [
+        _segment(f"seg-{i}", f"바닷가 마을 이야기 {i}번 문단입니다. 파도 소리를 들었다. " * 3)
+        for i in range(8)
+    ]
+    _, index = _index_with(tmp_path, segments)
+    result = index.search("바닷가 마을 이야기 파도 소리를 들었다.", top_k=8, evidence_limit=3)
+
+    reranked = [h for h in result.hits if h.reranked]
+    assert len(reranked) == 3
+    assert result.evidence_truncated is True
+    for hit in result.hits:
+        if not hit.reranked:
+            assert hit.evidence == [] and hit.coverage == 0.0 and hit.longest_span == 0
+
+
+def test_query_wrapper_still_returns_hits(tmp_path):
+    """기존 호출부 호환 — query() 는 여전히 list[PersistentHit]."""
+    _, index = _index_with(tmp_path, [_segment("seg-1", "바닷가 마을의 파도 소리.")])
+    hits = index.query("바닷가 마을의 파도 소리.", top_k=1)
+    assert isinstance(hits, list) and hits[0].segment_id == "seg-1"
+
+
+# ---------------------------------------------------------------------------
+# #5 참조 특징 캐시 + 마이그레이션
+# ---------------------------------------------------------------------------
+
+def test_v1_database_migrates_without_data_loss(tmp_path):
+    """lemmas_json/elements_json 없는 기존 DB 도 그대로 열려야 한다."""
+    import sqlite3
+
+    db = tmp_path / "old.sqlite3"
+    con = sqlite3.connect(db)
+    con.executescript(
+        """
+        CREATE TABLE corpus_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
+        CREATE TABLE documents (
+            document_id TEXT PRIMARY KEY, title TEXT NOT NULL, source_path TEXT,
+            source_sha256 TEXT, metadata_json TEXT NOT NULL DEFAULT '{}',
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP,
+            updated_at TEXT DEFAULT CURRENT_TIMESTAMP);
+        CREATE TABLE segments (
+            segment_id TEXT PRIMARY KEY, document_id TEXT NOT NULL,
+            ordinal TEXT NOT NULL, text TEXT NOT NULL, text_sha256 TEXT NOT NULL,
+            coordinate_scope TEXT NOT NULL, page_number INTEGER,
+            paragraph_number INTEGER, char_start INTEGER, char_end INTEGER,
+            source_locator TEXT, metadata_json TEXT NOT NULL DEFAULT '{}',
+            created_at TEXT DEFAULT CURRENT_TIMESTAMP,
+            UNIQUE(document_id, text_sha256));
+        INSERT INTO documents(document_id,title) VALUES('doc-1','옛 책');
+        INSERT INTO segments VALUES('seg-1','doc-1','1','옛 본문입니다.','h','episode',
+            NULL,NULL,0,7,NULL,'{}',CURRENT_TIMESTAMP);
+        """
+    )
+    con.commit()
+    con.close()
+
+    store = CorpusStore(db)
+    store.initialize()  # 마이그레이션
+    segments = list(store.iter_segments())
+    assert len(segments) == 1
+    assert segments[0].text == "옛 본문입니다."
+    assert segments[0].lemmas is None
+    assert store.count_missing_features() == 1
+
+
+def test_feature_roundtrip_and_backfill(tmp_path):
+    store = CorpusStore(tmp_path / "corpus.sqlite3")
+    store.upsert_document(DocumentRecord(document_id="doc-1", title="책"))
+    store.add_segments([_segment("seg-1", "홍길동은 활빈당을 만들었다.")])
+    assert store.count_missing_features() == 1
+
+    store.update_segment_features([("seg-1", ["홍길동", "활빈당"], {"characters": ["홍길동"]})])
+    assert store.count_missing_features() == 0
+    loaded = store.get_segments(["seg-1"])["seg-1"]
+    assert loaded.lemmas == ["홍길동", "활빈당"]
+    assert loaded.elements == {"characters": ["홍길동"]}
+
+
+def test_precomputed_features_reach_the_hit(tmp_path):
+    store = CorpusStore(tmp_path / "corpus.sqlite3")
+    store.upsert_document(DocumentRecord(document_id="doc-1", title="책"))
+    store.add_segments([_segment(
+        "seg-1", "나는 어린 시절 바닷가 마을에서 살았다.",
+        lemmas=["바닷가", "마을", "살다"],
+        elements={"characters": [], "motifs": [], "genre": None, "keywords": ["바닷가"]},
+    )])
+    index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index")
+    index.sync()
+    hit = index.query("나는 어린 시절 바닷가 마을에서 살았다.", top_k=1)[0]
+    assert hit.reference_lemmas == ["바닷가", "마을", "살다"]
+    assert hit.reference_elements["keywords"] == ["바닷가"]
+
+
+def test_detector_uses_cache_instead_of_recomputing(monkeypatch, tmp_path):
+    """캐시가 있으면 형태소 분석을 호출하지 않아야 한다 (#5 호출횟수 테스트)."""
+    from app.api.schemas import ExtractedElements
+    from app.engine import detector as det_module
+    from app.engine.persistent_index import PersistentHit
+
+    calls = {"lemmas": 0, "extract": 0}
+
+    def counting_lemmas(text, *a, **kw):
+        calls["lemmas"] += 1
+        return ["x"]
+
+    monkeypatch.setattr(det_module, "extract_lemmas", counting_lemmas)
+
+    detector = det_module.PlagiarismDetector.__new__(det_module.PlagiarismDetector)
+    detector._feature_cache = {}
+    detector._persistent = None
+
+    class _Extractor:
+        def extract(self, text):
+            calls["extract"] += 1
+            return ExtractedElements()
+
+    detector._extractor = _Extractor()
+
+    cached_hit = PersistentHit(
+        segment_id="seg-1", document_id="doc-1", title="책", score=0.9,
+        evidence=[], coverage=0.0, longest_span=0, source_locator=None,
+        coordinate_scope="episode", page_number=None, paragraph_number=None,
+        source_char_start=None, source_char_end=None, reference_text="본문",
+        reference_lemmas=["미리", "계산"],
+        reference_elements={"characters": [], "motifs": [], "genre": None, "keywords": []},
+    )
+    lemmas, _ = detector.reference_features(cached_hit)
+    assert lemmas == ["미리", "계산"]
+    assert calls == {"lemmas": 0, "extract": 0}, "캐시가 있는데 재계산했다"
+
+    uncached = PersistentHit(
+        segment_id="seg-2", document_id="doc-1", title="책", score=0.9,
+        evidence=[], coverage=0.0, longest_span=0, source_locator=None,
+        coordinate_scope="episode", page_number=None, paragraph_number=None,
+        source_char_start=None, source_char_end=None, reference_text="본문",
+    )
+    detector.reference_features(uncached)
+    assert calls == {"lemmas": 1, "extract": 1}
+    # 두 번째 호출은 프로세스 캐시로 처리
+    detector.reference_features(uncached)
+    assert calls == {"lemmas": 1, "extract": 1}
+
+
+# ---------------------------------------------------------------------------
+# #10 법적 맥락 + 판례 랭킹
+# ---------------------------------------------------------------------------
+
+def _precedents() -> list[Precedent]:
+    return [
+        Precedent("2020다1", "가", "https://x/1", ("literary",),
+                  ("reproduction",), (), "요지1"),
+        Precedent("2019다2", "나", "https://x/2", ("literary",),
+                  ("reproduction", "derivative_work"), (), "요지2"),
+        Precedent("2018다3", "다", "https://x/3", ("musical",),
+                  ("reproduction",), (), "요지3"),
+    ]
+
+
+def test_precedent_ranking_prefers_more_tag_overlap():
+    engine = LegalRiskEngine(_precedents())
+    out = engine.assess(
+        max_similarity=0.9, coverage=0.5, longest_span=200,
+        legal_tags=["reproduction", "derivative_work"], work_type="literary",
+    )
+    assert out.precedent_ids[0] == "2019다2", "태그 교집합이 큰 판례가 먼저"
+    assert "2018다3" not in out.precedent_ids, "work_type 이 다른 판례는 제외"
+
+
+def test_only_registered_precedents_are_returned():
+    engine = LegalRiskEngine(_precedents())
+    out = engine.assess(
+        max_similarity=0.9, coverage=0.5, longest_span=200,
+        legal_tags=["reproduction"], work_type="literary",
+    )
+    registered = {p.case_id for p in _precedents()}
+    assert set(out.precedent_ids) <= registered
+
+
+def test_empty_precedent_db_reports_insufficient():
+    out = LegalRiskEngine([]).assess(
+        max_similarity=0.9, coverage=0.9, longest_span=500, legal_tags=["reproduction"],
+    )
+    assert out.status == "insufficient_precedent_data"
+    assert out.risk_level is None
+    assert out.precedent_ids == ()
+
+
+def test_legal_context_clears_missing_factors():
+    engine = LegalRiskEngine(_precedents())
+    default = engine.assess(max_similarity=0.5, coverage=0.1, longest_span=20,
+                            legal_tags=["reproduction"])
+    assert len(default.missing_factors) == 3
+    assert default.access_evidence == "not_provided"
+
+    supplied = engine.assess(
+        max_similarity=0.5, coverage=0.1, longest_span=20, legal_tags=["reproduction"],
+        access_evidence=True, protected_expression_reviewed=True, rights_verified=True,
+    )
+    assert supplied.missing_factors == ()
+    assert supplied.access_evidence == "provided"
+    assert supplied.protected_expression == "reviewed"
+
+
+def test_legal_context_flows_through_detect_request():
+    from app.api.schemas import DetectRequest, LegalContext
+
+    req = DetectRequest(
+        doc_id="d", text="본문",
+        legal_context=LegalContext(work_type="musical", access_evidence=False,
+                                   protected_expression_reviewed=True),
+    )
+    assert req.legal_context.work_type == "musical"
+    assert req.legal_context.access_evidence is False
+    assert req.legal_context.rights_verified is False
+    # 미제공이 기본
+    assert DetectRequest(doc_id="d", text="본문").legal_context is None

tests/test_review_regressions.py

@@ -0,0 +1,144 @@
+"""이미 반영된 리뷰 항목의 회귀 방지 (#2 #4 #6 #7 #8)."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from app.engine.provenance import CorpusStore, DocumentRecord, SegmentRecord
+
+pytest.importorskip("scipy", reason="scipy 미설치")
+
+from app.engine.persistent_index import (  # noqa: E402
+    VECTORIZER_CONFIG,
+    PersistentCorpusIndex,
+)
+
+
+def _store_with(tmp_path, texts: list[str]) -> CorpusStore:
+    store = CorpusStore(tmp_path / "corpus.sqlite3")
+    store.upsert_document(DocumentRecord(document_id="doc-1", title="책"))
+    store.add_segments([
+        SegmentRecord(segment_id=f"seg-{i}", document_id="doc-1", text=t,
+                      ordinal=str(i), char_start=0, char_end=len(t))
+        for i, t in enumerate(texts)
+    ])
+    return store
+
+
+# --- #2 원자적 교체 -------------------------------------------------------
+
+def test_matrix_file_is_generation_scoped(tmp_path):
+    """세대별 파일명이라 교체 중 이전 인덱스가 덮어써지지 않는다."""
+    store = _store_with(tmp_path, ["첫 번째 세그먼트 본문입니다."])
+    index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index")
+    index.sync()
+    first = json.loads((tmp_path / "index" / "index.json").read_text())["matrix_file"]
+
+    store.add_segments([SegmentRecord(
+        segment_id="seg-9", document_id="doc-1", text="두 번째 세그먼트 본문입니다.",
+        ordinal="9", char_start=0, char_end=10,
+    )])
+    index.sync()
+    second = json.loads((tmp_path / "index" / "index.json").read_text())["matrix_file"]
+
+    assert first != second, "세대가 바뀌면 파일명도 바뀌어야 한다"
+    assert (tmp_path / "index" / first).exists(), "이전 세대 파일이 살아 있어야 한다"
+
+
+def test_replaced_index_object_is_self_consistent(tmp_path):
+    """load() 한 객체는 matrix 행수와 segment_ids 길이가 항상 일치한다."""
+    _store_with(tmp_path, [f"세그먼트 {i} 본문입니다. 파도 소리." for i in range(5)])
+    index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index")
+    index.sync()
+    fresh = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index").load()
+    assert fresh._matrix.shape[0] == len(fresh._meta["segment_ids"]) == 5
+
+
+# --- #8 vectorizer 설정 고정 ---------------------------------------------
+
+def test_index_with_different_vectorizer_config_is_rejected(tmp_path):
+    _store_with(tmp_path, ["본문입니다. 파도 소리를 들었다."])
+    index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index")
+    index.sync()
+
+    meta_path = tmp_path / "index" / "index.json"
+    meta = json.loads(meta_path.read_text())
+    meta["vectorizer_config"] = {**VECTORIZER_CONFIG, "ngram_range": [2, 6]}
+    meta_path.write_text(json.dumps(meta, ensure_ascii=False), encoding="utf-8")
+
+    with pytest.raises(ValueError, match="재빌드"):
+        PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index").load()
+
+
+def test_config_change_forces_rebuild_not_append(tmp_path):
+    _store_with(tmp_path, ["본문입니다. 파도 소리를 들었다."])
+    index_dir = tmp_path / "index"
+    index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", index_dir)
+    index.sync()
+    # n_features 가 달라지면 append 가 성립하지 않아야 한다
+    result = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", index_dir).sync(n_features=2**16)
+    assert result["mode"] == "rebuild"
+
+
+# --- #6 document_count 캐시 ----------------------------------------------
+
+def test_document_count_served_from_meta_without_table_scan(tmp_path, monkeypatch):
+    _store_with(tmp_path, ["본문입니다."])
+    index = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index")
+    index.sync()
+    loaded = PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index").load()
+
+    def explode():
+        raise AssertionError("document_count 가 매 요청 DB 를 스캔하고 있다")
+
+    monkeypatch.setattr(loaded.store, "document_count", explode)
+    assert loaded.document_count == 1
+
+
+# --- #7 API 업로드 청킹 ---------------------------------------------------
+
+def test_api_upload_is_chunked_not_single_segment(tmp_path, monkeypatch):
+    from app.core.config import Settings
+    from app.engine.detector import PlagiarismDetector
+
+    settings = Settings(
+        use_persistent_index=True,
+        corpus_db_path=str(tmp_path / "corpus.sqlite3"),
+        persistent_index_dir=str(tmp_path / "index"),
+        precedents_path=str(tmp_path / "none.jsonl"),
+        ai_detector_model_path=str(tmp_path / "none.joblib"),
+        use_clustering=False,
+        use_lsh_filter=False,
+    )
+    _store_with(tmp_path, ["씨앗 세그먼트입니다."])
+    PersistentCorpusIndex(tmp_path / "corpus.sqlite3", tmp_path / "index").sync()
+
+    detector = PlagiarismDetector(settings=settings)
+    assert detector.uses_persistent_index
+
+    long_text = "긴 원고 문장입니다. 계속 이어집니다. " * 200  # 약 4,000자
+    doc_id = detector.add_persistent_document(None, "업로드 원고", long_text)
+
+    segments = [s for s in detector._persistent.store.iter_segments()
+                if s.document_id == doc_id]
+    assert len(segments) > 1, "문서 전체가 세그먼트 1개로 저장되면 부분 표절을 못 잡는다"
+    assert all(len(s.text) <= 1000 for s in segments)
+    # 오프셋이 원문을 정확히 복원해야 한다
+    for seg in segments:
+        assert long_text.strip()[seg.char_start:seg.char_end] == seg.text
+
+
+# --- #4 detect 가 이벤트 루프를 막지 않는지 -------------------------------
+
+def test_detect_route_runs_in_threadpool():
+    """라우트가 동기 detect 를 threadpool 로 넘기는지 (소스 계약 확인)."""
+    import inspect
+
+    from app.api import routes
+
+    source = inspect.getsource(routes.detect)
+    assert "run_in_threadpool" in source
+    for name in ("corpus_upload_json", "corpus_upload_file", "corpus_delete"):
+        assert "run_in_threadpool" in inspect.getsource(getattr(routes, name)), name