feat(session): 세션 생성 API + FIRST 콜백 + SSE 첫 질문 push

backend/CLAUDE.md

@@ -87,7 +87,7 @@ com.stackup.stackup.{domain}/
 | `github` | GitHub API 연동, 레포 목록/등록/메타 동기화 | US-07, US-08 |
 | `resume` | 이력서 업로드(S3)·메타 저장·목록·삭제 | US-05, US-06 |
 | `document` | 분석 문서(이력서/레포 공통) 메타 + S3 경로 | US-09~12 |
-| `session` | 면접 세션·메시지·피드백 (가장 큰 도메인) | US-13~20, US-24~27 |
+| `session` | 면접 세션·메시지·콜백. Sprint 2 에서 application/presentation/infrastructure 도입 (create/submitAnswer/end + callback.questions FIRST/FOLLOWUP/END + SSE push). 피드백은 Sprint 3. | US-13~20 |
 | `log.activity` | 사용자 행동 로그 | US-31 |
 | `log.ai` | AI 요청/응답 로깅 | US-30 |
 | `common` | BaseEntity, 글로벌 예외 핸들러, util | — |
@@ -359,5 +359,10 @@ docker compose up -d
 - Flyway 미도입 → 첫 entity 작성 PR에서 도입
 - **Spring AI 미사용** — LLM·임베딩 호출은 모두 AI 서버 위임. Core는 RabbitMQ 발행만 담당.
 - **Redis 미사용** — 휘발성 데이터는 DB short-lived 레코드 또는 인메모리로.
+- (2026-05) session 도메인 Phase A·B·C·D·E·F·G·H 완료 — 도메인 메서드 + 메시지 DTO +
+  SessionService.create/submitAnswer/end + SessionController + callback.questions
+  (FIRST/FOLLOWUP/END) 컨슈머 + SSE SESSION_MESSAGE/STATE push + 답변 멱등 (Idempotency-Key).
+  풀 미사용, 매 턴 LLM 호출. DB 마이그레이션 없음. golden path Testcontainer IT 추가
+  (Docker 필요).
 
 각 도입 시 본 문서 §1, 관련 도메인 `CLAUDE.md` 갱신.

backend/src/main/java/com/stackup/stackup/session/CLAUDE.md

