feat(nav): 상단 nav 질문·이드 제거 + 메모 추가

데스크탑 상단 nav 와 모바일 하단 탭바 모두에서 질문(/ask)·이드(/chat) 링크 제거,
메모(/memos) 추가(모바일은 기존 존재). 라우트 코드는 보존(nav 노출만 제거).
미사용 아이콘 import(HelpCircle·MessageCircle) 정리.

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

app/ai/client.py

@@ -264,7 +264,7 @@ class AIClient:
         """벡터 임베딩 — GPU 서버 전용"""
         response = await self._http.post(
             self.ai.embedding.endpoint,
-            json={"model": self.ai.embedding.model, "prompt": text},
+            json={"model": self.ai.embedding.model, "prompt": text, "keep_alive": -1},  # bge-m3 GPU 상주(홈랩 sparse 검색 cold reload ~6s 방지)
         )
         response.raise_for_status()
         return response.json()["embedding"]

app/api/queue_overview.py

@@ -103,6 +103,21 @@ class StageRow(BaseModel):
     oldest_pending_age_sec: int | None
 
 
+class BackgroundJobItem(BaseModel):
+    """큐 밖 관리 스크립트(백필 등) 작업 — processing_queue 가 못 보는 사각지대 노출.
+    stale = running 인데 heartbeat 가 오래 끊김(프로세스 사망 추정)."""
+    id: int
+    kind: str
+    machine: str
+    label: str | None
+    state: Literal["running", "done", "failed"]
+    processed: int
+    total: int | None
+    elapsed_sec: int
+    stale: bool
+    error: str | None
+
+
 class QueueOverviewResponse(BaseModel):
     machines: list[MachineCard]
     stages: list[StageRow]
@@ -110,6 +125,7 @@ class QueueOverviewResponse(BaseModel):
     summarize_by_machine: SummarizeByMachine
     trend_24h: list[TrendBucket]
     totals: Totals
+    background_jobs: list[BackgroundJobItem] = []
 
 
 class FailedItem(BaseModel):

app/core/config.py

@@ -169,6 +169,14 @@ class Settings(BaseModel):
     # 1 = 구 single-inference 동작. 2 = continuous batching 활용 (llm_gate docstring 참조).
     mlx_gate_concurrency: int = 1
 
+    # digest/briefing 생성 LLM 호출 파라미터 (2026-06-15, 모델 교체 후 타임아웃 단일소스화).
+    # 구 하드코딩 25s(빠른 Gemma 기준)가 Qwen3.6-27B-6bit(콜당 ~90~300s) 교체 sweep 에서
+    # 누락돼 digest 600s 하드캡 초과·briefing 4/4 폴백을 유발 → config 단일소스로 이관.
+    # 동시성은 별 키 아님 — 전역 mlx_gate_concurrency(게이트 단일 budget)가 담당.
+    digest_llm_timeout_s: int = 200
+    digest_llm_attempts: int = 2
+    digest_pipeline_hard_cap_s: int = 1800
+
     # PR-MacMini-Derived-Worker-1: study explanation owner = Mac mini
     # GPU 측은 false 로 설정 (.env), explanation 분기 skip guard 트리거.
     study_explanation_enabled: bool = True
@@ -257,6 +265,9 @@ def load_settings() -> Settings:
 
     pipeline_held_stages: list[str] = []
     mlx_gate_concurrency = 1
+    digest_llm_timeout_s = 200
+    digest_llm_attempts = 2
+    digest_pipeline_hard_cap_s = 1800
     if config_path.exists() and raw and "pipeline" in raw:
         held_raw = (raw.get("pipeline") or {}).get("held_stages") or []
         # 스칼라(문자열) 오기입 시 char-split 방지 — 단일 항목 리스트로 수용.
@@ -269,6 +280,19 @@ def load_settings() -> Settings:
             )
         except (TypeError, ValueError):
             mlx_gate_concurrency = 1
+        _pl = raw.get("pipeline") or {}
+        try:
+            digest_llm_timeout_s = max(1, int(_pl.get("digest_llm_timeout_s", 200)))
+        except (TypeError, ValueError):
+            digest_llm_timeout_s = 200
+        try:
+            digest_llm_attempts = max(1, int(_pl.get("digest_llm_attempts", 2)))
+        except (TypeError, ValueError):
+            digest_llm_attempts = 2
+        try:
+            digest_pipeline_hard_cap_s = max(60, int(_pl.get("digest_pipeline_hard_cap_s", 1800)))
+        except (TypeError, ValueError):
+            digest_pipeline_hard_cap_s = 1800
 
     taxonomy = raw.get("taxonomy", {}) if config_path.exists() and raw else {}
     document_types = raw.get("document_types", []) if config_path.exists() and raw else []
@@ -300,6 +324,9 @@ def load_settings() -> Settings:
         internal_worker_token=internal_worker_token,
         pipeline_held_stages=pipeline_held_stages,
         mlx_gate_concurrency=mlx_gate_concurrency,
+        digest_llm_timeout_s=digest_llm_timeout_s,
+        digest_llm_attempts=digest_llm_attempts,
+        digest_pipeline_hard_cap_s=digest_pipeline_hard_cap_s,
     )
 
 

app/main.py

@@ -56,12 +56,15 @@ async def lifespan(app: FastAPI):
     from workers.mailplus_archive import run as mailplus_run
     from workers.statute_collector import run as statute_run
     from workers.news_collector import run as news_collector_run
+    from workers.arxiv_collector import run as arxiv_collector_run
+    from workers.openalex_collector import run as openalex_collector_run
+    from workers.paper_doi_reconcile import run as paper_doi_reconcile_run
     from workers.fulltext_worker import reconcile_unresolved as fulltext_reconcile_run
     from workers.kosha_collector import run as kosha_collector_run
     from workers.csb_collector import run as csb_collector_run
     from workers.api_standards_collector import run as api_standards_run
     from workers.ccps_collector import run as ccps_collector_run
-    from workers.queue_consumer import consume_queue, consume_fast_queue, consume_markdown_queue
+    from workers.queue_consumer import consume_queue, consume_fast_queue, consume_markdown_queue, consume_deep_queue
     from workers.study_queue_consumer import consume_study_queue
     from workers.study_session_queue_consumer import consume_study_session_queue
     from workers.study_memo_card_jobs_consumer import consume_study_memo_card_queue
@@ -98,6 +101,8 @@ async def lifespan(app: FastAPI):
     # 2026-06-12 fast-consumer split: embed/chunk(건당 <1s)를 LLM 사이클에서 분리 —
     # classify(~190s×3)가 사이클을 점유해 벡터 적재가 굶던 구조 캡 해소 (markdown 선례).
     scheduler.add_job(consume_fast_queue, "interval", minutes=1, id="fast_queue_consumer")
+    # 2026-06-15 deep-consumer split: deep_summary(70~300s)를 메인 루프에서 분리 (markdown/fast 선례).
+    scheduler.add_job(consume_deep_queue, "interval", minutes=1, id="deep_queue_consumer")
     scheduler.add_job(watch_inbox, "interval", minutes=5, id="file_watcher")
     scheduler.add_job(cleanup_orphan_uploads, "interval", minutes=10, id="upload_cleanup")
     # PR-4: study_questions 자동 임베딩 (status='none/failed/stale' 행을 batch=10 처리).
@@ -139,6 +144,9 @@ async def lifespan(app: FastAPI):
     # plan ds-s1-backend-1 B-4: dedup 컬럼(duplicate_of/duplicate_count) 야간 절대 재계산.
     # soft-delete 잔여 드리프트 정리(멱등, 드리프트 없으면 no-op). cron 03:30 (다른 잡과 비충돌).
     scheduler.add_job(dedup_reconcile_run, CronTrigger(hour=3, minute=30, timezone=KST), id="dedup_reconcile")
+    # B-3 PR4: 레거시 paper 행 arXiv DataCite DOI 스탬프(재유입 차단). keyless·in-DB·enqueue 0.
+    # dedup_reconcile(03:30)·fulltext_reconcile(03:40) 와 별 worker·비충돌 슬롯.
+    scheduler.add_job(paper_doi_reconcile_run, CronTrigger(hour=3, minute=50, timezone=KST), id="paper_doi_reconcile")
     # crawl-24x7 C-2: KOSHA 재해사례 diff + GUIDE 점진 백필 (daily, 새벽 잡들과 비충돌 슬롯).
     scheduler.add_job(kosha_collector_run, CronTrigger(hour=6, minute=40, timezone=KST), id="kosha_collector")
     # 사이클 3 C-2 잔여: CSB sitemap lastmod diff (weekly 월, cap 40 + 워터마크 점진 백필).
@@ -147,6 +155,12 @@ async def lifespan(app: FastAPI):
     scheduler.add_job(api_standards_run, CronTrigger(day=5, hour=7, minute=5, timezone=KST), id="api_standards_collector")
     # 사이클 3 C-2 잔여: CCPS Beacon 월간 PDF (playwright 익명 경유 — WAF 차단 시 health 로 가시화).
     scheduler.add_job(ccps_collector_run, CronTrigger(day=5, hour=7, minute=20, timezone=KST), id="ccps_collector")
+    # B-3 PR2: arXiv 키워드 필터 수집기 (daily 07:30 KST — statute 07:00 직후 빈 슬롯).
+    # signal-only 초록 색인, per-run cap 으로 임베드 큐 보호. keyless.
+    scheduler.add_job(arxiv_collector_run, CronTrigger(hour=7, minute=30, timezone=KST), id="arxiv_collector")
+    # B-3 PR3: OpenAlex 백본 수집기 (daily 07:45 KST). scaffold-first(키 부재 explicit-skip),
+    # signal-only 초록 색인, per-run cap + cursor watermark. 키=OPENALEX_API_KEY(credentials.env).
+    scheduler.add_job(openalex_collector_run, CronTrigger(hour=7, minute=45, timezone=KST), id="openalex_collector")
     scheduler.start()
 
     # Phase 2.1 (async 구조): QueryAnalyzer prewarm.

app/services/background_jobs.py

@@ -0,0 +1,93 @@
+"""off-queue 관리 스크립트(백필 등) 진행 가시화 — background_jobs (migration 357).
+
+processing_queue 는 파이프라인 stage 전용이라 hier_overnight_backfill /
+section_summary_pilot 같은 스크립트 작업은 대시보드 보드에 안 잡힌다. 이 모듈로
+스크립트가 진행상황을 남기면 queue_overview 가 "백그라운드 작업" 패널로 노출한다.
+
+설계 불변식:
+- **자율 트랜잭션**: 각 기록은 engine.begin() 짧은 트랜잭션으로 즉시 commit한다.
+  스크립트 본 작업은 별도 세션(긴 트랜잭션)이라, 같이 묶으면 commit 전까지 안 보여
+  실시간 가시화가 깨진다. 그래서 전용 connection 으로 독립 commit.
+- **best-effort**: 관측 기록 실패가 본 작업을 깨면 안 된다 — 모든 함수 try/except,
+  실패 시 warning 로그만. job_id=None 이면 조용히 no-op (start 실패해도 이어서 동작).
+"""
+
+import json
+import logging
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncEngine
+
+logger = logging.getLogger(__name__)
+
+
+async def start_job(
+    engine: AsyncEngine, kind: str, label: str | None = None, total: int | None = None
+) -> int | None:
+    """작업 시작 기록 → background_jobs.id (실패 시 None — 호출측은 그대로 진행)."""
+    try:
+        async with engine.begin() as conn:
+            row = (
+                await conn.execute(
+                    text(
+                        "INSERT INTO background_jobs (kind, label, total) "
+                        "VALUES (:k, :l, :t) RETURNING id"
+                    ),
+                    {"k": kind, "l": label, "t": total},
+                )
+            ).first()
+            return int(row[0]) if row else None
+    except Exception as exc:  # noqa: BLE001 — 관측은 부가, 본작업 보호
+        logger.warning(f"[background_jobs] start 실패(무시): {type(exc).__name__}: {exc}")
+        return None
+
+
+async def heartbeat(
+    engine: AsyncEngine,
+    job_id: int | None,
+    *,
+    processed: int | None = None,
+    total: int | None = None,
+    detail: dict | None = None,
+) -> None:
+    """진행 갱신(processed/total/detail). job_id=None 또는 실패 시 no-op."""
+    if job_id is None:
+        return
+    try:
+        async with engine.begin() as conn:
+            await conn.execute(
+                text(
+                    "UPDATE background_jobs SET "
+                    "processed = COALESCE(:p, processed), "
+                    "total = COALESCE(:t, total), "
+                    "detail = COALESCE(CAST(:d AS jsonb), detail), "
+                    "updated_at = now() WHERE id = :id"
+                ),
+                {
+                    "id": job_id,
+                    "p": processed,
+                    "t": total,
+                    "d": json.dumps(detail, ensure_ascii=False) if detail is not None else None,
+                },
+            )
+    except Exception as exc:  # noqa: BLE001
+        logger.warning(f"[background_jobs] heartbeat 실패(무시): {type(exc).__name__}: {exc}")
+
+
+async def finish_job(
+    engine: AsyncEngine, job_id: int | None, *, state: str = "done", error: str | None = None
+) -> None:
+    """종료 기록(done/failed). job_id=None 또는 실패 시 no-op."""
+    if job_id is None:
+        return
+    try:
+        async with engine.begin() as conn:
+            await conn.execute(
+                text(
+                    "UPDATE background_jobs SET state = :s, error = :e, "
+                    "finished_at = now(), updated_at = now() WHERE id = :id"
+                ),
+                {"id": job_id, "s": state, "e": (error or None)},
+            )
+    except Exception as exc:  # noqa: BLE001
+        logger.warning(f"[background_jobs] finish 실패(무시): {type(exc).__name__}: {exc}")

app/services/briefing/comparator.py

@@ -18,12 +18,14 @@ from typing import Any
 import numpy as np
 
 from ai.client import parse_json_response
+from core.config import settings
 from core.utils import setup_logger
 from services.clustering_common import normalize_vector
+from services.search.llm_gate import Priority, acquire_mlx_gate
 
 logger = setup_logger("briefing_comparator")
 
-LLM_CALL_TIMEOUT = 25                 # 초. Phase 4 와 동일
+LLM_CALL_TIMEOUT = settings.digest_llm_timeout_s   # 2026-06-15 config 단일소스 (Phase 4 와 동일 키)
 HISTORICAL_TOP_K = 5
 HISTORICAL_SIMILARITY_MIN = 0.70
 HISTORICAL_WINDOW_DAYS = 30
@@ -39,7 +41,6 @@ MAX_ARTICLE_IDS_PER_COUNTRY = 5       # country_perspectives[].article_ids 후
 FALLBACK_HEADLINE = "LLM 분석 실패로 원문 기사 묶음만 표시합니다."
 FALLBACK_TOPIC_LABEL = "주요 뉴스 묶음"
 
-_llm_sem = asyncio.Semaphore(1)
 _PROMPT_PATH = Path(__file__).resolve().parent.parent.parent / "prompts" / "briefing_comparative.txt"
 _PROMPT_TEMPLATE: str | None = None
 
