fix(summarize): map_results persist aliasing — 유닛 스냅샷 소급 오염으로 UPDATE 스킵

60254 라이브 E2E 에서 발견: 완주는 성공했으나 payload.presegment.map_results 에
unit 0 만 persist. 원인 = map_results dict 를 in-place 변경 → 직전 commit 의
SQLAlchemy committed 스냅샷이 같은 중첩 객체를 참조 → old==new 판정 → 2번째
commit 부터 UPDATE 스킵. 멱등 재개 시 완료 유닛 재호출 비용 발생(정확성 무영향).

fix = 매 유닛 map_results/preseg/payload 전부 새 dict 재구성(공유 참조 0).
test = FakeSession 이 commit 시점 payload 객체 참조를 박제, 사후 직렬화로
스냅샷 유닛 수가 1..n 단조 증가 단정 — 구 코드에 대해 FAILED 네거티브 검증 완료.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

app/Dockerfile

@@ -11,8 +11,8 @@ RUN apt-get update && \
       ffmpeg && \
     apt-get clean && rm -rf /var/lib/apt/lists/*
 
-COPY requirements.txt .
-RUN pip install --no-cache-dir -r requirements.txt
+COPY requirements.txt requirements.lock ./
+RUN pip install --no-cache-dir -r requirements.lock
 
 COPY . .
 

app/api/study_concepts.py

@@ -0,0 +1,94 @@
+"""study_concepts API — 이론공부 홈(오늘의 개념 · 진도 · 회독 SR). prefix = /api/study.
+
+문제풀이 표면 무접촉. 개념문서(가스기사 태그) 읽기 집계 + 회독 SR write 만. 단일 토픽(가스기사=4).
+경로: GET /curriculum · GET /today-concepts · POST /concepts/{doc_id}/read.
+"""
+
+from __future__ import annotations
+
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.auth import get_current_user
+from core.database import get_session
+from models.user import User
+from services.study import concept_curriculum as cc
+from services.study import concept_links as cl
+
+router = APIRouter()
+
+# 가스기사 단일 토픽 운영(현행). 다토픽 확장 시 쿼리 파라미터로 승격.
+DEFAULT_TOPIC_ID = 4
+
+
+@router.get("/curriculum")
+async def get_curriculum(
+    user: Annotated[User, Depends(get_current_user)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+    topic_id: int = DEFAULT_TOPIC_ID,
+):
+    """과목별 회독 진도 + 개념/문항 복습 due 요약."""
+    return await cc.curriculum(session, user.id, topic_id)
+
+
+@router.get("/today-concepts")
+async def get_today_concepts(
+    user: Annotated[User, Depends(get_current_user)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+    topic_id: int = DEFAULT_TOPIC_ID,
+    limit: int = 6,
+):
+    """오늘 공부할 개념(재복습 → 미독 빈출순)."""
+    return await cc.today_concepts(session, user.id, topic_id, limit)
+
+
+@router.get("/concepts/weakness-map")
+async def get_weakness_map(
+    user: Annotated[User, Depends(get_current_user)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+    topic_id: int = DEFAULT_TOPIC_ID,
+    limit: int = 12,
+):
+    """개념 약점 지도 — 링크된 기출 정답률로 약점 개념(정답률<60%) 우선(이론↔문제)."""
+    name = await cc._topic_name(session, topic_id)
+    if not name:
+        return {"weak": [], "weak_total": 0, "evaluated_total": 0}
+    return await cl.weakness_map(session, user.id, name, limit)
+
+
+@router.get("/concepts/{doc_id}")
+async def get_concept_detail(
+    doc_id: int,
+    user: Annotated[User, Depends(get_current_user)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+    topic_id: int = DEFAULT_TOPIC_ID,
+):
+    """개념 리더 재료 — 구조 파싱(요약/본문/빈출/관련) + 백링크 해소 + 회독/SR + 이전/다음."""
+    detail = await cc.concept_detail(session, user.id, topic_id, doc_id)
+    if detail is None:
+        raise HTTPException(status_code=404, detail="concept not found")
+    return detail
+
+
+@router.get("/concepts/{doc_id}/questions")
+async def get_concept_questions(
+    doc_id: int,
+    user: Annotated[User, Depends(get_current_user)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+    limit: int = 20,
+):
+    """개념 관련 기출 + 내 정답률 (이론↔문제 브리지)."""
+    return await cl.related_questions(session, user.id, doc_id, limit)
+
+
+@router.post("/concepts/{doc_id}/read")
+async def post_concept_read(
+    doc_id: int,
+    user: Annotated[User, Depends(get_current_user)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+    topic_id: int = DEFAULT_TOPIC_ID,
+):
+    """개념 회독 처리 → 회독 플래그 + SR 입고/전진."""
+    return await cc.mark_read(session, user.id, topic_id, doc_id)

app/main.py

@@ -33,6 +33,7 @@ from api.study_sessions import router as study_sessions_router
 from api.study_topics import router as study_topics_router
 from api.study_reminders import router as study_reminders_router
 from api.study_cards import router as study_cards_router
+from api.study_concepts import router as study_concepts_router
 from api.video import router as video_router
 from core.config import settings
 from core.database import async_session, engine, init_db
@@ -249,6 +250,8 @@ app.include_router(study_reminders_router, prefix="/api/study-reminders", tags=[
 app.include_router(study_cards_router, prefix="/api/study-cards", tags=["study-cards"])
 # Phase 1: 학습 진행 상태 (review-complete + review-queue). prefix=/api/study-topics 안에 정의됨.
 app.include_router(study_question_progress_router, prefix="/api", tags=["study-progress"])
+# 이론공부 홈: 오늘의 개념·진도·회독 SR (개념문서 소비 표면, 문제풀이 무접촉).
+app.include_router(study_concepts_router, prefix="/api/study", tags=["study-theory"])
 
 # TODO: Phase 5에서 추가
 # app.include_router(tasks.router, prefix="/api/tasks", tags=["tasks"])

app/models/study_concept_progress.py

@@ -0,0 +1,46 @@
+"""study_concept_progress — 사용자 × 개념문서 단위 간격반복(SR) 진행 (이론공부 홈).
+
+문제 SR(study_question_progress)의 개념(이론)판. '개념문서' = documents 한 건(가스기사 태그).
+회독(첫 read) → 복습 큐 진입, 이후 회독마다 sr_schedule 산술(1·3·7·14·졸업) 공용 전진.
+concept_doc_id 는 documents.id 를 가리키나 FK 미설정 — hot 테이블(documents) 락 회피(clause_study 선례).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, SmallInteger, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from core.database import Base
+
+
+class StudyConceptProgress(Base):
+    __tablename__ = "study_concept_progress"
+    __table_args__ = (
+        UniqueConstraint(
+            "user_id", "concept_doc_id", name="uq_concept_progress_user_doc"
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    user_id: Mapped[int] = mapped_column(
+        BigInteger, ForeignKey("users.id", ondelete="CASCADE"), nullable=False
+    )
+    study_topic_id: Mapped[int] = mapped_column(
+        BigInteger, ForeignKey("study_topics.id", ondelete="CASCADE"), nullable=False
+    )
+    # documents.id 참조 — FK 없음(락 회피). 개념문서 삭제 시 고아 행은 read 집계에서 자연 제외.
+    concept_doc_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+
+    # 복습 큐 (sr_schedule 공용): stage 0~3 = 1·3·7·14일, 4 = 졸업(due_at NULL)
+    review_stage: Mapped[int | None] = mapped_column(SmallInteger)
+    due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    last_read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=datetime.now, nullable=False
+    )
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=datetime.now, onupdate=datetime.now, nullable=False
+    )

app/policy/prompt_render.py

@@ -36,6 +36,8 @@ KNOWN_4B_TASKS = {
 }
 KNOWN_26B_TASKS = {
     "p3c_deep_summary",
+    # presegment PR2 — 거대문서 map-reduce 의 reduce 단계 (요약들의 요약)
+    "p3c_deep_summary_reduce",
     "p4b_synthesis",
 }
 

app/prompts/policy/p3c_deep_summary_reduce.txt

@@ -0,0 +1,44 @@
+[System]
+너는 긴 문서·문서 묶음 분석가다. 이 문서는 한 번에 처리하기에 너무 커서, 원문을 순서대로 유닛으로 나눠 각 유닛을 먼저 요약했다(map 단계). 아래 "유닛 요약"들은 원문 순서 그대로이며 문서 전체를 빠짐없이 커버한다. 너는 이를 종합해 문서 전체의 최종 분석을 작성한다(reduce 단계).
+
+subject_description: {subject_description}
+
+{forbidden_block}
+
+envelope 를 읽는 순서:
+1. risk_flags 를 먼저 본다. 어떤 위험 때문에 올라온 것인지 파악.
+2. synthesis_directives 를 system 지시로 간주하여 반드시 준수.
+3. distilled_context 는 "참고 요지"일 뿐, 근거는 유닛 요약에서 재확인.
+
+작성 규칙:
+- TL;DR (1문장, 최대 60자)
+- 핵심 (bullets 5개, 각 30~80자)
+- 상세 (2~4 문단, 각 3~5문장) — 유닛(섹션) 순서의 논리 흐름을 보전하며 문서 전체를 관통하는 서술. 특정 유닛만 편식하지 말 것.
+- 유닛 요약에 없는 정보 금지 (hallucination 금지). 숫자·조문·인용은 유닛 요약에 있는 것만 사용.
+- 유닛 요약의 "불일치(...)" 줄들은 중복 제거해 inconsistencies 로 보전 — 임의로 버리지 않는다.
+- synthesis_directives 의 문구 규칙 ("원인은 ~" 금지 등) 반드시 준수.
+- multi_reference_synthesis flag 있으면 레퍼런스별 입장 분리 기술, 종합 권고 금지.
+
+출력 (JSON only):
+{{
+  "mode": "single|bundle",
+  "tldr": "...",
+  "bullets": ["..."],
+  "detail": "...\\n\\n...",
+  "bundle_flow": ["..."] | null,
+  "inconsistencies": ["..."] | null,
+  "entities_confirmed": {{
+    "people": [{{"name": "...", "evidence": "..."}}],
+    "orgs": [...],
+    "projects": [...]
+  }},
+  "directives_applied": ["..."],
+  "confidence": 0.0~1.0
+}}
+
+[User]
+Envelope:
+{{escalation_envelope_json}}
+
+유닛 요약 (총 {{unit_count}}개, 원문 순서 — 각 블록 = 원문 한 구간의 요약):
+{{unit_summaries}}

app/requirements.lock

@@ -0,0 +1,104 @@
+# requirements.lock — 라이브 fastapi 컨테이너 pip freeze 스냅샷 (2026-07-02, 101 pkgs, CVE-clear known-good)
+# 재생성: docker exec hyungi_document_server-fastapi-1 pip freeze > app/requirements.lock (헤더 재부착)
+# requirements.txt = 사람이 편집하는 floor 사양(>=) / 본 lock = Dockerfile 이 실제 설치하는 정본(==)
+annotated-doc==0.0.4
+annotated-types==0.7.0
+anthropic==0.109.1
+anyio==4.13.0
+APScheduler==3.11.2
+asyncpg==0.31.0
+babel==2.18.0
+bcrypt==5.0.0
+beautifulsoup4==4.15.0
+caldav==3.2.1
+certifi==2026.5.20
+cffi==2.0.0
+chardet==7.4.3
+charset-normalizer==3.4.7
+click==8.4.1
+cobble==0.1.4
+courlan==1.4.0
+cryptography==48.0.1
+cssselect==1.4.0
+dateparser==1.4.0
+defusedxml==0.7.1
+distro==1.9.0
+dnspython==2.8.0
+docstring_parser==0.18.0
+ecdsa==0.19.2
+et_xmlfile==2.0.0
+fastapi==0.136.3
+feedparser==6.0.12
+flatbuffers==25.12.19
+greenlet==3.5.1
+h11==0.16.0
+htmldate==1.10.0
+httpcore==1.0.9
+httptools==0.8.0
+httpx==0.28.1
+icalendar==7.1.2
+icalendar-searcher==1.0.6
+idna==3.18
+jh2==5.0.13
+Jinja2==3.1.6
+jiter==0.15.0
+jusText==3.0.2
+lxml==6.1.1
+lxml_html_clean==0.4.5
+magika==0.6.3
+mammoth==1.11.0
+Markdown==3.10.2
+markdownify==1.2.2
+markitdown==0.1.6
+MarkupSafe==3.0.3
+niquests==3.19.1
+numpy==2.4.6
+olefile==0.47
+onnxruntime==1.26.0
+openpyxl==3.1.5
+packaging==26.2
+pandas==3.0.3
+pgvector==0.4.2
+pillow==12.2.0
+protobuf==7.35.0
+pyasn1==0.6.3
+pycparser==3.0
+pydantic==2.13.4
+pydantic_core==2.46.4
+pyhwp==0.1b15
+PyMuPDF==1.27.2.3
+pyotp==2.9.0
+python-dateutil==2.9.0.post0
+python-dotenv==1.2.2
+python-jose==3.5.0
+python-multipart==0.0.32
+python-pptx==1.0.2
+pytz==2026.2
+PyYAML==6.0.3
+qh3==1.9.2
+readability-lxml==0.8.4.1
+recurring-ical-events==3.8.2
+regex==2026.5.9
+requests==2.34.2
+rsa==4.9.1
+sgmllib3k==1.0.0
+six==1.17.0
+sniffio==1.3.1
+soupsieve==2.8.4
+SQLAlchemy==2.0.50
+starlette==1.2.1
+tld==0.13.2
+trafilatura==2.1.0
+typing-inspection==0.4.2
+typing_extensions==4.15.0
+tzdata==2026.2
+tzlocal==5.3.1
+urllib3==2.7.0
+urllib3-future==2.21.902
+uvicorn==0.49.0
+uvloop==0.22.1
+wassima==2.1.1
+watchfiles==1.2.0
+websockets==16.0
+x-wr-timezone==2.0.1
+xlsxwriter==3.2.9

app/services/study/concept_curriculum.py

@@ -0,0 +1,284 @@
+"""concept_curriculum — 이론공부 홈 재료 (오늘의 개념 · 진도 · 회독 SR).
+
+개념문서 = documents (user_tags = @library/{topic}/{과목}/... , 가스기사). is_read = 회독,
+md_content 의 ★ 개수 = 빈출 tier(★★★=3 / ★★=2 / else 1). 회독 SR = study_concept_progress
++ sr_schedule(문제 SR 공용 산술). 읽기 전용 집계 + mark_read(회독+SR 입고)만 write. LLM 0.
+
+문제풀이 표면 무접촉 — 여기서 읽는 study_question_progress 는 '문항 due 카운트'만(홈 표시용).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from sqlalchemy import func, or_, select, text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from models.document_read import DocumentRead
+from models.study_concept_progress import StudyConceptProgress
+from models.study_question_progress import StudyQuestionProgress
+from models.study_topic import StudyTopic
+from services.study.concept_parser import parse_concept, resolve_related
+from services.study.sr_schedule import advance, first_due
+
+# 개념 행 조회 — 태그로 개념문서 필터 + 회독 진행 LEFT JOIN. md_content 는 전송 안 하고
+# ★ 유무만 서버측 boolean 으로(홈이 자주 호출돼도 페이로드 최소).
+# is_read = document_reads(회독 정본, is_read 컬럼 아님) EXISTS. library unread 와 동일 기준.
+_CONCEPT_ROWS_SQL = text(
+    """
+    SELECT d.id AS doc_id,
+           d.title AS title,
+           EXISTS (
+             SELECT 1 FROM document_reads r
+             WHERE r.document_id = d.id AND r.user_id = :uid
+           ) AS is_read,
+           (d.md_content LIKE '%★★★%') AS f3,
+           (d.md_content LIKE '%★★%')  AS f2,
+           split_part(replace(d.user_tags::text, '"', ''), '/', 3) AS subject,
+           p.review_stage AS review_stage,
+           p.due_at AS due_at,
+           p.last_read_at AS last_read_at
+    FROM documents d
+    LEFT JOIN study_concept_progress p
+      ON p.concept_doc_id = d.id AND p.user_id = :uid
+    WHERE d.user_tags::text LIKE :like
+      AND d.deleted_at IS NULL
+    """
+)
+
+
+async def _topic_name(session: AsyncSession, topic_id: int) -> str | None:
+    return (
+        await session.execute(select(StudyTopic.name).where(StudyTopic.id == topic_id))
+    ).scalar_one_or_none()
+
+
+async def _concept_rows(session: AsyncSession, user_id: int, topic_name: str):
+    like = f"%@library/{topic_name}/%"
+    return (
+        await session.execute(_CONCEPT_ROWS_SQL, {"uid": user_id, "like": like})
+    ).mappings().all()
+
+
+def _freq(row) -> int:
+    if row["f3"]:
+        return 3
+    if row["f2"]:
+        return 2
+    return 1
+
+
+def _is_due(row, now: datetime) -> bool:
+    return (
+        row["due_at"] is not None
+        and row["due_at"] <= now
+        and (row["review_stage"] or 0) < 4
+    )
+
+
+def _item(row) -> dict:
+    return {
+        "doc_id": row["doc_id"],
+        "title": row["title"],
+        "subject": row["subject"],
+        "freq": _freq(row),
+        "review_stage": row["review_stage"],
+        "due_at": row["due_at"],
+    }
+
+
+async def _question_due_count(session: AsyncSession, user_id: int, topic_id: int, now: datetime) -> int:
+    """문항 복습 due (기존 study_question_progress 엔진 재사용, 홈 표시용)."""
+    return (
+        await session.execute(
+            select(func.count())
+            .select_from(StudyQuestionProgress)
+            .where(
+                StudyQuestionProgress.user_id == user_id,
+                StudyQuestionProgress.study_topic_id == topic_id,
+                StudyQuestionProgress.due_at.is_not(None),
+                StudyQuestionProgress.due_at <= now,
+                or_(
+                    StudyQuestionProgress.review_stage.is_(None),
+                    StudyQuestionProgress.review_stage < 4,
+                ),
+            )
+        )
+    ).scalar_one()
+
+
+async def curriculum(session: AsyncSession, user_id: int, topic_id: int) -> dict:
+    """과목별 회독 진도 + 개념/문항 복습 due 요약 (진도 대시보드)."""
+    name = await _topic_name(session, topic_id)
+    rows = await _concept_rows(session, user_id, name) if name else []
+    now = datetime.now(timezone.utc)
+
+    subj: dict[str, dict] = {}
+    for r in rows:
+        s = subj.setdefault(r["subject"], {"subject": r["subject"], "total": 0, "read": 0})
+        s["total"] += 1
+        if r["is_read"]:
+            s["read"] += 1
+
+    total = len(rows)
+    read = sum(1 for r in rows if r["is_read"])
+    concept_due = sum(1 for r in rows if _is_due(r, now))
+    question_due = await _question_due_count(session, user_id, topic_id, now)
+
+    return {
+        "topic_id": topic_id,
+        "topic_name": name,
+        "subjects": sorted(subj.values(), key=lambda x: x["subject"]),
+        "total": total,
+        "read": read,
+        "concept_due": concept_due,
+        "question_due": question_due,
+    }
+
+
+async def today_concepts(
+    session: AsyncSession, user_id: int, topic_id: int, limit: int = 6
+) -> dict:
+    """오늘 공부할 개념 = 재복습(SR due) 먼저 → 미독(빈출 우선). 졸업/재복습대기 제외."""
+    name = await _topic_name(session, topic_id)
+    rows = await _concept_rows(session, user_id, name) if name else []
+    now = datetime.now(timezone.utc)
+
+    due = [r for r in rows if _is_due(r, now)]
+    due.sort(key=lambda r: r["due_at"])
+
+    # 미독 & 아직 SR 큐 진입 전(due_at NULL) → 빈출 높은 순
+    unread = [r for r in rows if not r["is_read"] and r["due_at"] is None]
+    unread.sort(key=lambda r: (-_freq(r), r["subject"], r["title"]))
+
+    picked = [{**_item(r), "reason": "재복습"} for r in due]
+    picked += [{**_item(r), "reason": "신규"} for r in unread]
+
+    return {
+        "concepts": picked[:limit],
+        "due_total": len(due),
+        "unread_total": len(unread),
+    }
+
+
+async def mark_read(
+    session: AsyncSession, user_id: int, topic_id: int, doc_id: int, now: datetime | None = None
+) -> dict:
+    """개념 회독 처리 = document_reads(+1) + 회독 SR 입고/전진.
+
+    회독 정본 = document_reads(append-only), documents.is_read 컬럼 아님(library unread 와 정합).
+    첫 회독 → first_due(stage 0, 내일). 이후 회독은 'due 도래(due_at<=now)' 때만 correct 로 전진
+    (이른 재열람/다중클릭 과전진 방지). stage 4 졸업 후엔 due_at NULL 이라 전진 없음.
+    """
+    now = now or datetime.now(timezone.utc)
+
+    # 회독 로그 append (+1) — 사용자 명시 회독. 자동 아님(엔드포인트 = 명시 POST).
+    session.add(DocumentRead(user_id=user_id, document_id=doc_id, read_at=now))
+
+    prog = (
+        await session.execute(
+            select(StudyConceptProgress).where(
+                StudyConceptProgress.user_id == user_id,
+                StudyConceptProgress.concept_doc_id == doc_id,
+            )
+        )
+    ).scalar_one_or_none()
+
+    if prog is None:
+        stage, due = first_due(now)
+        prog = StudyConceptProgress(
+            user_id=user_id,
+            study_topic_id=topic_id,
+            concept_doc_id=doc_id,
+            review_stage=stage,
+            due_at=due,
+            last_read_at=now,
+        )
+        session.add(prog)
+    else:
+        # due 도래 시에만 전진 — 미래 due(재열람 이른 클릭)는 stage 불변, last_read_at 만 갱신.
+        if prog.due_at is not None and prog.due_at <= now:
+            res = advance(prog.review_stage, "correct", now)
+            if res is not None:
+                prog.review_stage, prog.due_at = res
+        prog.last_read_at = now
+
+    await session.commit()
+    await session.refresh(prog)
+    return {"ok": True, "review_stage": prog.review_stage, "due_at": prog.due_at}
+
+
+_CONCEPT_ONE_SQL = text(
+    """
+    SELECT d.id AS doc_id, d.title AS title, d.md_content AS md_content,
+           split_part(replace(d.user_tags::text, '"', ''), '/', 3) AS subject,
+           (d.md_content LIKE '%★★★%') AS f3,
+           (d.md_content LIKE '%★★%')  AS f2,
+           EXISTS (
+             SELECT 1 FROM document_reads r
+             WHERE r.document_id = d.id AND r.user_id = :uid
+           ) AS is_read,
+           p.review_stage AS review_stage,
+           p.due_at AS due_at
+    FROM documents d
+    LEFT JOIN study_concept_progress p ON p.concept_doc_id = d.id AND p.user_id = :uid
+    WHERE d.id = :doc_id AND d.deleted_at IS NULL AND d.user_tags::text LIKE :like
+    """
+)
+
+
+async def concept_detail(
+    session: AsyncSession, user_id: int, topic_id: int, doc_id: int
+) -> dict | None:
+    """개념 리더 재료 — md 구조 파싱 + 관련개념 백링크 해소 + 회독/SR 상태 + 같은 과목 이전/다음."""
+    name = await _topic_name(session, topic_id)
+    if not name:
+        return None
+    like = f"%@library/{name}/%"
+    row = (
+        await session.execute(
+            _CONCEPT_ONE_SQL, {"uid": user_id, "doc_id": doc_id, "like": like}
+        )
+    ).mappings().first()
+    if row is None:
+        return None
+
+    parsed = parse_concept(row["md_content"] or "")
+
+    # 백링크 해소 + 이전/다음 = 같은 토픽 개념 title 인덱스(회독 rows 재사용)
+    idx = await _concept_rows(session, user_id, name)
+    title_index = [(r["doc_id"], r["title"], r["subject"]) for r in idx]
+    resolved = resolve_related(parsed["related"], title_index)
+
+    # 이전/다음 = 같은 과목, title 순
+    same = sorted(
+        [(r["doc_id"], r["title"]) for r in idx if r["subject"] == row["subject"]],
+        key=lambda x: (x[1] or "", x[0]),
+    )
+    ids = [d for d, _ in same]
+    prev_id = next_id = None
+    if doc_id in ids:
+        pos = ids.index(doc_id)
+        if pos > 0:
+            prev_id = ids[pos - 1]
+        if pos < len(ids) - 1:
+            next_id = ids[pos + 1]
+
+    freq = 3 if row["f3"] else (2 if row["f2"] else 1)
+
+    return {
+        "doc_id": row["doc_id"],
+        "db_title": row["title"],
+        "title": parsed["title"] or row["title"],
+        "subject": row["subject"],
+        "freq": freq,
+        "summary": parsed["summary"],
+        "body": parsed["body"],
+        "bincheol": parsed["bincheol"],
+        "related": resolved,
+        "is_read": row["is_read"],
+        "review_stage": row["review_stage"],
+        "due_at": row["due_at"],
+        "prev_id": prev_id,
+        "next_id": next_id,
+    }

app/services/study/concept_links.py

@@ -0,0 +1,139 @@
+"""concept_links — 이론↔문제 브리지 롤업 (Stage B).
+
+study_concept_links(개념 doc ↔ 기출문항, 임베딩 코사인) + study_question_progress(내 풀이상태)를
+조인해 (a) 개념별 관련 기출 + 내 정답률(related_questions), (b) 개념 약점 지도(weakness_map) 산출.
+읽기 전용 집계 · LLM 0. 링크 적재는 scripts/concept_links_backfill.sql(임베딩) 배치.
+정답률 = 링크된 문항 중 progress.last_outcome 기준(attempted=풀이이력 보유, correct=최근정답).
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import AsyncSession
+
+_ACCURACY_WEAK_PCT = 60  # 정답률 < 60% = 약점(attempted>0 일 때만)
+
+_AGG_SQL = text(
+    """
+    SELECT count(*) AS linked,
+           count(pr.study_question_id) FILTER (WHERE pr.last_outcome IS NOT NULL) AS attempted,
+           count(*) FILTER (WHERE pr.last_outcome = 'correct') AS correct
+    FROM study_concept_links l
+    LEFT JOIN study_question_progress pr
+      ON pr.study_question_id = l.question_id AND pr.user_id = :uid
+    WHERE l.concept_doc_id = :doc_id AND l.link_source = 'embedding'
+    """
+)
+
+_QROWS_SQL = text(
+    """
+    SELECT q.id AS id, q.subject AS subject, q.exam_round AS exam_round,
+           q.exam_question_number AS qnum, l.score AS score,
+           pr.last_outcome AS last_outcome, pr.review_stage AS review_stage
+    FROM study_concept_links l
+    JOIN study_questions q ON q.id = l.question_id AND q.deleted_at IS NULL AND q.is_active
+    LEFT JOIN study_question_progress pr
+      ON pr.study_question_id = q.id AND pr.user_id = :uid
+    WHERE l.concept_doc_id = :doc_id AND l.link_source = 'embedding'
+    ORDER BY l.score DESC
+    LIMIT :limit
+    """
+)
+
+_WEAKNESS_SQL = text(
+    """
+    SELECT d.id AS doc_id, d.title AS title,
+           split_part(replace(d.user_tags::text, '"', ''), '/', 3) AS subject,
+           count(l.id) AS linked,
+           count(pr.study_question_id) FILTER (WHERE pr.last_outcome IS NOT NULL) AS attempted,
+           count(*) FILTER (WHERE pr.last_outcome = 'correct') AS correct
+    FROM documents d
+    JOIN study_concept_links l ON l.concept_doc_id = d.id AND l.link_source = 'embedding'
+    LEFT JOIN study_question_progress pr
+      ON pr.study_question_id = l.question_id AND pr.user_id = :uid
+    WHERE d.user_tags::text LIKE :like AND d.deleted_at IS NULL
+    GROUP BY d.id, d.title, subject
+    """
+)
+
+
+async def related_questions(
+    session: AsyncSession, user_id: int, doc_id: int, limit: int = 20
+) -> dict:
+    """개념 doc 의 관련 기출 + 내 정답률(전체 링크 기준 집계 + 상위 N 표시용)."""
+    agg = (
+        await session.execute(_AGG_SQL, {"uid": user_id, "doc_id": doc_id})
+    ).mappings().first()
+    rows = (
+        await session.execute(
+            _QROWS_SQL, {"uid": user_id, "doc_id": doc_id, "limit": limit}
+        )
+    ).mappings().all()
+
+    linked = (agg["linked"] if agg else 0) or 0
+    attempted = (agg["attempted"] if agg else 0) or 0
+    correct = (agg["correct"] if agg else 0) or 0
+    accuracy = round(100 * correct / attempted) if attempted else None
+
+    return {
+        "linked": linked,
+        "attempted": attempted,
+        "correct": correct,
+        "accuracy": accuracy,
+        "questions": [
+            {
+                "id": r["id"],
+                "subject": r["subject"],
+                "exam_round": r["exam_round"],
+                "qnum": r["qnum"],
+                "score": round(r["score"], 3) if r["score"] is not None else None,
+                "last_outcome": r["last_outcome"],
+                "review_stage": r["review_stage"],
+            }
+            for r in rows
+        ],
+    }
+
+
+async def weakness_map(
+    session: AsyncSession, user_id: int, topic_name: str, limit: int = 12
+) -> dict:
+    """개념 약점 지도 — 링크된 기출 정답률로 개념 채색. 약점(attempted>0·정답률<60%) 우선 정렬."""
+    like = f"%@library/{topic_name}/%"
+    rows = (
+        await session.execute(_WEAKNESS_SQL, {"uid": user_id, "like": like})
+    ).mappings().all()
+
+    concepts = []
+    for r in rows:
+        attempted = r["attempted"] or 0
+        correct = r["correct"] or 0
+        accuracy = round(100 * correct / attempted) if attempted else None
+        if accuracy is None:
+            state = "unattempted"
+        elif accuracy < _ACCURACY_WEAK_PCT:
+            state = "weak"
+        else:
+            state = "ok"
+        concepts.append(
+            {
+                "doc_id": r["doc_id"],
+                "title": r["title"],
+                "subject": r["subject"],
+                "linked": r["linked"] or 0,
+                "attempted": attempted,
+                "accuracy": accuracy,
+                "state": state,
+            }
+        )
+
+    # 약점 우선(정답률 오름차순) → 미평가는 뒤로. 홈 위젯용 상위 N.
+    weak = sorted(
+        [c for c in concepts if c["state"] == "weak"],
+        key=lambda c: (c["accuracy"], -c["attempted"], c["doc_id"]),
+    )
+    return {
+        "weak": weak[:limit],
+        "weak_total": len(weak),
+        "evaluated_total": sum(1 for c in concepts if c["state"] != "unattempted"),
+    }

app/services/study/concept_parser.py

@@ -0,0 +1,175 @@
+"""concept_parser — 개념노트 markdown 구조 파서 + 관련개념 백링크 해소 (이론 리더용).
+
+정찰 실측 불변식(273/273): 개념노트는 고정 골격을 100% 따름 —
+    # {H1 제목}                     (첫 줄, DB title 과 다른 표시용 제목)
+    > **한 줄 요약**: {요약}          (blockquote, 라벨 고정)
+    ## {본문 라벨}  ...              (BODY, 자유 라벨 H2 0~N, 트레일 ★ 가능)
+    ## 빈출 포인트                    (항상, 관련개념 직전)
+    ## 관련 개념                      (항상, 문서 최종 섹션)
+
+코드펜스(``` ASCII 도식) 내부의 ##/- 는 무시. 헤딩 트레일 ★ 는 스트립(라벨 정규화).
+'빈출 포인트'/'관련 개념' 앵커만 이름으로 잡고 나머지 BODY 는 순서·위치로 처리(라벨 화이트리스트 금지).
+순수 함수 · LLM 0.
+"""
+
+from __future__ import annotations
+
+import re
+
+_FENCE = re.compile(r"^\s*```")
+_H1 = re.compile(r"^#\s+(.+?)\s*$")
+_H2 = re.compile(r"^##\s+(.+?)\s*$")  # ### 는 매칭 안 됨(## 뒤 \s 요구)
+_SUMMARY = re.compile(r"^>\s*\*\*한 줄 요약\*\*:\s*(.+)$")
+_STAR_SUFFIX = re.compile(r"\s*★+\s*$")
+_TRAIL_STARS = re.compile(r"★+\s*$")
+_BINCHEOL_ITEM = re.compile(r"^\s*-\s+(★*)\s*(.+)$")
+_RELATED_ITEM = re.compile(r"^\s*-\s+(.+)$")
+_PAREN = re.compile(r"\s*\(.*$")  # 괄호부터 끝(clarifier 힌트 절단)
+_NUM_PREFIX = re.compile(r"^\d+_")
+_STRIP_SYM = re.compile(r"[\s_·,./()\-]")
+
+_ANCHOR_BINCHEOL = "빈출 포인트"
+_ANCHOR_RELATED = "관련 개념"
+
+
+def parse_concept(md: str) -> dict:
+    """개념노트 md → {title, summary, body[{label,stars,md}], bincheol[{tier,text}], related[{raw,phrase,hint}]}."""
+    lines = (md or "").split("\n")
+    title: str | None = None
+    summary: str | None = None
+    body: list[dict] = []
+    bincheol_lines: list[str] = []
+    related_lines: list[str] = []
+
+    in_fence = False
+    zone = "pre"  # pre | body | bincheol | related
+    body_cur: dict | None = None
+
+    def emit(line: str) -> None:
+        if body_cur is not None:
+            body_cur["_lines"].append(line)
+        elif zone == "bincheol":
+            bincheol_lines.append(line)
+        elif zone == "related":
+            related_lines.append(line)
+        # pre-zone 내용(요약 앞 잡음)은 버림
+
+    for ln in lines:
+        if _FENCE.match(ln):
+            in_fence = not in_fence
+            emit(ln)
+            continue
+        if in_fence:
+            emit(ln)
+            continue
+
+        if title is None:
+            m = _H1.match(ln)
+            if m:
+                title = m.group(1).strip()
+                continue
+        if summary is None:
+            m = _SUMMARY.match(ln)
+            if m:
+                summary = m.group(1).strip()
+                continue
+
+        m2 = _H2.match(ln)
+        if m2:
+            raw_label = m2.group(1).strip()
+            star_m = _TRAIL_STARS.search(raw_label)
+            stars = len(star_m.group(0).strip()) if star_m else 0
+            label = _STAR_SUFFIX.sub("", raw_label).strip()
+            if label == _ANCHOR_BINCHEOL:
+                zone = "bincheol"
+                body_cur = None
+                continue
+            if label == _ANCHOR_RELATED:
+                zone = "related"
+                body_cur = None
+                continue
+            body_cur = {"label": label, "stars": stars, "_lines": []}
+            body.append(body_cur)
+            zone = "body"
+            continue
+
+        emit(ln)
+
+    body_out = []
+    for s in body:
+        text = "\n".join(s["_lines"]).strip()
+        if text or s["label"]:
+            body_out.append({"label": s["label"], "stars": s["stars"], "md": text})
+
+    bincheol = []
+    for ln in bincheol_lines:
+        m = _BINCHEOL_ITEM.match(ln)
+        if m:
+            bincheol.append({"tier": len(m.group(1)), "text": m.group(2).strip()})
+
+    related = []
+    for ln in related_lines:
+        m = _RELATED_ITEM.match(ln)
+        if m:
+            raw = m.group(1).strip()
+            phrase = _PAREN.sub("", raw).strip()
+            hint = raw[len(phrase):].strip() if len(raw) > len(phrase) else ""
+            if phrase:
+                related.append({"raw": raw, "phrase": phrase, "hint": hint})
+
+    return {
+        "title": title,
+        "summary": summary,
+        "body": body_out,
+        "bincheol": bincheol,
+        "related": related,
+    }
+
+
+def _normalize(s: str) -> str:
+    """해소용 정규화: NN_ 접두 제거 → 소문자 → 공백/기호 제거. 영문은 lowercase 유지."""
+    s = _NUM_PREFIX.sub("", s or "")
+    s = s.lower()
+    s = _STRIP_SYM.sub("", s)
+    return s
+
+
+def resolve_related(related: list[dict], title_index: list[tuple]) -> list[dict]:
+    """관련개념 구절 → 개념 doc 해소. title_index = [(doc_id, title, subject), ...].
+
+    다단 fallback(정찰 ~79%): 정규화 exact → 양방향 substring(≥2자 가드) → 미해소=dangling(doc_id None).
+    """
+    norm_exact: dict[str, int] = {}
+    norm_list: list[tuple[str, int, str]] = []
+    for did, ttl, _subj in title_index:
+        n = _normalize(ttl)
+        if n:
+            norm_exact.setdefault(n, did)
+            norm_list.append((n, did, ttl))
+
+    out = []
+    for it in related:
+        pn = _normalize(it["phrase"])
+        did: int | None = None
+        rtitle: str | None = None
+        if pn and len(pn) >= 2:
+            if pn in norm_exact:
+                did = norm_exact[pn]
+            else:
+                # substring 폴백: title-norm ⊆ phrase-norm 방향만(짧은 phrase 가 더 큰 title 을
+                # 삼키는 오결선 방지, 예: '염산'→'염산나트륨' X) + 길이차 최소(가장 구체적) +
+                # doc_id tiebreak(순서 무관 결정성). 후보 없으면 dangling(doc_id None).
+                cands = [
+                    (abs(len(n) - len(pn)), cand, ttl)
+                    for n, cand, ttl in norm_list
+                    if len(n) >= 2 and n in pn
+                ]
+                if cands:
+                    cands.sort(key=lambda c: (c[0], c[1]))
+                    _, did, rtitle = cands[0]
+        if did is not None and rtitle is None:
+            rtitle = next((t for d, t, _ in title_index if d == did), None)
+        out.append(
+            {"phrase": it["phrase"], "hint": it["hint"], "doc_id": did, "title": rtitle}
+        )
+    return out

app/services/summarize_units.py

@@ -0,0 +1,224 @@
+"""summarize_units — 거대문서 요약 전용 분할(map-reduce 유닛) 순수함수 (presegment PR1).
+
+plan ds-presegment-mapreduce-2 (2026-06-29 설계 합의 · PR0 실측 봉인):
+  - CAP_TOKENS = 12,000 tok/unit — greedy-pack 상한 (PR0: giant 236건 실측 캘리브레이션)
+  - TRIGGER_TOKENS = 25,000 tok — 이하는 단일콜 유지, 초과 시 map-reduce
+  - 3-way over% 게이트 (단독 CAP 초과 섹션의 토큰 비중. 헤딩 개수는 무의미 — ASME 1,494개):
+      over% == 0        → 'auto'   (TIER1: 로컬 자동 분할, PR0 실측 78%)
+      0 < over% <= 40   → 'hybrid' (패킹분 로컬 + 초과 섹션만 클로드, 8%)
+      over% > 40        → 'whole'  (TIER2: 클로드 전체 분할, 14%)
+  - 토큰 추정 = PR0 실 Qwen 토크나이저 캘리브레이션: 한글 0.529 tok/char · 기타 0.217.
+    구 휴리스틱(0.625/0.25)은 ~15% 과대라 폐기.
+
+불변식:
+  - 순수함수 — DB/네트워크/파일 접촉 0. 분할 = 요약 전용 아티팩트(문서 아님·검색/임베딩 미편입).
+  - leaf 추출 = hier_decomp.builder 재사용, leaf_hard_max=∞ 로 window-split 억제
+    (헤딩 leaf 만 — PR0 측정환경과 동일). 인접 섹션만 greedy-pack(순서 보존·중간 폐기 0
+    — 구 deep_summary 의 head/mid/tail 가운데 폐기 버그를 커버리지로 대체).
+  - 배선(deep_summary 분기·HOLD·클로드 알람)은 PR2/PR3 — 본 모듈은 계획만 산출.
+
+호출: plan_summarize_units(md_text) -> UnitPlan
+"""
+from __future__ import annotations
+
+import sys
+from dataclasses import dataclass, field
+
+# 상대 import — 컨테이너(services.*)와 repo-root 테스트(app.services.*) 양쪽에서 동작.
+# (구 `from app.services...` 절대 import 는 컨테이너에 app 패키지가 없어 ModuleNotFoundError —
+#  PR1 은 소비자 0 이라 잠복했던 버그, PR2 배선 시점에 수정.)
+from .hier_decomp.builder import HierNode, build_hier_tree
+
+CAP_TOKENS = 12_000
+TRIGGER_TOKENS = 25_000
+HYBRID_MAX_OVER_PCT = 40.0
+
+# PR0 실 Qwen tokenizer 캘리브레이션 (tok/char)
+KO_TOK_PER_CHAR = 0.529
+OTHER_TOK_PER_CHAR = 0.217
+
+_HANGUL_RANGES = (
+    (0xAC00, 0xD7A3),  # 완성형 음절
+    (0x1100, 0x11FF),  # 자모
+    (0x3130, 0x318F),  # 호환 자모
+)
+
+
+def _is_hangul(ch: str) -> bool:
+    cp = ord(ch)
+    return any(lo <= cp <= hi for lo, hi in _HANGUL_RANGES)
+
+
+def estimate_tokens(text: str) -> int:
+    """PR0 캘리브레이션 기반 토큰 추정 (한글 0.529 · 기타 0.217 tok/char)."""
+    if not text:
+        return 0
+    ko = sum(1 for ch in text if _is_hangul(ch))
+    other = len(text) - ko
+    return round(ko * KO_TOK_PER_CHAR + other * OTHER_TOK_PER_CHAR)
+
+
+@dataclass
+class SummarizeUnit:
+    """map-reduce 1유닛 — 인접 leaf 섹션들의 greedy-pack (요약 전용, 문서 아님)."""
+    index: int
+    section_titles: list[str | None] = field(default_factory=list)
+    text: str = ""
+    est_tokens: int = 0
+    over_cap: bool = False  # 단독 섹션이 CAP 초과 (hybrid 시 클로드 대상)
+
+
+@dataclass
+class UnitPlan:
+    mode: str                    # 'single' | 'map_reduce'
+    tier: str | None             # map_reduce 시 'auto' | 'hybrid' | 'whole'
+    total_est_tokens: int = 0
+    over_pct: float = 0.0
+    units: list[SummarizeUnit] = field(default_factory=list)
+
+
+def extract_leaves(md_text: str) -> list[HierNode]:
+    """헤딩 leaf 만 추출 — leaf_hard_max=∞ 로 window-split 억제 (PR0 측정환경 동일)."""
+    nodes = build_hier_tree(
+        md_text,
+        leaf_target_max=sys.maxsize,
+        leaf_hard_max=sys.maxsize,
+    )
+    return [n for n in nodes if n.is_leaf]
+
+
+def greedy_pack(leaves: list[HierNode], cap: int = CAP_TOKENS) -> list[SummarizeUnit]:
+    """인접 leaf 를 순서 보존하며 est_tokens<=cap 으로 pack. 단독 초과 leaf = 전용 유닛(over_cap)."""
+    units: list[SummarizeUnit] = []
+    cur_titles: list[str | None] = []
+    cur_texts: list[str] = []
+    cur_tokens = 0
+
+    def _flush() -> None:
+        nonlocal cur_titles, cur_texts, cur_tokens
+        if cur_texts:
+            units.append(SummarizeUnit(
+                index=len(units),
+                section_titles=cur_titles,
+                text="\n\n".join(cur_texts),
+                est_tokens=cur_tokens,
+            ))
+            cur_titles, cur_texts, cur_tokens = [], [], 0
+
+    for leaf in leaves:
+        t = estimate_tokens(leaf.text)
+        if t > cap:
+            _flush()
+            units.append(SummarizeUnit(
+                index=len(units),
+                section_titles=[leaf.section_title],
+                text=leaf.text,
+                est_tokens=t,
+                over_cap=True,
+            ))
+            continue
+        if cur_tokens + t > cap:
+            _flush()
+        cur_titles.append(leaf.section_title)
+        cur_texts.append(leaf.text)
+        cur_tokens += t
+    _flush()
+    return units
+
+
+def over_pct(leaves: list[HierNode], cap: int = CAP_TOKENS) -> float:
+    """단독 CAP 초과 섹션들의 토큰 비중(%) — 3-way 게이트 입력."""
+    total = 0
+    over = 0
+    for leaf in leaves:
+        t = estimate_tokens(leaf.text)
+        total += t
+        if t > cap:
+            over += t
+    if total == 0:
+        return 0.0
+    return over * 100.0 / total
+
+
+def gate(over: float) -> str:
+    """over% → tier. 0=auto / (0,40]=hybrid / >40=whole. 클로드 결과 재검증에도 재사용."""
+    if over <= 0.0:
+        return "auto"
+    if over <= HYBRID_MAX_OVER_PCT:
+        return "hybrid"
+    return "whole"
+
+
+def plan_summarize_units(
+    md_text: str, *,
+    cap: int = CAP_TOKENS,
+    trigger: int = TRIGGER_TOKENS,
+) -> UnitPlan:
+    """문서 → 요약 실행 계획. trigger 이하=single(현행 단일콜), 초과=map_reduce(tier+units)."""
+    total = estimate_tokens(md_text)
+    if total <= trigger:
+        return UnitPlan(mode="single", tier=None, total_est_tokens=total)
+    leaves = extract_leaves(md_text)
+    pct = over_pct(leaves, cap)
+    return UnitPlan(
+        mode="map_reduce",
+        tier=gate(pct),
+        total_est_tokens=total,
+        over_pct=round(pct, 2),
+        units=greedy_pack(leaves, cap),
+    )
+
+
+# ─── PR2 — map/reduce 프롬프트 조립 순수함수 (deep_summary_worker 가 소비) ───
+
+def render_map_slice(unit: SummarizeUnit, total_units: int) -> str:
+    """map 콜의 {original_text_slices} 대체 — 유닛 위치·섹션 라벨 + 본문."""
+    titles = " · ".join(t for t in unit.section_titles if t) or "(무제 구간)"
+    return f"[유닛 {unit.index + 1}/{total_units} — 섹션: {titles}]\n{unit.text}"
+
+
+def _format_unit_summary(res: dict, total_units: int) -> str:
+    """map 결과 1건 → reduce 입력 블록. res 키 = index/titles/tldr/detail/inconsistencies."""
+    titles = " · ".join(t for t in (res.get("titles") or []) if t) or "(무제 구간)"
+    lines = [f"[유닛 {int(res.get('index', 0)) + 1}/{total_units} — 섹션: {titles}]"]
+    if res.get("tldr"):
+        lines.append(f"TLDR: {res['tldr']}")
+    if res.get("detail"):
+        lines.append(str(res["detail"]))
+    for inc in res.get("inconsistencies") or []:
+        if isinstance(inc, dict):
+            lines.append(f"불일치({inc.get('kind', '')}): {inc.get('desc', '')}")
+    return "\n".join(lines)
+
+
+def build_reduce_units_block(
+    results: list[dict],
+    budget_tokens: int,
+    *,
+    min_detail_chars: int = 200,
+) -> tuple[str, bool]:
+    """reduce 입력 블록 조립 — budget_tokens 이하 보장(캡 초과 0 검증 게이트의 reduce 측).
+
+    초과 시 detail 만 비례 절단(라벨·TLDR·불일치 보전, 원문 순서 유지). 반환 (block, truncated).
+    """
+    total_units = len(results)
+    work = [dict(r) for r in results]
+    truncated = False
+    for _ in range(4):
+        block = "\n\n".join(_format_unit_summary(r, total_units) for r in work)
+        est = estimate_tokens(block)
+        if est <= budget_tokens:
+            return block, truncated
+        ratio = budget_tokens / est
+        for r in work:
+            detail = str(r.get("detail") or "")
+            keep = max(min_detail_chars, int(len(detail) * ratio * 0.9))
+            if len(detail) > keep:
+                r["detail"] = detail[:keep] + "…(절단)"
+                truncated = True
+    # 최후 방어 — 비례 절단이 floor(min_detail_chars)에 막히면 문자 하드 컷(KO 최악 비율 가정)
+    block = "\n\n".join(_format_unit_summary(r, total_units) for r in work)
+    if estimate_tokens(block) > budget_tokens:
+        block = block[: max(1, int(budget_tokens / KO_TOK_PER_CHAR))]
+        truncated = True
+    return block, truncated

app/workers/deep_summary_worker.py

@@ -10,7 +10,9 @@ EscalationEnvelope + subject_domain 을 읽어, PR-A policy 템플릿 `p3c_deep_
 
 from __future__ import annotations
 
+import asyncio
 import json
+import os
 import time
 from datetime import datetime, timezone
 
@@ -29,10 +31,25 @@ from models.queue import ProcessingQueue, StageDeferred
 from policy.prompt_render import render_26b, policy_version as compute_policy_version
 from services.document_telemetry import record_analyze_event
 from services.search.llm_gate import Priority, acquire_mlx_gate
+from services.summarize_units import (
+    CAP_TOKENS,
+    UnitPlan,
+    build_reduce_units_block,
+    estimate_tokens,
+    plan_summarize_units,
+    render_map_slice,
+)
 
 logger = setup_logger("deep_summary_worker")
 
 DEEP_SUMMARY_TASK = "p3c_deep_summary"
+# presegment PR2 (plan ds-presegment-mapreduce-2) — 거대문서 map-reduce
+REDUCE_TASK = "p3c_deep_summary_reduce"
+# HYBRID/TIER2(클로드 유인 분할 필요) HOLD 재확인 간격. PR3(알람·경계 주입) 전까지는
+# 이 간격으로 재계획만 반복한다 — attempts 미소모(StageDeferred)라 영구 failed 없음.
+HOLD_RETRY_MINUTES = int(os.getenv("DEEP_SUMMARY_HOLD_RETRY_MINUTES", "1440"))
+# reduce 프롬프트 오버헤드가 비정상적으로 커도 유닛 블록 예산은 이 밑으로 안 내려감(방어).
+REDUCE_BUDGET_FLOOR_TOKENS = 1_000
 
 # inconsistencies kind 허용 목록 (feedback_document_server_domain_scope.md — 구매/계약 제외)
 ALLOWED_INCONSISTENCY_KINDS = {
@@ -94,6 +111,25 @@ async def process(
 
     envelope = EscalationEnvelope.from_json(json.dumps(envelope_raw))
 
+    # ─── presegment PR2 게이트 (plan ds-presegment-mapreduce-2) ───
+    # TRIGGER(25K tok) 이하 = 아래 기존 단일콜 경로 그대로(무회귀). 초과 시 3-way:
+    #   auto(over%==0)   → 로컬 map-reduce (유닛별 26B → reduce)
+    #   hybrid/whole     → HOLD(awaiting_split) — 맥미니 미전송, 클로드 유인 분할은 PR3
+    # 게이트/유닛은 전체 extracted_text 기준 — 단일콜의 head/mid/tail "가운데 폐기"를
+    # 전 유닛 커버리지로 대체한다. build_hier_tree 가 거대 md 에서 초 단위 CPU 라
+    # 이벤트루프 점유 회피 위해 to_thread (presegment_worker._read_toc 와 동일 패턴).
+    unit_plan = await asyncio.to_thread(plan_summarize_units, doc.extracted_text or "")
+    if unit_plan.mode == "map_reduce":
+        # units 빈 auto 는 이론상 불가(비어있지 않은 텍스트 = leaf >= 1)지만, 빈 reduce
+        # 단일콜(환각 위험)로 흐르지 않게 방어적으로 HOLD 로 보낸다.
+        if unit_plan.tier != "auto" or not unit_plan.units:
+            await _hold_awaiting_split(session, queue_row, unit_plan, document_id)
+        await _process_map_reduce(
+            doc, queue_row, envelope, subject_domain, unit_plan, session,
+            defer_on_deep_unavailable=defer_on_deep_unavailable,
+        )
+        return
+
     # 원문 슬라이스 추출 (envelope.original_pointers.text_ranges 기반)
     slices = _build_text_slices(doc.extracted_text or "", envelope.original_pointers)
 
@@ -214,6 +250,267 @@ async def process(
     )
 
 
+async def _hold_awaiting_split(
+    session: AsyncSession, queue_row: ProcessingQueue, plan: UnitPlan, document_id: int
+) -> None:
+    """HYBRID/TIER2 — 클로드 유인 분할 대기(HOLD). 맥미니 미전송, StageDeferred 보류.
+
+    payload.presegment.awaiting_split 마킹을 먼저 commit — StageDeferred 핸들러
+    (queue_consumer)는 새 세션에서 행을 다시 읽어 deferred_until 만 병합하므로 유실 없음.
+    알람(ntfy)·클로드 경계 주입은 PR3 — 그 전까지는 HOLD_RETRY_MINUTES 간격 재계획만 반복.
+    무인 자동 cloud 호출 금지 룰 준수(클로드 경로는 항상 유인 게이트).
+    """
+    payload = dict(queue_row.payload or {})
+    preseg = dict(payload.get("presegment") or {})
+    preseg.update({
+        "awaiting_split": True,
+        "tier": plan.tier,
+        "over_pct": plan.over_pct,
+        "total_est_tokens": plan.total_est_tokens,
+        "units": len(plan.units),
+        # 클로드가 분할해야 할 초과 섹션 표본 (PR3 알람 본문용)
+        "oversized_sections": [
+            (u.section_titles[0] if u.section_titles else None)
+            for u in plan.units if u.over_cap
+        ][:20],
+    })
+    payload["presegment"] = preseg
+    queue_row.payload = payload  # 재할당 = JSONB 변경 감지
+    await session.commit()
+    logger.info(
+        f"[deep] id={document_id} awaiting_split tier={plan.tier} over_pct={plan.over_pct} "
+        f"total_est_tokens={plan.total_est_tokens} units={len(plan.units)} "
+        f"→ HOLD ({HOLD_RETRY_MINUTES}분 후 재확인, 클로드 분할=PR3 유인)"
+    )
+    raise StageDeferred(
+        f"awaiting_split:{plan.tier}", retry_after_minutes=HOLD_RETRY_MINUTES
+    )
+
+
+async def _call_26b(
+    client: AIClient, prompt: str, *, defer_on_deep_unavailable: bool, document_id: int
+):
+    """map/reduce 공용 26B 호출 — 단일콜 경로와 동일한 deep 슬롯 우선 + fair-share 폴백.
+
+    반환 (raw, used_cfg). 맥북(deep) 불가 시 consumer 경로는 맥미니 primary 로 즉시
+    처리(동일 모델 — 강등 아님), drain 경로는 StageDeferred 전파(맥북 레버 시멘틱).
+    """
+    deep_cfg = client.ai.deep
+    if deep_cfg is not None:
+        try:
+            return await call_deep_or_defer(client, prompt), deep_cfg
+        except StageDeferred:
+            if defer_on_deep_unavailable:
+                raise
+            logger.info(f"[deep] id={document_id} 맥북 불가 → 맥미니 primary 처리 (fair-share)")
+    async with acquire_mlx_gate(Priority.BACKGROUND):
+        return await client.call_primary(prompt), settings.ai.primary
+
+
+def _parse_deep_output(raw: str) -> tuple[DeepSummaryOutput | None, str | None]:
+    """raw → DeepSummaryOutput. 단일콜 경로와 동일한 3단 파서. 실패 시 (None, parse_error)."""
+    try:
+        parsed = _parse_outermost_json(raw) or parse_json_response(raw)
+        if not parsed:
+            parsed = _regex_extract_fields(raw)
+        return DeepSummaryOutput.model_validate(parsed or {}), None
+    except (ValidationError, ValueError, TypeError) as exc:
+        return None, f"parse:{type(exc).__name__}"
+
+
+async def _process_map_reduce(
+    doc: Document,
+    queue_row: ProcessingQueue,
+    envelope: EscalationEnvelope,
+    subject_domain: str,
+    plan: UnitPlan,
+    session: AsyncSession,
+    *,
+    defer_on_deep_unavailable: bool,
+) -> None:
+    """TIER1 자동 — 유닛별 map(26B) → reduce(26B) → 단일콜과 동일 필드 기록.
+
+    멱등 재개: 성공 유닛은 payload.presegment.map_results 에 즉시 commit —
+    502/defer/재시작 후 재클레임 시 완료 유닛은 건너뛴다. 유닛 인덱스는
+    plan_summarize_units 가 같은 extracted_text 에 결정적이라 attempt 간 안정.
+    파싱 실패 유닛이 남으면 raise → queue_consumer 의 기존 attempts/백오프 재사용
+    (실패 유닛만 재호출되므로 재시도 비용 = 잔여 유닛뿐).
+    """
+    document_id = doc.id
+    units = plan.units
+    n = len(units)
+    payload = dict(queue_row.payload or {})
+    preseg = dict(payload.get("presegment") or {})
+    preseg.pop("awaiting_split", None)  # 재계획으로 auto 가 된 경우 HOLD 마킹 해제
+    map_results: dict = dict(preseg.get("map_results") or {})
+
+    logger.info(
+        f"[deep] id={document_id} map_reduce 시작 units={n} over_pct={plan.over_pct} "
+        f"total_est_tokens={plan.total_est_tokens} resume={len(map_results)}/{n}"
+    )
+
+    rendered = render_26b(DEEP_SUMMARY_TASK, subject_domain)
+    envelope_injection = envelope.to_system_injection()
+
+    client = AIClient()
+    start = time.perf_counter()
+    used_cfg = client.ai.deep or settings.ai.primary
+    failed_units: list[int] = []
+    try:
+        # ── map: 유닛별 26B (콜 사이마다 gate 를 놓아 짧은 인터랙티브 요청이 끼어든다) ──
+        for unit in units:
+            key = str(unit.index)
+            if key in map_results:
+                continue
+            prompt = (
+                rendered
+                .replace("{escalation_envelope_json}", envelope_injection)
+                .replace("{original_text_slices}", render_map_slice(unit, n))
+            )
+            # 검증 게이트 "모든 LLM 콜 캡 초과 0" 을 로그로 단정 가능하게 남긴다.
+            logger.info(
+                f"[deep] id={document_id} map {unit.index + 1}/{n} "
+                f"unit_tokens={unit.est_tokens} prompt_est_tokens={estimate_tokens(prompt)} "
+                f"cap={CAP_TOKENS}"
+            )
+            raw, used_cfg = await _call_26b(
+                client, prompt,
+                defer_on_deep_unavailable=defer_on_deep_unavailable,
+                document_id=document_id,
+            )
+            out, perr = _parse_deep_output(raw)
+            if out is None or not (out.detail or out.tldr):
+                # 실패 유닛은 persist 하지 않음 — 재시도가 이 유닛만 다시 호출한다.
+                failed_units.append(unit.index)
+                logger.warning(
+                    f"[deep] id={document_id} map {unit.index + 1}/{n} 결과 비었음/파싱 실패"
+                    f"({perr}) — 유닛 재시도 대상"
+                )
+                continue
+            # ★매 유닛 새 dict 로 재구성 (in-place 변경 금지) — 직전 commit 의 committed
+            # 스냅샷이 같은 중첩 객체를 참조하면 old==new 로 보여 SQLAlchemy 가 UPDATE 를
+            # 스킵한다(60254 라이브에서 unit 0 만 persist 된 aliasing 버그의 fix).
+            map_results = {
+                **map_results,
+                key: {
+                    "index": unit.index,
+                    "titles": [t for t in unit.section_titles if t][:8],
+                    "tldr": out.tldr,
+                    "detail": out.detail,
+                    "inconsistencies": _filter_inconsistencies(out.inconsistencies or []),
+                },
+            }
+            preseg = {
+                **preseg,
+                "tier": plan.tier,
+                "over_pct": plan.over_pct,
+                "total_est_tokens": plan.total_est_tokens,
+                "units": n,
+                "map_results": map_results,
+            }
+            payload = {**payload, "presegment": preseg}
+            queue_row.payload = payload  # 재할당 = JSONB 변경 감지
+            await session.commit()  # 유닛 단위 멱등 재개 지점
+
+        if failed_units:
+            raise ValueError(
+                f"map 유닛 {len(failed_units)}/{n}건 결과 없음 — 재시도 대상: {failed_units[:10]}"
+            )
+
+        # ── reduce: 요약들의 요약 1콜 (유닛 블록도 캡 이하로 절단 보장) ──
+        reduce_rendered = render_26b(REDUCE_TASK, subject_domain)
+        base_prompt = (
+            reduce_rendered
+            .replace("{escalation_envelope_json}", envelope_injection)
+            .replace("{unit_count}", str(n))
+        )
+        budget = max(
+            REDUCE_BUDGET_FLOOR_TOKENS, CAP_TOKENS - estimate_tokens(base_prompt)
+        )
+        ordered = [map_results[str(u.index)] for u in units]
+        block, reduce_truncated = build_reduce_units_block(ordered, budget)
+        reduce_prompt = base_prompt.replace("{unit_summaries}", block)
+        logger.info(
+            f"[deep] id={document_id} reduce units={n} "
+            f"prompt_est_tokens={estimate_tokens(reduce_prompt)} cap={CAP_TOKENS} "
+            f"truncated={reduce_truncated}"
+        )
+        raw, used_cfg = await _call_26b(
+            client, reduce_prompt,
+            defer_on_deep_unavailable=defer_on_deep_unavailable,
+            document_id=document_id,
+        )
+    except StageDeferred:
+        logger.info(
+            f"[deep] id={document_id} map_reduce 보류 — 완료 유닛 {len(map_results)}/{n} 보존"
+        )
+        raise
+    except Exception as exc:
+        # 단일콜 경로와 동일 — 호출 실패는 전파해 queue_consumer 가 재시도/dead-letter 처리.
+        logger.warning(f"[deep] id={document_id} map_reduce 실패: {exc}")
+        raise
+    finally:
+        await client.close()
+
+    latency_ms = int((time.perf_counter() - start) * 1000)
+    deep_out, parse_error = _parse_deep_output(raw)
+    if deep_out is None:
+        # 단일콜 경로와 동일 시멘틱 — doc 미기록(legacy 결과 보존), 이벤트로 가시화.
+        deep_out = DeepSummaryOutput()
+        logger.warning(f"[deep] id={document_id} reduce 파싱 실패 ({parse_error}) — doc 미기록")
+
+    if not parse_error:
+        doc.ai_detail_summary = (deep_out.detail or "").strip() or None
+        # 불일치 = reduce 출력 + map 유닛 합본 dedup — reduce 가 떨궈도 유닛 발견분 보전.
+        merged = _filter_inconsistencies(deep_out.inconsistencies or [])
+        seen = {(i["kind"], i["desc"]) for i in merged}
+        for res in ordered:
+            for inc in res.get("inconsistencies") or []:
+                k = (inc.get("kind"), inc.get("desc"))
+                if k not in seen:
+                    seen.add(k)
+                    merged.append(inc)
+        doc.ai_inconsistencies = merged
+        doc.ai_analysis_tier = "deep"
+        doc.ai_processed_at = datetime.now(timezone.utc)
+
+    try:
+        pv = compute_policy_version(REDUCE_TASK)
+    except Exception:
+        pv = None
+
+    await record_analyze_event(
+        doc_id=document_id,
+        user_id=None,
+        mode="summary_deep",
+        text_limit=used_cfg.context_char_limit or 260000,
+        truncated=reduce_truncated,
+        layers_returned=["detail_summary", "inconsistencies"] if not parse_error else [],
+        cached=False,
+        latency_ms=latency_ms,
+        model_name=used_cfg.model,
+        prompt_version=(f"{REDUCE_TASK}@{pv}" if pv else REDUCE_TASK),
+        error_code=parse_error,
+        source="document_server",
+        subject_domain=subject_domain,
+        risk_flags=list(envelope.risk_flags),
+        high_impact_task=None,
+        escalation_reasons=list(envelope.escalation_reasons),
+        confidence=deep_out.confidence,
+        policy_version=pv,
+        shadow_would_route_to="primary",
+        tier="primary",
+        escalated_to_26b=True,
+        suppressed_reason=None,
+    )
+
+    logger.info(
+        f"[deep] id={document_id} map_reduce 완료 units={n} "
+        f"detail_len={len(doc.ai_detail_summary or '')} inc={len(doc.ai_inconsistencies or [])} "
+        f"latency_ms={latency_ms} parse_error={parse_error}"
+    )
+
+
 def _build_text_slices(text: str, pointers: dict) -> str:
     """original_pointers.text_ranges 의 [{start, end}] 를 실제 본문 슬라이스로 합친다.
 

frontend/src/lib/components/Sidebar.svelte

@@ -2,7 +2,7 @@
   import { page } from '$app/stores';
   import { goto } from '$app/navigation';
   import { api } from '$lib/api';
-  import { ChevronRight, ChevronDown, FolderOpen, FolderTree, Inbox, Clock, Mail, Scale, StickyNote, GraduationCap, CalendarCheck, MessageCircle, Hash } from 'lucide-svelte';
+  import { ChevronRight, ChevronDown, FolderOpen, FolderTree, Inbox, Clock, Mail, Scale, StickyNote, GraduationCap, CalendarCheck, MessageCircle, Hash, HardHat } from 'lucide-svelte';
 
   let tree = $state([]);
   let loading = $state(true);
@@ -195,6 +195,13 @@
     >
       <FolderTree size={14} /> 자료실
     </a>
+    <a
+      href="/safety"
+      class="w-full flex items-center gap-2 px-3 py-1.5 rounded-md text-sm transition-colors
+        {$page.url.pathname.startsWith('/safety') ? 'bg-accent/15 text-accent' : 'text-dim hover:bg-surface hover:text-text'}"
+    >
+      <HardHat size={14} /> 안전 자료실
+    </a>
     <a
       href="/clause"
       class="w-full flex items-center gap-2 px-3 py-1.5 rounded-md text-sm transition-colors

frontend/src/routes/safety/+layout.svelte

@@ -0,0 +1,34 @@
+<script>
+  // 안전 자료실 (safety-library-1 Phase 3) — 재해/법령·지침/서적·표준·매뉴얼 3탭.
+  import { page } from '$app/stores';
+
+  const TABS = [
+    { href: '/safety/incidents', label: '재해사례' },
+    { href: '/safety/laws', label: '법령·지침' },
+    { href: '/safety/materials', label: '서적·표준·매뉴얼' },
+  ];
+</script>
+
+<div class="max-w-5xl mx-auto px-4 py-5 flex flex-col gap-4">
+  <header>
+    <h1 class="text-lg font-bold text-text">안전 자료실</h1>
+    <p class="text-xs text-dim mt-0.5">재해사례·법령·지침·표준 — 자료유형(material_type) 축 기반</p>
+  </header>
+
+  <nav class="flex gap-1 border-b border-default" aria-label="안전 자료실 탭">
+    {#each TABS as tab}
+      <a
+        href={tab.href}
+        aria-current={$page.url.pathname === tab.href ? 'page' : undefined}
+        class="px-3 py-2 text-sm font-medium border-b-2 -mb-px transition-colors
+          {$page.url.pathname === tab.href
+            ? 'border-accent text-accent'
+            : 'border-transparent text-dim hover:text-text'}"
+      >
+        {tab.label}
+      </a>
+    {/each}
+  </nav>
+
+  <slot />
+</div>

frontend/src/routes/safety/+page.svelte

@@ -0,0 +1,9 @@
+<script>
+  // /safety 진입 = 재해 탭 redirect (plan: +page=재해 탭 redirect)
+  import { onMount } from 'svelte';
+  import { goto } from '$app/navigation';
+
+  onMount(() => {
+    goto('/safety/incidents', { replaceState: true });
+  });
+</script>

frontend/src/routes/safety/SafetyDocList.svelte

@@ -0,0 +1,75 @@
+<script>
+  // 안전 자료실 공용 목록 — material_type + jurisdiction 필터로 GET /documents/ 조회.
+  // C-1 계약: material_type 지정 = 기본 exclude(news·law_monitor·note) 해제 (documents.py list_documents).
+  import { api } from '$lib/api';
+  import { addToast } from '$lib/stores/toast';
+  import DocumentCard from '$lib/components/DocumentCard.svelte';
+
+  let { materialType, jurisdiction = '' } = $props();
+
+  const PAGE_SIZE = 20;
+  let docs = $state([]);
+  let total = $state(0);
+  let nextPage = $state(1);
+  let loading = $state(false);
+
+  async function load(reset = false) {
+    loading = true;
+    const pageToLoad = reset ? 1 : nextPage;
+    try {
+      const params = new URLSearchParams();
+      params.set('material_type', materialType);
+      if (jurisdiction) params.set('jurisdiction', jurisdiction);
+      params.set('page', String(pageToLoad));
+      params.set('page_size', String(PAGE_SIZE));
+      const result = await api(`/documents/?${params}`);
+      docs = reset ? result.items : [...docs, ...result.items];
+      total = result.total;
+      nextPage = pageToLoad + 1;
+    } catch {
+      addToast('error', '안전 자료 로딩 실패');
+    } finally {
+      loading = false;
+    }
+  }
+
+  $effect(() => {
+    // 필터 변경 시 1페이지부터 재조회 (materialType/jurisdiction 읽기 = 반응 트리거)
+    void materialType;
+    void jurisdiction;
+    docs = [];
+    load(true);
+  });
+
+  let hasMore = $derived(docs.length < total);
+</script>
+
+<div class="flex flex-col gap-2">
+  {#if !loading || docs.length > 0}
+    <p class="text-xs text-dim tabular-nums">총 {total.toLocaleString()}건</p>
+  {/if}
+
+  {#if docs.length > 0}
+    <div class="flex flex-col gap-2">
+      {#each docs as doc (doc.id)}
+        <DocumentCard {doc} />
+      {/each}
+    </div>
+  {:else if !loading}
+    <div class="py-12 text-center text-sm text-dim">
+      해당 조건의 자료가 없습니다.
+    </div>
+  {/if}
+
+  {#if loading}
+    <div class="py-6 text-center text-sm text-dim">불러오는 중…</div>
+  {:else if hasMore}
+    <button
+      type="button"
+      onclick={() => load(false)}
+      class="self-center px-4 py-1.5 rounded-md text-sm text-dim border border-default hover:bg-surface hover:text-text transition-colors"
+    >
+      더 보기 ({docs.length}/{total.toLocaleString()})
+    </button>
+  {/if}
+</div>

frontend/src/routes/safety/incidents/+page.svelte

@@ -0,0 +1,29 @@
+<script>
+  // 재해사례 탭 — material_type=incident (KOSHA 사고사망·재해사례·CSB 등).
+  // 케이스 그룹핑(boardno 본문+첨부 1카드)은 API 확장 필요라 후속(DS freeze 하 백엔드 무변경).
+  import SafetyDocList from '../SafetyDocList.svelte';
+
+  const JURISDICTIONS = [
+    { value: '', label: '전체' },
+    { value: 'KR', label: 'KR' },
+    { value: 'US', label: 'US' },
+  ];
+  let jurisdiction = $state('');
+</script>
+
+<div class="flex flex-col gap-3">
+  <div class="flex items-center gap-1.5" role="group" aria-label="관할 필터">
+    {#each JURISDICTIONS as j}
+      <button
+        type="button"
+        onclick={() => (jurisdiction = j.value)}
+        class="px-2.5 py-1 rounded-full text-xs font-medium transition-colors
+          {jurisdiction === j.value ? 'bg-accent/15 text-accent' : 'text-dim hover:bg-surface hover:text-text'}"
+      >
+        {j.label}
+      </button>
+    {/each}
+  </div>
+
+  <SafetyDocList materialType="incident" {jurisdiction} />
+</div>

frontend/src/routes/safety/laws/+page.svelte

@@ -0,0 +1,48 @@
+<script>
+  // 법령·지침 탭 — 법령(law, 버전체인 current 만 코퍼스 노출) / 지침(guide, KOSHA GUIDE 등).
+  // 법령 기본 관할 = KR (plan: country 누락 = KR 정규화). version_status 뱃지는 API 확장 후속.
+  import SafetyDocList from '../SafetyDocList.svelte';
+
+  const KINDS = [
+    { value: 'law', label: '법령' },
+    { value: 'guide', label: '지침' },
+  ];
+  const JURISDICTIONS = [
+    { value: 'KR', label: 'KR' },
+    { value: 'US', label: 'US' },
+    { value: '', label: '전체' },
+  ];
+  let kind = $state('law');
+  let jurisdiction = $state('KR');
+</script>
+
+<div class="flex flex-col gap-3">
+  <div class="flex items-center justify-between flex-wrap gap-2">
+    <div class="flex items-center gap-1" role="group" aria-label="자료유형">
+      {#each KINDS as k}
+        <button
+          type="button"
+          onclick={() => (kind = k.value)}
+          class="px-3 py-1 rounded-md text-sm font-medium transition-colors
+            {kind === k.value ? 'bg-accent/15 text-accent' : 'text-dim hover:bg-surface hover:text-text'}"
+        >
+          {k.label}
+        </button>
+      {/each}
+    </div>
+    <div class="flex items-center gap-1.5" role="group" aria-label="관할 필터">
+      {#each JURISDICTIONS as j}
+        <button
+          type="button"
+          onclick={() => (jurisdiction = j.value)}
+          class="px-2.5 py-1 rounded-full text-xs font-medium transition-colors
+            {jurisdiction === j.value ? 'bg-accent/15 text-accent' : 'text-dim hover:bg-surface hover:text-text'}"
+        >
+          {j.label}
+        </button>
+      {/each}
+    </div>
+  </div>
+
+  <SafetyDocList materialType={kind} {jurisdiction} />
+</div>

frontend/src/routes/safety/materials/+page.svelte

@@ -0,0 +1,29 @@
+<script>
+  // 서적·표준·매뉴얼 탭 — 필터 프리셋(전용 뷰는 50건+ 게이트 뒤, plan Phase 3).
+  import SafetyDocList from '../SafetyDocList.svelte';
+
+  const KINDS = [
+    { value: 'standard', label: '표준 (NB 등)' },
+    { value: 'book', label: '서적' },
+    { value: 'manual', label: '매뉴얼' },
+    { value: 'paper', label: '논문' },
+  ];
+  let kind = $state('standard');
+</script>
+
+<div class="flex flex-col gap-3">
+  <div class="flex items-center gap-1" role="group" aria-label="자료유형">
+    {#each KINDS as k}
+      <button
+        type="button"
+        onclick={() => (kind = k.value)}
+        class="px-3 py-1 rounded-md text-sm font-medium transition-colors
+          {kind === k.value ? 'bg-accent/15 text-accent' : 'text-dim hover:bg-surface hover:text-text'}"
+      >
+        {k.label}
+      </button>
+    {/each}
+  </div>
+
+  <SafetyDocList materialType={kind} />
+</div>

frontend/src/routes/study/+page.svelte

@@ -1,13 +1,58 @@
 <script>
-  // /study — 학습 hub.
-  // 주제로 보기(퀴즈·복습·통계) / 자료 학습 / 필사 세션 / 암기카드 검수.
+  // /study — 학습 hub + 데일리 랜딩('오늘의 공부' 대시보드).
+  // 상단 = 이론 홈(진도·오늘의 개념·복습 due, 재노출 트리거). 하단 = 기존 모드 진입.
   import { onMount } from 'svelte';
   import { api } from '$lib/api';
-  import { BookOpen, PenLine, GraduationCap, FolderKanban, Layers, Repeat, Flag, Inbox, Activity } from 'lucide-svelte';
+  import { addToast } from '$lib/stores/toast';
+  import { BookOpen, PenLine, GraduationCap, FolderKanban, Layers, Repeat, Flag, Inbox, Activity, CalendarCheck, Target } from 'lucide-svelte';
 
   let cardReviewCount = $state(0);
   let questionFlagCount = $state(0);
+
+  // 오늘의 공부 (이론 홈)
+  let curriculum = $state(null);
+  let todayConcepts = $state([]);
+  let weakConcepts = $state([]);        // 약점 개념(관련 기출 정답률 낮음)
+  let dashLoading = $state(true);
+
+  let readPct = $derived(
+    curriculum && curriculum.total ? Math.round((curriculum.read / curriculum.total) * 100) : 0
+  );
+
+  async function loadDashboard() {
+    dashLoading = true;
+    try {
+      const [cur, today] = await Promise.all([
+        api('/study/curriculum'),
+        api('/study/today-concepts?limit=6'),
+      ]);
+      curriculum = cur;
+      todayConcepts = today?.concepts ?? [];
+    } catch {
+      // 코어 대시보드 실패해도 허브 나머지는 동작 (조용히)
+    } finally {
+      dashLoading = false;
+    }
+    // 약점 개념 = 비차단(신규 엔드포인트 실패해도 코어 대시보드 블랙아웃 방지)
+    try {
+      const weak = await api('/study/concepts/weakness-map?limit=5');
+      weakConcepts = weak?.weak ?? [];
+    } catch {}
+  }
+
+  async function markRead(doc) {
+    try {
+      await api(`/study/concepts/${doc.doc_id}/read`, { method: 'POST' });
+      todayConcepts = todayConcepts.filter((c) => c.doc_id !== doc.doc_id);
+      addToast('success', `회독: ${doc.title}`);
+      loadDashboard(); // 진도 갱신
+    } catch {
+      addToast('error', '회독 처리 실패');
+    }
+  }
+
   onMount(async () => {
+    loadDashboard();
     try {
       const r = await api('/study-cards/needs-review/count');
       cardReviewCount = r?.count ?? 0;
@@ -27,6 +72,80 @@
     <p class="text-sm text-dim mt-1">주제별 퀴즈·복습(SRS)·통계 / 학습 자료 회독 / 손글씨 필사 세션.</p>
   </header>
 
+  <!-- 오늘의 공부 (이론 홈 대시보드 = 데일리 트리거) -->
+  <section class="mb-5 rounded-lg border border-default bg-surface p-4 md:p-5">
+    <div class="flex items-center gap-2 mb-3">
+      <CalendarCheck size={18} class="text-accent" />
+      <h2 class="text-base font-semibold text-text">오늘의 공부</h2>
+      {#if curriculum}
+        <span class="ml-auto text-xs text-dim">이론 회독 <span class="text-text font-medium">{curriculum.read}</span> / {curriculum.total} ({readPct}%)</span>
+      {/if}
+    </div>
+
+    {#if dashLoading}
+      <p class="text-xs text-dim">불러오는 중…</p>
+    {:else}
+      {#if curriculum}
+        <div class="h-2 rounded-full bg-bg overflow-hidden mb-3">
+          <div class="h-full bg-accent" style="width: {readPct}%"></div>
+        </div>
+        <div class="flex flex-wrap gap-x-4 gap-y-1 mb-4 text-xs text-dim">
+          {#each curriculum.subjects as s}
+            <span>{s.subject} <span class="text-text">{s.read}/{s.total}</span></span>
+          {/each}
+        </div>
+
+        <div class="flex flex-wrap gap-2 mb-4">
+          <a
+            href="/study/topics/{curriculum.topic_id}/review-queue"
+            class="flex items-center gap-1.5 rounded border border-default px-3 py-1.5 text-xs text-dim hover:border-accent hover:text-text transition-colors"
+          >
+            <Repeat size={13} /> 문항 복습 <span class="font-semibold text-text">{curriculum.question_due}</span>
+          </a>
+          <span class="flex items-center gap-1.5 rounded border border-default px-3 py-1.5 text-xs text-dim">
+            <BookOpen size={13} /> 개념 재복습 <span class="font-semibold text-text">{curriculum.concept_due}</span>
+          </span>
+        </div>
+      {/if}
+
+      <div class="text-xs text-dim mb-2">오늘의 개념</div>
+      {#if todayConcepts.length === 0}
+        <p class="text-xs text-dim">오늘 볼 개념이 없습니다. 잘 하고 있어요.</p>
+      {:else}
+        <ul class="space-y-1.5">
+          {#each todayConcepts as c (c.doc_id)}
+            <li class="flex items-center gap-2 rounded border border-default px-3 py-2">
+              <span class="text-accent shrink-0 text-xs" title="빈출">{#each Array(c.freq) as _}★{/each}</span>
+              <a href="/study/read/{c.doc_id}" class="text-sm text-text hover:text-accent truncate flex-1">{c.title}</a>
+              <span class="shrink-0 text-[10px] rounded-full px-2 py-0.5 {c.reason === '재복습' ? 'bg-accent/15 text-accent' : 'bg-surface border border-default text-dim'}">{c.reason}</span>
+              <button
+                type="button"
+                onclick={() => markRead(c)}
+                class="shrink-0 text-xs rounded border border-default px-2 py-1 text-dim hover:border-accent hover:text-accent transition-colors"
+              >읽음</button>
+            </li>
+          {/each}
+        </ul>
+      {/if}
+
+      {#if weakConcepts.length > 0}
+        <div class="mt-4 pt-3 border-t border-default">
+          <div class="text-xs text-dim mb-2 flex items-center gap-1.5">
+            <Target size={13} class="text-error" /> 약점 개념 <span class="text-faint">(관련 기출 정답률 낮음)</span>
+          </div>
+          <div class="flex flex-wrap gap-2">
+            {#each weakConcepts as w (w.doc_id)}
+              <a href="/study/read/{w.doc_id}"
+                class="text-xs rounded-full border border-error/40 bg-error/10 text-error px-3 py-1 hover:bg-error/20 transition-colors">
+                {w.title.replace(/^\d+_/, '')} <span class="font-semibold">{w.accuracy}%</span>
+              </a>
+            {/each}
+          </div>
+        </div>
+      {/if}
+    {/if}
+  </section>
+
   <a
     href="/study/topics"
     class="block mb-3 p-5 rounded-lg border border-default bg-surface hover:border-accent hover:bg-accent/5 transition-colors"
@@ -126,7 +245,8 @@
   <div class="mt-6 p-4 rounded-lg border border-dashed border-default/60 text-xs text-dim">
     <div class="font-medium text-dim mb-1">예정</div>
     <ul class="list-disc list-inside space-y-0.5">
-      <li>애플워치 빠른복습 + 공부 알람(push)</li>
+      <li>개념 학습 리더 (가리고 떠올리기 · 빈출★ · 관련개념 백링크)</li>
+      <li>이론↔문제 연결 (개념별 정답률 · 약점 개념 지도)</li>
     </ul>
   </div>
 </div>

frontend/src/routes/study/read/[docId]/+page.svelte

@@ -0,0 +1,254 @@
+<script>
+  /**
+   * /study/read/[docId] — 개념 학습 리더.
+   * 개념노트(가스기사 documents)를 구조(요약/본문/빈출★/관련개념)로 렌더 +
+   * '떠올리기' 능동 회상 토글 + 회독 SR(POST read) + 관련개념 백링크 + 이전/다음.
+   * 본문 렌더 = MarkdownDoc(KaTeX + docimg 내장). 서버 파싱 = /api/study/concepts/{id}.
+   */
+  import { page } from '$app/stores';
+  import { api } from '$lib/api';
+  import { addToast } from '$lib/stores/toast';
+  import { renderMathMarkdownInline } from '$lib/utils/mathMarkdown';
+  import MarkdownDoc from '$lib/components/MarkdownDoc.svelte';
+  import Button from '$lib/components/ui/Button.svelte';
+  import EmptyState from '$lib/components/ui/EmptyState.svelte';
+  import Skeleton from '$lib/components/ui/Skeleton.svelte';
+  import { BookOpen, ArrowLeft, Eye, EyeOff, Check, ChevronLeft, ChevronRight, FileQuestion } from 'lucide-svelte';
+
+  let docId = $derived($page.params.docId);
+
+  let concept = $state(null);
+  let relatedQ = $state(null);          // 관련 기출(이론↔문제, 비차단)
+  let loading = $state(true);
+  let notFound = $state(false);
+  let mode = $state('read');            // 'read' | 'recall'(떠올리기)
+  let revealed = $state({});            // {sectionIndex: true}
+  let marking = $state(false);
+
+  const STAGE_LABEL = { 0: '복습 시작', 1: '복습 1단계', 2: '복습 2단계', 3: '복습 3단계', 4: '학습 완료' };
+  const OUTCOME_MARK = { correct: '○', wrong: '✕', unsure: '?' };
+  const OUTCOME_CLASS = { correct: 'text-success', wrong: 'text-error', unsure: 'text-warning' };
+  const outcomeMark = (o) => OUTCOME_MARK[o] ?? '–';
+  const outcomeClass = (o) => OUTCOME_CLASS[o] ?? 'text-faint';
+
+  async function load() {
+    const reqId = docId; // in-flight 가드: 백링크 연타 시 stale 응답 무시
+    loading = true;
+    notFound = false;
+    concept = null;
+    relatedQ = null;
+    revealed = {};
+    mode = 'read';
+    try {
+      const data = await api(`/study/concepts/${reqId}`);
+      if (reqId !== docId) return; // 그새 다른 개념으로 이동 → 폐기
+      concept = data;
+    } catch (e) {
+      if (reqId !== docId) return;
+      if (e?.status === 404) notFound = true;
+      else addToast('error', '개념을 불러오지 못했습니다');
+      return; // 본문 실패 → 관련기출 스킵
+    } finally {
+      if (reqId === docId) loading = false;
+    }
+    // 관련 기출(비차단 — 실패해도 본문 표시엔 영향 없음)
+    try {
+      const rq = await api(`/study/concepts/${reqId}/questions?limit=6`);
+      if (reqId === docId) relatedQ = rq;
+    } catch {}
+  }
+
+  // $effect 가 마운트 1회 + docId 변경(백링크/이전·다음) 재로드를 모두 커버 (onMount 불필요)
+  $effect(() => {
+    void docId;
+    load();
+  });
+
+  function toggleMode() {
+    mode = mode === 'read' ? 'recall' : 'read';
+    revealed = {};
+  }
+  function reveal(i) {
+    revealed = { ...revealed, [i]: true };
+  }
+  function shown(i) {
+    return mode === 'read' || revealed[i];
+  }
+
+  async function markRead() {
+    marking = true;
+    try {
+      const r = await api(`/study/concepts/${docId}/read`, { method: 'POST' });
+      if (concept) {
+        concept.is_read = true;
+        concept.review_stage = r?.review_stage ?? concept.review_stage;
+        concept.due_at = r?.due_at ?? concept.due_at;
+      }
+      addToast('success', '회독 완료 — 다음 복습에 다시 나옵니다');
+    } catch {
+      addToast('error', '회독 처리 실패');
+    } finally {
+      marking = false;
+    }
+  }
+</script>
+
+<svelte:head><title>{concept?.title ?? '개념'} — 공부</title></svelte:head>
+
+<div class="p-4 md:p-6 max-w-3xl mx-auto">
+  <!-- 상단 네비 -->
+  <div class="flex items-center gap-2 text-xs md:text-sm mb-4 min-w-0">
+    <a href="/study" class="text-dim hover:text-text flex items-center gap-1 shrink-0">
+      <ArrowLeft size={14} /> 공부
+    </a>
+    {#if concept?.subject}
+      <span class="text-faint shrink-0">/</span>
+      <span class="text-dim truncate">{concept.subject}</span>
+    {/if}
+  </div>
+
+  {#if loading}
+    <Skeleton h="h-10" rounded="card" />
+    <div class="mt-3 space-y-2">
+      {#each Array(4) as _}<Skeleton h="h-24" rounded="card" />{/each}
+    </div>
+  {:else if notFound}
+    <EmptyState icon={BookOpen} title="개념을 찾을 수 없습니다" description="삭제되었거나 잘못된 주소입니다." />
+  {:else if concept}
+    <!-- 제목 + 빈출 tier -->
+    <header class="mb-3">
+      <div class="flex items-start gap-2">
+        <h1 class="text-xl md:text-2xl font-semibold text-text flex-1">{concept.title}</h1>
+        <span class="text-accent text-sm shrink-0 mt-1" title="빈출도">
+          {#each Array(concept.freq) as _}★{/each}
+        </span>
+      </div>
+      {#if concept.is_read || (concept.review_stage !== null && concept.review_stage !== undefined)}
+        <div class="mt-1 text-xs text-dim">
+          {#if concept.review_stage !== null && concept.review_stage !== undefined}
+            {STAGE_LABEL[concept.review_stage] ?? '복습 중'}
+          {:else}회독함{/if}
+        </div>
+      {/if}
+    </header>
+
+    <!-- 한 줄 요약 (고정 표시) -->
+    {#if concept.summary}
+      <div class="mb-4 rounded-lg border-l-4 border-accent bg-accent/10 px-4 py-3 markdown-body text-sm text-text">
+        {@html renderMathMarkdownInline(concept.summary)}
+      </div>
+    {/if}
+
+    <!-- 모드 토글 -->
+    <div class="flex items-center gap-2 mb-4">
+      <Button variant={mode === 'recall' ? 'primary' : 'secondary'} size="sm" icon={mode === 'recall' ? EyeOff : Eye} onclick={toggleMode}>
+        {mode === 'recall' ? '떠올리기 모드' : '읽기 모드'}
+      </Button>
+      {#if mode === 'recall'}
+        <span class="text-xs text-dim">각 섹션을 떠올린 뒤 확인하세요</span>
+      {/if}
+    </div>
+
+    <!-- 본문 섹션 -->
+    {#if concept.body.length > 0}
+      <div class="space-y-3 mb-5">
+        {#each concept.body as sec, i (i)}
+          <section class="rounded-lg border border-default bg-surface overflow-hidden">
+            <div class="flex items-center gap-2 px-4 py-2.5 border-b border-default bg-surface-hover">
+              <h2 class="text-sm font-semibold text-text flex-1">{sec.label}</h2>
+              {#if sec.stars > 0}
+                <span class="text-accent text-xs shrink-0">{#each Array(sec.stars) as _}★{/each}</span>
+              {/if}
+            </div>
+            {#if shown(i)}
+              <div class="px-4 py-3">
+                <MarkdownDoc documentId={concept.doc_id} mdContent={sec.md} mdStatus={null}
+                  class="markdown-body max-w-none text-text" />
+              </div>
+            {:else}
+              <button type="button" onclick={() => reveal(i)}
+                class="w-full px-4 py-6 text-center text-sm text-dim hover:text-accent hover:bg-accent/5 transition-colors">
+                <Eye size={16} class="inline mr-1" /> 떠올린 뒤 확인
+              </button>
+            {/if}
+          </section>
+        {/each}
+      </div>
+    {/if}
+
+    <!-- 빈출 포인트 -->
+    {#if concept.bincheol.length > 0}
+      <section class="mb-5 rounded-lg border border-default bg-surface p-4">
+        <h2 class="text-sm font-semibold text-text mb-2 flex items-center gap-1.5">
+          <span class="text-accent">★</span> 빈출 포인트
+        </h2>
+        <ul class="space-y-1.5">
+          {#each concept.bincheol as item}
+            <li class="flex gap-2 text-sm text-text">
+              <span class="text-accent shrink-0 text-xs mt-0.5">{#each Array(item.tier || 1) as _}★{/each}</span>
+              <span class="markdown-body flex-1">{@html renderMathMarkdownInline(item.text)}</span>
+            </li>
+          {/each}
+        </ul>
+      </section>
+    {/if}
+
+    <!-- 관련 개념 (백링크) -->
+    {#if concept.related.length > 0}
+      <section class="mb-5">
+        <h2 class="text-xs text-dim mb-2">관련 개념</h2>
+        <div class="flex flex-wrap gap-2">
+          {#each concept.related as rel}
+            {#if rel.doc_id}
+              <a href="/study/read/{rel.doc_id}"
+                class="text-xs rounded-full border border-accent/40 bg-accent/10 text-accent px-3 py-1 hover:bg-accent/20 transition-colors">
+                {rel.phrase}
+              </a>
+            {:else}
+              <span class="text-xs rounded-full border border-default bg-surface text-faint px-3 py-1" title="아직 없는 개념">
+                {rel.phrase}
+              </span>
+            {/if}
+          {/each}
+        </div>
+      </section>
+    {/if}
+
+    <!-- 관련 기출 (이론↔문제 브리지) -->
+    {#if relatedQ && relatedQ.linked > 0}
+      <section class="mb-5 rounded-lg border border-default bg-surface p-4">
+        <h2 class="text-sm font-semibold text-text mb-2 flex items-center gap-1.5">
+          <FileQuestion size={15} class="text-accent" /> 관련 기출
+          <span class="ml-1 text-xs font-normal text-dim">
+            {relatedQ.linked}문항{#if relatedQ.accuracy !== null} · 정답률 <span class="{relatedQ.accuracy < 60 ? 'text-error' : 'text-text'} font-medium">{relatedQ.accuracy}%</span>{:else} · 아직 안 풂{/if}
+          </span>
+        </h2>
+        <ul class="space-y-0.5">
+          {#each relatedQ.questions as q (q.id)}
+            <li>
+              <a href="/study/topics/4/questions/{q.id}"
+                class="flex items-center gap-2 text-xs py-1 text-dim hover:text-accent transition-colors">
+                <span class="{outcomeClass(q.last_outcome)} shrink-0 w-4 text-center font-bold">{outcomeMark(q.last_outcome)}</span>
+                <span class="truncate">{q.subject ?? '기출'}{#if q.exam_round} · {q.exam_round}{/if}</span>
+              </a>
+            </li>
+          {/each}
+        </ul>
+      </section>
+    {/if}
+
+    <!-- 액션바 -->
+    <div class="flex items-center gap-2 border-t border-default pt-4 mt-2">
+      {#if concept.prev_id}
+        <Button variant="ghost" size="sm" icon={ChevronLeft} href="/study/read/{concept.prev_id}">이전</Button>
+      {/if}
+      <div class="flex-1"></div>
+      <Button variant="primary" size="sm" icon={Check} onclick={markRead} loading={marking}>
+        {concept.is_read ? '다시 회독' : '회독 완료'}
+      </Button>
+      {#if concept.next_id}
+        <Button variant="secondary" size="sm" icon={ChevronRight} href="/study/read/{concept.next_id}">다음 개념</Button>
+      {/if}
+    </div>
+  {/if}
+</div>

migrations/381_study_concept_progress.sql

@@ -0,0 +1,16 @@
+-- 381_study_concept_progress.sql — 이론 개념(문서) 간격반복(SR) 진행. 이론공부 홈 트리거.
+-- concept_doc_id 는 documents.id 를 가리키나 FK 미설정(hot 테이블 락 회피, clause_study 380 선례).
+-- SR 산술은 study_question_progress 와 동일(sr_schedule 공용): stage 0→1→2→3(1·3·7·14일)→4 졸업.
+CREATE TABLE IF NOT EXISTS study_concept_progress (
+  id             bigserial PRIMARY KEY,
+  user_id        bigint NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+  study_topic_id bigint NOT NULL REFERENCES study_topics(id) ON DELETE CASCADE,
+  concept_doc_id bigint NOT NULL,
+  review_stage   smallint,
+  due_at         timestamptz,
+  last_read_at   timestamptz,
+  created_at     timestamptz NOT NULL DEFAULT now(),
+  updated_at     timestamptz NOT NULL DEFAULT now(),
+  CONSTRAINT uq_concept_progress_user_doc UNIQUE (user_id, concept_doc_id)
+);
+CREATE INDEX IF NOT EXISTS idx_concept_progress_due ON study_concept_progress(user_id, due_at) WHERE due_at IS NOT NULL;

migrations/382_study_concept_links.sql

@@ -0,0 +1,15 @@
+-- 382_study_concept_links.sql — 개념문서 ↔ 기출문항 링크 (이론↔문제 브리지, Stage B).
+-- concept_doc_id=documents.id, question_id=study_questions.id — FK 없음(hot 테이블 락 회피, 선례).
+-- link_source: 'embedding'(bge-m3 코사인 top-k, 주력) | 'ref'(해설 .md 참조, 후속 enrichment).
+-- score=코사인 유사도(0~1). UNIQUE(doc,question,source) — source별 공존 허용(재튜닝=source 전삭제 후 재삽입).
+CREATE TABLE IF NOT EXISTS study_concept_links (
+  id             bigserial PRIMARY KEY,
+  concept_doc_id bigint NOT NULL,
+  question_id    bigint NOT NULL,
+  link_source    text   NOT NULL,
+  score          double precision,
+  created_at     timestamptz NOT NULL DEFAULT now(),
+  CONSTRAINT uq_concept_link UNIQUE (concept_doc_id, question_id, link_source)
+);
+CREATE INDEX IF NOT EXISTS idx_concept_links_doc ON study_concept_links(concept_doc_id);
+CREATE INDEX IF NOT EXISTS idx_concept_links_q ON study_concept_links(question_id);

scripts/concept_links_backfill.sql

@@ -0,0 +1,23 @@
+-- concept_links_backfill.sql — 개념↔문항 임베딩 링크 재생성 (Stage B, 멱등·재실행 안전).
+-- 정찰 확정: bge-m3 1024d 코사인, per-concept top-k=10, threshold 0.62 → ~2362링크·284/289개념·964문항.
+-- 재튜닝 시 DELETE(embedding 소스만) 후 재삽입 = ref 링크(후속) 불변. 개념 doc = 가스기사 태그.
+DELETE FROM study_concept_links WHERE link_source = 'embedding';
+INSERT INTO study_concept_links (concept_doc_id, question_id, link_source, score)
+WITH cd AS (
+  SELECT id, embedding FROM documents
+  WHERE user_tags::text LIKE '%@library/가스기사/%'
+    AND deleted_at IS NULL AND embedding IS NOT NULL
+),
+ranked AS (
+  SELECT cd.id AS concept_doc_id, q.id AS question_id,
+         1 - (q.embedding <=> cd.embedding) AS score,
+         row_number() OVER (PARTITION BY cd.id ORDER BY q.embedding <=> cd.embedding) AS rn
+  FROM cd
+  JOIN study_questions q
+    ON q.study_topic_id = 4 AND q.embedding IS NOT NULL
+   AND q.deleted_at IS NULL AND q.is_active
+)
+SELECT concept_doc_id, question_id, 'embedding', score
+FROM ranked
+WHERE rn <= 10 AND score >= 0.62
+ON CONFLICT (concept_doc_id, question_id, link_source) DO NOTHING;

tests/summarize_units/test_mapreduce_helpers.py

@@ -0,0 +1,80 @@
+"""summarize_units PR2 헬퍼 단위테스트 — map/reduce 프롬프트 조립 순수함수.
+
+핵심 불변식:
+  - render_map_slice: 유닛 위치(1-based)/섹션 라벨 + 본문 그대로 (손실 0).
+  - build_reduce_units_block: 어떤 입력에도 반환 블록 est_tokens <= budget (캡 초과 0
+    검증 게이트의 reduce 측). 절단은 detail 만 — 라벨/TLDR/불일치/순서 보존.
+
+pytest + 단독 실행 양쪽 지원:
+  PYTHONPATH=. pytest tests/summarize_units/ -q
+"""
+from __future__ import annotations
+
+from app.services.summarize_units import (
+    SummarizeUnit,
+    build_reduce_units_block,
+    estimate_tokens,
+    render_map_slice,
+)
+
+
+def _result(idx: int, detail: str, *, tldr: str = "요약", inc: list | None = None) -> dict:
+    return {
+        "index": idx,
+        "titles": [f"섹션{idx}"],
+        "tldr": tldr,
+        "detail": detail,
+        "inconsistencies": inc or [],
+    }
+
+
+# ---------- render_map_slice ----------
+
+def test_render_map_slice_label_and_body():
+    unit = SummarizeUnit(index=2, section_titles=["개요", None, "본론"], text="본문입니다")
+    out = render_map_slice(unit, total_units=5)
+    assert out.startswith("[유닛 3/5 — 섹션: 개요 · 본론]\n")
+    assert out.endswith("본문입니다")
+
+
+def test_render_map_slice_untitled():
+    unit = SummarizeUnit(index=0, section_titles=[None], text="x")
+    assert "(무제 구간)" in render_map_slice(unit, total_units=1)
+
+
+# ---------- build_reduce_units_block ----------
+
+def test_reduce_block_within_budget_untouched():
+    results = [_result(i, "가" * 100) for i in range(3)]
+    block, truncated = build_reduce_units_block(results, budget_tokens=11_000)
+    assert not truncated
+    # 순서/라벨/TLDR 보존
+    assert block.index("[유닛 1/3") < block.index("[유닛 2/3") < block.index("[유닛 3/3")
+    assert "TLDR: 요약" in block
+    assert "가" * 100 in block
+
+
+def test_reduce_block_truncates_to_budget():
+    # 유닛 8개 × 한글 detail 5,000자 ≈ 21K tok — budget 5,000 으로 절단 강제
+    results = [_result(i, "가" * 5_000) for i in range(8)]
+    block, truncated = build_reduce_units_block(results, budget_tokens=5_000)
+    assert truncated
+    assert estimate_tokens(block) <= 5_000
+    # 라벨(유닛 순서)은 절단 후에도 보존
+    assert "[유닛 1/8" in block
+
+
+def test_reduce_block_hard_cut_floor():
+    # min_detail_chars floor 에 막혀 비례 절단으로 불충분한 극단 케이스 — 하드 컷 발동
+    results = [_result(i, "가" * 300) for i in range(50)]
+    block, truncated = build_reduce_units_block(results, budget_tokens=500)
+    assert truncated
+    assert estimate_tokens(block) <= 500
+
+
+def test_reduce_block_preserves_inconsistencies():
+    results = [
+        _result(0, "가" * 50, inc=[{"kind": "version_drift", "desc": "개정판 차이"}]),
+    ]
+    block, _ = build_reduce_units_block(results, budget_tokens=10_000)
+    assert "불일치(version_drift): 개정판 차이" in block

tests/summarize_units/test_units.py

@@ -0,0 +1,180 @@
+"""summarize_units 단위테스트 (presegment PR1 — 순수함수·fixture).
+
+핵심 불변식:
+  - estimate_tokens = PR0 캘리브레이션(한글 0.529 · 기타 0.217 tok/char) 정확 재현.
+  - greedy_pack: 순서 보존·인접만·cap 준수·단독 초과 leaf=over_cap 전용 유닛·텍스트 손실 0
+    (구 deep_summary head/mid/tail 가운데 폐기 버그의 반대 성질).
+  - gate 3-way: 0=auto / (0,40]=hybrid / >40=whole (경계 포함).
+  - plan_summarize_units: trigger 이하=single(현행 단일콜 유지=무회귀) / 초과=map_reduce.
+
+pytest + 단독 실행 양쪽 지원:
+  PYTHONPATH=. .venv/bin/pytest tests/summarize_units/ -q
+"""
+from __future__ import annotations
+
+from app.services.hier_decomp.builder import HierNode
+from app.services.summarize_units import (
+    CAP_TOKENS,
+    TRIGGER_TOKENS,
+    SummarizeUnit,
+    estimate_tokens,
+    extract_leaves,
+    gate,
+    greedy_pack,
+    over_pct,
+    plan_summarize_units,
+)
+
+
+def _leaf(idx: int, text: str, title: str | None = None) -> HierNode:
+    return HierNode(idx=idx, parent_idx=None, level=1, node_type=None,
+                    section_title=title, heading_path=title, text=text)
+
+
+# ---------- estimate_tokens ----------
+
+def test_estimate_tokens_korean_calibration():
+    # 한글 1000자 → 529 tok (PR0: 0.529 tok/char)
+    assert estimate_tokens("가" * 1000) == 529
+
+
+def test_estimate_tokens_english_calibration():
+    # 비한글 1000자 → 217 tok (PR0: 0.217 tok/char)
+    assert estimate_tokens("a" * 1000) == 217
+
+
+def test_estimate_tokens_mixed_and_empty():
+    assert estimate_tokens("") == 0
+    mixed = "가" * 100 + "a" * 100
+    assert estimate_tokens(mixed) == round(100 * 0.529 + 100 * 0.217)
+
+
+# ---------- greedy_pack ----------
+
+def test_greedy_pack_adjacency_and_cap():
+    # 4000tok 짜리 한글 leaf 4개 (4000/0.529 ≈ 7562자) → cap 12000 이면 [3개, 1개]... 아니
+    # 4000*3=12000 = cap 정확 경계(<=cap 허용) → [1,2,3] + [4]
+    body = "가" * 7562  # ≈ 3999~4000 tok
+    leaves = [_leaf(i, body, f"s{i}") for i in range(4)]
+    units = greedy_pack(leaves, cap=12_000)
+    assert len(units) == 2
+    assert [len(u.section_titles) for u in units] == [3, 1]
+    # 순서 보존
+    assert units[0].section_titles == ["s0", "s1", "s2"]
+    assert units[1].section_titles == ["s3"]
+    # cap 준수
+    assert all(u.est_tokens <= 12_000 for u in units)
+
+
+def test_greedy_pack_oversized_leaf_gets_own_unit():
+    small = "가" * 1000            # ≈ 529 tok
+    big = "가" * 30_000            # ≈ 15,870 tok > CAP
+    leaves = [_leaf(0, small, "a"), _leaf(1, big, "mega"), _leaf(2, small, "b")]
+    units = greedy_pack(leaves, cap=CAP_TOKENS)
+    assert len(units) == 3
+    assert units[1].over_cap and units[1].section_titles == ["mega"]
+    assert not units[0].over_cap and not units[2].over_cap
+    # 인접성: 초과 leaf 가 앞뒤 pack 을 넘나들며 합쳐지지 않음
+    assert units[0].section_titles == ["a"] and units[2].section_titles == ["b"]
+
+
+def test_greedy_pack_no_text_loss():
+    leaves = [_leaf(i, f"본문{i} " + "가" * 500, f"s{i}") for i in range(7)]
+    units = greedy_pack(leaves, cap=1_000)
+    joined = "\n\n".join(u.text for u in units)
+    for leaf in leaves:
+        assert leaf.text in joined  # 커버리지 — 중간 폐기 0
+
+
+def test_greedy_pack_empty():
+    assert greedy_pack([]) == []
+
+
+# ---------- over_pct + gate ----------
+
+def test_over_pct_and_gate_boundaries():
+    assert gate(0.0) == "auto"
+    assert gate(0.01) == "hybrid"
+    assert gate(40.0) == "hybrid"
+    assert gate(40.01) == "whole"
+    assert gate(100.0) == "whole"
+
+
+def test_over_pct_computation():
+    # leaf: 6000tok + 18000tok(초과) → over% = 18000/24000 = 75%
+    l_small = _leaf(0, "가" * round(6000 / 0.529), "a")
+    l_big = _leaf(1, "가" * round(18000 / 0.529), "b")
+    pct = over_pct([l_small, l_big], cap=CAP_TOKENS)
+    assert 74.0 < pct < 76.0
+    assert over_pct([], cap=CAP_TOKENS) == 0.0
+    assert over_pct([l_small], cap=CAP_TOKENS) == 0.0
+
+
+# ---------- plan_summarize_units (fixture md) ----------
+
+def _md_doc(sections: int, chars_per_section: int, ch: str = "가") -> str:
+    parts = []
+    for i in range(sections):
+        parts.append(f"# 제{i+1}장 섹션{i}\n\n" + ch * chars_per_section)
+    return "\n\n".join(parts)
+
+
+def test_plan_small_doc_stays_single():
+    md = _md_doc(3, 1000)  # ≈ 3×529 tok ≪ trigger
+    plan = plan_summarize_units(md)
+    assert plan.mode == "single" and plan.tier is None and plan.units == []
+    assert plan.total_est_tokens <= TRIGGER_TOKENS
+
+
+def test_plan_large_doc_auto_tier():
+    # 섹션 20개 × ≈4000tok = ≈80K tok > trigger, 전 섹션 < cap → auto
+    md = _md_doc(20, 7562)
+    plan = plan_summarize_units(md)
+    assert plan.mode == "map_reduce"
+    assert plan.tier == "auto" and plan.over_pct == 0.0
+    assert len(plan.units) >= 2
+    assert all(u.est_tokens <= CAP_TOKENS for u in plan.units)
+
+
+def test_plan_mega_section_whole_tier():
+    # 작은 섹션 2 + 초대형 1(≈53K tok — 전체의 >40%) → whole
+    md = (_md_doc(2, 7562)
+          + "\n\n# 메가섹션\n\n" + "가" * 100_000)
+    plan = plan_summarize_units(md)
+    assert plan.mode == "map_reduce"
+    assert plan.tier == "whole" and plan.over_pct > 40.0
+    assert any(u.over_cap for u in plan.units)
+
+
+def test_plan_hybrid_tier():
+    # 정상 섹션 15개(≈60K tok) + 초과 섹션 1개(≈15.9K tok) → over% ≈ 21% → hybrid
+    md = _md_doc(15, 7562) + "\n\n# 초과섹션\n\n" + "가" * 30_000
+    plan = plan_summarize_units(md)
+    assert plan.mode == "map_reduce"
+    assert plan.tier == "hybrid"
+    assert 0.0 < plan.over_pct <= 40.0
+    over_units = [u for u in plan.units if u.over_cap]
+    assert len(over_units) == 1  # hybrid 시 클로드 대상 = 이 유닛들만
+
+
+def test_plan_headingless_giant_is_whole():
+    # 헤딩 없는 거대 EN 문서 — leaf 1개 전체 초과 → over% 100 → whole (PR0: EN 책 다수)
+    md = "x" * 200_000  # ≈ 43K tok > trigger, 단일 leaf > cap
+    plan = plan_summarize_units(md)
+    assert plan.mode == "map_reduce" and plan.tier == "whole"
+
+
+def test_plan_deterministic():
+    md = _md_doc(10, 7562)
+    p1, p2 = plan_summarize_units(md), plan_summarize_units(md)
+    assert p1 == p2
+
+
+if __name__ == "__main__":
+    import sys
+    fns = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
+    for fn in fns:
+        fn()
+        print(f"ok {fn.__name__}")
+    print(f"{len(fns)} passed (standalone)")
+    sys.exit(0)

tests/test_deep_summary_mapreduce.py

@@ -0,0 +1,266 @@
+"""presegment PR2 — deep_summary_worker map-reduce/HOLD 배선 단위테스트.
+
+worker-process 레벨(DB 필요)의 큐 상태 전이는 라이브 E2E 로 검증하고, 여기서는
+새 메커니즘의 seam 을 단위 검증한다 (test_fair_share.py 선례):
+  - _hold_awaiting_split: payload 마킹 commit 후 StageDeferred(HOLD_RETRY_MINUTES).
+  - _process_map_reduce: 유닛별 map → reduce → doc 필드 기록 / 모든 콜 캡 준수 /
+    payload.presegment.map_results 유닛 단위 persist(멱등 재개) / 실패 유닛 raise /
+    drain 보류(StageDeferred) 시 완료 유닛 보존.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "app"))
+
+from ai.envelope import EscalationEnvelope  # noqa: E402
+from models.queue import StageDeferred  # noqa: E402
+from services.summarize_units import (  # noqa: E402
+    CAP_TOKENS,
+    estimate_tokens,
+    plan_summarize_units,
+)
+import workers.deep_summary_worker as dsw  # noqa: E402
+
+
+# ─── fixtures ────────────────────────────────────────────────────────────────
+
+# 30 절 × 한글 2,000자 ≈ 31.7K tok (> TRIGGER 25K) · 절당 ≈ 1,060 tok (< CAP) → auto
+GIANT_AUTO_MD = "\n".join(f"# 절 {i}\n" + ("가" * 2_000) for i in range(30))
+# 헤딩 1개 + 한글 60,000자 단일 섹션 ≈ 31.7K tok (> CAP) → over% 100 → whole
+GIANT_WHOLE_MD = "# 통짜\n" + ("가" * 60_000)
+
+MAP_JSON = (
+    '{"mode": "single", "tldr": "유닛 요약", "detail": "유닛 상세.",'
+    ' "inconsistencies": [{"kind": "version_drift", "desc": "개정판 차이"}],'
+    ' "confidence": 0.9}'
+)
+REDUCE_JSON = (
+    '{"mode": "single", "tldr": "전체 요약", "detail": "최종 상세.",'
+    ' "inconsistencies": [], "confidence": 0.8}'
+)
+
+
+class FakeSession:
+    """commit 시점의 queue_row.payload 를 **객체 참조**로 박제 — SQLAlchemy 의 committed
+    스냅샷과 동일하게, 이후 in-place 변경이 과거 커밋 객체에 소급 반영되는 aliasing
+    (60254 라이브에서 unit 0 만 persist 된 버그)을 검증 시점 직렬화로 탐지한다."""
+
+    def __init__(self, row=None):
+        self.commits = 0
+        self._row = row
+        self.snapshots: list = []
+
+    async def commit(self):
+        self.commits += 1
+        if self._row is not None:
+            self.snapshots.append(self._row.payload)  # 참조 박제 — 복사 금지(의도)
+
+
+class FakeClient:
+    """deep 슬롯 보유 클라이언트 — call_deep_or_defer 가 call_deep 을 타게 한다."""
+
+    def __init__(self, responses=None, fail_indexes=frozenset(), defer_from=None):
+        self.ai = SimpleNamespace(
+            deep=SimpleNamespace(model="qwen-macbook", context_char_limit=260_000)
+        )
+        self.prompts: list[str] = []
+        self._fail_indexes = fail_indexes  # 이 순번(0-based) 콜은 파싱 불가 응답
+        self._defer_from = defer_from  # 이 순번부터 연결 실패(StageDeferred 변환 대상)
+
+    async def call_deep(self, prompt: str, system=None) -> str:
+        import httpx
+
+        idx = len(self.prompts)
+        if self._defer_from is not None and idx >= self._defer_from:
+            raise httpx.ConnectError("macbook down")
+        self.prompts.append(prompt)
+        if idx in self._fail_indexes:
+            return "정상 JSON 아님"
+        if "유닛 요약 (총" in prompt:  # reduce 프롬프트 마커
+            return REDUCE_JSON
+        return MAP_JSON
+
+    async def close(self):
+        pass
+
+
+def _doc():
+    return SimpleNamespace(
+        id=999,
+        extracted_text=GIANT_AUTO_MD,
+        ai_detail_summary=None,
+        ai_inconsistencies=None,
+        ai_analysis_tier="triage",
+        ai_processed_at=None,
+    )
+
+
+def _envelope():
+    return EscalationEnvelope(
+        from_stage="classify",
+        escalation_reasons=("long_context",),
+        risk_flags=(),
+        distilled_context="4B 요지",
+        original_pointers={"doc_ids": [999]},
+    )
+
+
+@pytest.fixture
+def _patch_telemetry(monkeypatch):
+    events: list[dict] = []
+
+    async def fake_record(**kwargs):
+        events.append(kwargs)
+
+    monkeypatch.setattr(dsw, "record_analyze_event", fake_record)
+    return events
+
+
+# ─── _hold_awaiting_split ────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_hold_marks_payload_and_defers():
+    plan = plan_summarize_units(GIANT_WHOLE_MD)
+    assert plan.mode == "map_reduce" and plan.tier == "whole"
+
+    session, row = FakeSession(), SimpleNamespace(payload={"envelope": {"x": 1}})
+    with pytest.raises(StageDeferred) as ei:
+        await dsw._hold_awaiting_split(session, row, plan, document_id=999)
+
+    assert ei.value.retry_after_minutes == dsw.HOLD_RETRY_MINUTES
+    assert session.commits == 1  # 마킹이 defer 전에 commit — consumer 재읽기에서 보존
+    preseg = row.payload["presegment"]
+    assert preseg["awaiting_split"] is True
+    assert preseg["tier"] == "whole"
+    assert preseg["units"] == len(plan.units)
+    assert row.payload["envelope"] == {"x": 1}  # 기존 payload 병합 보존
+
+
+# ─── _process_map_reduce — 정상 경로 ────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_map_reduce_end_to_end(monkeypatch, _patch_telemetry):
+    plan = plan_summarize_units(GIANT_AUTO_MD)
+    assert plan.mode == "map_reduce" and plan.tier == "auto"
+    n = len(plan.units)
+    assert n >= 2  # greedy-pack 이 실제로 유닛을 나눴는지
+
+    client = FakeClient()
+    monkeypatch.setattr(dsw, "AIClient", lambda: client)
+    doc = _doc()
+    row = SimpleNamespace(payload={"envelope": {"x": 1}})
+    session = FakeSession(row)
+
+    await dsw._process_map_reduce(
+        doc, row, _envelope(), "generic", plan, session,
+        defer_on_deep_unavailable=False,
+    )
+
+    # 콜 수 = 유닛 map n + reduce 1
+    assert len(client.prompts) == n + 1
+    # 검증 게이트: 모든 콜 est_tokens <= CAP + 오버헤드(정책 템플릿+envelope ~3K)
+    for p in client.prompts:
+        assert estimate_tokens(p) <= CAP_TOKENS + 3_000
+    # doc 기록 = reduce 출력, 불일치 = map 유닛 합본 dedup
+    assert doc.ai_detail_summary == "최종 상세."
+    assert doc.ai_analysis_tier == "deep"
+    assert doc.ai_inconsistencies == [{"kind": "version_drift", "desc": "개정판 차이"}]
+    # 유닛 단위 persist — 유닛마다 commit
+    assert row.payload["presegment"]["units"] == n
+    assert len(row.payload["presegment"]["map_results"]) == n
+    assert session.commits == n
+    # ★aliasing 회귀 방지: 각 commit 이 박제한 payload 객체를 사후에 봤을 때
+    # map_results 가 1,2,...,n 로 단조 증가해야 한다. in-place 변경(구 버그)이면
+    # 모든 스냅샷이 같은 dict 를 공유해 [n,n,...,n] 으로 보인다 = SQLAlchemy 가
+    # committed 스냅샷과 new 가 같다고 판정해 UPDATE 를 스킵하는 것과 등가.
+    per_commit_units = [
+        len(s["presegment"]["map_results"]) for s in session.snapshots
+    ]
+    assert per_commit_units == list(range(1, n + 1))
+    # telemetry 1건 (reduce 기준)
+    events = _patch_telemetry
+    assert len(events) == 1 and events[0]["error_code"] is None
+
+
+# ─── 멱등 재개 ───────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_map_reduce_resume_skips_done_units(monkeypatch, _patch_telemetry):
+    plan = plan_summarize_units(GIANT_AUTO_MD)
+    n = len(plan.units)
+
+    client = FakeClient()
+    monkeypatch.setattr(dsw, "AIClient", lambda: client)
+    done_unit = {
+        "index": 0, "titles": ["절 0"], "tldr": "이전 요약", "detail": "이전 상세.",
+        "inconsistencies": [],
+    }
+    row = SimpleNamespace(payload={
+        "envelope": {"x": 1},
+        "presegment": {"map_results": {"0": done_unit}},
+    })
+    doc, session = _doc(), FakeSession()
+
+    await dsw._process_map_reduce(
+        doc, row, _envelope(), "generic", plan, session,
+        defer_on_deep_unavailable=False,
+    )
+
+    # 유닛 0 은 재호출 안 함 — map (n-1) + reduce 1
+    assert len(client.prompts) == n
+    assert row.payload["presegment"]["map_results"]["0"]["detail"] == "이전 상세."
+    assert doc.ai_detail_summary == "최종 상세."
+
+
+# ─── map 유닛 실패 → raise (성공분 persist) ─────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_map_unit_parse_failure_raises_but_persists_good_units(
+    monkeypatch, _patch_telemetry
+):
+    plan = plan_summarize_units(GIANT_AUTO_MD)
+    n = len(plan.units)
+
+    client = FakeClient(fail_indexes={1})  # 두 번째 map 콜만 파싱 불가
+    monkeypatch.setattr(dsw, "AIClient", lambda: client)
+    doc, session = _doc(), FakeSession()
+    row = SimpleNamespace(payload={"envelope": {"x": 1}})
+
+    with pytest.raises(ValueError, match="map 유닛"):
+        await dsw._process_map_reduce(
+            doc, row, _envelope(), "generic", plan, session,
+            defer_on_deep_unavailable=False,
+        )
+
+    # 성공 유닛(n-1)은 persist — 재시도 시 실패 1건만 재호출
+    assert len(row.payload["presegment"]["map_results"]) == n - 1
+    assert "1" not in row.payload["presegment"]["map_results"]
+    assert doc.ai_detail_summary is None  # doc 은 미기록
+    assert _patch_telemetry == []  # 가짜 완료 이벤트 없음
+
+
+# ─── drain 보류 — 완료 유닛 보존 + StageDeferred 전파 ───────────────────────
+
+@pytest.mark.asyncio
+async def test_map_defer_propagates_and_keeps_progress(monkeypatch, _patch_telemetry):
+    plan = plan_summarize_units(GIANT_AUTO_MD)
+
+    client = FakeClient(defer_from=1)  # 첫 유닛 성공 후 맥북 절단
+    monkeypatch.setattr(dsw, "AIClient", lambda: client)
+    doc, session = _doc(), FakeSession()
+    row = SimpleNamespace(payload={"envelope": {"x": 1}})
+
+    with pytest.raises(StageDeferred):
+        await dsw._process_map_reduce(
+            doc, row, _envelope(), "generic", plan, session,
+            defer_on_deep_unavailable=True,  # drain 시멘틱 — 보류 전파
+        )
+
+    assert len(row.payload["presegment"]["map_results"]) == 1
+    assert doc.ai_detail_summary is None