feat(study): Phase 1 학습 루프 데이터 계층 — progress 캐시 + finalize + review API

vision (풀이 → 확인 → 학습 → 복습 → 다음 풀이 가중치) 의 데이터 계층. 데이터 모델 (migrations 222~225): - study_question_progress 테이블 — user × topic × question 단위 현재 상태 캐시 - 마지막 시도: last_outcome, last_attempted_at, last_attempt_id - 검토 상태: last_reviewed_at - 복습 큐: due_at, review_stage - 패턴 분류 (derived): pattern_state, pattern_updated_at, pattern_window_attempts - 3 partial idx (due / topic_pattern / pending_review) — 탭별 빠른 조회 패턴 분류 (services/study/learning_pattern.py): - 7 분류: unattempted/unsure/chronic_wrong/regressed/recovered/stable/unstable - 윈도우 = 최근 3회 + 과거 correct/wrong 존재 여부 - chronic_wrong > regressed > recovered 우선순위 (보수적 학습) - 가드: wrong 1회만으로 regressed 안 됨 (이전 correct 이력 필요) - stable 은 3 연속 correct 부터 세션 종료 집계 (services/study/session_finalize.py): - attempts append-only 원본 보존, progress upsert 만 - 마지막 attempt 직후 finalize hook 자동 발동 - finalize 는 last_* + pattern_state 만 갱신, due_at 미진입 문제는 NULL 유지 - 이미 due_at 박힌 문제는 finalize 가 stage 갱신 (correct → +1 / wrong → 리셋) API (api/study_question_progress.py): - POST /study-topics/{tid}/questions/{qid}/review-complete → last_reviewed_at + (wrong/unsure 인 경우만) due_at 최초 부여 - GET /study-topics/{tid}/review-queue?tab=due_today|pending_review|chronic|regressed|mastered → 5 탭 paginated 조회 → pending_review 는 last_reviewed_at < last_attempted_at 까지 포함 (이전 확인완료 후 다시 wrong 잡힘) Phase 1-E (풀이 선별 알고리즘) 은 후속 commit. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-01 09:28:46 +09:00
parent e66addf975
commit e5982ebde4
11 changed files with 724 additions and 0 deletions
@@ -0,0 +1,212 @@
+"""학습 진행 상태 (progress) API — review-complete + review-queue (Phase 1).
+
+review-complete: 사용자가 오답/모르겠음 문제를 검토했음을 표시. due_at 최초 부여.
+review-queue: 5 탭 (due_today / pending_review / chronic / regressed / mastered) 으로 progress 조회.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.auth import get_current_user
+from core.database import get_session
+from models.study_question import StudyQuestion
+from models.study_question_progress import StudyQuestionProgress
+from models.study_topic import StudyTopic
+from models.user import User
+
+router = APIRouter(prefix="/study-topics", tags=["study-progress"])
+
+# 1차 due_at 부여 시 디폴트 1일 뒤
+DEFAULT_FIRST_DUE_DAYS = 1
+
+
+def _verify_topic_owner(topic: StudyTopic | None, user: User) -> None:
+    if topic is None or topic.deleted_at is not None or topic.user_id != user.id:
+        raise HTTPException(status_code=404, detail="주제를 찾을 수 없습니다")
+
+
+@router.post("/{topic_id}/questions/{question_id}/review-complete", status_code=204)
+async def review_complete(
+    topic_id: int,
+    question_id: int,
+    user: Annotated[User, Depends(get_current_user)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+):
+    """확인완료 처리 — last_reviewed_at + (wrong/unsure 인 경우) due_at 최초 부여.
+
+    이미 due_at 박힌 문제면 due_at 그대로 유지 (큐 위치 보존).
+    정답 맞춘 문제면 due_at 박지 않음 (큐 폭발 방지).
+    """
+    topic = await session.get(StudyTopic, topic_id)
+    _verify_topic_owner(topic, user)
+
+    q = await session.get(StudyQuestion, question_id)
+    if q is None or q.deleted_at is not None or q.user_id != user.id or q.study_topic_id != topic_id:
+        raise HTTPException(status_code=404, detail="문제를 찾을 수 없습니다")
+
+    progress = (
+        await session.execute(
+            select(StudyQuestionProgress).where(
+                StudyQuestionProgress.user_id == user.id,
+                StudyQuestionProgress.study_topic_id == topic_id,
+                StudyQuestionProgress.study_question_id == question_id,
+            )
+        )
+    ).scalar_one_or_none()
+    if progress is None:
+        # attempt 없는데 review-complete 시도. 진척 상태가 없어 의미 없음.
+        raise HTTPException(status_code=409, detail="아직 시도한 적이 없는 문제입니다")
+
+    now = datetime.now(timezone.utc)
+    progress.last_reviewed_at = now
+
+    # due_at 최초 부여는 wrong/unsure 일 때만. 이미 박혀있으면 유지.
+    if progress.last_outcome in ("wrong", "unsure") and progress.due_at is None:
+        progress.review_stage = 0
+        progress.due_at = now + timedelta(days=DEFAULT_FIRST_DUE_DAYS)
+
+    await session.commit()
+
+
+# ─── review-queue ───
+
+
+class ReviewQueueItem(BaseModel):
+    question_id: int
+    question_text: str
+    subject: str | None
+    scope: str | None
+    exam_round: str | None
+    exam_question_number: int | None
+    last_outcome: str | None
+    last_attempted_at: datetime | None
+    last_reviewed_at: datetime | None
+    due_at: datetime | None
+    review_stage: int | None
+    pattern_state: str | None
+
+
+class ReviewQueueResponse(BaseModel):
+    tab: str
+    total: int
+    items: list[ReviewQueueItem]
+    page: int
+    page_size: int
+
+
+def _truncate(text: str, n: int = 80) -> str:
+    if not text:
+        return ""
+    s = text.strip()
+    return s if len(s) <= n else s[:n].rstrip() + "…"
+
+
+@router.get("/{topic_id}/review-queue", response_model=ReviewQueueResponse)
+async def review_queue(
+    topic_id: int,
+    user: Annotated[User, Depends(get_current_user)],
+    session: Annotated[AsyncSession, Depends(get_session)],
+    tab: str = Query(..., pattern="^(due_today|pending_review|chronic|regressed|mastered)$"),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(50, ge=1, le=200),
+):
+    """5 탭 진행 상태 조회.
+
+    - due_today:       progress.due_at <= now() AND review_stage < 4
+    - pending_review:  last_outcome IN (wrong, unsure)
+                       AND (last_reviewed_at IS NULL OR last_reviewed_at < last_attempted_at)
+    - chronic:         pattern_state = 'chronic_wrong'
+    - regressed:       pattern_state = 'regressed'
+    - mastered:        review_stage >= 4
+    """
+    topic = await session.get(StudyTopic, topic_id)
+    _verify_topic_owner(topic, user)
+
+    base = (
+        select(StudyQuestionProgress, StudyQuestion)
+        .join(StudyQuestion, StudyQuestion.id == StudyQuestionProgress.study_question_id)
+        .where(
+            StudyQuestionProgress.user_id == user.id,
+            StudyQuestionProgress.study_topic_id == topic_id,
+            StudyQuestion.deleted_at.is_(None),
+        )
+    )
+
+    now = datetime.now(timezone.utc)
+    if tab == "due_today":
+        base = base.where(
+            StudyQuestionProgress.due_at.is_not(None),
+            StudyQuestionProgress.due_at <= now,
+            or_(
+                StudyQuestionProgress.review_stage.is_(None),
+                StudyQuestionProgress.review_stage < 4,
+            ),
+        ).order_by(StudyQuestionProgress.due_at.asc())
+    elif tab == "pending_review":
+        base = base.where(
+            StudyQuestionProgress.last_outcome.in_(("wrong", "unsure")),
+            or_(
+                StudyQuestionProgress.last_reviewed_at.is_(None),
+                and_(
+                    StudyQuestionProgress.last_reviewed_at.is_not(None),
+                    StudyQuestionProgress.last_attempted_at.is_not(None),
+                    StudyQuestionProgress.last_reviewed_at
+                    < StudyQuestionProgress.last_attempted_at,
+                ),
+            ),
+        ).order_by(StudyQuestionProgress.last_attempted_at.desc().nulls_last())
+    elif tab == "chronic":
+        base = base.where(
+            StudyQuestionProgress.pattern_state == "chronic_wrong",
+        ).order_by(StudyQuestionProgress.last_attempted_at.desc().nulls_last())
+    elif tab == "regressed":
+        base = base.where(
+            StudyQuestionProgress.pattern_state == "regressed",
+        ).order_by(StudyQuestionProgress.last_attempted_at.desc().nulls_last())
+    elif tab == "mastered":
+        base = base.where(
+            StudyQuestionProgress.review_stage.is_not(None),
+            StudyQuestionProgress.review_stage >= 4,
+        ).order_by(StudyQuestionProgress.last_attempted_at.desc().nulls_last())
+
+    # total
+    total_row = await session.execute(
+        select(func.count()).select_from(base.subquery())
+    )
+    total = int(total_row.scalar() or 0)
+
+    # paged
+    rows = (
+        await session.execute(
+            base.offset((page - 1) * page_size).limit(page_size)
+        )
+    ).all()
+
+    items = [
+        ReviewQueueItem(
+            question_id=q.id,
+            question_text=_truncate(q.question_text, 80),
+            subject=q.subject,
+            scope=q.scope,
+            exam_round=q.exam_round,
+            exam_question_number=q.exam_question_number,
+            last_outcome=p.last_outcome,
+            last_attempted_at=p.last_attempted_at,
+            last_reviewed_at=p.last_reviewed_at,
+            due_at=p.due_at,
+            review_stage=p.review_stage,
+            pattern_state=p.pattern_state,
+        )
+        for (p, q) in rows
+    ]
+
+    return ReviewQueueResponse(
+        tab=tab, total=total, items=items, page=page, page_size=page_size,
+    )
@@ -940,6 +940,7 @@ async def submit_attempt(
    )
    session.add(attempt)

+    session_just_finished = False
    if quiz_session is not None:
        quiz_session.cursor = quiz_session.cursor + 1
        if outcome == "correct":
@@ -951,11 +952,33 @@ async def submit_attempt(
        if quiz_session.cursor >= len(quiz_session.question_ids or []):
            quiz_session.status = "done"
            quiz_session.finished_at = datetime.now(timezone.utc)
+            session_just_finished = True
        quiz_session.updated_at = datetime.now(timezone.utc)

+    # 1차 commit: attempt + quiz_session.status (실패 시 attempt 조차 안 박힘 — 사용자 재시도)
    await session.commit()
    await session.refresh(attempt)

+    # Phase 1: 세션 종료 시 progress upsert + pattern 갱신 + (있는 경우) 복습 stage 갱신.
+    # 별도 트랜잭션 — finalize 실패가 attempt 보존을 위협하지 않게 분리. 멱등성 보장 (다음 review/queue
+    # 호출 시 stale 진단 가능). 첫 차 commit 의 attempts 가 이미 가시화된 상태에서 시작.
+    if session_just_finished and quiz_session is not None:
+        from services.study.session_finalize import finalize_session
+        try:
+            await finalize_session(
+                session,
+                user_id=user.id,
+                study_topic_id=q.study_topic_id,
+                quiz_session_id=quiz_session.id,
+            )
+            await session.commit()
+        except Exception:
+            import logging
+            logging.getLogger(__name__).exception(
+                "finalize_session_failed quiz_session_id=%s", quiz_session.id
+            )
+            await session.rollback()
+
    stats = await _attempt_stats(session, user.id, question_id)
    return AttemptResponse(
        id=attempt.id,
@@ -19,6 +19,7 @@ from api.memos import router as memos_router
 from api.news import router as news_router
 from api.search import router as search_router
 from api.setup import router as setup_router
+from api.study_question_progress import router as study_question_progress_router
 from api.study_questions import router as study_questions_router
 from api.study_sessions import router as study_sessions_router
 from api.study_topics import router as study_topics_router
@@ -130,6 +131,8 @@ app.include_router(study_sessions_router, prefix="/api/study-sessions", tags=["s
 app.include_router(study_topics_router, prefix="/api/study-topics", tags=["study-topics"])
 # study_questions: 라우터 안에서 /study-topics/{id}/questions 와 /study-questions/{id} 두 줄기를 모두 정의하므로 prefix=/api 로 등록
 app.include_router(study_questions_router, prefix="/api", tags=["study-questions"])
+# Phase 1: 학습 진행 상태 (review-complete + review-queue). prefix=/api/study-topics 안에 정의됨.
+app.include_router(study_question_progress_router, prefix="/api", tags=["study-progress"])

 # TODO: Phase 5에서 추가
 # app.include_router(tasks.router, prefix="/api/tasks", tags=["tasks"])
@@ -0,0 +1,73 @@
+"""study_question_progress — 사용자 × 토픽 × 문제 단위 현재 상태 캐시 (Phase 1).
+
+attempts (append-only 원본 로그) 와 분리. 한 번 박힌 attempts 는 절대 update 안 함.
+progress 는 마지막 시도 / 사용자 검토 / 복습 큐 / 패턴 분류 derived 4 차원 메타.
+
+세션 종료 시 finalize 가 다음 갱신:
+    - last_outcome / last_attempted_at / last_attempt_id
+    - pattern_state / pattern_updated_at / pattern_window_attempts
+    - (이미 due_at 박힌 행만) review_stage / due_at  ← 복습 stage 갱신
+
+review-complete 가 다음 갱신:
+    - last_reviewed_at
+    - (wrong/unsure 인 경우) due_at 최초 부여
+
+study_question_id 는 단일 topic 소속 전제 (현재 가스기사 토픽 4 단일 운영). 향후 question
+재사용/N:M 가능성 대비 unique 키는 (user_id, study_topic_id, study_question_id) 3 키.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import BigInteger, DateTime, ForeignKey, Integer, SmallInteger, String, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from core.database import Base
+
+
+class StudyQuestionProgress(Base):
+    __tablename__ = "study_question_progress"
+    __table_args__ = (
+        UniqueConstraint(
+            "user_id", "study_topic_id", "study_question_id",
+            name="uq_progress_user_topic_question",
+        ),
+    )
+
+    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
+    )
+    study_question_id: Mapped[int] = mapped_column(
+        BigInteger, ForeignKey("study_questions.id", ondelete="RESTRICT"), nullable=False
+    )
+
+    # 마지막 시도 요약
+    last_outcome: Mapped[str | None] = mapped_column(String(20))
+    last_attempted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    last_attempt_id: Mapped[int | None] = mapped_column(
+        BigInteger, ForeignKey("study_question_attempts.id", ondelete="SET NULL")
+    )
+
+    # 사용자 검토 상태
+    last_reviewed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+
+    # 복습 큐
+    due_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    review_stage: Mapped[int | None] = mapped_column(SmallInteger)
+
+    # 패턴 분류 (derived)
+    pattern_state: Mapped[str | None] = mapped_column(String(30))
+    pattern_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    pattern_window_attempts: Mapped[int | None] = mapped_column(SmallInteger)
+
+    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
+    )
@@ -0,0 +1,92 @@
+"""학습 패턴 분류 (Phase 1).
+
+attempts 시간 순 리스트 → pattern_state 7 분류:
+  unattempted / unsure / chronic_wrong / regressed / recovered / stable / unstable
+
+윈도우 = 최근 3회 attempts + 과거 correct/wrong 존재 여부.
+recent3 만으로는 regressed/recovered 판정 불가능 — 과거 이력 함께 필요.
+
+우선순위 (위→아래, 첫 매칭 박힘):
+  1. unattempted (attempts == 0)
+  2. unsure (latest == 'unsure')
+  3. chronic_wrong (recent3 wrong ≥ 2)
+  4. regressed (latest wrong + 이전 correct 이력)
+  5. recovered (latest 2 연속 correct + 이전 wrong 이력)
+  6. stable (recent3 모두 correct, 3회 충족)
+  7. unstable (그 외)
+
+UI 라벨 (가칭):
+  unattempted="미학습" / unsure="모르겠음" / chronic_wrong="반복 오답 경향"
+  regressed="퇴행" / recovered="회복" / stable="안정" / unstable="불안정"
+
+chronic_wrong 은 [wrong,wrong,correct] 같이 latest correct 케이스도 잡힘.
+"지금 틀림" 의미 X, "최근 3회 오답 비중 높음" 의미 — UI 라벨 주의.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+PATTERN_UNATTEMPTED = "unattempted"
+PATTERN_UNSURE = "unsure"
+PATTERN_CHRONIC_WRONG = "chronic_wrong"
+PATTERN_REGRESSED = "regressed"
+PATTERN_RECOVERED = "recovered"
+PATTERN_STABLE = "stable"
+PATTERN_UNSTABLE = "unstable"
+
+WINDOW_SIZE = 3
+
+
+@dataclass
+class _AttemptOutcome:
+    """compute_pattern_state 입력용 가벼운 dataclass. 시간 순 (oldest → newest)."""
+    outcome: str  # correct / wrong / unsure / skipped
+
+
+def compute_pattern_state(attempts: list[_AttemptOutcome]) -> tuple[str, int]:
+    """attempts (oldest → newest) → (pattern_state, window_size).
+
+    window_size = min(WINDOW_SIZE, len(attempts)).
+    """
+    if not attempts:
+        return PATTERN_UNATTEMPTED, 0
+
+    latest = attempts[-1].outcome
+    if latest == "unsure":
+        return PATTERN_UNSURE, min(WINDOW_SIZE, len(attempts))
+
+    recent_window = attempts[-WINDOW_SIZE:]
+    wrong_in_window = sum(1 for a in recent_window if a.outcome == "wrong")
+
+    # chronic_wrong: 최근 3회 중 wrong ≥ 2 (또는 attempts 가 2회뿐이고 둘 다 wrong)
+    if wrong_in_window >= 2:
+        return PATTERN_CHRONIC_WRONG, len(recent_window)
+
+    # regressed: latest wrong + 이전 correct 이력 1건 이상
+    if latest == "wrong" and any(a.outcome == "correct" for a in attempts[:-1]):
+        return PATTERN_REGRESSED, len(recent_window)
+
+    # recovered: latest 2 연속 correct + 그 이전 wrong 이력
+    if (
+        len(attempts) >= 3
+        and attempts[-1].outcome == "correct"
+        and attempts[-2].outcome == "correct"
+        and any(a.outcome == "wrong" for a in attempts[:-2])
+    ):
+        return PATTERN_RECOVERED, len(recent_window)
+
+    # stable: 최근 3회 모두 correct (3회 충족)
+    if len(recent_window) >= WINDOW_SIZE and all(a.outcome == "correct" for a in recent_window):
+        return PATTERN_STABLE, WINDOW_SIZE
+
+    # 그 외: unstable (1~2회 correct, wrong 1회 + correct 이력 없음, 등)
+    return PATTERN_UNSTABLE, len(recent_window)
+
+
+def attempts_from_rows(rows) -> list[_AttemptOutcome]:
+    """SQL row (outcome 컬럼) → _AttemptOutcome 리스트. 호출자가 정렬은 미리.
+
+    rows 는 oldest → newest 정렬되어 들어와야 함 (answered_at asc).
+    """
+    return [_AttemptOutcome(outcome=r.outcome) for r in rows]
@@ -0,0 +1,259 @@
+"""세션 종료 시 progress upsert + pattern 갱신 + 복습 stage 갱신 (Phase 1).
+
+호출 시점: 마지막 attempt 가 박혀 status='done' 으로 전이된 직후.
+한 트랜잭션 안에서 처리 — caller 가 commit 책임.
+
+책임:
+  1. 이 세션의 attempts qid 별 그룹
+  2. 각 qid 별 progress upsert:
+     - last_outcome / last_attempted_at / last_attempt_id 갱신
+     - 최근 3회 attempts (이번 세션 + 과거) 로 pattern_state 계산
+     - pattern_window_attempts 갱신
+  3. 만약 progress.due_at IS NOT NULL (이미 복습 큐에 있던 문제):
+     - correct → review_stage += 1, due_at 다음 단계 (1→3→7→14일)
+     - stage ≥ 4 → due_at = NULL (학습완료)
+     - wrong/unsure → review_stage = 0, due_at = now()+1일
+  4. progress.due_at IS NULL 인 일반 풀이 결과는 due_at/stage 건드리지 않음
+  5. 세션 결과 요약 dict 반환
+
+attempts 원본 수정 0건. progress upsert 만.
+"""
+
+from __future__ import annotations
+
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from models.study_question import StudyQuestion, StudyQuestionAttempt
+from models.study_question_progress import StudyQuestionProgress
+from services.study.learning_pattern import (
+    PATTERN_CHRONIC_WRONG,
+    PATTERN_RECOVERED,
+    PATTERN_REGRESSED,
+    PATTERN_STABLE,
+    attempts_from_rows,
+    compute_pattern_state,
+)
+
+# review_stage 별 다음 due_at interval (days)
+REVIEW_INTERVAL_DAYS = {1: 3, 2: 7, 3: 14}
+REVIEW_STAGE_MASTERED = 4
+DEFAULT_FIRST_DUE_DAYS = 1
+
+
+@dataclass
+class SessionSummary:
+    """세션 결과 요약 — 응답 payload 직전 단계."""
+    correct: int
+    wrong: int
+    unsure: int
+    skipped: int
+    # 상태 변화 카운트
+    newly_correct: int       # 처음 푸는데 맞힘 (이전 attempts 0건)
+    relapsed: int            # 이전 정답이었으나 이번 wrong
+    recovered: int           # pattern_state = recovered 새로 박힘
+    chronic_remaining: int   # 이번 세션 후 chronic_wrong 으로 박힌 행 수
+    # 바로 할 일 후보
+    pending_review_count: int  # wrong/unsure + 미확인 (last_reviewed_at < last_attempted_at)
+    chronic_count: int         # pattern_state = chronic_wrong
+    regressed_count: int       # pattern_state = regressed
+
+
+async def finalize_session(
+    session: AsyncSession,
+    *,
+    user_id: int,
+    study_topic_id: int,
+    quiz_session_id: int,
+) -> SessionSummary:
+    """세션 종료 후처리. caller 가 commit. 멱등성 — 두 번 호출되어도 결과 같음."""
+    # 1. 이 세션 attempts 시간순
+    session_attempts = (
+        await session.execute(
+            select(StudyQuestionAttempt)
+            .where(
+                StudyQuestionAttempt.quiz_session_id == quiz_session_id,
+                StudyQuestionAttempt.user_id == user_id,
+            )
+            .order_by(StudyQuestionAttempt.answered_at.asc(), StudyQuestionAttempt.id.asc())
+        )
+    ).scalars().all()
+    if not session_attempts:
+        return SessionSummary(0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
+
+    # qid 그룹: 한 세션 안에 같은 qid 가 두 번 박힐 가능성 거의 없지만 안전하게.
+    last_per_qid: dict[int, StudyQuestionAttempt] = {}
+    for a in session_attempts:
+        last_per_qid[a.study_question_id] = a  # 시간 순 정렬이라 마지막이 최종
+
+    qids = list(last_per_qid.keys())
+
+    # 2. 기존 progress 행 일괄 fetch
+    existing = (
+        await session.execute(
+            select(StudyQuestionProgress)
+            .where(
+                StudyQuestionProgress.user_id == user_id,
+                StudyQuestionProgress.study_topic_id == study_topic_id,
+                StudyQuestionProgress.study_question_id.in_(qids),
+            )
+        )
+    ).scalars().all()
+    progress_by_qid: dict[int, StudyQuestionProgress] = {p.study_question_id: p for p in existing}
+
+    # 3. qid 별 과거 attempts 일괄 fetch (이 세션 attempts 까지 포함, 시간순)
+    history_rows = (
+        await session.execute(
+            select(StudyQuestionAttempt)
+            .where(
+                StudyQuestionAttempt.user_id == user_id,
+                StudyQuestionAttempt.study_topic_id == study_topic_id,
+                StudyQuestionAttempt.study_question_id.in_(qids),
+            )
+            .order_by(
+                StudyQuestionAttempt.study_question_id.asc(),
+                StudyQuestionAttempt.answered_at.asc(),
+                StudyQuestionAttempt.id.asc(),
+            )
+        )
+    ).scalars().all()
+    history_by_qid: dict[int, list[StudyQuestionAttempt]] = defaultdict(list)
+    for r in history_rows:
+        history_by_qid[r.study_question_id].append(r)
+
+    now = datetime.now(timezone.utc)
+
+    # 카운터 초기화
+    correct = wrong = unsure = skipped = 0
+    newly_correct = relapsed = recovered_count = 0
+    chronic_remaining = 0
+
+    for qid, last_attempt in last_per_qid.items():
+        outcome = last_attempt.outcome
+        if outcome == "correct":
+            correct += 1
+        elif outcome == "wrong":
+            wrong += 1
+        elif outcome == "unsure":
+            unsure += 1
+        elif outcome == "skipped":
+            skipped += 1
+
+        history = history_by_qid.get(qid, [])
+        # 이번 세션 이전 attempts 만 (status 변화 판단용)
+        prior = [a for a in history if a.id != last_attempt.id]
+
+        # 상태 변화 카운트
+        if not prior:
+            if outcome == "correct":
+                newly_correct += 1
+        else:
+            prior_last_outcome = prior[-1].outcome
+            if prior_last_outcome == "correct" and outcome == "wrong":
+                relapsed += 1
+
+        # pattern 계산
+        pattern_state, window_size = compute_pattern_state(attempts_from_rows(history))
+        if pattern_state == PATTERN_RECOVERED and (
+            qid not in progress_by_qid
+            or progress_by_qid[qid].pattern_state != PATTERN_RECOVERED
+        ):
+            recovered_count += 1
+        if pattern_state == PATTERN_CHRONIC_WRONG:
+            chronic_remaining += 1
+
+        # progress upsert
+        progress = progress_by_qid.get(qid)
+        if progress is None:
+            progress = StudyQuestionProgress(
+                user_id=user_id,
+                study_topic_id=study_topic_id,
+                study_question_id=qid,
+            )
+            session.add(progress)
+            progress_by_qid[qid] = progress
+
+        progress.last_outcome = outcome
+        progress.last_attempted_at = last_attempt.answered_at
+        progress.last_attempt_id = last_attempt.id
+        progress.pattern_state = pattern_state
+        progress.pattern_updated_at = now
+        progress.pattern_window_attempts = window_size
+
+        # 복습 stage 갱신 — 이미 due_at 박힌 문제만
+        if progress.due_at is not None:
+            if outcome == "correct":
+                progress.review_stage = (progress.review_stage or 0) + 1
+                if progress.review_stage >= REVIEW_STAGE_MASTERED:
+                    progress.due_at = None  # 학습완료
+                else:
+                    days = REVIEW_INTERVAL_DAYS[progress.review_stage]
+                    progress.due_at = now + timedelta(days=days)
+            elif outcome in ("wrong", "unsure"):
+                progress.review_stage = 0
+                progress.due_at = now + timedelta(days=DEFAULT_FIRST_DUE_DAYS)
+            # skipped 는 due_at 그대로 (큐 유지, stage 변경 안 함)
+        # progress.due_at IS NULL 일반 풀이 → stage 건드리지 않음
+
+    # 4. 바로 할 일 카운트 (요약 응답용) — finalize 직후 progress 상태 기준 SQL 한 번
+    pending_review_count = await _count_pending_review(session, user_id, study_topic_id)
+    chronic_count = await _count_pattern(session, user_id, study_topic_id, PATTERN_CHRONIC_WRONG)
+    regressed_count = await _count_pattern(session, user_id, study_topic_id, PATTERN_REGRESSED)
+
+    return SessionSummary(
+        correct=correct,
+        wrong=wrong,
+        unsure=unsure,
+        skipped=skipped,
+        newly_correct=newly_correct,
+        relapsed=relapsed,
+        recovered=recovered_count,
+        chronic_remaining=chronic_remaining,
+        pending_review_count=pending_review_count,
+        chronic_count=chronic_count,
+        regressed_count=regressed_count,
+    )
+
+
+async def _count_pending_review(session: AsyncSession, user_id: int, topic_id: int) -> int:
+    """pending_review 탭 조건 기반 카운트."""
+    from sqlalchemy import and_, func, or_
+    row = await session.execute(
+        select(func.count())
+        .select_from(StudyQuestionProgress)
+        .where(
+            StudyQuestionProgress.user_id == user_id,
+            StudyQuestionProgress.study_topic_id == topic_id,
+            StudyQuestionProgress.last_outcome.in_(("wrong", "unsure")),
+            or_(
+                StudyQuestionProgress.last_reviewed_at.is_(None),
+                and_(
+                    StudyQuestionProgress.last_reviewed_at.is_not(None),
+                    StudyQuestionProgress.last_attempted_at.is_not(None),
+                    StudyQuestionProgress.last_reviewed_at
+                    < StudyQuestionProgress.last_attempted_at,
+                ),
+            ),
+        )
+    )
+    return int(row.scalar() or 0)
+
+
+async def _count_pattern(
+    session: AsyncSession, user_id: int, topic_id: int, pattern: str
+) -> int:
+    from sqlalchemy import func
+    row = await session.execute(
+        select(func.count())
+        .select_from(StudyQuestionProgress)
+        .where(
+            StudyQuestionProgress.user_id == user_id,
+            StudyQuestionProgress.study_topic_id == topic_id,
+            StudyQuestionProgress.pattern_state == pattern,
+        )
+    )
+    return int(row.scalar() or 0)
@@ -46,6 +46,7 @@ from models.user import User  # noqa: F401
 from models.document import Document  # noqa: F401
 from models.study_topic import StudyTopic  # noqa: F401
 from models.study_session import StudySession  # noqa: F401
+from models.study_question_progress import StudyQuestionProgress  # noqa: F401

 logger = logging.getLogger("study_question_embed_worker")

@@ -0,0 +1,42 @@
+-- 222_study_question_progress.sql
+-- Phase 1: 학습 루프 데이터 계층 — 사용자 × 토픽 × 문제 단위 현재 상태 캐시.
+--
+-- 책임 분리:
+--   attempts (append-only 원본 로그) → progress (현재 상태 캐시) → pattern derived
+--
+-- 컬럼 분류:
+--   마지막 시도: last_outcome / last_attempted_at / last_attempt_id
+--   사용자 검토: last_reviewed_at
+--   복습 큐: due_at / review_stage
+--   패턴 분류 (derived): pattern_state / pattern_updated_at / pattern_window_attempts
+--
+-- finalize 가 last_* + pattern_state 만 갱신. due_at 최초 부여는 review-complete 단계.
+-- 이미 due_at 박힌 문제 다시 풀면 finalize 가 stage 갱신.
+--
+-- study_question_id 는 현재 단일 topic 소속 전제. 향후 N:M 도입되어도 user×topic×question
+-- 3 키 unique 가 안전 → 처음부터 3 키.
+
+CREATE TABLE IF NOT EXISTS study_question_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,
+    study_question_id        BIGINT       NOT NULL REFERENCES study_questions(id) ON DELETE RESTRICT,
+
+    last_outcome             VARCHAR(20),
+    last_attempted_at        TIMESTAMPTZ,
+    last_attempt_id          BIGINT REFERENCES study_question_attempts(id) ON DELETE SET NULL,
+
+    last_reviewed_at         TIMESTAMPTZ,
+
+    due_at                   TIMESTAMPTZ,
+    review_stage             SMALLINT,
+
+    pattern_state            VARCHAR(30),
+    pattern_updated_at       TIMESTAMPTZ,
+    pattern_window_attempts  SMALLINT,
+
+    created_at               TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at               TIMESTAMPTZ NOT NULL DEFAULT now(),
+
+    CONSTRAINT uq_progress_user_topic_question UNIQUE (user_id, study_topic_id, study_question_id)
+);
@@ -0,0 +1,6 @@
+-- 223_progress_due_idx.sql
+-- 복습 큐 due_today 탭 — 사용자별 due_at 오름차순 조회 빠르게.
+
+CREATE INDEX IF NOT EXISTS idx_progress_due
+    ON study_question_progress (user_id, due_at)
+    WHERE due_at IS NOT NULL;
@@ -0,0 +1,5 @@
+-- 224_progress_topic_pattern_idx.sql
+-- 패턴별 필터 — chronic/regressed/recovered 탭 조회 빠르게.
+
+CREATE INDEX IF NOT EXISTS idx_progress_topic_pattern
+    ON study_question_progress (user_id, study_topic_id, pattern_state);
@@ -0,0 +1,8 @@
+-- 225_progress_pending_review_idx.sql
+-- pending_review 탭 — 미확인 오답 조회 빠르게.
+-- partial 조건은 last_outcome 만, reviewed/attempted 비교는 쿼리에서 (이전 확인완료한
+-- 문제가 다시 wrong/unsure 됐을 때 잡히도록 last_reviewed_at < last_attempted_at 필터).
+
+CREATE INDEX IF NOT EXISTS idx_progress_pending_review
+    ON study_question_progress (user_id, study_topic_id, last_attempted_at DESC)
+    WHERE last_outcome IN ('wrong', 'unsure');