perf(search): synthesis 캐시 TTL enforce + chunk news_source DB 필터 (R10)

- synthesis_service: _CACHE 가 ts 없이 result 만 저장해 CACHE_TTL(1h) 미적용 → 원문 수정돼도
  CACHE_MAXSIZE 찰 때까지 stale answer 반환. (ts, result) tuple + get_cached 에서 만료 pop
  (query_rewriter expire_at 정본 복제).
- chunk_worker: 문서마다 news_sources 전량 로드 후 Python prefix 루프 → DB 필터 푸시다운
  ((name==source_name) | startswith(source_name+' ')). split[0]==source_name 과 동치, autoescape.

검증: py_compile 통과.

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

app/services/search/synthesis_service.py

@@ -95,8 +95,10 @@ except FileNotFoundError:
     )
 
 
-# ─── in-memory LRU (FIFO 근사, query_analyzer 패턴 복제) ─
-_CACHE: dict[str, SynthesisResult] = {}
+# ─── in-memory 캐시 (FIFO eviction + TTL, query_analyzer 패턴 복제) ─
+# R10: (ts, result) 저장 — TTL 미적용으로 원문 수정돼도 CACHE_MAXSIZE 찰 때까지 stale answer
+# 반환하던 결함 수정. query_rewriter 의 expire_at TTL enforce 정본 복제.
+_CACHE: dict[str, tuple[float, SynthesisResult]] = {}
 
 
 def _model_version() -> str:
@@ -122,10 +124,11 @@ def get_cached(query: str, chunk_ids: list[int], backend_name: str = "gemma-macm
     entry = _CACHE.get(key)
     if entry is None:
         return None
-    # TTL 체크는 elapsed_ms 를 악용할 수 없으므로 별도 저장
-    # 여기서는 단순 policy 로 처리: entry 가 있으면 반환 (eviction 은 FIFO 시점)
-    # 정확한 TTL 이 필요하면 (ts, result) tuple 로 저장해야 함.
-    return entry
+    ts, result = entry
+    if time.time() - ts > CACHE_TTL:
+        _CACHE.pop(key, None)  # 만료 — 삭제 후 miss
+        return None
+    return result
 
 
 def _should_cache(result: SynthesisResult) -> bool:
@@ -143,8 +146,9 @@ def set_cached(query: str, chunk_ids: list[int], result: SynthesisResult, backen
     if not _should_cache(result):
         return
     key = _cache_key(query, chunk_ids, backend_name)
+    now = time.time()
     if key in _CACHE:
-        _CACHE[key] = result
+        _CACHE[key] = (now, result)
         return
     if len(_CACHE) >= CACHE_MAXSIZE:
         try:
@@ -152,7 +156,7 @@ def set_cached(query: str, chunk_ids: list[int], result: SynthesisResult, backen
             _CACHE.pop(oldest, None)
         except StopIteration:
             pass
-    _CACHE[key] = result
+    _CACHE[key] = (now, result)
 
 
 def cache_stats() -> dict[str, int]:

app/workers/chunk_worker.py

@@ -272,15 +272,20 @@ async def _lookup_news_source(
     if not source_name:
         return None, None, None
 
-    # news_sources에서 이름이 일치하는 레코드 찾기 (prefix match)
-    result = await session.execute(select(NewsSource))
-    sources = result.scalars().all()
-    for src in sources:
-        if source_name and (
-            src.name.split(" ")[0] == source_name
-            or src.name.startswith(source_name + " ")
-        ):
-            return src.country, src.name, src.language
+    # news_sources prefix 매칭 — R10: 전체 로드+Python 루프 대신 DB 필터 푸시다운.
+    # (name == source_name) OR (name 이 "source_name " 로 시작) = 기존 split[0]==source_name 동치
+    # (첫 토큰 일치 = 정확일치 또는 'source_name ' prefix). autoescape 로 %/_ 안전.
+    result = await session.execute(
+        select(NewsSource)
+        .where(
+            (NewsSource.name == source_name)
+            | NewsSource.name.startswith(source_name + " ", autoescape=True)
+        )
+        .limit(1)
+    )
+    src = result.scalars().first()
+    if src is not None:
+        return src.country, src.name, src.language
 
     logger.warning(
         f"[chunk] news_source 매핑 실패: doc_id={doc.id} ai_sub_group={source_name!r} "