@@ -0,0 +1,72 @@
+# `session` — 면접 세션 도메인 가이드
+
+> 상위 컨텍스트: [`/backend/CLAUDE.md`](../../../../../CLAUDE.md), [`/backend/src/main/java/com/stackup/stackup/CLAUDE.md`](../CLAUDE.md)
+
+---
+
+## 1. Aggregate
+
+`InterviewSession` (root). 연관:
+- `InterviewMessage` (질문/답변 시퀀스 — `(session_id, sequence_number) UNIQUE`)
+- `SessionContext` (세션 ↔ `AnalyzedDocument` N:M)
+- `MessageVoiceAnalysis` (Phase 2 — 음성)
+- `SessionFeedback` (Sprint 3 — 종합 평가)
+
+`InterviewMessage`, `SessionContext` 등 자식 aggregate 는 Repository 가 별도이며 service 가 명시적으로 INSERT/조회.
+
+## 2. 상태 전이 (`SessionStatus`)
+
+```
+       create()
+            ↓
+READY ──markInProgress()──> IN_PROGRESS ──end()──> COMPLETED
+  │                              │
+  └──cancel()──> CANCELLED       └──cancel()──> CANCELLED
+                                 │
+                                 └──(network 등)──> INTERRUPTED  *(현재 코드 진입 없음)*
+```
+
+전이 가드는 `InterviewSession` 도메인 메서드 내부 `IllegalStateException`. 서비스 계층에서는 `SessionInvalidStateException`(`SESSION_INVALID_STATE` → HTTP 422) 으로 사전 차단.
+
+## 3. API endpoints (`/api/sessions`)
+
+| Method | Path | 설명 |
+|--------|------|------|
+| POST | `/` | 세션 생성 (READY) + `generate.questions` 발행 |
+| POST | `/{id}/messages` | 답변 제출 (`Idempotency-Key` 헤더). `generate.followup` 발행 |
+| POST | `/{id}/end` | 세션 종료 / 취소 |
+| GET | `/{id}` | 단건 조회 |
+| GET | `/` | 페이지 목록 (생성일 DESC) |
+| GET | `/{id}/messages` | 메시지 시퀀스 ASC |
+
+## 4. 메시지
+
+발행 (`session.application.SessionService`):
+- `generate.questions` — 세션 생성 commit 후 AFTER_COMMIT 이벤트 (첫 질문 요청)
+- `generate.followup` — 답변 commit 후 AFTER_COMMIT (꼬리질문 요청)
+
+소비 (`session.infrastructure.SessionCallbackHandler` → `session.application.SessionCallbackService`):
+- `core.callback.questions` 큐 — `kind=FIRST/FOLLOWUP/END` 분기
+  - `FIRST` → `markInProgress` + 첫 INTERVIEWER 메시지 INSERT + SSE
+  - `FOLLOWUP` → INTERVIEWER 메시지 INSERT + max 도달 시 자동 종료 + SSE
+  - `END` → AI 가 조기 종료 신호. `session.end()` + SSE
+  - 멱등: `processed_messages` (`messageId` PK)
+
+## 5. 답변 멱등 정책
+
+`POST /messages` 의 `Idempotency-Key` 헤더 → `processed_messages` 의 PK `session.answer:{uuid}` 로 24h 캐시. 중복 호출 시 가장 최근 INTERVIEWEE 메시지를 그대로 반환 (멱등 충돌 시 시퀀스 중복 방지).
+
+## 6. SSE 이벤트
+
+- `SESSION_MESSAGE` — 새 INTERVIEWER 메시지 도착 시 (FIRST/FOLLOWUP)
+- `SESSION_STATE` — 상태 전이 또는 카운트 변화 시
+- `ERROR` — 콜백 페이로드 비정상 (예: 빈 question)
+
+전송 경로: Core 인메모리 (`SseEventPublisher.publishToSession`). RealTime 서버 미사용 (Phase 1).
+
+## 7. 안티패턴
+
+- 컨트롤러에서 `@Transactional` (ArchUnit 차단)
+- 엔티티에 `@Setter` (ArchUnit 차단)
+- Service 에 클래스 레벨 `@Transactional` + `@TransactionalEventListener` 혼용 (Spring 7 제약 — `AnalysisRequestService` 패턴 참조)
+- `Map.of(...)` 에 null 값 (NPE)

backend/src/main/java/com/stackup/stackup/session/application/SessionCallbackService.java