@@ -112,7 +113,8 @@ def retrieve_historical(
 
 
 async def _try_call_llm(client: Any, prompt: str) -> str:
-    async with _llm_sem:
+    # 전역 MLX gate(BACKGROUND) 경유 — 영구 룰(llm_gate): 새 Semaphore 금지, timeout 은 gate 안쪽.
+    async with acquire_mlx_gate(Priority.BACKGROUND):
         return await asyncio.wait_for(
             client.call_primary(prompt),
             timeout=LLM_CALL_TIMEOUT,
@@ -282,7 +284,7 @@ async def compare_cluster_with_fallback(
     historical_docs = historical_docs or []
     prompt = build_prompt(selected, historical_docs)
 
-    for attempt in range(2):
+    for attempt in range(settings.digest_llm_attempts):  # 2026-06-15 config 단일소스
         try:
             raw = await _try_call_llm(client, prompt)
         except asyncio.TimeoutError:

app/services/briefing/pipeline.py

@@ -6,6 +6,7 @@
 regenerate 정책: briefing_date UNIQUE 충돌 시 transaction 안에서 DELETE+INSERT.
 """
 
+import asyncio
 import time
 from datetime import date, datetime, timedelta, timezone
 from typing import Any
@@ -15,7 +16,9 @@ from sqlalchemy import delete
 
 from ai.client import AIClient
 from core.database import async_session
+from core.database import engine as db_engine
 from core.utils import setup_logger
+from services import background_jobs as bgj
 from models.briefing import BriefingTopic, MorningBriefing
 from services.briefing.clustering import LAMBDA, cluster_global
 from services.briefing.comparator import (
@@ -33,7 +36,6 @@ KST = ZoneInfo("Asia/Seoul")
 NIGHT_WINDOW_HOURS = 5             # KST 00:00 ~ 05:00
 SELECT_K = 7                       # Plan §"Clustering 파라미터" briefing K_PER_CLUSTER=7
 SELECT_LAMBDA_MMR = 0.6            # Plan briefing MMR lambda 0.6
-PIPELINE_HARD_CAP = 600            # 초. Phase 4 와 동일
 
 
 def _compute_window(target_date: date | None = None) -> tuple[datetime, datetime, date]:
@@ -143,7 +145,7 @@ async def _save_briefing(
         return new.id
 
 
-async def run_briefing_pipeline(target_date: date | None = None) -> dict[str, Any]:
+async def run_briefing_pipeline(target_date: date | None = None, job_id: int | None = None) -> dict[str, Any]:
     """야간 뉴스 브리핑 1회 실행. cron 또는 수동 regenerate API 에서 호출.
 
     Returns:
@@ -206,16 +208,36 @@ async def run_briefing_pipeline(target_date: date | None = None) -> dict[str, An
     usable_count = 0
 
     try:
+        # 2026-06-15: cluster 호출 gather 동시 실행. 실동시성 = 전역 MLX gate
+        # (config.mlx_gate_concurrency, BACKGROUND 우선순위). rank/순서 보존.
+        jobs = []
         for rank, cluster in enumerate(clusters, start=1):
             selected = select_for_llm(cluster, k=SELECT_K, lambda_mmr=SELECT_LAMBDA_MMR)
             historical_docs = (
                 retrieve_historical(cluster, historical_candidates)
                 if historical_enabled() else []
             )
-            llm_calls += 1
-            envelope = await compare_cluster_with_fallback(
+            jobs.append((rank, cluster, selected, historical_docs))
+
+        if job_id is not None:
+            await bgj.heartbeat(db_engine, job_id, total=len(jobs))
+        _prog = {"n": 0}
+
+        async def _run_one(cluster, selected, historical_docs):
+            r = await compare_cluster_with_fallback(
                 client, cluster, selected, historical_docs=historical_docs
             )
+            if job_id is not None:
+                _prog["n"] += 1
+                await bgj.heartbeat(db_engine, job_id, processed=_prog["n"])
+            return r
+
+        results = await asyncio.gather(
+            *[_run_one(c, s, h) for (_, c, s, h) in jobs]
+        )
+
+        for (rank, cluster, selected, historical_docs), envelope in zip(jobs, results):
+            llm_calls += 1
             if envelope.get("llm_fallback_used"):
                 llm_failures += 1
             if _is_usable_topic(envelope, envelope["topic_label"]):

app/services/digest/pipeline.py

@@ -10,6 +10,7 @@ Step:
     7. start/end 로그 + generation_ms + fallback 비율 health metric
 """
 
+import asyncio
 import hashlib
 import time
 from datetime import datetime, timedelta, timezone
@@ -19,7 +20,9 @@ from sqlalchemy import delete
 
 from ai.client import AIClient
 from core.database import async_session
+from core.database import engine as db_engine
 from core.utils import setup_logger
+from services import background_jobs as bgj
 from models.digest import DigestTopic, GlobalDigest
 
 from .clustering import LAMBDA, cluster_country
@@ -73,7 +76,7 @@ def _build_topic_row(
     )
 
 
-async def run_digest_pipeline() -> dict:
+async def run_digest_pipeline(job_id: int | None = None) -> dict:
     """전체 파이프라인 실행. worker entry 에서 호출.
 
     Returns:
@@ -107,20 +110,37 @@ async def run_digest_pipeline() -> dict:
     stats = {"llm_calls": 0, "fallback_used": 0}
 
     try:
+        # 2026-06-15: cluster 호출을 gather 로 동시 실행. 실제 동시성은 전역 MLX gate
+        # (config.mlx_gate_concurrency, BACKGROUND 우선순위) 가 제한한다. rank/순서 보존.
+        jobs = []
         for country, docs in docs_by_country.items():
             clusters = cluster_country(country, docs)
             if not clusters:
                 continue  # sparse country 자동 제외
-
             for rank, cluster in enumerate(clusters, start=1):
                 selected = select_for_llm(cluster)
-                stats["llm_calls"] += 1
-                llm_result = await summarize_cluster_with_fallback(client, cluster, selected)
-                if llm_result["llm_fallback_used"]:
-                    stats["fallback_used"] += 1
-                all_topic_rows.append(
-                    _build_topic_row(country, rank, cluster, selected, llm_result, primary_model)
-                )
+                jobs.append((country, rank, cluster, selected))
+
+        if job_id is not None:
+            await bgj.heartbeat(db_engine, job_id, total=len(jobs))
+        _prog = {"n": 0}
+
+        async def _run_one(cluster, selected):
+            r = await summarize_cluster_with_fallback(client, cluster, selected)
+            if job_id is not None:
+                _prog["n"] += 1
+                await bgj.heartbeat(db_engine, job_id, processed=_prog["n"])
+            return r
+
+        results = await asyncio.gather(*[_run_one(c, s) for (_, _, c, s) in jobs])
+
+        for (country, rank, cluster, selected), llm_result in zip(jobs, results):
+            stats["llm_calls"] += 1
+            if llm_result["llm_fallback_used"]:
+                stats["fallback_used"] += 1
+            all_topic_rows.append(
+                _build_topic_row(country, rank, cluster, selected, llm_result, primary_model)
+            )
     finally:
         await client.close()
 

app/services/digest/summarizer.py

@@ -2,8 +2,8 @@
 
 핵심 결정:
 - AIClient._call_chat 직접 호출 (client.py 수정 회피, fallback 로직 재사용)
-- Semaphore(1) 로 MLX 과부하 회피
-- Per-call timeout 25초 (asyncio.wait_for) — MLX hang / fallback Claude API stall 방어
+- 전역 MLX gate(BACKGROUND) 경유로 동시성 제어 (services.search.llm_gate 단일 게이트)
+- Per-call timeout = config.digest_llm_timeout_s (asyncio.wait_for, gate 안쪽)
 - JSON 파싱 실패 → 1회 재시도 → 그래도 실패 시 minimal fallback (drop 금지)
 - fallback: topic_label="주요 뉴스 묶음", summary = top member ai_summary[:200]
 """
@@ -13,15 +13,16 @@ from pathlib import Path
 from typing import Any
 
 from ai.client import parse_json_response
+from core.config import settings
 from core.utils import setup_logger
+from services.search.llm_gate import Priority, acquire_mlx_gate
 
 logger = setup_logger("digest_summarizer")
 
-LLM_CALL_TIMEOUT = 25       # 초. MLX 평균 5초 + tail latency 마진
+# 2026-06-15: config 단일소스 (구 하드코딩 25s = 빠른 Gemma 기준, Qwen 27B 교체 후 누락).
+LLM_CALL_TIMEOUT = settings.digest_llm_timeout_s
 FALLBACK_SUMMARY_LIMIT = 200
 
-_llm_sem = asyncio.Semaphore(1)
-
 _PROMPT_PATH = Path(__file__).resolve().parent.parent.parent / "prompts" / "digest_topic.txt"
 _PROMPT_TEMPLATE: str | None = None
 
@@ -48,8 +49,12 @@ def build_prompt(selected: list[dict]) -> str:
 
 
 async def _try_call_llm(client: Any, prompt: str) -> str:
-    """Semaphore + per-call timeout 으로 감싼 단일 호출."""
-    async with _llm_sem:
+    """전역 MLX gate(BACKGROUND) + per-call timeout 으로 감싼 단일 호출.
+
+    영구 룰(llm_gate): Mac mini endpoint 는 단일 게이트 공유, 새 Semaphore 금지.
+    동시성 lever = config.mlx_gate_concurrency. timeout 은 gate 안쪽에서만.
+    """
+    async with acquire_mlx_gate(Priority.BACKGROUND):
         return await asyncio.wait_for(
             client._call_chat(client.ai.primary, prompt),
             timeout=LLM_CALL_TIMEOUT,
@@ -86,7 +91,7 @@ async def summarize_cluster_with_fallback(
     """
     prompt = build_prompt(selected)
 
-    for attempt in range(2):  # 1회 재시도 포함
+    for attempt in range(settings.digest_llm_attempts):  # config 단일소스 (기본 2 = 1회 재시도)
         try:
             raw = await _try_call_llm(client, prompt)
         except asyncio.TimeoutError:

app/services/papers/__init__.py

@@ -0,0 +1,5 @@
+"""B-3 논문 수집 트랙 공유 모듈 (plan safety-library-b3-1).
+
+doi    — DOI 정규화·dedup 키·2-Document(holder/parent_doi child) extract_meta 계약 (순수).
+holder — 서지 holder 공유 dedup 조회 (DB).
+"""

app/services/papers/doi.py

@@ -0,0 +1,141 @@
+"""B-3 논문 DOI 코어 — 정규화·dedup 키·2-Document(서지 holder / parent_doi child) 계약.
+
+plan safety-library-b3-1 PR1 (keyless·마이그 0).
+
+핵심 계약(모든 논문 수집기·reconcile·구매 PDF 스탬프가 공유):
+- DOI 정규화는 이 단일 함수(normalize_doi) 경유 — **저장=조회 동일 함수**
+  (migration 351 주석 명시, news_collector._normalize_url 의 store=lookup 불변식 선례).
+  같은 논문이 다른 표기(https://doi.org/ vs doi: vs 대문자)로 들어와도 한 holder 로 붕괴.
+- dedup 키 = lower(extract_meta #>> '{paper,doi}') — 라이브 partial-unique 인덱스
+  uq_documents_paper_doi(WHERE material_type='paper' AND ... IS NOT NULL)가 강제.
+- 2-Document(R2-B1): paper.doi 는 **서지 Document 단일 보유**. OA/구매 전문 PDF 는
+  doi 없이 paper.parent_doi 로 holder 링크(NULL doi 라 인덱스 밖 → 다중행 무충돌).
+  holder 와 child 는 doi/parent_doi 를 **상호 배타**로 가진다.
+"""
+
+import hashlib
+import re
+
+# 소문자화 후 비교하므로 전부 소문자 prefix. 긴 것부터(dx.doi.org 가 doi.org 보다 먼저).
+_DOI_PREFIXES = (
+    "https://dx.doi.org/",
+    "http://dx.doi.org/",
+    "https://doi.org/",
+    "http://doi.org/",
+    "dx.doi.org/",
+    "doi.org/",
+    "doi:",
+)
+
+
+def normalize_doi(raw: str | None) -> str | None:
+    """DOI 정규화 — 소문자 + URL/doi: prefix 제거 + 양끝 공백·잡음 제거. 단일 함수(저장=조회).
+
+    유효 DOI(10. 으로 시작)가 아니면 None. 저장측·조회측·dedup 키 생성이 모두 이 함수를
+    공유해야 dedup 이 성립한다(raw 를 그대로 저장하고 정규화로 조회하면 영구 미스).
+    """
+    if not raw:
+        return None
+    s = raw.strip().lower()
+    for p in _DOI_PREFIXES:
+        if s.startswith(p):
+            s = s[len(p):]
+            break
+    s = s.strip()
+    # 인용문 끝 잡음(마침표/쉼표/세미콜론)만 제거. 괄호 '()' 는 DOI 일부일 수 있어 보존한다
+    # (예: 10.1016/s0010-8650(00)80003-2) — 과삭제는 서로 다른 논문을 한 holder 로 병합하는
+    # 데이터 손상이라 near-dup(과소삭제)보다 위험. API 소스(OpenAlex/arXiv)의 doi 는 이미 깨끗.
+    s = s.rstrip(".,;")
+    if not s.startswith("10."):
+        return None
+    return s
+
+
+# arXiv id: 신형 'YYMM.NNNNN'(+vN) 또는 구형 'archive(.SUBJ)/NNNNNNN'. 'arXiv:' 접두 흡수.
+_ARXIV_ID_RE = re.compile(
+    r"arxiv:\s*([a-z\-]+(?:\.[a-z]{2})?/\d{7}|\d{4}\.\d{4,5})(v\d+)?", re.IGNORECASE
+)
+
+
+def parse_arxiv_id(text: str | None) -> str | None:
+    """본문/제목에서 arXiv id(versionless) 추출. 없으면 None. 레거시 reconcile 의 입력."""
+    if not text:
+        return None
+    m = _ARXIV_ID_RE.search(text)
+    return m.group(1) if m else None
+
+
+def arxiv_doi(arxiv_id: str | None) -> str | None:
+    """arXiv DataCite DOI = 10.48550/arxiv.{id} (정규화). 저널 DOI 없는 프리프린트의 canonical
+    paper.doi 통일 키 — OpenAlex 가 프리프린트에 동일 DOI 부여(실측 확인). 모든 수집기·reconcile 가
+    같은 함수로 같은 DOI 를 써야 교차소스 dedup 이 성립."""
+    if not arxiv_id:
+        return None
+    return normalize_doi(f"10.48550/arXiv.{arxiv_id}")
+
+
+_DOI_IN_TEXT_RE = re.compile(r"10\.\d{4,9}/[^\s\"'<>]+", re.IGNORECASE)
+
+
+def parse_doi_from_text(text: str | None) -> str | None:
+    """본문에서 첫 DOI 추출(정규화). 구매 PDF 의 paper.parent_doi 링크용(PDF 구조 무관 — 전체 스캔).
+    DOI 끝 구두점은 normalize_doi 가 정리. 없으면 None."""
+    if not text:
+        return None
+    m = _DOI_IN_TEXT_RE.search(text)
+    return normalize_doi(m.group(0)) if m else None
+
+
+def paper_doi_hash(normalized_doi: str) -> str:
+    """서지 holder 의 Document.file_hash — sha256('paper|{doi}')[:32].
+
+    statute 의 'statute|{jur}|{native_id}|{version_key}' 다중부 키 선례를 따른다.
+    인자는 normalize_doi() 출력(정규화 완료값)이어야 한다 — raw 를 넣으면 dedup 이 깨진다.
+    """
+    if not normalized_doi:
+        raise ValueError("paper_doi_hash 는 정규화된 DOI 필요 (normalize_doi 먼저)")
+    return hashlib.sha256(f"paper|{normalized_doi}".encode()).hexdigest()[:32]
+
+
+def read_paper_doi(extract_meta: dict | None) -> str | None:
+    """holder 의 정규화 DOI 읽기 — 인덱스 식 lower(extract_meta #>> '{paper,doi}') 의 조회측 거울.
+
+    방어적 재정규화(이미 정규화돼 저장되지만 레거시·외부 주입 대비).
+    """
+    if not extract_meta:
+        return None
+    paper = extract_meta.get("paper")
+    if not isinstance(paper, dict):
+        return None
+    return normalize_doi(paper.get("doi"))
+
+
+def with_paper_doi(extract_meta: dict | None, normalized_doi: str) -> dict:
+    """서지 holder 의 extract_meta 에 paper.doi 주입 (merge-safe, 타 키 보존).
+
+    holder 전용 — parent_doi 는 제거(상호 배타). 반환값은 새 dict(입력 비변경).
+    """
+    if not normalized_doi:
+        raise ValueError("with_paper_doi 는 정규화된 DOI 필요")
+    meta = dict(extract_meta or {})
+    paper = dict(meta.get("paper") or {})
+    paper["doi"] = normalized_doi
+    paper.pop("parent_doi", None)
+    meta["paper"] = paper
+    return meta
+
+
+def with_parent_doi(extract_meta: dict | None, parent_normalized_doi: str) -> dict:
+    """child(OA/구매 전문 PDF)의 extract_meta 에 paper.parent_doi 주입 (merge-safe, 타 키 보존).
+
+    child 는 paper.doi 를 갖지 않는다(NULL → partial-unique 인덱스 밖, 2-Document 무충돌).
+    반환값은 새 dict(입력 비변경).
+    """
+    if not parent_normalized_doi:
+        raise ValueError("with_parent_doi 는 정규화된 DOI 필요")
+    meta = dict(extract_meta or {})
+    paper = dict(meta.get("paper") or {})
+    paper["parent_doi"] = parent_normalized_doi
+    paper.pop("doi", None)
+    meta["paper"] = paper
+    return meta

app/services/papers/holder.py

@@ -0,0 +1,38 @@
+"""B-3 논문 서지 holder 공유 dedup 조회.
+
+모든 논문 수집기(OpenAlex/arXiv/KoreaScience/J-STAGE)·reconcile·구매 PDF 스탬프가
+ingest 전 이 함수로 holder 존재를 확인한다(있으면 skip 또는 child 링크).
+
+- 조회 키 = lower(extract_meta #>> '{paper,doi}') == normalize_doi(...) — 라이브 partial-unique
+  인덱스 uq_documents_paper_doi 와 동일 식(인덱스 사용).
+- .scalars().first() — 교차게시·다중 landing-page 로 2행 이상 매칭 시 MultipleResultsFound
+  raise 방지(scalar_one_or_none 금지, 2026-06 BBC 수집 중단 선례 / news_collector 동일 규율).
+- 서지 holder Document 의 **생성**은 각 수집기/스탬프 경로가 소유한다(초록 signal 문서 vs 구매
+  최소 holder 로 shape 가 다름). 이 모듈은 dedup 조회만 공유한다.
+
+DB 조회라 본 모듈은 PR2(arXiv 실수집)에서 라이브 검증한다 — PR1 단위 테스트 대상은 doi.py(순수).
+"""
+
+from sqlalchemy import func, select
+
+from models.document import Document
+from services.papers.doi import normalize_doi
+
+# 인덱스 식과 동일: lower(extract_meta #>> '{paper,doi}')
+_DOI_EXPR = func.lower(Document.extract_meta[("paper", "doi")].astext)
+
+
+async def find_paper_holder(session, raw_or_normalized_doi):
+    """정규화 DOI 로 서지 holder Document 조회. 없으면 None.
+
+    인자는 raw 든 정규화든 받아 normalize_doi 로 통일(저장=조회 동일 함수 보장).
+    """
+    doi = normalize_doi(raw_or_normalized_doi)
+    if not doi:
+        return None
+    result = await session.execute(
+        select(Document)
+        .where(Document.material_type == "paper", _DOI_EXPR == doi)
+        .limit(1)
+    )
+    return result.scalars().first()

app/services/queue_overview.py

@@ -412,7 +412,7 @@ async def build_overview(session: AsyncSession) -> dict:
         for row in current_result
     ]
 
-    return compose_overview(
+    result = compose_overview(
         rows_to_stage_stats(stage_rows),
         rows_to_summarize_split(split_rows),
         {row[0]: int(row[1]) for row in inflow_rows},
@@ -421,6 +421,55 @@ async def build_overview(session: AsyncSession) -> dict:
         deep_enabled=deep_enabled,
         now_kst=now_kst,
     )
+    # 큐 밖 관리 스크립트(백필 등) = background_jobs (migration 357). 테이블 부재 시 graceful([]).
+    result["background_jobs"] = await _fetch_background_jobs(session)
+    return result
+
+
+# kind -> 처리 머신 (보드 머신 카드 귀속용). 미상 kind = gpu(오케스트레이션 호스트).
+_BG_JOB_MACHINE = {
+    "global_digest": "macmini",
+    "morning_briefing": "macmini",
+    "section_summary": "macmini",
+    "hier_backfill": "gpu",
+    "hier_redecompose": "gpu",
+}
+
+
+_BACKGROUND_JOBS_SQL = """
+    SELECT id, kind, label, state, processed, total,
+           EXTRACT(EPOCH FROM (now() - started_at))::int AS elapsed_sec,
+           (state = 'running' AND updated_at < now() - interval '5 minutes') AS stale,
+           error
+    FROM background_jobs
+    WHERE state = 'running' OR finished_at > now() - interval '6 hours'
+    ORDER BY (state = 'running') DESC, started_at DESC
+    LIMIT 20
+"""
+
+
+async def _fetch_background_jobs(session: AsyncSession) -> list[dict]:
+    """running + 최근 6h 완료 background_jobs. 테이블 없거나 오류면 [] (보드 무영향).
+
+    요청 세션과 **별도 connection**으로 조회한다 — 테이블 부재(마이그 357 미적용 등) 시
+    SELECT 실패가 요청 세션의 트랜잭션을 오염시키지 않도록 물리적으로 분리(실패 시 그
+    임시 connection만 폐기). 관측은 부가 기능이라 보드 본체를 절대 깨면 안 된다.
+    """
+    try:
+        async with session.bind.connect() as conn:  # 풀에서 독립 connection
+            rows = (await conn.execute(text(_BACKGROUND_JOBS_SQL))).mappings().all()
+    except Exception:  # noqa: BLE001 — 관측 부가, 보드 본체 보호
+        return []
+    return [
+        {
+            "id": r["id"], "kind": r["kind"], "label": r["label"], "state": r["state"],
+            "processed": int(r["processed"] or 0), "total": r["total"],
+            "elapsed_sec": int(r["elapsed_sec"] or 0), "stale": bool(r["stale"]),
+            "error": r["error"],
+            "machine": _BG_JOB_MACHINE.get(r["kind"], "gpu"),
+        }
+        for r in rows
+    ]
 
 
 # ─── 실패 처리 (plan ds-board-engines-1) ─────────────────────────────────────

app/services/search/retrieval_service.py

@@ -361,7 +361,7 @@ async def search_text(
                        + similarity(coalesce(d.ai_tags::text, ''), :q) * 2.5
                        + similarity(coalesce(d.user_note, ''), :q) * 2.0
                        + similarity(coalesce(d.ai_summary, ''), :q) * 1.5
-                       + similarity(coalesce(d.extracted_text, ''), :q) * 1.0
+                       + similarity(left(coalesce(d.extracted_text, ''), 2000), :q) * 1.0
                        -- FTS 보너스 (idx_documents_fts_full 활용)
                        + coalesce(ts_rank(
                            to_tsvector('simple',
@@ -369,7 +369,7 @@ async def search_text(
                                coalesce(d.ai_tags::text, '') || ' ' ||
                                coalesce(d.ai_summary, '') || ' ' ||
                                coalesce(d.user_note, '') || ' ' ||
-                               coalesce(d.extracted_text, '')
+                               left(coalesce(d.extracted_text, ''), 2000)
                            ),
                            plainto_tsquery('simple', :q)
                        ), 0) * 2.0
@@ -380,7 +380,7 @@ async def search_text(
                        WHEN similarity(coalesce(d.ai_tags::text, ''), :q) >= 0.3 THEN 'tags'
                        WHEN similarity(coalesce(d.user_note, ''), :q) >= 0.3 THEN 'note'
                        WHEN similarity(coalesce(d.ai_summary, ''), :q) >= 0.3 THEN 'summary'
-                       WHEN similarity(coalesce(d.extracted_text, ''), :q) >= 0.3 THEN 'content'
+                       WHEN similarity(left(coalesce(d.extracted_text, ''), 2000), :q) >= 0.3 THEN 'content'
                        ELSE 'fts'
                    END AS match_reason,
                    d.material_type, d.jurisdiction, d.published_date

app/workers/arxiv_collector.py

@@ -0,0 +1,370 @@
+"""arXiv 키워드 필터 수집기 — B-3 PR2 (plan safety-library-b3-1).
+
+bespoke arXiv API(Atom) 수집기. 카테고리 RSS 통째(firehose)가 아니라
+cat:{category} AND (abs:키워드 ...) 로 안전/신뢰성/압력용기 관련분만 좁혀 수집한다.
+
+- signal-only: 초록만 색인(embed+chunk), summarize 절대 미enqueue — 맥미니 Qwen 큐 무접촉.
+- DOI 보유 → paper.doi(서지 holder, partial-unique 인덱스 진입). 없으면 versionless arXiv id 로
+  dedup(향후 PR4 reconcile 가 DOI 백필).
+- etiquette: 요청 간 ≥3s + HTTP 429 지수 백오프. 카테고리별 submittedDate 워터마크로 증분.
+- per-run insert cap(_RUN_CAP) — 광역 수집이 GPU bge-m3 embed 큐를 범람시키지 않게(적대리뷰 A major).
+  잔여는 silent-cap 금지(csb idiom): 누락 건수 로깅.
+- keyless. enabled=False news_sources 행(6h 뉴스 사이클 비대상) + main.py CronTrigger(자체 폴링).
+- arXiv API 는 https 필수(http=301). UA = CRAWL_UA.
+"""
+
+import asyncio
+import hashlib
+import re
+import xml.etree.ElementTree as ET
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+import httpx
+from sqlalchemy import select
+
+from core.crawl_politeness import CRAWL_UA
+from core.database import async_session
+from core.utils import setup_logger
+from models.document import Document
+from models.news_source import NewsSource
+from models.queue import enqueue_stage
+from services.papers.doi import arxiv_doi, normalize_doi
+from services.papers.holder import find_paper_holder
+from workers.news_collector import (
+    FeedError,
+    _get_or_create_health,
+    _record_failure,
+    _record_success,
+)
+
+logger = setup_logger("arxiv_collector")
+
+_ARXIV_API = "https://export.arxiv.org/api/query"
+_SOURCE_NAME = "arXiv 안전·공학 (keyword)"
+
+# 신규 카테고리만 — 기존 RSS 행(id 62 physics.app-ph, id 64 cond-mat.mtrl-sci)과 비중복.
+_CATEGORIES = (
+    "eess.SY",        # systems & control
+    "physics.flu-dyn",  # 유체 — 압력/유동
+    "physics.comp-ph",  # 전산물리
+    "math.OC",        # 최적화·제어
+    "math.NA",        # 수치해석 (FEM 등)
+    "stat.AP",        # 응용통계 — 신뢰성
+    "cs.CE",          # computational engineering
+)
+# 압력용기·공정안전·구조건전성 도메인 키워드(abs: OR 게이트). 좁게 유지 = 관련성↑·볼륨↓ (튜너블).
+_KEYWORDS = (
+    "pressure vessel",
+    "process safety",
+    "structural integrity",
+    "fracture mechanics",
+    "fatigue life",
+    "corrosion",
+)
+
+_RUN_CAP = 80           # 1회 run 신규 적재 상한(임베드 큐 보호). bulk 시 해제.
+_PAGE_SIZE = 50         # max_results per request
+_MAX_PAGES_PER_CAT = 4  # 카테고리당 최대 페이지(증분이라 보통 1페이지에 워터마크 도달)
+_REQ_SLEEP = 3.0        # arXiv etiquette ≥3s
+_MAX_RETRY = 4
+_BACKOFF_BASE = 5.0
+
+_NS = {
+    "a": "http://www.w3.org/2005/Atom",
+    "arxiv": "http://arxiv.org/schemas/atom",
+    "opensearch": "http://a9.com/-/spec/opensearch/1.1/",
+}
+_ABS_ID_RE = re.compile(r"arxiv\.org/abs/(.+?)(v\d+)?$")
+_WS_RE = re.compile(r"\s+")
+
+
+# ───────────────────────── 순수 파서 (fixture 단위 테스트 대상) ─────────────────────────
+
+@dataclass
+class ArxivEntry:
+    arxiv_id: str            # versionless, 예: "1209.2405"
+    version: str | None      # "v1" 또는 None
+    title: str
+    summary: str             # 초록
+    published: datetime | None
+    doi: str | None          # normalize_doi 적용
+    journal_ref: str | None
+    primary_category: str | None
+    categories: list = field(default_factory=list)
+    abs_url: str | None = None
+    pdf_url: str | None = None
+
+
+def _clean(text: str | None) -> str:
+    return _WS_RE.sub(" ", text).strip() if text else ""
+
+
+def _parse_id(raw_id: str | None) -> tuple[str | None, str | None]:
+    """'http://arxiv.org/abs/1209.2405v1' → ('1209.2405', 'v1'). versionless id 가 dedup 키."""
+    m = _ABS_ID_RE.search((raw_id or "").strip())
+    if not m:
+        return None, None
+    return m.group(1), m.group(2)
+
+
+def _parse_dt(s: str | None) -> datetime | None:
+    if not s:
+        return None
+    try:
+        return datetime.fromisoformat(s.replace("Z", "+00:00"))
+    except ValueError:
+        return None
+
+
+def build_search_query(category: str, keywords=_KEYWORDS) -> str:
+    """cat:{category} AND (abs:kw1 OR abs:"kw with space" ...). 공백 키워드는 따옴표 구절."""
+    kw = " OR ".join(f'abs:"{k}"' if " " in k else f"abs:{k}" for k in keywords)
+    return f"cat:{category} AND ({kw})"
+
+
+def parse_arxiv_feed(xml_text: str) -> tuple[int, list[ArxivEntry]]:
+    """arXiv Atom 응답 → (total_results, [ArxivEntry]). 순수 함수."""
+    root = ET.fromstring(xml_text)
+    raw_total = root.findtext("opensearch:totalResults", default="0", namespaces=_NS)
+    try:
+        total = int(raw_total)
+    except (TypeError, ValueError):
+        total = 0
+    entries: list[ArxivEntry] = []
+    for e in root.findall("a:entry", _NS):
+        aid, ver = _parse_id(e.findtext("a:id", namespaces=_NS))
+        if not aid:
+            continue
+        prim = e.find("arxiv:primary_category", _NS)
+        abs_url = pdf_url = None
+        for ln in e.findall("a:link", _NS):
+            if ln.get("rel") == "alternate" and (ln.get("type") or "").startswith("text/html"):
+                abs_url = ln.get("href")
+            elif ln.get("title") == "pdf":
+                pdf_url = ln.get("href")
+        entries.append(ArxivEntry(
+            arxiv_id=aid,
+            version=ver,
+            title=_clean(e.findtext("a:title", namespaces=_NS)),
+            summary=_clean(e.findtext("a:summary", namespaces=_NS)),
+            published=_parse_dt(e.findtext("a:published", namespaces=_NS)),
+            doi=normalize_doi(e.findtext("arxiv:doi", namespaces=_NS)),
+            journal_ref=_clean(e.findtext("arxiv:journal_ref", namespaces=_NS)) or None,
+            primary_category=prim.get("term") if prim is not None else None,
+            categories=[c.get("term") for c in e.findall("a:category", _NS)],
+            abs_url=abs_url,
+            pdf_url=pdf_url,
+        ))
+    return total, entries
+
+
+# ───────────────────────── 적재 (DB — PR2 라이브 검증) ─────────────────────────
+
+def _build_paper_meta(source: NewsSource, entry: ArxivEntry, doi: str | None) -> dict:
+    """extract_meta — license + source + paper 식별. 서지 holder 는 paper.doi(있으면) 보유."""
+    paper: dict = {"arxiv_id": entry.arxiv_id}
+    if doi:
+        paper["doi"] = doi          # partial-unique 인덱스 진입 (교차소스 dedup)
+    if entry.journal_ref:
+        paper["journal_ref"] = entry.journal_ref
+    if entry.primary_category:
+        paper["primary_category"] = entry.primary_category
+    meta: dict = {
+        "source_id": source.id,
+        "source_name": source.name,
+        "source_region": "INT",   # arXiv = 국제 preprint. paper.jurisdiction 은 NULL 유지(A-2).
+        "paper": paper,
+        # arXiv 기본 라이선스 = 비배포(보수적). restricted 부재 → 초록은 RAG 사용 가능.
+        # (명시 CC 검출은 OAI 인터페이스 필요 — Atom API 미포함, PR 후속/관찰.)
+        "license": {"scheme": "arxiv", "redistribute": False, "attribution": "arXiv"},
+    }
+    if entry.published:
+        meta["published_at"] = entry.published.isoformat()
+    return meta
+
+
+async def _ingest_entry(session, source: NewsSource, entry: ArxivEntry) -> bool:
+    """1건 적재. 반환 = 신규 여부. signal-only(embed+chunk, summarize 없음)."""
+    arxiv_hash = hashlib.sha256(f"arxiv|{entry.arxiv_id}".encode()).hexdigest()[:32]
+    # 재수집 dedup(arXiv id) — .first()(다중행 방어)
+    dup = await session.execute(
+        select(Document.id).where(Document.file_hash == arxiv_hash).limit(1)
+    )
+    if dup.scalars().first():
+        return False
+    # arXiv canonical DOI = 저널 DOI 또는 arXiv DataCite DOI(프리프린트도 paper.doi 보유 → PR3 와 dedup)
+    doi = entry.doi or arxiv_doi(entry.arxiv_id)
+    # 교차소스 dedup(DOI holder 이미 존재 — partial-unique 인덱스 백스톱 선제 회피)
+    if doi and await find_paper_holder(session, doi):
+        return False
+
+    body = entry.summary or entry.title
+    doc = Document(
+        file_path=f"crawl/arxiv/{entry.arxiv_id}",
+        file_hash=arxiv_hash,
+        file_format="article",
+        file_size=len(body.encode()),
+        file_type="note",
+        title=entry.title,
+        extracted_text=f"{entry.title}\n\n{body}",
+        extracted_at=datetime.now(timezone.utc),
+        extractor_version="arxiv-api-signal",
+        md_status="skipped",
+        md_extraction_error="arXiv abstract: signal-only, markdown 비대상",
+        source_channel="crawl",
+        data_origin="external",
+        edit_url=entry.abs_url,
+        review_status="approved",
+        material_type="paper",
+        jurisdiction=None,   # paper = NULL 불변(A-2). 지역은 extract_meta.paper.source_region.
+        published_date=entry.published.date() if entry.published else None,
+        extract_meta=_build_paper_meta(source, entry, doi),
+    )
+    session.add(doc)
+    await session.flush()
+    # signal-only: 검색 색인만. summarize/fulltext 절대 enqueue 안 함(맥미니 큐 무접촉).
+    await enqueue_stage(session, doc.id, "embed")
+    await enqueue_stage(session, doc.id, "chunk")
+    return True
+
+
+async def _get_or_create_source(session) -> NewsSource:
+    result = await session.execute(
+        select(NewsSource).where(NewsSource.name == _SOURCE_NAME)
+    )
+    source = result.scalars().first()
+    if source is None:
+        source = NewsSource(
+            name=_SOURCE_NAME, feed_url=_ARXIV_API, feed_type="atom",
+            fetch_method="signal-only", fulltext_policy="none",
+            source_channel="crawl", category="Engineering", language="en",
+            country=None,            # paper → jurisdiction NULL (country 미전파)
+            material_type="paper",
+            license_scheme="arxiv", license_redistribute=False,
+            enabled=False,           # 6h 뉴스 사이클 비대상 — 본 워커가 자체 폴링
+        )
+        session.add(source)
+        await session.flush()
+    return source
+
+
+def _watermark(source: NewsSource, category: str) -> datetime | None:
+    raw = (source.selector_override or {}).get("arxiv_watermark", {}).get(category)
+    if not raw:
+        return None
+    return _parse_dt(raw)
+
+
+def _set_watermark(source: NewsSource, category: str, value: datetime) -> None:
+    cfg = dict(source.selector_override or {})
+    wm = dict(cfg.get("arxiv_watermark") or {})
+    wm[category] = value.isoformat()
+    cfg["arxiv_watermark"] = wm
+    source.selector_override = cfg   # JSONB 변경 감지 위해 재할당
+
+
+async def _fetch(client: httpx.AsyncClient, query: str, start: int) -> str:
+    params = {
+        "search_query": query, "start": start, "max_results": _PAGE_SIZE,
+        "sortBy": "submittedDate", "sortOrder": "descending",
+    }
+    for attempt in range(_MAX_RETRY):
+        resp = await client.get(_ARXIV_API, params=params)
+        if resp.status_code == 429:
+            await asyncio.sleep(_BACKOFF_BASE * (2 ** attempt))
+            continue
+        resp.raise_for_status()
+        return resp.text
+    raise FeedError(f"arXiv 429 재시도 초과: {query[:48]}")
+
+
+async def run(bulk: bool = False, limit: int = 0) -> None:
+    """daily 진입점(스케줄러). bulk/limit 은 CLI 전용(bulk=cap 해제·깊은 페이징)."""
+    now = datetime.now(timezone.utc)
+    async with async_session() as session:
+        source = await _get_or_create_source(session)
+        await session.commit()
+        source_id = source.id
+
+    run_cap = (limit or 10**9) if bulk else (min(limit, _RUN_CAP) if limit else _RUN_CAP)
+    inserted = 0
+    seen = 0
+    failures: list[str] = []
+
+    async with httpx.AsyncClient(
+        timeout=30.0, headers={"User-Agent": CRAWL_UA}, follow_redirects=True
+    ) as client:
+        for category in _CATEGORIES:
+            if inserted >= run_cap:
+                break
+            query = build_search_query(category)
+            async with async_session() as session:
+                src = await session.get(NewsSource, source_id)
+                watermark = _watermark(src, category)
+            newest_seen: datetime | None = None
+            max_pages = (10**6 if bulk else _MAX_PAGES_PER_CAT)
+            try:
+                for page in range(max_pages):
+                    if inserted >= run_cap:
+                        break
+                    xml_text = await _fetch(client, query, page * _PAGE_SIZE)
+                    total, entries = parse_arxiv_feed(xml_text)
+                    if not entries:
+                        break
+                    stop = False
+                    for entry in entries:
+                        seen += 1
+                        if entry.published:
+                            newest_seen = max(newest_seen or entry.published, entry.published)
+                            # 증분: 워터마크 이하 도달 시 이 카테고리 종료(이미 본 구간)
+                            if watermark and not bulk and entry.published <= watermark:
+                                stop = True
+                                break
+                        async with async_session() as session:
+                            src = await session.get(NewsSource, source_id)
+                            if await _ingest_entry(session, src, entry):
+                                inserted += 1
+                                await session.commit()
+                            else:
+                                await session.rollback()
+                        if inserted >= run_cap:
+                            break
+                    await asyncio.sleep(_REQ_SLEEP)
+                    if stop or (page + 1) * _PAGE_SIZE >= total:
+                        break
+                # 카테고리 워터마크 전진(이번 run 최신 발행일)
+                if newest_seen:
+                    async with async_session() as session:
+                        src = await session.get(NewsSource, source_id)
+                        _set_watermark(src, category, newest_seen)
+                        await session.commit()
+            except (httpx.HTTPError, FeedError, ET.ParseError) as e:
+                msg = f"[{category}] {e or repr(e)}"
+                logger.error(f"[arxiv] {msg}")
+                failures.append(msg)
+
+    async with async_session() as session:
+        health = await _get_or_create_health(session, source_id)
+        if failures and inserted == 0:
+            _record_failure(health, "; ".join(failures)[:500], now)
+        else:
+            _record_success(health, inserted, False, now)
+        await session.commit()
+
+    deferred = "" if inserted < run_cap else f" (cap {run_cap} 도달 — 잔여는 다음 run 이월)"
+    logger.info(
+        f"[arxiv] {len(_CATEGORIES)}개 카테고리 스캔 {seen}건 → 신규 {inserted}건{deferred}"
+        + (f" / 실패 {len(failures)}건" if failures else "")
+    )
+
+
+if __name__ == "__main__":
+    # CLI = 수동/백필 전용. --bulk = cap 해제·깊은 페이징, --limit N = 상한 N(라이브 검증용).
+    import argparse
+
+    parser = argparse.ArgumentParser(description="arXiv 안전·공학 키워드 수집기")
+    parser.add_argument("--bulk", action="store_true", help="cap 해제 + 깊은 페이징 백필")
+    parser.add_argument("--limit", type=int, default=0, help="신규 적재 상한(0=기본 cap)")
+    args = parser.parse_args()
+    asyncio.run(run(bulk=args.bulk, limit=args.limit))

app/workers/briefing_worker.py

@@ -9,12 +9,15 @@ import asyncio
 from datetime import date
 
 from core.config import settings
+from core.database import engine as db_engine
 from core.utils import setup_logger
+from services.background_jobs import finish_job, start_job
 from services.briefing.pipeline import run_briefing_pipeline
 
 logger = setup_logger("briefing_worker")
 
-PIPELINE_HARD_CAP = 600
+# 2026-06-15: config 단일소스 (digest 와 공유 키). 구 600s = 빠른 Gemma 기준.
+PIPELINE_HARD_CAP = settings.digest_pipeline_hard_cap_s
 
 
 async def run(target_date: date | None = None) -> dict | None:
@@ -26,19 +29,24 @@ async def run(target_date: date | None = None) -> dict | None:
     if "briefing" in settings.pipeline_held_stages:
         logger.info("[briefing] 보류 (pipeline.held_stages) — 이번 실행 skip")
         return None
+    # 보드 가시화: 큐 밖 cron 생성 작업이라 background_jobs 로 노출 (best-effort, 맥미니 귀속)
+    job_id = await start_job(db_engine, "morning_briefing", label="조간 브리핑 생성")
     try:
         result = await asyncio.wait_for(
-            run_briefing_pipeline(target_date),
+            run_briefing_pipeline(target_date, job_id=job_id),
             timeout=PIPELINE_HARD_CAP,
         )
+        await finish_job(db_engine, job_id, state="done")
         logger.info(f"[briefing] 워커 완료: {result}")
         return result
     except asyncio.TimeoutError:
+        await finish_job(db_engine, job_id, state="failed", error=f"HARD CAP {PIPELINE_HARD_CAP}s 초과")
         logger.error(
             f"[briefing] HARD CAP {PIPELINE_HARD_CAP}s 초과 — 워커 강제 중단. "
             f"기존 briefing 은 commit 시점에만 갱신되므로 그대로 유지됨."
         )
     except Exception as e:
+        await finish_job(db_engine, job_id, state="failed", error=str(e)[:300])
         logger.exception(f"[briefing] 워커 실패: {e}")
     return None
 

app/workers/classify_worker.py

@@ -411,6 +411,15 @@ async def process(
         logger.info(f"doc {document_id}: devonagent → classify skip")
         return
 
+    # 논문(material_type='paper') — 요약/분류 LLM 스킵(맥미니 큐 무접촉, B-3 signal-only 유지).
+    # embed/chunk/markdown 은 queue_consumer 가 chain (early-return 후에도 다음 stage enqueue).
+    if doc.material_type == "paper":
+        if not doc.ai_domain:
+            doc.ai_domain = "논문"
+        await session.commit()
+        logger.info(f"doc {document_id}: paper → classify skip (no summarize)")
+        return
+
     if not doc.extracted_text:
         raise ValueError(f"문서 ID {document_id}: extracted_text가 비어있음")
 

app/workers/digest_worker.py

@@ -11,12 +11,15 @@ global_digests / digest_topics 테이블에 저장한다.
 import asyncio
 
 from core.config import settings
+from core.database import engine as db_engine
 from core.utils import setup_logger
+from services.background_jobs import finish_job, start_job
 from services.digest.pipeline import run_digest_pipeline
 
 logger = setup_logger("digest_worker")
 
-PIPELINE_HARD_CAP = 600  # 10분 hard cap
+# 2026-06-15: config 단일소스 (구 600s = 빠른 Gemma 기준, Qwen 27B 교체 후 누락 → 초과).
+PIPELINE_HARD_CAP = settings.digest_pipeline_hard_cap_s
 
 
 async def run() -> None:
@@ -28,19 +31,24 @@ async def run() -> None:
     if "digest" in settings.pipeline_held_stages:
         logger.info("[global_digest] 보류 (pipeline.held_stages) — 이번 실행 skip")
         return
+    # 보드 가시화: 큐 밖 cron 생성 작업이라 background_jobs 로 노출 (best-effort, 맥미니 귀속)
+    job_id = await start_job(db_engine, "global_digest", label="글로벌 다이제스트 생성")
     try:
         result = await asyncio.wait_for(
-            run_digest_pipeline(),
+            run_digest_pipeline(job_id=job_id),
             timeout=PIPELINE_HARD_CAP,
         )
+        await finish_job(db_engine, job_id, state="done")
         logger.info(f"[global_digest] 워커 완료: {result}")
     except asyncio.TimeoutError:
+        await finish_job(db_engine, job_id, state="failed", error=f"HARD CAP {PIPELINE_HARD_CAP}s 초과")
         logger.error(
             f"[global_digest] HARD CAP {PIPELINE_HARD_CAP}s 초과 — 워커 강제 중단. "
             f"기존 digest 는 commit 시점에만 갱신되므로 그대로 유지됨. "
             f"다음 cron 실행에서 재시도."
         )
     except Exception as e:
+        await finish_job(db_engine, job_id, state="failed", error=str(e)[:300])
         logger.exception(f"[global_digest] 워커 실패: {e}")
 
 

app/workers/openalex_collector.py

@@ -0,0 +1,393 @@
+"""OpenAlex 백본 수집기 — B-3 PR3 (plan safety-library-b3-1).
+
+OpenAlex = 발견+dedup 글로벌 백본(JP/EU/US 논문 다 색인 + 정본 DOI). 전문은 안 줌(oa_url 포인터만).
+- scaffold-first: OPENALEX_API_KEY 부재 시 FeedError(explicit-skip, silent fallback 금지). 키=무료.
+- signal-only: 초록(inverted-index 복원)만 색인(embed+chunk), summarize 절대 미enqueue(맥미니 큐 무접촉).
+  PDF 는 절대 OpenAlex 경유로 안 받음(oa_url 은 링크/신호일 뿐).
+- 관련성 사전필터 = title_and_abstract.search 키워드(서버측) + per-run insert cap(임베드 firehose 차단,
+  적대리뷰 A major). cursor 페이징 + from_publication_date 워터마크로 증분.
+- 초록 없는 thin 레코드(주로 비-OA 메타)는 skip — Phase-1 재료 품질 유지.
+- DOI → paper.doi(holder, partial-unique 인덱스, 교차소스 dedup). 없으면 openalex id fallback.
+- license: 명시 CC → redistribute=true / 그 외 OA·closed → false(restricted 부재 = 초록 RAG 사용 가능).
+- enabled=False news_sources 행 + main.py CronTrigger(자체 폴링). list+filter 비용 미미($1/일 크레딧).
+"""
+
+import asyncio
+import hashlib
+import json
+import os
+from dataclasses import dataclass
+from datetime import date, datetime, timezone
+
+import httpx
+from sqlalchemy import select
+
+from core.crawl_politeness import CRAWL_UA
+from core.database import async_session
+from core.utils import setup_logger
+from models.document import Document
+from models.news_source import NewsSource
+from models.queue import enqueue_stage
+from services.papers.doi import normalize_doi
+from services.papers.holder import find_paper_holder
+from workers.news_collector import (
+    FeedError,
+    _get_or_create_health,
+    _record_failure,
+    _record_success,
+)
+
+logger = setup_logger("openalex_collector")
+
+_API = "https://api.openalex.org/works"
+_SOURCE_NAME = "OpenAlex 안전·공학 (keyword)"
+_ENV_KEY = "OPENALEX_API_KEY"
+
+# 압력용기·공정안전·구조건전성 도메인 키워드(키워드별 1쿼리 = 관련성 사전필터).
+_KEYWORDS = (
+    "pressure vessel safety",
+    "process safety",
+    "structural integrity",
+    "fracture mechanics",
+    "fatigue life assessment",
+)
+
+# 도메인 직결 저널 ISSN 시드(OpenAlex sources 실측 확인) — 키워드 매칭 누락분까지 전수 커버.
+# KR 안전/가스/기계 + JP 고압. KR/JP 관심 = OpenAlex 깨끗한 API 로 직접(KoreaScience/J-STAGE 전용
+# 스크래퍼 불요 — Phase-1 메타는 OpenAlex 와 중복, 전용 수집기의 유니크 가치=무료 전문 PDF=Phase-2).
+_JOURNAL_ISSNS = (
+    ("한국안전학회지", "1738-3803"),
+    ("한국가스학회지", "1226-8402"),
+    ("대한기계학회논문집 A", "1226-4873"),
+    ("대한기계학회논문집 B", "1226-4881"),
+    ("KSME International J.", "1226-4865"),
+    ("Review of High Pressure Sci&Tech (JP)", "0917-639X"),
+)
+
+_RUN_CAP = 60           # 1회 run 신규 적재 상한(임베드 큐 보호). bulk 시 해제.
+_PER_PAGE = 50
+_MAX_PAGES_PER_KW = 4   # 키워드당 최대 페이지(증분이라 보통 1페이지에 워터마크 도달)
+_REQ_SLEEP = 1.0        # 페이지 간 polite 간격
+_MAX_RETRY = 4
+_BACKOFF_BASE = 5.0
+
+
+# ───────────────────────── 순수 파서 (fixture 단위 테스트 대상) ─────────────────────────
+
+@dataclass
+class OpenAlexWork:
+    openalex_id: str           # "W2910511816"
+    doi: str | None            # normalize_doi 적용
+    title: str
+    abstract: str              # inverted-index 복원 (없으면 "")
+    publication_date: str | None
+    oa_status: str | None      # closed/green/bronze/hybrid/gold/diamond
+    oa_url: str | None
+    is_oa: bool
+    license: str | None        # cc-by / cc-by-nc-nd / None
+    source_name: str | None
+    primary_topic: str | None
+    work_type: str | None
+
+
+def _clean(text):
+    return " ".join(text.split()).strip() if text else ""
+
+
+def _reconstruct_abstract(inv: dict | None) -> str:
+    """abstract_inverted_index({word:[positions]}) → 평문 초록. 없으면 ''."""
+    if not inv:
+        return ""
+    positions = [(pos, word) for word, idxs in inv.items() for pos in idxs]
+    positions.sort()
+    return " ".join(w for _, w in positions)
+
+
+def license_meta(license_str: str | None, is_oa: bool, source_name: str | None) -> dict:
+    """extract_meta.license — 명시 CC/public-domain 만 redistribute=true. restricted 부재(초록 색인 자유).
+
+    redistribute=false 라도 restricted 가 없으면 RAG 사용 가능(초록). 비-CC 전문의 RAG verbatim 차단은
+    Phase-2 전문 승격 단계가 restricted=true 로 처리(L-1) — Phase-1(초록)은 무해.
+    """
+    attribution = source_name or "OpenAlex"
+    if license_str and (license_str.startswith("cc") or license_str == "public-domain"):
+        return {"scheme": license_str, "redistribute": True, "attribution": attribution}
+    return {
+        "scheme": "open-unspecified" if is_oa else "proprietary",
+        "redistribute": False,
+        "attribution": attribution,
+    }
+
+
+def parse_openalex_works(json_text: str) -> tuple[int, str | None, list[OpenAlexWork]]:
+    """OpenAlex /works 응답 → (count, next_cursor, [OpenAlexWork]). 순수 함수."""
+    d = json.loads(json_text)
+    meta = d.get("meta") or {}
+    count = meta.get("count") or 0
+    next_cursor = meta.get("next_cursor")
+    works: list[OpenAlexWork] = []
+    for w in d.get("results") or []:
+        oid = (w.get("id") or "").rstrip("/").rsplit("/", 1)[-1]
+        if not oid:
+            continue
+        oa = w.get("open_access") or {}
+        pl = w.get("primary_location") or {}
+        pt = w.get("primary_topic") or {}
+        works.append(OpenAlexWork(
+            openalex_id=oid,
+            doi=normalize_doi(w.get("doi")),
+            title=_clean(w.get("title")),
+            abstract=_reconstruct_abstract(w.get("abstract_inverted_index")),
+            publication_date=w.get("publication_date"),
+            oa_status=oa.get("oa_status"),
+            oa_url=oa.get("oa_url") or None,
+            is_oa=bool(oa.get("is_oa")),
+            license=pl.get("license"),
+            source_name=(pl.get("source") or {}).get("display_name"),
+            primary_topic=pt.get("display_name"),
+            work_type=w.get("type"),
+        ))
+    return count, next_cursor, works
+
+
+def build_filter(keyword: str, from_date: str | None = None) -> str:
+    f = f"title_and_abstract.search:{keyword}"
+    if from_date:
+        f += f",from_publication_date:{from_date}"
+    return f
+
+
+def build_issn_filter(issn: str, from_date: str | None = None) -> str:
+    f = f"primary_location.source.issn:{issn}"
+    if from_date:
+        f += f",from_publication_date:{from_date}"
+    return f
+
+
+def _seeds() -> list[tuple[str, str, str]]:
+    """수집 시드 = (라벨, 워터마크키, 종류). 도메인 저널 ISSN 우선(cap 우선권) → 키워드."""
+    s: list[tuple[str, str, str]] = [(label, issn, "issn") for label, issn in _JOURNAL_ISSNS]
+    s += [(kw, kw, "kw") for kw in _KEYWORDS]
+    return s
+
+
+# ───────────────────────── 적재 (DB — PR3 라이브 검증) ─────────────────────────
+
+def _build_paper_meta(source: NewsSource, w: OpenAlexWork) -> dict:
+    paper: dict = {"openalex_id": w.openalex_id}
+    if w.doi:
+        paper["doi"] = w.doi          # partial-unique 인덱스 진입(교차소스 dedup)
+    if w.oa_status:
+        paper["oa_status"] = w.oa_status
+    if w.oa_url:
+        paper["oa_url"] = w.oa_url     # 링크/신호 — 자동 fetch 안 함
+    if w.primary_topic:
+        paper["topic"] = w.primary_topic
+    meta: dict = {
+        "source_id": source.id,
+        "source_name": source.name,
+        "source_region": "INT",        # OpenAlex = 글로벌. paper.jurisdiction 은 NULL 유지(A-2).
+        "paper": paper,
+        "license": license_meta(w.license, w.is_oa, w.source_name),
+    }
+    if w.publication_date:
+        meta["published_at"] = w.publication_date
+    return meta
+
+
+async def _ingest_work(session, source: NewsSource, w: OpenAlexWork) -> bool:
+    """1건 적재. 반환 = 신규 여부. signal-only. 초록 없으면 skip(thin 레코드 배제)."""
+    if not w.abstract:
+        return False  # 초록 없는 thin 레코드(주로 비-OA 메타) — Phase-1 재료 품질 유지
+    oid_hash = hashlib.sha256(f"openalex|{w.openalex_id}".encode()).hexdigest()[:32]
+    dup = await session.execute(
+        select(Document.id).where(Document.file_hash == oid_hash).limit(1)
+    )
+    if dup.scalars().first():
+        return False
+    if w.doi and await find_paper_holder(session, w.doi):
+        return False  # 교차소스 dedup(arXiv 등이 이미 holder 보유)
+
+    pub_date = None
+    if w.publication_date:
+        try:
+            pub_date = date.fromisoformat(w.publication_date)
+        except ValueError:
+            pub_date = None
+    body = w.abstract
+    doc = Document(
+        file_path=f"crawl/openalex/{w.openalex_id}",
+        file_hash=oid_hash,
+        file_format="article",
+        file_size=len(body.encode()),
+        file_type="note",
+        title=w.title,
+        extracted_text=f"{w.title}\n\n{body}",
+        extracted_at=datetime.now(timezone.utc),
+        extractor_version="openalex-signal",
+        md_status="skipped",
+        md_extraction_error="OpenAlex abstract: signal-only, markdown 비대상",
+        source_channel="crawl",
+        data_origin="external",
+        edit_url=w.oa_url or f"https://openalex.org/{w.openalex_id}",
+        review_status="approved",
+        material_type="paper",
+        jurisdiction=None,
+        published_date=pub_date,
+        extract_meta=_build_paper_meta(source, w),
+    )
+    session.add(doc)
+    await session.flush()
+    await enqueue_stage(session, doc.id, "embed")
+    await enqueue_stage(session, doc.id, "chunk")
+    return True
+
+
+async def _get_or_create_source(session) -> NewsSource:
+    result = await session.execute(
+        select(NewsSource).where(NewsSource.name == _SOURCE_NAME)
+    )
+    source = result.scalars().first()
+    if source is None:
+        source = NewsSource(
+            name=_SOURCE_NAME, feed_url=_API, feed_type="json",
+            fetch_method="signal-only", fulltext_policy="none",
+            source_channel="crawl", category="Engineering", language="en",
+            country=None, material_type="paper",
+            license_scheme="openalex", license_redistribute=False,
+            enabled=False,
+        )
+        session.add(source)
+        await session.flush()
+    return source
+
+
+def _api_key() -> str:
+    key = os.getenv(_ENV_KEY, "").strip()
+    if not key:
+        raise FeedError(f"{_ENV_KEY} 미설정 — OpenAlex 수집 불가 (scaffold-first explicit-skip)")
+    return key
+
+
+def _watermark(source: NewsSource, keyword: str) -> str | None:
+    return (source.selector_override or {}).get("openalex_watermark", {}).get(keyword)
+
+
+def _set_watermark(source: NewsSource, keyword: str, value: str) -> None:
+    cfg = dict(source.selector_override or {})
+    wm = dict(cfg.get("openalex_watermark") or {})
+    wm[keyword] = value
+    cfg["openalex_watermark"] = wm
+    source.selector_override = cfg
+
+
+async def _fetch(client: httpx.AsyncClient, key: str, filter_str: str, cursor: str) -> str:
+    params = {
+        "filter": filter_str, "per-page": _PER_PAGE, "cursor": cursor,
+        "sort": "publication_date:desc", "api_key": key,
+    }
+    for attempt in range(_MAX_RETRY):
+        resp = await client.get(_API, params=params)
+        if resp.status_code == 429:
+            await asyncio.sleep(_BACKOFF_BASE * (2 ** attempt))
+            continue
+        resp.raise_for_status()
+        return resp.text
+    raise FeedError(f"OpenAlex 429 재시도 초과: {filter_str[:48]}")
+
+
+async def run(bulk: bool = False, limit: int = 0) -> None:
+    """daily 진입점(스케줄러). 키 부재 = explicit-skip(health 실패 기록)."""
+    now = datetime.now(timezone.utc)
+    async with async_session() as session:
+        source = await _get_or_create_source(session)
+        await session.commit()
+        source_id = source.id
+
+    try:
+        key = _api_key()
+    except FeedError as e:
+        logger.warning(f"[openalex] {e}")
+        async with async_session() as session:
+            health = await _get_or_create_health(session, source_id)
+            _record_failure(health, str(e), now)
+            await session.commit()
+        return
+
+    run_cap = (limit or 10**9) if bulk else (min(limit, _RUN_CAP) if limit else _RUN_CAP)
+    inserted = 0
+    seen = 0
+    failures: list[str] = []
+
+    async with httpx.AsyncClient(
+        timeout=30.0, headers={"User-Agent": CRAWL_UA}, follow_redirects=True
+    ) as client:
+        for label, wm_key, kind in _seeds():
+            if inserted >= run_cap:
+                break
+            async with async_session() as session:
+                src = await session.get(NewsSource, source_id)
+                watermark = None if bulk else _watermark(src, wm_key)
+            filter_str = (build_issn_filter(wm_key, watermark) if kind == "issn"
+                          else build_filter(wm_key, watermark))
+            newest: str | None = None
+            cursor = "*"
+            max_pages = (10**6 if bulk else _MAX_PAGES_PER_KW)
+            try:
+                for _page in range(max_pages):
+                    if inserted >= run_cap:
+                        break
+                    text = await _fetch(client, key, filter_str, cursor)
+                    _count, next_cursor, works = parse_openalex_works(text)
+                    if not works:
+                        break
+                    for w in works:
+                        seen += 1
+                        if w.publication_date and (newest is None or w.publication_date > newest):
+                            newest = w.publication_date
+                        async with async_session() as session:
+                            src = await session.get(NewsSource, source_id)
+                            if await _ingest_work(session, src, w):
+                                inserted += 1
+                                await session.commit()
+                            else:
+                                await session.rollback()
+                        if inserted >= run_cap:
+                            break
+                    await asyncio.sleep(_REQ_SLEEP)
+                    if not next_cursor:
+                        break
+                    cursor = next_cursor
+                if newest:
+                    async with async_session() as session:
+                        src = await session.get(NewsSource, source_id)
+                        _set_watermark(src, wm_key, newest)
+                        await session.commit()
+            except (httpx.HTTPError, FeedError, ValueError) as e:
+                msg = f"[{label}] {e or repr(e)}"
+                logger.error(f"[openalex] {msg}")
+                failures.append(msg)
+
+    async with async_session() as session:
+        health = await _get_or_create_health(session, source_id)
+        if failures and inserted == 0:
+            _record_failure(health, "; ".join(failures)[:500], now)
+        else:
+            _record_success(health, inserted, False, now)
+        await session.commit()
+
+    deferred = "" if inserted < run_cap else f" (cap {run_cap} 도달 — 잔여 다음 run 이월)"
+    logger.info(
+        f"[openalex] {len(_seeds())}개 시드(ISSN+키워드) 스캔 {seen}건 → 신규 {inserted}건{deferred}"
+        + (f" / 실패 {len(failures)}건" if failures else "")
+    )
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(description="OpenAlex 안전·공학 키워드 백본 수집기")
+    parser.add_argument("--bulk", action="store_true", help="cap 해제 + 깊은 cursor 페이징 백필")
+    parser.add_argument("--limit", type=int, default=0, help="신규 적재 상한(0=기본 cap)")
+    args = parser.parse_args()
+    asyncio.run(run(bulk=args.bulk, limit=args.limit))

app/workers/paper_doi_reconcile.py

@@ -0,0 +1,102 @@
+"""paper DOI reconcile — B-3 PR4(레거시 arXiv) + PR5(구매 PDF) (plan safety-library-b3-1).
+
+paper.doi/parent_doi 둘 다 없는 paper 행을 두 갈래로 정리:
+- 레거시 arXiv 초록(holder): arXiv id → arxiv_doi(10.48550/arxiv.{id}) 스탬프 → partial-unique
+  인덱스 편입 → 재유입 차단('동일-DOI 재유입 차단만').
+- 구매 PDF(child, license.restricted=true — Papers_Purchased 드롭): 본문 DOI 파싱 → paper.parent_doi
+  링크(서지 holder 와 DOI 공유로 연결). child 는 doi 미보유(인덱스 밖) → unique 무충돌.
+
+- KEYLESS·결정적(OpenAlex 호출 0)·in-DB·enqueue 0(콘텐츠 무변경). dedup_reconcile(file_hash 캐시)와
+  별 worker(적대리뷰 B·C major). 선재 DOI holder 존재 시 arXiv 행도 parent_doi 마킹(unique 위반 회피).
+"""
+
+import asyncio
+
+from sqlalchemy import select
+
+from core.database import async_session
+from core.utils import setup_logger
+from models.document import Document
+from services.papers.doi import (
+    arxiv_doi,
+    parse_arxiv_id,
+    parse_doi_from_text,
+    with_paper_doi,
+    with_parent_doi,
+)
+from services.papers.holder import find_paper_holder
+
+logger = setup_logger("paper_doi_reconcile")
+
+_DOI_TEXT = Document.extract_meta[("paper", "doi")].astext
+_PARENT_DOI_TEXT = Document.extract_meta[("paper", "parent_doi")].astext
+
+
+def _is_restricted(meta: dict) -> bool:
+    return (meta.get("license") or {}).get("restricted") in (True, "true")
+
+
+async def run(limit: int = 0) -> None:
+    """paper.doi/parent_doi 없는 paper 행 reconcile(멱등). limit=0 = 전건."""
+    stamped = marked_dup = skipped_no_arxiv = 0
+    linked_purchased = skipped_purchased_no_doi = 0
+    async with async_session() as session:
+        q = (
+            select(Document)
+            .where(
+                Document.material_type == "paper",
+                _DOI_TEXT.is_(None),
+                _PARENT_DOI_TEXT.is_(None),
+            )
+            .order_by(Document.id)
+        )
+        if limit:
+            q = q.limit(limit)
+        rows = (await session.execute(q)).scalars().all()
+
+        for row in rows:
+            meta = dict(row.extract_meta or {})
+            paper = dict(meta.get("paper") or {})
+
+            # PR5: 구매 PDF(restricted) = child → 본문 DOI 파싱 → parent_doi 링크
+            if _is_restricted(meta):
+                doi = parse_doi_from_text(row.extracted_text)
+                if not doi:
+                    skipped_purchased_no_doi += 1
+                    continue
+                row.extract_meta = with_parent_doi(meta, doi)
+                linked_purchased += 1
+                continue
+
+            # PR4: 레거시 arXiv 초록(holder) = arXiv DataCite DOI 스탬프
+            arxiv_id = paper.get("arxiv_id") or parse_arxiv_id(row.extracted_text)
+            doi = arxiv_doi(arxiv_id)
+            if not doi:
+                skipped_no_arxiv += 1
+                continue
+            paper["arxiv_id"] = arxiv_id
+            meta["paper"] = paper
+            holder = await find_paper_holder(session, doi)
+            if holder is not None and holder.id != row.id:
+                row.extract_meta = with_parent_doi(meta, doi)  # 선재 중복 → child 마킹
+                marked_dup += 1
+            else:
+                row.extract_meta = with_paper_doi(meta, doi)   # holder 스탬프, 인덱스 진입
+                stamped += 1
+            # 콘텐츠 무변경 → enqueue 없음(summarize/embed/chunk 0)
+        await session.commit()
+
+    logger.info(
+        f"[paper_doi_reconcile] {len(rows)}행 → arXiv 스탬프 {stamped} · 선재중복 {marked_dup} · "
+        f"arXiv id 없음 skip {skipped_no_arxiv} / 구매PDF parent_doi 링크 {linked_purchased} · "
+        f"구매PDF DOI 없음 skip {skipped_purchased_no_doi}"
+    )
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(description="paper DOI reconcile (arXiv 레거시 + 구매 PDF, keyless)")
+    parser.add_argument("--limit", type=int, default=0, help="처리 상한(0=전건)")
+    args = parser.parse_args()
+    asyncio.run(run(limit=args.limit))

app/workers/paper_fulltext_promote.py

@@ -0,0 +1,123 @@
+"""논문 arXiv 전문 승격 (in-place) — B-3 Phase-2 P2-PR1 (plan safety-library-b3-1).
+
+arXiv 프리프린트 초록 행(file_format='article', signal-only)을 전문 PDF로 **in-place 승격**:
+PDF 다운로드 → file_format/file_type/file_path/md_status 갱신 → 'extract' enqueue → 기존 파이프라인
+(extract → classify[paper skip summarize] → embed/chunk/markdown)이 전문 검색 청크 + md_content(marker 표시)
++ hier 절구조를 생성. 1-Document(2행 분리 회피, 기존 display 스택 재사용).
+
+- arXiv = 공개 프리프린트(arxiv.org/pdf/{id}, friendly host) → 전문 검색/RAG 무난, restricted 불요.
+  (유료 구매 논문은 Papers_Purchased 경로가 restricted=true 로 별개 처리.)
+- per-run cap (marker GPU ~10GB + embed 부하 보호, 4070 16GB 빡빡 → idle-unload·증분). keyless.
+- 요약 0 (classify paper-skip 가드). file_hash·extract_meta.paper 보존(수집기 dedup 무영향).
+- CLI 전용(Phase-2 deliberate 승격, GPU 부하 사용자 통제). 스케줄 잡 미등록.
+"""
+
+import argparse
+import asyncio
+import random
+from pathlib import Path
+
+import httpx
+from sqlalchemy import or_, select
+
+from core.config import settings
+from core.crawl_politeness import CRAWL_UA
+from core.database import async_session
+from core.utils import setup_logger
+from models.document import Document
+from models.queue import enqueue_stage
+
+logger = setup_logger("paper_fulltext_promote")
+
+_ARXIV_PDF = "https://arxiv.org/pdf/{id}"
+_MAX_FILE_BYTES = 50 * 1024 * 1024
+_DOWNLOAD_DELAY = (2.0, 5.0)
+_RUN_CAP = 10  # 1회 승격 상한(marker/embed GPU 보호). bulk 시 해제.
+
+_ARXIV_ID_EXPR = Document.extract_meta[("paper", "arxiv_id")].astext
+_OA_URL_EXPR = Document.extract_meta[("paper", "oa_url")].astext
+_OA_STATUS_EXPR = Document.extract_meta[("paper", "oa_status")].astext
+_REAL_OA = ("gold", "hybrid", "green", "diamond")
+
+
+async def _download(url: str, dest: Path) -> int:
+    """arXiv PDF 다운로드 — 크기 cap + PDF 헤더 검증 + 연속 간격(kosha 패턴)."""
+    await asyncio.sleep(random.uniform(*_DOWNLOAD_DELAY))
+    async with httpx.AsyncClient(timeout=60, follow_redirects=True) as client:
+        resp = await client.get(url, headers={"User-Agent": CRAWL_UA})
+    if resp.status_code != 200:
+        raise RuntimeError(f"arXiv PDF {resp.status_code}: {url}")
+    if len(resp.content) > _MAX_FILE_BYTES:
+        raise RuntimeError(f"크기 초과 {len(resp.content)}b: {url}")
+    if resp.content[:5] != b"%PDF-":
+        raise RuntimeError(f"PDF 아님(헤더 {resp.content[:8]!r}): {url}")
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    dest.write_bytes(resp.content)
+    return len(resp.content)
+
+
+async def run(bulk: bool = False, limit: int = 0) -> None:
+    """미승격 arXiv 논문(file_format='article')을 전문 PDF로 in-place 승격."""
+    cap = (limit or 10**9) if bulk else (min(limit, _RUN_CAP) if limit else _RUN_CAP)
+    async with async_session() as session:
+        q = (
+            select(Document.id)
+            .where(
+                Document.material_type == "paper",
+                Document.file_format == "article",
+                or_(
+                    _ARXIV_ID_EXPR.isnot(None),
+                    Document.extract_meta[("paper", "oa_url")].astext.isnot(None),
+                ),
+            )
+            .order_by(Document.id.desc())
+            .limit(cap)
+        )
+        ids = [r[0] for r in (await session.execute(q)).all()]
+
+    promoted = failed = 0
+    for doc_id in ids:
+        async with async_session() as session:
+            doc = await session.get(Document, doc_id)
+            if doc is None or doc.file_format != "article":
+                continue
+            paper = (doc.extract_meta or {}).get("paper") or {}
+            arxiv_id = paper.get("arxiv_id")
+            oa_status = (paper.get("oa_status") or "").lower()
+            if arxiv_id:
+                url = _ARXIV_PDF.format(id=arxiv_id)
+                key = arxiv_id.replace("/", "_")
+            elif paper.get("oa_url") and oa_status in _REAL_OA:
+                url = paper["oa_url"]  # doi.org/KISTI/PMC (friendly OA). 비-OA·paywall 은 헤더검증서 skip
+                key = (paper.get("openalex_id") or paper.get("doi") or "oa").replace("/", "_")
+            else:
+                continue
+            rel_path = f"crawl_raw/papers/{key}.pdf"
+            dest = Path(settings.nas_mount_path) / rel_path
+            try:
+                size = await _download(url, dest)
+            except Exception as e:  # noqa: BLE001 — 다운로드 실패 격리
+                logger.error(f"[promote] {key} 다운로드 실패: {e}")
+                failed += 1
+                continue
+            # in-place 승격: 초록 행 → 전문 PDF 행 (file_hash·extract_meta.paper 보존)
+            doc.file_path = rel_path
+            doc.file_format = "pdf"
+            doc.file_type = "immutable"
+            doc.file_size = size
+            doc.md_status = "pending"  # marker 재실행(기존 'skipped' 해제)
+            doc.md_extraction_error = None
+            await enqueue_stage(session, doc.id, "extract")
+            await session.commit()
+            promoted += 1
+            logger.info(f"[promote] {key} → 전문 PDF in-place (doc {doc.id}, {size}b)")
+
+    logger.info(f"[paper_fulltext_promote] 승격 {promoted} · 실패 {failed} (cap {cap})")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="논문 arXiv 전문 승격 (in-place, keyless)")
+    parser.add_argument("--bulk", action="store_true", help="cap 해제(전건 백필 — GPU 부하 주의)")
+    parser.add_argument("--limit", type=int, default=0, help="승격 상한(0=기본 cap 10)")
+    args = parser.parse_args()
+    asyncio.run(run(bulk=args.bulk, limit=args.limit))

app/workers/queue_consumer.py

@@ -47,10 +47,15 @@ MARKDOWN_STALE_THRESHOLD_MINUTES = int(os.getenv("MARKDOWN_STALE_MINUTES", "120"
 # STT 도 장기 작업 가능성이 있으나 본 PR 범위 밖 — main 에 유지(follow-up).
 MAIN_QUEUE_STAGES = [
     "extract", "classify", "summarize",
-    "preview", "stt", "thumbnail", "deep_summary", "fulltext",
+    "preview", "stt", "thumbnail", "fulltext",
 ]
 MARKDOWN_QUEUE_STAGES = ["markdown"]
 
+# 2026-06-15: deep_summary(26B, 콜당 70~300s)를 메인 루프에서 분리 (markdown/fast 선례).
+# 단일 deep 호출이 1분 틱을 초과해 메인 consume_queue 가 영구 coalesce 되고 extract/
+# classify 등 경량 stage 까지 굶던 문제 제거. 집합 disjoint(자기 집합만 stale reset).
+DEEP_QUEUE_STAGES = ["deep_summary"]
+
 # 고속(비-LLM·경량 GPU) stage — LLM 사이클(분 단위)에서 분리해 1분 잡 전용 소비.
 # embed/chunk 는 건당 <1s 라 main 루프에 두면 classify(~190s×3) 뒤에서 굶는다
 # (2026-06-12 실측: 적체 3,570 · 4070 가동률 0%). markdown 분리(05-01)와 동일 패턴.
@@ -405,3 +410,24 @@ async def consume_markdown_queue():
 
     for stage in MARKDOWN_QUEUE_STAGES:
         await _process_stage(stage, workers[stage])
+
+
+async def consume_deep_queue():
+    """deep_summary 전용 큐 소비자 (2026-06-15) — 26B 심층요약을 메인 파이프라인과 분리.
+
+    deep_summary 1콜이 70~300s(맥미니 Qwen 27B 폴백)라 메인 consume_queue(1분 틱) 안에
+    있으면 매 틱이 interval 을 초과해 영구 "maximum running instances" coalesce 되고
+    extract/classify 등 경량 stage 까지 함께 굶었다. 분리 후 = deep 만 자기 1분 잡에서
+    coalesce, 나머지 메인 루프는 틱 내 완료. max_instances=1 로 동시 deep 2건은 방지.
+    """
+    workers = _load_workers()
+
+    try:
+        await reset_stale_items(DEEP_QUEUE_STAGES, STALE_THRESHOLD_MINUTES)
+    except Exception:
+        logger.exception("deep stale reset failed, but continuing queue consumption")
+
+    for stage in DEEP_QUEUE_STAGES:
+        if stage in settings.pipeline_held_stages:
+            continue
+        await _process_stage(stage, workers[stage])

config.yaml

@@ -13,7 +13,7 @@ ai:
 
     # triage: 상시 분류·요약·근거 선별. Mac mini Qwen 27B (primary 와 동일 endpoint, 짧은 max_tokens).
     triage:
-      endpoint: "http://100.76.254.116:8801/v1/chat/completions"
+      endpoint: "http://100.76.254.116:8890/v1/chat/completions"
       model: "mlx-community/Qwen3.6-27B-6bit"
       max_tokens: 4096
       timeout: 480  # 프리필 실측 ~112 tok/s — 120K자 장문 커버 (2026-06-11)
@@ -22,7 +22,7 @@ ai:
 
     # primary: 에스컬레이션 전용. Qwen 27B MLX (맥미니 Semaphore(1) 보호 대상).
     primary:
-      endpoint: "http://100.76.254.116:8801/v1/chat/completions"
+      endpoint: "http://100.76.254.116:8890/v1/chat/completions"
       model: "mlx-community/Qwen3.6-27B-6bit"
       max_tokens: 8192
       timeout: 900  # 프리필 실측 ~112 tok/s — 260K자 상한 장문 커버 (2026-06-11)
@@ -72,7 +72,7 @@ ai:
     # Phase 3.5a answerability classifier. 2026-05-14 GPU LLM 제거 후 Mac mini 26B 로 swap.
     # classifier_service 가 hasattr 체크로 optional 이므로 이 섹션 제거 시 classifier gate 는 자동 skip (score-only).
     classifier:
-      endpoint: "http://100.76.254.116:8801/v1/chat/completions"
+      endpoint: "http://100.76.254.116:8890/v1/chat/completions"
       model: "mlx-community/Qwen3.6-27B-6bit"  # 2026-06-11 B안 동승 — gemma id 잔존 시 mlx 서버가 Gemma 를 재로드(이중 적재) 위험
       max_tokens: 512
       timeout: 30  # 2026-05-17: 15s 도 동시 부하 시 elapsed 14.4s 직전이라 tight — 30s 로 2x 마진. classifier_service.LLM_TIMEOUT_MS=30000 와 align (초과 = score-only skip, graceful)
@@ -199,8 +199,14 @@ schedule:
 # 이력: 2026-06-11 맥미니 모델 확정까지 8키 홀드 → 同日 Qwen3.6-27B-6bit 전환과 함께 해제([]).
 pipeline:
   held_stages: []
-  # mlx gate 동시 실행 상한 (2026-06-12 fair-share): 구 "1 고정" 룰의 전제(single-inference
-  # 서버)가 소멸 — 현 mlx_vlm 은 continuous batching (2026-06-11 밤 6~8 concurrent 실측 정상).
-  # 2 = 워커 LLM 호출과 인터랙티브(ask/eid)가 서로 안 막힘 + 집계 throughput ~1.8배.
-  # 게이트(상한+우선순위)는 유지 — thundering herd 방지. 1 로 되돌리면 구 동작.
+  # mlx gate 동시 실행 상한 (config.mlx_gate_concurrency). 현 mlx_vlm = continuous batching
+  # (2026-06-11 밤 6~8 concurrent 실측 정상). 2026-06-15: 2→4 — digest/briefing 합성을
+  # 이 단일 게이트(BACKGROUND 우선순위)로 라우팅하며 digest(클러스터 44~68)가 하드캡 내
+  # 완료되도록 동시성 확보. ask/eid(FOREGROUND)는 큐 점프라 영향 최소. 되돌리면 구 동작.
   mlx_gate_concurrency: 2
+  # 2026-06-15: digest/briefing 생성 LLM 파라미터 (모델 교체 후 단일소스, 상세 = config.py).
+  # 구 하드코딩 25s(빠른 Gemma)가 Qwen 27B(콜당 ~90~300s) 교체 sweep 누락 → digest 600s
+  # 초과·briefing 4/4 폴백. 동시성은 위 mlx_gate_concurrency 가 담당(별 키 없음).
+  digest_llm_timeout_s: 300
+  digest_llm_attempts: 2
+  digest_pipeline_hard_cap_s: 5400

frontend/src/app.css

@@ -213,3 +213,14 @@ body {
 
 /* Phase 1C: frontmatter 박스 — 본문 위 메타 표시 */
 .md-frontmatter dt { font-weight: 500; }
+
+/* AI 요약(TL;DR 등) 마크다운 렌더 — 좁은 카드에 맞게 문단/리스트 마진 압축 */
+.summary-md > :first-child { margin-top: 0; }
+.summary-md > :last-child { margin-bottom: 0; }
+.summary-md p { margin: 0 0 0.45em; }
+.summary-md ul, .summary-md ol { margin: 0.25em 0; padding-left: 1.2em; }
+.summary-md ul { list-style: disc; }
+.summary-md ol { list-style: decimal; }
+.summary-md li { margin: 0.1em 0; }
+.summary-md strong { font-weight: 700; }
+.summary-md code { background: rgba(0, 0, 0, 0.05); padding: 0 0.3em; border-radius: 3px; }

frontend/src/lib/components/AnalysisPanel.svelte

@@ -12,6 +12,7 @@
 -->
 <script lang="ts">
   import { api } from '$lib/api';
+  import { renderDocMarkdown } from '$lib/utils/docMarkdown';
   import Badge from '$lib/components/ui/Badge.svelte';
   import Button from '$lib/components/ui/Button.svelte';
   import Skeleton from '$lib/components/ui/Skeleton.svelte';
@@ -104,9 +105,7 @@
       </div>
 
       {#if tldr}
-        <p class="text-xs font-medium text-text leading-relaxed mb-2">
-          {tldr}
-        </p>
+        <div class="summary-md text-xs font-medium text-text leading-relaxed mb-2">{@html renderDocMarkdown(tldr)}</div>
       {/if}
 
       {#if bullets && bullets.length > 0}

frontend/src/lib/components/ProcessingFlowBoard.svelte

@@ -210,6 +210,23 @@
   // 맥북이 요약을 실제로 가져가는 중인가 (합류 표식 게이트)
   const offloadActive = $derived(split.macbook.done_1h > 0);
 
+  // ─── 백그라운드 작업 (큐 밖 스크립트 backfill) — processing_queue 사각지대 노출 ───
+  const bgJobs = $derived(overview.background_jobs ?? []);
+  const runningBg = $derived(bgJobs.filter((j) => j.state === 'running'));
+  function bgForMachine(key: string) {
+    return runningBg.filter((j) => j.machine === key);
+  }
+  function fmtElapsed(s: number): string {
+    if (s < 60) return `${s}s`;
+    if (s < 3600) return `${Math.floor(s / 60)}m`;
+    return `${Math.floor(s / 3600)}h${Math.floor((s % 3600) / 60)}m`;
+  }
+  function bgDot(j: { state: string; stale: boolean }): string {
+    if (j.state === 'running') return j.stale ? 'bg-warning' : 'bg-success';
+    if (j.state === 'failed') return 'bg-error';
+    return 'bg-faint';
+  }
+
   // ─── 지배 백로그 = 요약. 정직 ETA(유입 차감) — summarize_eta ───
   const eta = $derived(overview.summarize_eta);
   // 정직 ETA 라벨: eta_minutes null = 유입이 소화를 앞섬(소진 불가)
@@ -320,10 +337,11 @@
     {#each lanes as lane (lane.key)}
       <div class="bg-surface border border-default rounded-card px-3.5 py-2.5">
         <div class="flex items-center gap-2 flex-wrap mb-2">
-          <span class="w-2 h-2 rounded-full shrink-0 {dotClass(lane.card?.state ?? 'idle')}"></span>
+          <span class="w-2 h-2 rounded-full shrink-0 {dotClass(bgForMachine(lane.key).length > 0 ? 'active' : (lane.card?.state ?? 'idle'))}"></span>
           <span class="text-[9px] font-bold rounded px-1.5 py-px mtag-{lane.key}">{lane.meta.label}</span>
           <span class="text-[10px] text-faint font-mono">{lane.meta.model}</span>
           <span class="text-[11px] text-dim tabular-nums ml-1">{formatRate(lane.card?.done_1h ?? 0)}/h</span>
+          {#each bgForMachine(lane.key) as j (j.id)}<span class="text-[10px] font-semibold text-success tabular-nums ml-1">생성 중: {j.label ?? j.kind}{#if j.total} {j.processed}/{j.total}{/if}</span>{/each}
           {#if lane.key === 'macbook' && (lane.card?.deferred_pending ?? 0) > 0}
             <span class="text-[10px] font-semibold text-warning tabular-nums">보류 {lane.card?.deferred_pending}</span>
           {/if}
@@ -466,6 +484,32 @@
     </div>
   {/if}
 
+  <!-- 백그라운드 작업 (큐 밖 스크립트 backfill 등 — processing_queue 가 못 보는 사각지대) -->
+  {#if bgJobs.length > 0}
+    <div class="mt-3">
+      <div class="text-[11px] font-bold text-dim uppercase tracking-wider mb-2">백그라운드 작업</div>
+      <div class="grid gap-2">
+        {#each bgJobs as j (j.id)}
+          <div class="bg-surface border rounded-card px-3.5 py-2.5 {j.stale ? 'border-warning' : j.state === 'failed' ? 'border-error' : 'border-default'}">
+            <div class="flex items-center gap-2 flex-wrap">
+              <span class="w-2 h-2 rounded-full shrink-0 {bgDot(j)}"></span>
+              <span class="text-[9px] font-bold rounded px-1.5 py-px bg-default text-dim font-mono">{j.kind}</span>
+              <span class="text-xs font-semibold text-text truncate">{j.label ?? '작업'}</span>
+              <span class="text-[11px] text-dim tabular-nums ml-auto">
+                {#if j.total}{j.processed.toLocaleString()}/{j.total.toLocaleString()}{:else}{j.processed.toLocaleString()}건{/if} · {fmtElapsed(j.elapsed_sec)}
+              </span>
+            </div>
+            {#if j.stale}
+              <div class="text-[10px] text-warning mt-1.5">heartbeat 끊김 — 프로세스 중단 추정 (재개 필요할 수 있음)</div>
+            {:else if j.state === 'failed'}
+              <div class="text-[10px] text-error mt-1.5 truncate">실패{#if j.error} · {j.error}{/if}</div>
+            {/if}
+          </div>
+        {/each}
+      </div>
+    </div>
+  {/if}
+
   <!-- 실패 처리 드로어 -->
   {#if failOpen}
     <div class="border border-error/40 rounded-card mt-3 overflow-hidden bg-surface">

frontend/src/lib/types/queue.ts

@@ -75,6 +75,21 @@ export interface QueueStageRow {
   oldest_pending_age_sec: number | null;
 }
 
+/** 큐 밖 관리 스크립트(백필 등) 작업 — processing_queue 가 못 보는 사각지대.
+ *  stale = running 인데 heartbeat 끊김(프로세스 사망 추정). */
+export interface BackgroundJob {
+  id: number;
+  kind: string;
+  label: string | null;
+  state: 'running' | 'done' | 'failed';
+  machine: string;
+  processed: number;
+  total: number | null;
+  elapsed_sec: number;
+  stale: boolean;
+  error: string | null;
+}
+
 export interface QueueOverview {
   machines: MachineOverview[];
   summarize_eta: SummarizeEta;
@@ -82,6 +97,7 @@ export interface QueueOverview {
   trend_24h: TrendPoint[];
   stages: QueueStageRow[];
   totals: QueueTotals;
+  background_jobs?: BackgroundJob[];
 }
 
 /** ─── 실패 처리 (ds-board-engines-1) — GET /api/queue/failed · POST /retry|/skip ─── */

frontend/src/lib/utils/docMarkdown.ts

@@ -15,6 +15,7 @@
 
 import DOMPurify from 'dompurify';
 import { Marked } from 'marked';
+import katex from 'katex';
 // @ts-ignore — 타입 정의 누락 시 무시
 import markedKatex from 'marked-katex-extension';
 // @ts-ignore — 타입 정의 누락 시 무시
@@ -88,10 +89,59 @@ const SANITIZE_OPTS = {
   ALLOW_UNKNOWN_PROTOCOLS: false,
 } as const;
 
+// ── 수식 pre-render ──────────────────────────────────────────────────────────
+// marked-katex-extension 의 토크나이저는 `$$` 가 블록 선두에 있어야 발화하는데,
+// (1) 개요 anchor splice 가 `$$` 직전에 <span id="sec-N"> 를 끼우면 `$$` 가 문단 중간으로
+// 밀려 블록 규칙이 깨지고, (2) 빌드/런타임 환경에 따라 확장 토크나이저가 발화하지 않으면
+// `$$` 가 평문으로 새어 marked 의 백슬래시 이스케이프(\% → %, \, → ,)에 망가진다.
+// → marked 가 손대기 *전에* 수식을 katex 로 직접 렌더해 placeholder 로 보호한 뒤 복원한다.
+//   위치·인접 상황과 무관(전역 정규식)하므로 위 두 경우를 모두 우회한다.
+const _MATH_SLOT = (i: number) => `KX0MATHSLOT${i}MATHKX0`; // marked-안전(영숫자) + 충돌 불가
+const _MATH_SLOT_RE = /KX0MATHSLOT(\d+)MATHKX0/g;
+const _BLOCK_MATH_RE = /\$\$([\s\S]+?)\$\$/g;
+// 인라인 $...$ — 통화($5)·이스케이프(\$)·`$$` 회피. $ 직후 비공백, $ 직전 비공백.
+const _INLINE_MATH_RE = /(?<![\\$\d])\$(?!\s)([^$\n]*?[^$\n\s])\$(?!\d)/g;
+
+function _protectMath(text: string, slots: string[]): string {
+  const render = (tex: string, displayMode: boolean): string => {
+    slots.push(
+      katex.renderToString(tex.trim(), { displayMode, throwOnError: false, output: 'html' }),
+    );
+    return _MATH_SLOT(slots.length - 1);
+  };
+  return text
+    .replace(_BLOCK_MATH_RE, (m, tex) => {
+      try {
+        return render(String(tex), true);
+      } catch {
+        return m;
+      }
+    })
+    .replace(_INLINE_MATH_RE, (m, tex) => {
+      try {
+        return render(String(tex), false);
+      } catch {
+        return m;
+      }
+    });
+}
+
 export function renderDocMarkdown(text: string | null | undefined): string {
   if (!text) return '';
   try {
-    const html = docMarked.parse(text) as string;
+    const slots: string[] = [];
+    const protectedText = _protectMath(text, slots);
+    let html = docMarked.parse(protectedText) as string;
+    if (slots.length) {
+      // 블록 수식이 단독 문단이면 marked 가 <p> 로 감싸므로 그 <p> 를 벗겨 블록 수식이 문단에
+      // 매몰되지 않게 한다. (katex-display 는 block 이라 <p> 안에 두면 브라우저가 자동 분리.)
+      html = html
+        .replace(
+          new RegExp(`<p>\\s*KX0MATHSLOT(\\d+)MATHKX0\\s*</p>`, 'g'),
+          (m, i) => slots[Number(i)] ?? m,
+        )
+        .replace(_MATH_SLOT_RE, (m, i) => slots[Number(i)] ?? m);
+    }
     return DOMPurify.sanitize(html, SANITIZE_OPTS);
   } catch {
     // 마지막 안전망: 모든 태그 제거 후 escape

frontend/src/routes/+layout.svelte

@@ -3,7 +3,7 @@
   import { browser } from '$app/environment';
   import { page } from '$app/stores';
   import { goto } from '$app/navigation';
-  import { Menu, EllipsisVertical, ChevronDown, FileText, Newspaper, HelpCircle, StickyNote, Inbox, PanelLeft, MessageCircle } from 'lucide-svelte';
+  import { Menu, EllipsisVertical, ChevronDown, FileText, Newspaper, StickyNote, Inbox, PanelLeft } from 'lucide-svelte';
   import { isAuthenticated, user, tryRefresh, logout } from '$lib/stores/auth';
   import { toasts, removeToast } from '$lib/stores/toast';
   import { refresh as refreshPublicConfig } from '$lib/stores/config';
@@ -151,8 +151,7 @@
             {/if}
           </div>
 
-          <a href="/ask" class="px-3 py-1.5 rounded-md text-sm font-semibold transition-colors {isActive('/ask') ? 'text-accent bg-accent/12' : 'text-dim hover:text-text hover:bg-surface'}">질문</a>
-          <a href="/chat" class="px-3 py-1.5 rounded-md text-sm font-semibold transition-colors {isActive('/chat') ? 'text-accent bg-accent/12' : 'text-dim hover:text-text hover:bg-surface'}">이드</a>
+          <a href="/memos" class="px-3 py-1.5 rounded-md text-sm font-semibold transition-colors {isActive('/memos') ? 'text-accent bg-accent/12' : 'text-dim hover:text-text hover:bg-surface'}">메모</a>
           <SystemStatusDot />
         </div>
 
@@ -212,8 +211,6 @@
       <nav class="lg:hidden shrink-0 flex border-t border-default bg-sidebar" aria-label="하단 탭">
         <a href="/documents" aria-current={docsActive ? 'page' : undefined} class="flex-1 flex flex-col items-center justify-center gap-1 py-2 text-[10px] font-semibold transition-colors {docsActive ? 'text-accent' : 'text-dim'}"><FileText size={18} strokeWidth={1.9} /> 문서</a>
         <a href="/news" aria-current={newsActive ? 'page' : undefined} class="flex-1 flex flex-col items-center justify-center gap-1 py-2 text-[10px] font-semibold transition-colors {newsActive ? 'text-accent' : 'text-dim'}"><Newspaper size={18} strokeWidth={1.9} /> 뉴스</a>
-        <a href="/ask" aria-current={isActive('/ask') ? 'page' : undefined} class="flex-1 flex flex-col items-center justify-center gap-1 py-2 text-[10px] font-semibold transition-colors {isActive('/ask') ? 'text-accent' : 'text-dim'}"><HelpCircle size={18} strokeWidth={1.9} /> 질문</a>
-        <a href="/chat" aria-current={isActive('/chat') ? 'page' : undefined} class="flex-1 flex flex-col items-center justify-center gap-1 py-2 text-[10px] font-semibold transition-colors {isActive('/chat') ? 'text-accent' : 'text-dim'}"><MessageCircle size={18} strokeWidth={1.9} /> 이드</a>
         <a href="/memos" aria-current={isActive('/memos') ? 'page' : undefined} class="flex-1 flex flex-col items-center justify-center gap-1 py-2 text-[10px] font-semibold transition-colors {isActive('/memos') ? 'text-accent' : 'text-dim'}"><StickyNote size={18} strokeWidth={1.9} /> 메모</a>
         <button onclick={() => ui.openDrawer('sidebar')} class="flex-1 flex flex-col items-center justify-center gap-1 py-2 text-[10px] font-semibold text-dim"><Menu size={18} strokeWidth={1.9} /> 더보기</button>
       </nav>

frontend/src/routes/documents/[id]/+page.svelte

@@ -16,6 +16,7 @@
   import Skeleton from '$lib/components/ui/Skeleton.svelte';
   import HandwriteCanvas from '$lib/components/HandwriteCanvas.svelte';
   import MarkdownDoc from '$lib/components/MarkdownDoc.svelte';
+  import { renderDocMarkdown } from '$lib/utils/docMarkdown';
   import MarkdownStatusBadge from '$lib/components/MarkdownStatusBadge.svelte';
   import NoteEditor from '$lib/components/editors/NoteEditor.svelte';
   import EditUrlEditor from '$lib/components/editors/EditUrlEditor.svelte';
@@ -263,7 +264,7 @@
     {#if doc.ai_tldr || doc.ai_summary}
       <div style="background:#f4f7f1;border:1px solid #dde3d6;border-radius:14px;padding:13px;">
         <div style="font-size:10.5px;font-weight:700;color:#697061;letter-spacing:.4px;margin-bottom:7px;">TL;DR</div>
-        <div style="font-size:12px;line-height:1.5;color:#23291f;">{doc.ai_tldr || doc.ai_summary}</div>
+        <div class="summary-md" style="font-size:12px;line-height:1.5;color:#23291f;">{@html renderDocMarkdown(doc.ai_tldr || doc.ai_summary)}</div>
       </div>
     {/if}
     {#if doc.ai_bullets && doc.ai_bullets.length}

migrations/357_background_jobs.sql

@@ -0,0 +1,19 @@
+-- 2026-06-14 PR-Background-Jobs-Observability: 큐 밖 관리 스크립트(백필 등) 진행 가시화.
+-- processing_queue 는 파이프라인 stage 전용 — hier_overnight_backfill / section_summary_pilot
+-- 같은 off-queue 관리 스크립트는 여기에 진행상황을 남겨 대시보드 보드가 노출한다.
+-- worker_jobs(user_id NOT NULL, worker-pool 전용)와 별개 — 이건 owner 없는 관리 작업 heartbeat.
+-- 단일 statement (asyncpg multi-statement 불허 컨벤션). 인덱스는 소량 테이블이라 생략.
+CREATE TABLE IF NOT EXISTS background_jobs (
+    id           BIGSERIAL PRIMARY KEY,
+    kind         TEXT NOT NULL,                 -- 'hier_redecompose' | 'section_summary' | ...
+    label        TEXT,                          -- 사람이 읽는 대상 표기 (예: 'doc 5210 (Sec VIII)')
+    state        TEXT NOT NULL DEFAULT 'running'
+                    CHECK (state IN ('running', 'done', 'failed')),
+    processed    INTEGER NOT NULL DEFAULT 0,    -- 처리한 단위 수 (절/leaf 등)
+    total        INTEGER,                       -- 전체 단위 수 (미상이면 NULL)
+    detail       JSONB NOT NULL DEFAULT '{}'::jsonb,
+    error        TEXT,
+    started_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    finished_at  TIMESTAMPTZ
+);

migrations/358_documents_embedding_hnsw.sql

@@ -0,0 +1,11 @@
+-- 358: documents.embedding HNSW 벡터 인덱스 + hnsw.ef_search (검색 latency T3, 2026-06-15)
+-- PROD 적용 = CREATE INDEX CONCURRENTLY 로 수동 빌드(40k rows 무중단, /dev/shm 회피 위해 단일 스레드)
+--   + schema_migrations(358) 수동 기록 완료. runner 는 단일 트랜잭션이라 CONCURRENTLY 불가.
+-- 본 파일 = fresh-init/재현용: non-concurrent IF NOT EXISTS (빈 테이블 init 시 즉시, 기존 index 존재 시 no-op).
+CREATE INDEX IF NOT EXISTS idx_documents_embedding_hnsw
+  ON documents USING hnsw (embedding vector_cosine_ops)
+  WHERE (deleted_at IS NULL AND embedding IS NOT NULL);
+
+-- docs vector leg LIMIT = limit*4 (기본 80) → HNSW recall 위해 ef_search >= 80 필요.
+-- ivfflat.probes=20 과 동일하게 DB 레벨 GUC (ALTER DATABASE) 로 설정.
+ALTER DATABASE pkm SET hnsw.ef_search = 100;

scripts/hier_overnight_backfill.py

@@ -32,6 +32,7 @@ from core.config import settings
 from services.hier_decomp.builder import build_hier_tree
 from services.hier_decomp.persist import persist_hier_tree
 from services.search.llm_gate import Priority, acquire_mlx_gate
+from services.background_jobs import finish_job, heartbeat, start_job
 
 # 단일 진실: 절 분석 상수/헬퍼 (PROMPT_VERSION 일치 = 멱등 보존)
 from section_summary_pilot import (
@@ -140,8 +141,10 @@ def _make_engine():
     return create_async_engine(os.environ["DATABASE_URL"], pool_pre_ping=True)
 
 
-async def _analyze_doc_leaves(session, client, doc_id, doc_domain, model_name, stop_at):
-    """doc 의 미분석 hier leaf 분석 → upsert. stop_at(epoch) 넘으면 leaf 경계 중단."""
+async def _analyze_doc_leaves(session, client, doc_id, doc_domain, model_name, stop_at,
+                              engine=None, job_id=None, base_processed=0):
+    """doc 의 미분석 hier leaf 분석 → upsert. stop_at(epoch) 넘으면 leaf 경계 중단.
+    engine/job_id 주어지면 background_jobs 에 ~10절마다 진행 heartbeat(보드 가시화)."""
     rows = (await session.execute(LEAF_SQL, {"doc": doc_id, "pv": PROMPT_VERSION})).mappings().all()
     ok = fail = skip = 0
     timings, types = [], []
@@ -187,6 +190,8 @@ async def _analyze_doc_leaves(session, client, doc_id, doc_domain, model_name, s
             "content_hash": r["content_hash"], "error": err,
         })
         await session.commit()
+        if job_id and (ok + fail + skip) % 10 == 0:
+            await heartbeat(engine, job_id, processed=base_processed + ok + fail + skip)
     await session.commit()
     return {"ok": ok, "fail": fail, "skip": skip, "leaves": len(rows),
             "timings": timings, "types": types, "aborted": aborted}
@@ -256,6 +261,12 @@ async def cmd_run(args):
                      _candidate_params(allowlist, doc_ids))).mappings().all()
         _log(f"후보 doc {len(cands)} 선별. 시작.")
 
+        # 관측: 큐 밖 작업이라 대시보드 보드가 못 보므로 background_jobs 에 진행 노출(best-effort)
+        _job_kind = "hier_redecompose" if reprocess else "hier_backfill"
+        _job_label = (f"doc {args.doc} {'재분해' if reprocess else '분해'}" if doc_ids
+                      else f"{len(cands)}개 문서 {'재분해' if reprocess else '분해'}")
+        job_id = await start_job(engine, _job_kind, _job_label, total=None)
+
         for c in cands:
             if time.time() >= stop_at:
                 _log(f"⏰ deadline 버퍼 도달 — doc 경계에서 중단 (처리 {tot_docs} doc)")
@@ -272,7 +283,10 @@ async def cmd_run(args):
                              "timings": [], "types": [], "aborted": False}
                 else:
                     async with sm() as session:
-                        astat = await _analyze_doc_leaves(session, client, doc_id, doc_domain, model_name, stop_at)
+                        astat = await _analyze_doc_leaves(
+                            session, client, doc_id, doc_domain, model_name, stop_at,
+                            engine=engine, job_id=job_id,
+                            base_processed=(tot_ok + tot_fail + tot_skip))
             except Exception as exc:
                 _log(f"  ✗ doc={doc_id} 처리 실패(건너뜀): {type(exc).__name__}: {repr(exc)[:160]}")
                 continue
@@ -280,6 +294,8 @@ async def cmd_run(args):
             tot_docs += 1
             tot_ok += astat["ok"]; tot_fail += astat["fail"]; tot_skip += astat["skip"]
             all_timings += astat["timings"]; all_types += astat["types"]
+            await heartbeat(engine, job_id, processed=(tot_ok + tot_fail + tot_skip),
+                            total=tot_leaves_created)
             avg = statistics.mean(astat["timings"]) if astat["timings"] else 0
             _log(f"  ✓ doc={doc_id} ({len(body):,}자 {doc_domain.split('/')[0]}) "
                  f"leaf생성={leaves_created} 분석ok={astat['ok']} fail={astat['fail']} skip={astat['skip']} "
@@ -287,6 +303,7 @@ async def cmd_run(args):
             if astat["aborted"]:
                 _log("⏰ leaf 분석 중 deadline 도달 — 중단")
                 break
+        await finish_job(engine, job_id, state="done")
     finally:
         await client.close()
         await engine.dispose()

tests/fixtures/arxiv_search_pressure_vessel.xml

@@ -0,0 +1,383 @@
+<?xml version='1.0' encoding='UTF-8'?>
+<feed xmlns:opensearch="http://a9.com/-/spec/opensearch/1.1/" xmlns:arxiv="http://arxiv.org/schemas/atom" xmlns="http://www.w3.org/2005/Atom">
+  <id>https://arxiv.org/api/m9A/71G4hH6NGyarIQjqA3n6Zzk</id>
+  <title>arXiv Query: search_query=abs:"pressure vessel"&amp;id_list=&amp;start=0&amp;max_results=10</title>
+  <updated>2026-06-13T21:57:59Z</updated>
+  <link href="https://arxiv.org/api/query?search_query=abs:%22pressure+vessel%22&amp;start=0&amp;max_results=10&amp;id_list=" type="application/atom+xml"/>
+  <opensearch:itemsPerPage>10</opensearch:itemsPerPage>
+  <opensearch:totalResults>89</opensearch:totalResults>
+  <opensearch:startIndex>0</opensearch:startIndex>
+  <entry>
+    <id>http://arxiv.org/abs/1209.2405v1</id>
+    <title>A Survey of Pressure Vessel Code Compliance for Superconducting RF Cryomodules</title>
+    <updated>2012-09-11T19:34:46Z</updated>
+    <link href="https://arxiv.org/abs/1209.2405v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/1209.2405v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>Superconducting radio frequency (SRF) cavities made from niobium and cooled with liquid helium are becoming key components of many particle accelerators. The helium vessels surrounding the RF cavities, portions of the niobium cavities themselves, and also possibly the vacuum vessels containing these assemblies, generally fall under the scope of local and national pressure vessel codes. In the U.S., Department of Energy rules require national laboratories to follow national consensus pressure vessel standards or to show "a level of safety greater than or equal to" that of the applicable standard. Thus, while used for its superconducting properties, niobium ends up being treated as a low-temperature pressure vessel material. Niobium material is not a code listed material and therefore requires the designer to understand the mechanical properties for material used in each pressure vessel fabrication; compliance with pressure vessel codes therefore becomes a problem. This report summarizes the approaches that various institutions have taken in order to bring superconducting RF cryomodules into compliance with pressure vessel codes.</summary>
+    <category term="physics.acc-ph" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2012-09-11T19:34:46Z</published>
+    <arxiv:comment>7 pp</arxiv:comment>
+    <arxiv:primary_category term="physics.acc-ph"/>
+    <author>
+      <name>Thomas Peterson</name>
+      <arxiv:affiliation>Fermilab</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Arkadiy Klebaner</name>
+      <arxiv:affiliation>Fermilab</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Tom Nicol</name>
+      <arxiv:affiliation>Fermilab</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Jay Theilacker</name>
+      <arxiv:affiliation>Fermilab</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Hitoshi Hayano</name>
+      <arxiv:affiliation>KEK, Tsukuba</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Eiji Kako</name>
+      <arxiv:affiliation>KEK, Tsukuba</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Hirotaka Nakai</name>
+      <arxiv:affiliation>KEK, Tsukuba</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Akira Yamamoto</name>
+      <arxiv:affiliation>KEK, Tsukuba</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Kay Jensch</name>
+      <arxiv:affiliation>DESY</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Axel Matheisen</name>
+      <arxiv:affiliation>DESY</arxiv:affiliation>
+    </author>
+    <author>
+      <name>John Mammosser</name>
+      <arxiv:affiliation>Jefferson Lab</arxiv:affiliation>
+    </author>
+    <arxiv:doi>10.1063/1.4707088</arxiv:doi>
+    <link rel="related" href="https://doi.org/10.1063/1.4707088" title="doi"/>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/2003.02057v1</id>
+    <title>Investigation of Unit-1 Nuclear Reactor of the Fukushima Daiichi by Cosmic Muon Radiography</title>
+    <updated>2020-03-03T03:21:53Z</updated>
+    <link href="https://arxiv.org/abs/2003.02057v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/2003.02057v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>We have investigated the status of the nuclear fuel assemblies in Unit-1 reactor of the Fukushima Daiichi Nuclear Power plant by the method called Cosmic Muon Radiography. In this study, muon tracking detectors were placed outside of the reactor building. We succeeded in identifying the inner structure of the reactor complex such as the reactor containment vessel, pressure vessel, and other structures of the reactor building, through the concrete wall of the reactor building. We found that a large amount of fuel assemblies was missing in the original fuel loading zone inside the pressure vessel. It can be naturally interpreted that most of the nuclear fuel was melt and dropped down to the bottom of the pressure vessel or even below.</summary>
+    <category term="physics.ins-det" scheme="http://arxiv.org/schemas/atom"/>
+    <category term="hep-ex" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2020-03-03T03:21:53Z</published>
+    <arxiv:comment>14 pages, 17 figures</arxiv:comment>
+    <arxiv:primary_category term="physics.ins-det"/>
+    <author>
+      <name>Hirofumi Fujii</name>
+      <arxiv:affiliation>High Energy Accelerator Research Organization</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Kazuhiko Hara</name>
+      <arxiv:affiliation>University of Tsukuba</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Kohei Hayashi</name>
+      <arxiv:affiliation>High Energy Accelerator Research Organization</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Hidekazu Kakuno</name>
+      <arxiv:affiliation>Tokyo Metropolitan University</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Hideyo Kodama</name>
+      <arxiv:affiliation>High Energy Accelerator Research Organization</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Kanetada Nagamine</name>
+      <arxiv:affiliation>High Energy Accelerator Research Organization</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Kotaro Sato</name>
+      <arxiv:affiliation>High Energy Accelerator Research Organization</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Shin-Hong Kim</name>
+      <arxiv:affiliation>University of Tsukuba</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Atsuto Suzuki</name>
+      <arxiv:affiliation>High Energy Accelerator Research Organization</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Takayuki Sumiyoshi</name>
+      <arxiv:affiliation>Tokyo Metropolitan University</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Kazuki Takahashi</name>
+      <arxiv:affiliation>University of Tsukuba</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Fumihiko Takasaki</name>
+      <arxiv:affiliation>High Energy Accelerator Research Organization</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Shuji Tanaka</name>
+      <arxiv:affiliation>High Energy Accelerator Research Organization</arxiv:affiliation>
+    </author>
+    <author>
+      <name>Satoru Yamashita</name>
+      <arxiv:affiliation>University of Tokyo</arxiv:affiliation>
+    </author>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/1609.07515v1</id>
+    <title>Low Background Stainless Steel for the Pressure Vessel in the PandaX-II Dark Matter Experiment</title>
+    <updated>2016-09-21T10:33:04Z</updated>
+    <link href="https://arxiv.org/abs/1609.07515v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/1609.07515v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>We report on the custom produced low radiation background stainless steel and the welding rod for the PandaX experiment, one of the deep underground experiments to search for dark matter and neutrinoless double beta decay using xenon. The anthropogenic 60 Co concentration in these samples is at the range of 1 mBq/kg or lower. We also discuss the radioactivity of nuclear-grade stainless steel from TISCO which has a similar background rate. The PandaX-II pressure vessel was thus fabricated using the stainless steel from CISRI and TISCO. Based on the analysis of the radioactivity data, we also made discussions on potential candidate for low background metal materials for future pressure vessel development.</summary>
+    <category term="physics.ins-det" scheme="http://arxiv.org/schemas/atom"/>
+    <category term="hep-ex" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2016-09-21T10:33:04Z</published>
+    <arxiv:primary_category term="physics.ins-det"/>
+    <author>
+      <name>Tao Zhang</name>
+    </author>
+    <author>
+      <name>Changbo Fu</name>
+    </author>
+    <author>
+      <name>Xiangdong Ji</name>
+    </author>
+    <author>
+      <name>Jianglai Liu</name>
+    </author>
+    <author>
+      <name>Xiang Liu</name>
+    </author>
+    <author>
+      <name>Xuming Wang</name>
+    </author>
+    <author>
+      <name>Chunfa Yao</name>
+    </author>
+    <author>
+      <name>Xunhua Yuan</name>
+    </author>
+    <arxiv:doi>10.1088/1748-0221/11/09/T09004</arxiv:doi>
+    <link rel="related" href="https://doi.org/10.1088/1748-0221/11/09/T09004" title="doi"/>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/2308.09786v1</id>
+    <title>Mechanical design of the optical modules intended for IceCube-Gen2</title>
+    <updated>2023-08-18T19:20:09Z</updated>
+    <link href="https://arxiv.org/abs/2308.09786v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/2308.09786v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>IceCube-Gen2 is an expansion of the IceCube neutrino observatory at the South Pole that aims to increase the sensitivity to high-energy neutrinos by an order of magnitude. To this end, about 10,000 new optical modules will be installed, instrumenting a fiducial volume of about 8 km^3. Two newly developed optical module types increase current sensitivity per module by a factor of three by integrating 16 and 18 newly developed four-inch PMTs in specially designed 12.5-inch diameter pressure vessels. Both designs use conical silicone gel pads to optically couple the PMTs to the pressure vessel to increase photon collection efficiency. The outside portion of gel pads are pre-cast onto each PMT prior to integration, while the interiors are filled and cast after the PMT assemblies are installed in the pressure vessel via a pushing mechanism. This paper presents both the mechanical design, as well as the performance of prototype modules at high pressure (70 MPa) and low temperature (-40 degree Celsius), characteristic of the environment inside the South Pole ice.</summary>
+    <category term="astro-ph.IM" scheme="http://arxiv.org/schemas/atom"/>
+    <category term="astro-ph.HE" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2023-08-18T19:20:09Z</published>
+    <arxiv:comment>Presented at the 38th International Cosmic Ray Conference (ICRC2023). See arXiv:2307.13048 for all IceCube-Gen2 contributions</arxiv:comment>
+    <arxiv:primary_category term="astro-ph.IM"/>
+    <author>
+      <name>Yuya Makino</name>
+      <arxiv:affiliation>for the IceCube-Gen2 Collaboration</arxiv:affiliation>
+    </author>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/0804.0261v1</id>
+    <title>Circulation in Blowdown Flows</title>
+    <updated>2008-04-01T22:22:32Z</updated>
+    <link href="https://arxiv.org/abs/0804.0261v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/0804.0261v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>  The blowdown of high pressure gas in a pressure vessel produces rapid adiabatic cooling of the gas remaining in the vessel. The gas near the wall is warmed by conduction from the wall, producing radial temperature and density gradients that affect the flow, the mass efflux rate and the thermodynamic states of both the outflowing and the contained gas. The resulting buoyancy-driven flow circulates gas through the vessel and reduces, but does not eliminate, these gradients. The purpose of this note is to estimate when blowdown cooling is rapid enough that the gas in the pressure vessel is neither isothermal nor isopycnic, though it remains isobaric. I define a dimensionless number, the buoyancy circulation number BC, that parametrizes these effects.</summary>
+    <category term="physics.flu-dyn" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2008-04-01T22:22:32Z</published>
+    <arxiv:comment>5 pp., no figures</arxiv:comment>
+    <arxiv:primary_category term="physics.flu-dyn"/>
+    <arxiv:journal_ref>J. Pressure Vessel Tech. 131, 034501 (2009)</arxiv:journal_ref>
+    <author>
+      <name>J. I. Katz</name>
+    </author>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/1204.0234v1</id>
+    <title>Substantiation of Thermodynamic Criteria of Explosion Safety in Process of Severe Accidents in Pressure Vessel Reactors</title>
+    <updated>2012-03-27T11:21:14Z</updated>
+    <link href="https://arxiv.org/abs/1204.0234v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/1204.0234v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>The paper represents original development of thermodynamic criteria of occurrence conditions of steam-gas explosions in the process of severe accidents. The received results can be used for modelling of processes of severe accidents in pressure vessel reactors.</summary>
+    <category term="physics.gen-ph" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2012-03-27T11:21:14Z</published>
+    <arxiv:comment>5 pages, 1 figure</arxiv:comment>
+    <arxiv:primary_category term="physics.gen-ph"/>
+    <author>
+      <name>V. I. Skalozubov</name>
+    </author>
+    <author>
+      <name>V. N. Vashchenko</name>
+    </author>
+    <author>
+      <name>S. S. Jarovoj</name>
+    </author>
+    <author>
+      <name>V. Yu. Kochnyeva</name>
+    </author>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/2511.11485v1</id>
+    <title>Data-efficient U-Net for Segmentation of Carbide Microstructures in SEM Images of Steel Alloys</title>
+    <updated>2025-11-14T17:01:02Z</updated>
+    <link href="https://arxiv.org/abs/2511.11485v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/2511.11485v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>Understanding reactor-pressure-vessel steel microstructure is crucial for predicting mechanical properties, as carbide precipitates both strengthen the alloy and can initiate cracks. In scanning electron microscopy images, gray-value overlap between carbides and matrix makes simple thresholding ineffective. We present a data-efficient segmentation pipeline using a lightweight U-Net (30.7~M parameters) trained on just \textbf{10 annotated scanning electron microscopy images}. Despite limited data, our model achieves a \textbf{Dice-Sørensen coefficient of 0.98}, significantly outperforming the state-of-the-art in the field of metallurgy (classical image analysis: 0.85), while reducing annotation effort by one order of magnitude compared to the state-of-the-art data efficient segmentation model. This approach enables rapid, automated carbide quantification for alloy design and generalizes to other steel types, demonstrating the potential of data-efficient deep learning in reactor-pressure-vessel steel analysis.</summary>
+    <category term="cs.LG" scheme="http://arxiv.org/schemas/atom"/>
+    <category term="cond-mat.mtrl-sci" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2025-11-14T17:01:02Z</published>
+    <arxiv:primary_category term="cs.LG"/>
+    <arxiv:journal_ref>Machine Learning and the Physical Sciences Workshop @ NeurIPS 2025 https://openreview.net/forum?id=xYY5pn4f8N</arxiv:journal_ref>
+    <author>
+      <name>Alinda Ezgi Gerçek</name>
+    </author>
+    <author>
+      <name>Till Korten</name>
+    </author>
+    <author>
+      <name>Paul Chekhonin</name>
+    </author>
+    <author>
+      <name>Maleeha Hassan</name>
+    </author>
+    <author>
+      <name>Peter Steinbach</name>
+    </author>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/2511.09689v1</id>
+    <title>An ASME-Compliant Helium-4 Evaporation Refrigerator for the SpinQuest Experiment</title>
+    <updated>2025-11-12T19:45:47Z</updated>
+    <link href="https://arxiv.org/abs/2511.09689v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/2511.09689v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>This paper presents the design, safety basis, and commissioning results of a 1 K liquid helium-4 (4He) evaporation refrigerator developed for the Fermilab SpinQuest Experiment (E1039). The system represents the first high power helium evaporation refrigerator operated in a fixed target scattering experiment at Fermilab and was engineered to comply with the Fermilab ES\&amp;H Manual (FESHM) requirements governing pressure vessels, piping, cryogenic systems, and vacuum vessels. The design is mapped to ASME B31.3 (Process Piping) and the ASME Boiler and Pressure Vessel Code (BPVC) for pressure boundary integrity and overpressure protection, with documented compliance to FESHM Chapters 5031 (Pressure Vessels), 5031.1 (Piping Systems), and 5033 (Vacuum Vessels). This work documents the methodology used to reach compliance and approval for the 4He evaporation refrigerator at Fermilab which the field lacks. Design considerations specific to the high radiation target-cave environment including remotely located instrumentation approximately 20 m from the cryostat are summarized, together with the relief-system sizing methodology used to accommodate transient heat loads from dynamic nuclear polarization microwaves and the high-intensity proton beam. Commissioning data from July 2024 confirms that the system satisfies all thermal performance and safety objectives.</summary>
+    <category term="physics.ins-det" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2025-11-12T19:45:47Z</published>
+    <arxiv:comment>For IEEE Transactions in Nuclear Physics, 11 pages, 14 figures</arxiv:comment>
+    <arxiv:primary_category term="physics.ins-det"/>
+    <author>
+      <name>Jordan D. Roberts</name>
+    </author>
+    <author>
+      <name>Vibodha Bandara</name>
+    </author>
+    <author>
+      <name>Kenichi Nakano</name>
+    </author>
+    <author>
+      <name>Dustin Keller</name>
+    </author>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/1507.04072v1</id>
+    <title>High-Voltage Terminal Test of Test Stand for 1-MV Electrostatic Accelerator</title>
+    <updated>2015-07-15T02:41:11Z</updated>
+    <link href="https://arxiv.org/abs/1507.04072v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/1507.04072v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>The Korea Multipurpose Accelerator Complex (KOMAC) has been developing a 300-kV test stand for a 1-MV electrostatic accelerator ion source. The ion source and accelerating tube will be installed in a high-pressure vessel. The ion source in the high-pressure vessel is required to have a high reliability. The test stand has been proposed and developed to confirm the stable operating conditions of the ion source. The ion source will be tested at the test stand to verify the long-time operating conditions. The test stand comprises a 300-kV high-voltage terminal, a battery for the ion-source power, a 60-Hz inverter, 200-MHz RF power, a 5-kV extraction power supply, a 300-kV accelerating tube, and a vacuum system. The results of the 300-kV high-voltage terminal tests are presented in this paper.</summary>
+    <category term="physics.acc-ph" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2015-07-15T02:41:11Z</published>
+    <arxiv:comment>International Conference on Accelerators and Beam Utilization (ICABU2014)</arxiv:comment>
+    <arxiv:primary_category term="physics.acc-ph"/>
+    <arxiv:journal_ref>Yong-Sub Cho KNS (2014); W. Sima IEEE (2004) 480-483; LA-UR-87-126 (1987); Jeong-tae Kim KNS (2014)</arxiv:journal_ref>
+    <author>
+      <name>Sae-Hoon Park</name>
+    </author>
+    <author>
+      <name>Yu-Seok Kim</name>
+    </author>
+    <arxiv:doi>10.3938/jkps</arxiv:doi>
+    <link rel="related" href="https://doi.org/10.3938/jkps" title="doi"/>
+  </entry>
+  <entry>
+    <id>http://arxiv.org/abs/2005.05585v1</id>
+    <title>Investigation of the Status of Unit 2 Nuclear Reactor of the Fukushima Daiichi by the Cosmic Muon Radiography</title>
+    <updated>2020-05-12T07:26:37Z</updated>
+    <link href="https://arxiv.org/abs/2005.05585v1" rel="alternate" type="text/html"/>
+    <link href="https://arxiv.org/pdf/2005.05585v1" rel="related" type="application/pdf" title="pdf"/>
+    <summary>We have investigated the status of the nuclear debris in the Unit-2 Nuclear Reactor of the Fukushima Daiichi Nuclear Power plant by the method called Cosmic Muon Radiography. In this measurement, the muon detector was placed outside of the reactor building as was the case of the measurement for the Unit-1 Reactor. Compared to the previous measurements, the detector was down-sized, which made us possible to locate it closer to the reactor and to investigate especially the lower part of the fuel loading zone. We identified the inner structures of the reactor such as the containment vessel, pressure vessel and other objects through the thick concrete wall of the reactor building. Furthermore, the observation showed existence of heavy material at the bottom of the pressure vessel, which can be interpreted as the debris of melted nuclear fuel dropped from the loading zone.</summary>
+    <category term="physics.ins-det" scheme="http://arxiv.org/schemas/atom"/>
+    <published>2020-05-12T07:26:37Z</published>
+    <arxiv:comment>11 figures and 2 tables</arxiv:comment>
+    <arxiv:primary_category term="physics.ins-det"/>
+    <author>
+      <name>Hirofumi Fujii</name>
+    </author>
+    <author>
+      <name>Kazuhiko Hara</name>
+    </author>
+    <author>
+      <name>Shugo Hashimoto</name>
+    </author>
+    <author>
+      <name>Kohei Hayashi</name>
+    </author>
+    <author>
+      <name>Hidekazu Kakuno</name>
+    </author>
+    <author>
+      <name>Hideyo Kodama</name>
+    </author>
+    <author>
+      <name>Gi Meiki</name>
+    </author>
+    <author>
+      <name>Masato Mizokami</name>
+    </author>
+    <author>
+      <name>Shinya Mizokami</name>
+    </author>
+    <author>
+      <name>Kanetada Nagamine</name>
+    </author>
+    <author>
+      <name>Kotaro Sato</name>
+    </author>
+    <author>
+      <name>Shunsuke Sekita</name>
+    </author>
+    <author>
+      <name>Hiroshi Shirai</name>
+    </author>
+    <author>
+      <name>Shin-Hong Kim</name>
+    </author>
+    <author>
+      <name>Takayuki Sumiyoshi</name>
+    </author>
+    <author>
+      <name>Atsuto Suzuki</name>
+    </author>
+    <author>
+      <name>Yoshihisa Takada</name>
+    </author>
+    <author>
+      <name>Kazuki Takahashi</name>
+    </author>
+    <author>
+      <name>Yu Takahashi</name>
+    </author>
+    <author>
+      <name>Fumihiko Takasaki</name>
+    </author>
+    <author>
+      <name>Daichi Yamada</name>
+    </author>
+    <author>
+      <name>Satoru Yamashita</name>
+    </author>
+  </entry>
+</feed>

tests/test_arxiv_collector_units.py

@@ -0,0 +1,75 @@
+"""B-3 PR2 — arXiv 파서·쿼리빌더 순수 단위 테스트 (plan safety-library-b3-1).
+
+fixture = arXiv API 실응답 박제(abs:"pressure vessel" relevance 10건 —
+DOI 보유 / journal_ref 만 보유 / 둘 다 없음 3경로 포함). run()/적재(DB)는 PR2 라이브 검증.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "app"))
+
+from workers.arxiv_collector import (  # noqa: E402
+    build_search_query,
+    parse_arxiv_feed,
+)
+
+FIX = Path(__file__).parent / "fixtures" / "arxiv_search_pressure_vessel.xml"
+
+
+def _entries():
+    total, entries = parse_arxiv_feed(FIX.read_text(encoding="utf-8"))
+    return total, {e.arxiv_id: e for e in entries}, entries
+
+
+# ─── 피드 레벨 ───
+
+def test_feed_total_and_count():
+    total, by_id, entries = _entries()
+    assert total == 89                                 # fixture totalResults (페이징 재료)
+    assert len(entries) == 10
+
+
+def test_versionless_ids():
+    _, by_id, entries = _entries()
+    # arxiv_id 는 versionless (버전 접미는 .version 으로 분리)
+    assert all("/" not in e.arxiv_id for e in entries)
+    assert "1209.2405" in by_id and by_id["1209.2405"].version == "v1"
+
+
+# ─── DOI 보유 entry ───
+
+def test_entry_with_doi():
+    _, by_id, _ = _entries()
+    e = by_id["1209.2405"]
+    assert e.doi == "10.1063/1.4707088"            # normalize_doi 적용(소문자·정규화)
+    assert e.journal_ref is None
+    assert e.primary_category == "physics.acc-ph"
+    assert e.title.startswith("A Survey of Pressure Vessel")
+    assert len(e.summary) > 200                     # 초록 본문
+    assert e.published is not None
+    assert e.abs_url and "/abs/" in e.abs_url
+    assert e.pdf_url and "pdf" in e.pdf_url
+
+
+# ─── journal_ref 만 (DOI 없음) — 압력용기 저널 출판분 ───
+
+def test_entry_journal_ref_without_doi():
+    _, by_id, _ = _entries()
+    e = by_id["0804.0261"]
+    assert e.doi is None
+    assert e.journal_ref and "Pressure Vessel" in e.journal_ref
+
+
+# ─── 둘 다 없음(최근 preprint) 경로도 존재 ───
+
+def test_entry_neither_doi_nor_journal_ref_exists():
+    _, _, entries = _entries()
+    assert any(e.doi is None and e.journal_ref is None for e in entries)
+
+
+# ─── 쿼리 빌더 ───
+
+def test_build_search_query():
+    q = build_search_query("eess.SY", ["pressure vessel", "safety"])
+    assert q == 'cat:eess.SY AND (abs:"pressure vessel" OR abs:safety)'

tests/test_openalex_collector_units.py

@@ -0,0 +1,106 @@
+"""B-3 PR3 — OpenAlex 파서·초록복원·license 순수 단위 테스트 (plan safety-library-b3-1).
+
+fixture = OpenAlex /works 실응답 박제(process safety/pressure vessel OA 5건 —
+cc-by/cc-by-nc-nd/license None, 초록 있음/없음). run()/적재(DB)는 PR3 라이브 검증.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "app"))
+
+from workers.openalex_collector import (  # noqa: E402
+    _reconstruct_abstract,
+    _seeds,
+    build_filter,
+    build_issn_filter,
+    license_meta,
+    parse_openalex_works,
+)
+
+FIX = Path(__file__).parent / "fixtures" / "openalex_works_response.json"
+
+
+def _works():
+    count, cursor, works = parse_openalex_works(FIX.read_text(encoding="utf-8"))
+    return count, {w.openalex_id: w for w in works}, works
+
+
+# ─── 피드 레벨 ───
+
+def test_count_and_results():
+    count, by_id, works = _works()
+    assert count == 1111
+    assert len(works) == 5
+    assert all(w.openalex_id.startswith("W") and "/" not in w.openalex_id for w in works)
+
+
+# ─── 초록 보유 + CC 라이선스 ───
+
+def test_work_with_abstract_and_cc():
+    _, by_id, _ = _works()
+    w = by_id["W2910511816"]
+    assert w.doi and w.doi.startswith("10.") and w.doi == w.doi.lower()  # normalize_doi
+    assert len(w.abstract) > 50            # inverted-index 복원
+    assert w.oa_status == "diamond" and w.is_oa is True
+    assert w.license == "cc-by"
+    assert license_meta(w.license, w.is_oa, w.source_name)["redistribute"] is True
+
+
+# ─── 초록 없는 thin 레코드(skip 대상) ───
+
+def test_work_without_abstract():
+    _, by_id, _ = _works()
+    w = by_id["W3107397139"]
+    assert w.abstract == ""                # inverted-index 부재 → 빈 초록
+    lm = license_meta(w.license, w.is_oa, w.source_name)
+    assert lm["redistribute"] is False     # license None → 비배포
+
+
+# ─── cc-by-nc-nd 도 CC 계열 → redistribute True ───
+
+def test_cc_variant_redistribute():
+    _, by_id, _ = _works()
+    w = by_id["W4391130399"]
+    assert w.license == "cc-by-nc-nd"
+    assert license_meta(w.license, w.is_oa, w.source_name)["redistribute"] is True
+
+
+# ─── 초록 inverted-index 복원 순서 ───
+
+def test_reconstruct_abstract_order():
+    inv = {"Safety": [0], "of": [1, 4], "pressure": [2], "vessels": [3], "design": [5]}
+    assert _reconstruct_abstract(inv) == "Safety of pressure vessels of design"
+    assert _reconstruct_abstract(None) == ""
+    assert _reconstruct_abstract({}) == ""
+
+
+# ─── license_meta 분기 ───
+
+def test_license_meta_branches():
+    assert license_meta("cc-by", True, "X")["redistribute"] is True
+    assert license_meta("cc0", True, "X")["redistribute"] is True
+    none_oa = license_meta(None, True, "X")
+    assert none_oa["redistribute"] is False and none_oa["scheme"] == "open-unspecified"
+    closed = license_meta(None, False, "X")
+    assert closed["redistribute"] is False and closed["scheme"] == "proprietary"
+
+
+# ─── 쿼리 빌더 ───
+
+def test_build_filter():
+    assert build_filter("process safety") == "title_and_abstract.search:process safety"
+    assert build_filter("process safety", "2026-06-01") == \
+        "title_and_abstract.search:process safety,from_publication_date:2026-06-01"
+
+
+# ─── PR6: ISSN 소스 시드 (KR/JP 안전 저널 직접 커버) ───
+
+def test_build_issn_filter_and_seeds():
+    assert build_issn_filter("1738-3803") == "primary_location.source.issn:1738-3803"
+    assert build_issn_filter("1738-3803", "2026-01-01") == \
+        "primary_location.source.issn:1738-3803,from_publication_date:2026-01-01"
+    seeds = _seeds()
+    kinds = [k for _, _, k in seeds]
+    assert kinds[0] == "issn"  # ISSN 시드가 키워드보다 먼저(cap 우선권)
+    assert any(v == "1738-3803" and k == "issn" for _, v, k in seeds)  # 한국안전학회지 포함

tests/test_paper_doi_units.py

@@ -0,0 +1,141 @@
+"""B-3 PR1 — 논문 DOI 코어 순수 단위 테스트 (plan safety-library-b3-1).
+
+holder.find_paper_holder(DB 조회)는 PR2 arXiv 실수집 시 라이브 검증 — 여기선 순수 함수만.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent.parent / "app"))
+
+from services.papers.doi import (  # noqa: E402
+    arxiv_doi,
+    normalize_doi,
+    paper_doi_hash,
+    parse_arxiv_id,
+    parse_doi_from_text,
+    read_paper_doi,
+    with_paper_doi,
+    with_parent_doi,
+)
+
+
+# ─── normalize_doi: 단일 함수(저장=조회) ───
+
+def test_normalize_strips_url_and_lowercases():
+    assert normalize_doi("https://doi.org/10.1585/PFR.15.2402039") == "10.1585/pfr.15.2402039"
+    assert normalize_doi("http://dx.doi.org/10.1115/1.4045678") == "10.1115/1.4045678"
+    assert normalize_doi("doi:10.1016/j.jlp.2020.104321") == "10.1016/j.jlp.2020.104321"
+    assert normalize_doi("DOI: 10.1234/ABC") == "10.1234/abc"
+
+
+def test_normalize_trims_whitespace_and_citation_noise():
+    assert normalize_doi("  https://doi.org/10.1234/abc  ") == "10.1234/abc"
+    assert normalize_doi("10.1234/abc.") == "10.1234/abc"
+    assert normalize_doi("10.1234/abc;") == "10.1234/abc"
+
+
+def test_normalize_preserves_parens_in_doi():
+    # 괄호는 DOI 일부일 수 있어 보존 (과삭제 = 다른 논문 병합 = 데이터 손상, near-dup 보다 위험)
+    assert normalize_doi("10.1016/s0010-8650(00)80003-2") == "10.1016/s0010-8650(00)80003-2"
+    assert normalize_doi("https://doi.org/10.1016/S0010-8650(00)80003-2") == "10.1016/s0010-8650(00)80003-2"
+
+
+def test_normalize_rejects_non_doi():
+    assert normalize_doi(None) is None
+    assert normalize_doi("") is None
+    assert normalize_doi("   ") is None
+    assert normalize_doi("not-a-doi") is None
+    assert normalize_doi("arXiv:2606.08108") is None  # arXiv id 는 DOI 아님
+
+
+def test_normalize_is_idempotent_store_equals_lookup():
+    # 저장측·조회측이 같은 함수를 거치면 표기 차이가 한 값으로 붕괴 (dedup 성립 조건)
+    forms = [
+        "https://doi.org/10.1/X",
+        "doi:10.1/x",
+        "10.1/X",
+        "  HTTPS://DOI.ORG/10.1/x ",
+    ]
+    assert {normalize_doi(f) for f in forms} == {"10.1/x"}
+    assert normalize_doi(normalize_doi("https://doi.org/10.1/X")) == "10.1/x"  # 멱등
+
+
+# ─── paper_doi_hash: holder file_hash 키 ───
+
+def test_paper_doi_hash_deterministic_len32():
+    h = paper_doi_hash("10.1234/abc")
+    assert len(h) == 32
+    assert h == paper_doi_hash("10.1234/abc")
+
+
+def test_paper_doi_hash_distinct_per_doi():
+    assert paper_doi_hash("10.1/a") != paper_doi_hash("10.1/b")
+
+
+# ─── 2-Document extract_meta 계약 (holder doi / child parent_doi 상호 배타) ───
+
+def test_with_paper_doi_holder_shape_and_merge_safe():
+    meta = with_paper_doi({"license": {"scheme": "cc_by"}, "source_id": 7}, "10.1/x")
+    assert meta["paper"]["doi"] == "10.1/x"
+    assert "parent_doi" not in meta["paper"]
+    assert meta["license"]["scheme"] == "cc_by"  # 타 키 보존
+    assert meta["source_id"] == 7
+
+
+def test_with_parent_doi_child_shape_no_doi():
+    meta = with_parent_doi({"license": {"scheme": "proprietary"}}, "10.1/holder")
+    assert meta["paper"]["parent_doi"] == "10.1/holder"
+    assert "doi" not in meta["paper"]  # child 는 doi 미보유 (partial-unique 인덱스 밖)
+    assert meta["license"]["scheme"] == "proprietary"
+
+
+def test_holder_child_mutually_exclusive():
+    child = with_parent_doi({}, "10.1/p")
+    promoted = with_paper_doi(child, "10.1/self")
+    assert promoted["paper"]["doi"] == "10.1/self"
+    assert "parent_doi" not in promoted["paper"]
+
+
+def test_input_not_mutated():
+    src = {"paper": {"doi": "10.1/old"}}
+    with_parent_doi(src, "10.1/new")
+    assert src["paper"]["doi"] == "10.1/old"  # 원본 dict 불변
+
+
+# ─── read_paper_doi: 인덱스 식의 조회측 거울 ───
+
+def test_read_paper_doi():
+    assert read_paper_doi({"paper": {"doi": "10.1/x"}}) == "10.1/x"
+    assert read_paper_doi({"paper": {"doi": "https://doi.org/10.1/X"}}) == "10.1/x"  # 방어적 재정규화
+    assert read_paper_doi({}) is None
+    assert read_paper_doi(None) is None
+    assert read_paper_doi({"paper": {"parent_doi": "10.1/p"}}) is None  # child 는 doi 없음
+    assert read_paper_doi({"paper": {}}) is None
+
+
+# ─── PR4: arXiv id 파싱 + arXiv DataCite DOI (교차소스 dedup 통일 키) ───
+
+def test_parse_arxiv_id():
+    assert parse_arxiv_id("Title arXiv:2606.10236v1 Announce Type: new Abstract") == "2606.10236"
+    assert parse_arxiv_id("see arXiv:2601.02852 for details") == "2601.02852"
+    assert parse_arxiv_id("arXiv:cond-mat/0703470v2") == "cond-mat/0703470"
+    assert parse_arxiv_id("no arxiv here") is None
+    assert parse_arxiv_id(None) is None
+
+
+def test_arxiv_doi_canonical():
+    # OpenAlex canonical 실측 일치: 10.48550/arxiv.{id} (소문자)
+    assert arxiv_doi("2606.10236") == "10.48550/arxiv.2606.10236"
+    assert arxiv_doi(None) is None
+    # 수집기·reconcile 가 같은 함수 → 같은 paper.doi (교차소스 dedup 성립)
+    assert arxiv_doi(parse_arxiv_id("x arXiv:2606.10236v1 y")) == "10.48550/arxiv.2606.10236"
+
+
+# ─── PR5: 구매 PDF 본문 DOI 파싱 (parent_doi 링크용, PDF 구조 무관) ───
+
+def test_parse_doi_from_text():
+    assert parse_doi_from_text("ref https://doi.org/10.1016/j.jlp.2024.105474 end") == "10.1016/j.jlp.2024.105474"
+    assert parse_doi_from_text("DOI 10.1115/1.4045678. Next.") == "10.1115/1.4045678"
+    assert parse_doi_from_text("no doi here") is None
+    assert parse_doi_from_text(None) is None

tests/test_pipeline_hold.py

@@ -26,7 +26,8 @@ def _fake_consumer_env(monkeypatch, held):
         lambda: {
             s: object()
             for s in (queue_consumer.MAIN_QUEUE_STAGES
-                      + queue_consumer.FAST_QUEUE_STAGES + ["markdown"])
+                      + queue_consumer.FAST_QUEUE_STAGES
+                      + queue_consumer.DEEP_QUEUE_STAGES + ["markdown"])
         },
     )
     monkeypatch.setattr(queue_consumer, "_hold_logged", False)
@@ -83,13 +84,37 @@ async def test_fast_consumer_respects_hold(monkeypatch):
     assert processed == ["chunk"]
 
 
+@pytest.mark.asyncio
+async def test_deep_consumer_processes_deep_only(monkeypatch):
+    """deep 컨슈머(2026-06-15 분리) = deep_summary 전용 (메인 루프와 디커플)."""
+    processed = _fake_consumer_env(monkeypatch, [])
+
+    await queue_consumer.consume_deep_queue()
+
+    assert processed == ["deep_summary"]
+
+
+@pytest.mark.asyncio
+async def test_deep_consumer_respects_hold(monkeypatch):
+    """deep_summary 홀드 시 deep 컨슈머가 claim 안 함."""
+    processed = _fake_consumer_env(monkeypatch, ["deep_summary"])
+
+    await queue_consumer.consume_deep_queue()
+
+    assert processed == []
+
+
 def test_fast_split_invariants():
-    """세 컨슈머 stage 집합 disjoint + embed/chunk 배치 상향 회귀 가드."""
+    """네 컨슈머 stage 집합 disjoint + embed/chunk 배치 상향 + deep split 회귀 가드."""
     main = set(queue_consumer.MAIN_QUEUE_STAGES)
     fast = set(queue_consumer.FAST_QUEUE_STAGES)
     md = set(queue_consumer.MARKDOWN_QUEUE_STAGES)
+    deep = set(queue_consumer.DEEP_QUEUE_STAGES)
     assert not (main & fast) and not (main & md) and not (fast & md)
+    assert not (main & deep) and not (fast & deep) and not (md & deep)
     assert fast == {"embed", "chunk"}
+    assert deep == {"deep_summary"}
+    assert "deep_summary" not in main  # 2026-06-15 split 회귀 가드
     assert queue_consumer.BATCH_SIZE["embed"] >= 10
     assert queue_consumer.BATCH_SIZE["chunk"] >= 10