@@ -0,0 +1,176 @@
+package com.stackup.stackup.session.application;
+
+import com.stackup.stackup.common.messaging.domain.ProcessedMessage;
+import com.stackup.stackup.common.messaging.domain.ProcessedMessageRepository;
+import com.stackup.stackup.common.sse.SseEventPublisher;
+import com.stackup.stackup.common.sse.SseEventType;
+import com.stackup.stackup.session.application.dto.MessageResult;
+import com.stackup.stackup.session.application.dto.QuestionsCallbackEnvelope;
+import com.stackup.stackup.session.application.dto.QuestionsCallbackPayload;
+import com.stackup.stackup.session.domain.InterviewMessage;
+import com.stackup.stackup.session.domain.InterviewMessageRepository;
+import com.stackup.stackup.session.domain.InterviewSession;
+import com.stackup.stackup.session.domain.InterviewSessionRepository;
+import com.stackup.stackup.session.domain.SessionStatus;
+import java.util.Map;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+@Service
+@RequiredArgsConstructor
+public class SessionCallbackService {
+
+    private static final Logger log = LoggerFactory.getLogger(SessionCallbackService.class);
+    private static final String CONSUMER_NAME = "core.callback.questions";
+
+    private final InterviewSessionRepository sessionRepo;
+    private final InterviewMessageRepository messageRepo;
+    private final ProcessedMessageRepository processedRepo;
+    private final SseEventPublisher sse;
+
+    @Transactional
+    public void apply(QuestionsCallbackEnvelope envelope) {
+        if (envelope == null || envelope.payload() == null) {
+            log.warn("callback.questions envelope or payload is null — skip");
+            return;
+        }
+        if (isProcessed(envelope.messageId())) {
+            log.info("callback.questions duplicate, skip. messageId={}", envelope.messageId());
+            return;
+        }
+        QuestionsCallbackPayload p = envelope.payload();
+        log.info("callback.questions received. messageId={}, sessionId={}, kind={}, traceId={}",
+            envelope.messageId(), p.sessionId(), p.kind(), envelope.traceId());
+
+        if (p.isFirst()) {
+            applyFirstQuestion(p);
+        } else if (p.isFollowup()) {
+            applyFollowup(p); // Task G1 에서 구현
+        } else if (p.isEnd()) {
+            applyEnd(p);      // Task G1 에서 구현
+        } else {
+            log.warn("unknown kind={}, skip. messageId={}", p.kind(), envelope.messageId());
+        }
+        markProcessed(envelope.messageId());
+    }
+
+    private void applyFirstQuestion(QuestionsCallbackPayload p) {
+        InterviewSession session = sessionRepo.findByIdAndIsDeletedFalse(p.sessionId()).orElse(null);
+        if (session == null) {
+            log.warn("session not found. id={}", p.sessionId());
+            return;
+        }
+        if (session.getStatus() != SessionStatus.READY) {
+            log.warn("session not READY, drop FIRST. id={}, status={}",
+                session.getId(), session.getStatus());
+            return;
+        }
+        if (p.question() == null || p.question().isBlank()) {
+            log.warn("first question is blank. sessionId={}", p.sessionId());
+            sse.publishToSession(session.getId(), SseEventType.ERROR,
+                Map.of("code", "FIRST_QUESTION_EMPTY",
+                       "message", "첫 질문 생성에 실패했습니다."));
+            return;
+        }
+
+        session.markInProgress();
+
+        int nextSeq = messageRepo.findMaxSequenceBySessionId(session.getId()) + 1;
+        InterviewMessage msg = messageRepo.save(
+            InterviewMessage.interviewer(session, nextSeq, p.question(), null));
+        session.incrementQuestionCount();
+
+        sse.publishToSession(session.getId(), SseEventType.SESSION_MESSAGE,
+            MessageResult.from(msg));
+        sse.publishToSession(session.getId(), SseEventType.SESSION_STATE,
+            Map.of("sessionId", session.getId(), "state", session.getStatus().name(),
+                   "totalQuestionCount", session.getTotalQuestionCount()));
+    }
+
+    private void applyFollowup(QuestionsCallbackPayload p) {
+        InterviewSession session = sessionRepo.findByIdAndIsDeletedFalse(p.sessionId()).orElse(null);
+        if (session == null) {
+            log.warn("session not found. id={}", p.sessionId());
+            return;
+        }
+        if (session.getStatus() != SessionStatus.IN_PROGRESS) {
+            log.warn("session not in progress, drop followup. id={}, status={}",
+                session.getId(), session.getStatus());
+            return;
+        }
+        if (p.parentMessageId() == null) {
+            log.warn("followup missing parentMessageId. sessionId={}", session.getId());
+            return;
+        }
+        if (p.question() == null || p.question().isBlank()) {
+            log.warn("followup question blank. sessionId={}", session.getId());
+            return;
+        }
+
+        InterviewMessage parent = messageRepo.findById(p.parentMessageId()).orElse(null);
+        if (parent == null) {
+            log.warn("followup parent not found. id={}", p.parentMessageId());
+            return;
+        }
+        if (!parent.getSession().getId().equals(session.getId())) {
+            log.warn("followup parent belongs to different session. parentId={}, parent.sessionId={}, expected.sessionId={}",
+                parent.getId(), parent.getSession().getId(), session.getId());
+            return;
+        }
+
+        int nextSeq = messageRepo.findMaxSequenceBySessionId(session.getId()) + 1;
+        InterviewMessage q = messageRepo.save(
+            InterviewMessage.interviewer(session, nextSeq, p.question(), parent));
+        session.incrementQuestionCount();
+
+        sse.publishToSession(session.getId(), SseEventType.SESSION_MESSAGE,
+            MessageResult.from(q));
+
+        if (session.isMaxReached()) {
+            session.end();
+            sse.publishToSession(session.getId(), SseEventType.SESSION_STATE,
+                Map.of("sessionId", session.getId(), "state", session.getStatus().name(),
+                       "totalQuestionCount", session.getTotalQuestionCount(),
+                       "endedAt", session.getEndedAt()));
+        } else {
+            sse.publishToSession(session.getId(), SseEventType.SESSION_STATE,
+                Map.of("sessionId", session.getId(), "state", session.getStatus().name(),
+                       "totalQuestionCount", session.getTotalQuestionCount()));
+        }
+    }
+
+    private void applyEnd(QuestionsCallbackPayload p) {
+        InterviewSession session = sessionRepo.findByIdAndIsDeletedFalse(p.sessionId()).orElse(null);
+        if (session == null) {
+            log.warn("session not found. id={}", p.sessionId());
+            return;
+        }
+        if (session.getStatus() != SessionStatus.IN_PROGRESS) {
+            log.warn("session not in progress, drop END. id={}, status={}",
+                session.getId(), session.getStatus());
+            return;
+        }
+        session.end();
+        sse.publishToSession(session.getId(), SseEventType.SESSION_STATE,
+            Map.of("sessionId", session.getId(), "state", session.getStatus().name(),
+                   "totalQuestionCount", session.getTotalQuestionCount(),
+                   "endedAt", session.getEndedAt()));
+    }
+
+    private boolean isProcessed(String messageId) {
+        return messageId != null && !messageId.isBlank() && processedRepo.existsById(messageId);
+    }
+
+    private void markProcessed(String messageId) {
+        if (messageId == null || messageId.isBlank()) return;
+        try {
+            processedRepo.save(ProcessedMessage.of(messageId, CONSUMER_NAME));
+        } catch (DataIntegrityViolationException ignored) {
+            // race: 무시 (AnalysisCallbackService 와 동일 패턴)
+        }
+    }
+}

backend/src/main/java/com/stackup/stackup/session/application/SessionQueryService.java

@@ -0,0 +1,51 @@
+package com.stackup.stackup.session.application;
+
+import com.stackup.stackup.session.application.dto.MessageResult;
+import com.stackup.stackup.session.application.dto.SessionResult;
+import com.stackup.stackup.session.application.exception.SessionForbiddenException;
+import com.stackup.stackup.session.application.exception.SessionNotFoundException;
+import com.stackup.stackup.session.domain.InterviewMessage;
+import com.stackup.stackup.session.domain.InterviewMessageRepository;
+import com.stackup.stackup.session.domain.InterviewSession;
+import com.stackup.stackup.session.domain.InterviewSessionRepository;
+import lombok.RequiredArgsConstructor;
+import org.springframework.data.domain.Page;
+import org.springframework.data.domain.Pageable;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+
+@Service
+@RequiredArgsConstructor
+@Transactional(readOnly = true)
+public class SessionQueryService {
+
+    private final InterviewSessionRepository sessionRepo;
+    private final InterviewMessageRepository messageRepo;
+
+    public SessionResult get(Long sessionId, Long userId) {
+        InterviewSession s = sessionRepo.findByIdAndIsDeletedFalse(sessionId)
+            .orElseThrow(() -> new SessionNotFoundException(sessionId));
+        if (!s.getUser().getId().equals(userId)) {
+            throw new SessionForbiddenException(sessionId);
+        }
+        return SessionResult.from(s);
+    }
+
+    public Page<SessionResult> list(Long userId, Pageable pageable) {
+        return sessionRepo
+            .findByUser_IdAndIsDeletedFalseOrderByCreatedAtDesc(userId, pageable)
+            .map(SessionResult::from);
+    }
+
+    public List<MessageResult> getMessages(Long sessionId, Long userId) {
+        InterviewSession s = sessionRepo.findByIdAndIsDeletedFalse(sessionId)
+            .orElseThrow(() -> new SessionNotFoundException(sessionId));
+        if (!s.getUser().getId().equals(userId)) {
+            throw new SessionForbiddenException(sessionId);
+        }
+        return messageRepo.findBySession_IdOrderBySequenceNumberAsc(sessionId)
+            .stream().map(MessageResult::from).toList();
+    }
+}