Fix/#61 company follow question

docker-compose.loadtest.yml

@@ -2,7 +2,7 @@ services:
   redis-loadtest:
     image: redis:7-alpine
     ports:
-      - "6380:6379"
+      - "127.0.0.1:6380:6379"
 
   wiremock:
     image: wiremock/wiremock:3.13.2

docker-compose.yml

@@ -3,7 +3,7 @@ services:
     image: redis:7-alpine
     container_name: repit-redis
     ports:
-      - "6379:6379"
+      - "127.0.0.1:6379:6379"
     command: ["redis-server", "--appendonly", "yes", "--appendfsync", "always", "--maxmemory-policy", "noeviction"]
     volumes:
       - redis-data:/data
@@ -26,6 +26,10 @@ services:
       INTERVIEW_RESULT_RETRY_ENABLED: ${INTERVIEW_RESULT_RETRY_ENABLED:-false}
       ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY}
       ANTHROPIC_MODEL: ${ANTHROPIC_MODEL}
+      ANTHROPIC_COMPANY_FOLLOW_UP_ENABLED: ${ANTHROPIC_COMPANY_FOLLOW_UP_ENABLED:-false}
+      ANTHROPIC_COMPANY_FOLLOW_UP_TOTAL_TIMEOUT: ${ANTHROPIC_COMPANY_FOLLOW_UP_TOTAL_TIMEOUT:-5s}
+      ANTHROPIC_COMPANY_FOLLOW_UP_GENERATION_TIMEOUT: ${ANTHROPIC_COMPANY_FOLLOW_UP_GENERATION_TIMEOUT:-2s}
+      ANTHROPIC_COMPANY_FOLLOW_UP_MAX_TOKENS: ${ANTHROPIC_COMPANY_FOLLOW_UP_MAX_TOKENS:-800}
       TTS_ENABLED: ${TTS_ENABLED:-false}
       ELEVENLABS_API_KEY: ${ELEVENLABS_API_KEY:-}
       ELEVENLABS_MAX_IN_MEMORY_SIZE: ${ELEVENLABS_MAX_IN_MEMORY_SIZE:-8388608}

docs/chat-reliability.md

@@ -51,6 +51,10 @@ BLOCKED 작업은 토큰 또는 payload 계약의 원인을 해결한 뒤 기존
 ## Redis 영속성 및 배포
 
 compose의 Redis는 named volume과 AOF, `appendfsync always`, `noeviction`을 사용한다.
+호스트 포트는 `127.0.0.1:6379`에만 열어 같은 호스트의 로컬 실행에서 접속할 수 있다.
+`docker compose up -d redis` 후 기본 프로필로 실행하면 `localhost:6379`를 사용한다.
+Compose의 Chat 컨테이너에서 같은 Redis를 사용할 때는 `.env`에 `REDIS_HOST=redis`,
+`REDIS_PORT=6379`를 지정한다. 컨테이너의 `localhost`는 Redis 컨테이너를 가리키지 않는다.
 기존 컨테이너에 볼륨 없이 저장된 데이터를 사용 중이라면 교체 전에 백업·이관해야 한다.
 `REDIS_HOST`가 외부 Redis를 가리키면 compose의 설정이 적용되지 않으므로 외부 Redis에도
 동등한 디스크 영속성·퇴거 방지 설정과 백업이 필요하다.

load-test/k6/lib/interview.js

@@ -170,6 +170,11 @@ function prepareInterview(sessionId) {
         interviewId: __VU * 1000000 + __ITER,
         userId: __VU,
         status: 'IN_PROGRESS',
+        mode: 'SOLO',
+        personality: 'FRIENDLY',
+        tone: 'GENTLE',
+        major: 'BACKEND',
+        level: 'NORMAL',
         questions: [
             {
                 id: 1,

src/main/java/com/person/repit/interview/service/InterviewResultDeliveryService.java

@@ -4,6 +4,7 @@
 import com.person.repit.interview.dto.request.ChatInterviewAllRequest;
 import com.person.repit.interview.dto.response.ChatProgressResponse;
 import com.person.repit.interview.exception.ApiServerResponseException;
+import com.person.repit.interview.exception.InterviewSessionAlreadyExistsException;
 import com.person.repit.interview.type.InterviewStatus;
 import io.micrometer.core.instrument.MeterRegistry;
 import lombok.RequiredArgsConstructor;
@@ -22,6 +23,13 @@ public class InterviewResultDeliveryService {
     private final InterviewResultRetryPolicy policy;
     private final MeterRegistry meters;
 
+    public Mono<Void> ensureNotFinalized(String sessionId) {
+        return outbox.find(sessionId)
+                .flatMap(item -> Mono.<Void>error(new InterviewSessionAlreadyExistsException(
+                        "이미 종료 기록이 등록된 면접 세션입니다.")))
+                .then();
+    }
+
     public Mono<ChatProgressResponse> resume(String sessionId) {
         return outbox.find(sessionId).flatMap(this::requireTerminalStatus).flatMap(item ->
                 item.getState() == InterviewResultDelivery.State.BLOCKED

src/main/java/com/person/repit/interview/service/impl/ChatInterviewServiceImpl.java

@@ -67,7 +67,15 @@ public class ChatInterviewServiceImpl implements ChatInterviewService {
 
     @Override
     public Mono<ChatInterviewResponse> prepareInterview(ChatInterviewPrepareRequest request) {
+        return operationLock.execute(request.getSessionId(), () ->
+                resultDelivery.ensureNotFinalized(request.getSessionId()).then(prepareOnce(request)));
+    }
+
+    private Mono<ChatInterviewResponse> prepareOnce(ChatInterviewPrepareRequest request) {
         return Mono.defer(() -> {
+            if (request.getStatus() != InterviewStatus.IN_PROGRESS) {
+                return Mono.error(new InvalidRequestException("새 면접은 진행 중 상태로만 준비할 수 있습니다."));
+            }
             validateQuestionIds(request);
             validatePersonas(request);
             log.debug("sessionId={}, interviewId={}", request.getSessionId(), request.getInterviewId());
@@ -111,7 +119,7 @@ public Mono<ChatInterviewResponse> prepareInterview(ChatInterviewPrepareRequest
                     .ttsProfileVersion(profiles == null || profiles.isEmpty()
                             ? null
                             : profiles.values().iterator().next().getProfileVersion())
-                    .status(request.getStatus())
+                    .status(InterviewStatus.IN_PROGRESS)
                     .questions(new ArrayList<>(questions))
                     .answers(new ArrayList<>())
                     .currentQuestionIndex(0)
@@ -180,7 +188,17 @@ private Mono<ChatProgressResponse> submitAnswerOnce(String sessionId, ChatAnswer
             if (currentQuestion == null) {
                 return finishInterview(session, InterviewStatus.COMPLETED);
             }
+            ChatAnswer previous = session.getAnswers().stream()
+                    .filter(item -> item.getQuestionId().equals(request.getQuestionId()))
+                    .findFirst().orElse(null);
+            if (previous != null && (!Objects.equals(previous.getContent(), request.getContent())
+                    || !Objects.equals(previous.getResponseTime(), request.getResponseTime()))) {
+                return Mono.error(new InvalidRequestException("이미 접수한 답변과 다릅니다."));
+            }
             if (!currentQuestion.getQuestionId().equals(request.getQuestionId())) {
+                if (previous != null) {
+                    return Mono.just(ChatProgressResponse.next(currentQuestion));
+                }
                 return Mono.error(new InterviewQuestionMismatchException(
                         "현재 질문과 요청한 질문이 일치하지 않습니다."
                 ));
@@ -195,14 +213,8 @@ private Mono<ChatProgressResponse> submitAnswerOnce(String sessionId, ChatAnswer
                     .createdAt(LocalDateTime.now())
                     .build();
 
-            ChatAnswer previous = session.getAnswers().stream()
-                    .filter(item -> item.getQuestionId().equals(request.getQuestionId()))
-                    .findFirst().orElse(null);
             if (previous == null) {
                 session.getAnswers().add(answer);
-            } else if (!java.util.Objects.equals(previous.getContent(), request.getContent())
-                    || !java.util.Objects.equals(previous.getResponseTime(), request.getResponseTime())) {
-                return Mono.error(new InvalidRequestException("이미 접수한 답변과 다릅니다."));
             }
             log.info(
                     "[ANSWER ACCEPTED] sessionId={}, interviewId={}, questionId={}",

src/main/java/com/person/repit/tts/domain/TtsJob.java

@@ -11,6 +11,7 @@ public class TtsJob implements Comparable<TtsJob> {
     private final long sequence;
     private final TtsInput input;
     private final Sinks.One<AudioState> sink = Sinks.one();
+    private final Sinks.Empty<Void> cancellation = Sinks.empty();
     private volatile int priority;
     private volatile boolean cancelled;
 
@@ -40,6 +41,8 @@ public void changePriority(int priority) {
 
     public void cancel() {
         cancelled = true;
+        cancellation.tryEmitEmpty();
+        sink.tryEmitEmpty();
     }
 
     @Override

src/main/java/com/person/repit/tts/service/InterviewTtsScheduler.java

@@ -161,10 +161,12 @@ private void start(TtsJob job) {
         generating.setUpdatedAt(Instant.now());
 
         stateRepository.save(generating)
-                .then(synthesize(job))
-                .flatMap(bytes -> upload(job, bytes))
+                .then(Mono.defer(() -> synthesize(job)))
+                .flatMap(bytes -> Mono.defer(() -> job.isCancelled()
+                        ? Mono.empty() : upload(job, bytes)))
                 .flatMap(stored -> completeReady(job, generating, stored))
                 .onErrorResume(error -> completeFailed(job, generating, error))
+                .takeUntilOther(job.getCancellation().asMono())
                 .doFinally(ignored -> finish(job))
                 .subscribe();
     }

src/main/java/com/person/repit/tts/service/TtsRecoveryService.java

@@ -1,9 +1,12 @@
 package com.person.repit.tts.service;
 
+import com.person.repit.common.exception.AlreadyUsingException;
 import com.person.repit.interview.domain.ChatInterviewSession;
 import com.person.repit.interview.domain.ChatQuestion;
 import com.person.repit.interview.domain.PersonaSnapshot;
+import com.person.repit.interview.service.InterviewOperationLock;
 import com.person.repit.interview.type.InterviewRole;
+import com.person.repit.interview.type.InterviewStatus;
 import com.person.repit.tts.domain.AudioState;
 import com.person.repit.tts.domain.TtsProfile;
 import lombok.RequiredArgsConstructor;
@@ -13,6 +16,9 @@
 import org.springframework.data.redis.core.ReactiveRedisTemplate;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Mono;
+import reactor.util.retry.Retry;
+
+import java.time.Duration;
 
 @Slf4j
 @Component
@@ -24,23 +30,31 @@ public class TtsRecoveryService {
     private final AudioStateRepository stateRepository;
     private final InterviewTtsScheduler scheduler;
     private final PersonaTtsProfileResolver profileResolver;
+    private final InterviewOperationLock operationLock;
 
     @EventListener(ApplicationReadyEvent.class)
     public void recover() {
         stateRepository.findRecoverable()
-                .flatMap(this::recoverState)
-                .onErrorContinue((error, ignored) -> log.warn("TTS 작업 복구 실패", error))
+                .concatMap(state -> recoverState(state)
+                        .retryWhen(Retry.fixedDelay(Long.MAX_VALUE, Duration.ofSeconds(1))
+                                .filter(AlreadyUsingException.class::isInstance))
+                        .onErrorResume(error -> {
+                            log.warn("TTS 작업 복구 실패", error);
+                            return Mono.empty();
+                        }))
                 .subscribe();
     }
 
     private Mono<Void> recoverState(AudioState state) {
-        return redisTemplate.opsForValue().get(SESSION_KEY_PREFIX + state.getSessionId())
+        return operationLock.execute(state.getSessionId(), () ->
+                redisTemplate.opsForValue().get(SESSION_KEY_PREFIX + state.getSessionId())
                 .ofType(ChatInterviewSession.class)
+                .filter(session -> session.getStatus() == InterviewStatus.IN_PROGRESS)
                 .flatMap(session -> recoverQuestion(session, state).thenReturn(true))
                 .defaultIfEmpty(false)
                 .flatMap(recovered -> recovered
                         ? Mono.empty()
-                        : stateRepository.delete(state.getSessionId(), state.getQuestionId()));
+                        : stateRepository.delete(state.getSessionId(), state.getQuestionId())));
     }
 
     private Mono<Void> recoverQuestion(ChatInterviewSession session, AudioState state) {
@@ -54,7 +68,7 @@ private Mono<Void> recoverQuestion(ChatInterviewSession session, AudioState stat
         TtsProfile profile = profileFor(session, question.getAskedByPersonaId());
         int order = session.getQuestions().indexOf(question);
         return stateRepository.delete(state.getSessionId(), state.getQuestionId())
-                .then(scheduler.register(session.getSessionId(), question, profile, 1, order))
+                .then(Mono.defer(() -> scheduler.register(session.getSessionId(), question, profile, 1, order)))
                 .then();
     }
 

src/test/java/com/person/repit/interview/service/InterviewResultRedisTest.java

@@ -286,6 +286,61 @@ void stopRedis() throws Exception {
         }
     }
 
+    @ParameterizedTest
+    @EnumSource(value = InterviewStatus.class, names = {"COMPLETED", "ABANDONED"})
+    void 종료전송과_준비요청이_경합해도_같은_세션을_재생성하지_않는다(InterviewStatus status) {
+        var objects = new RedisConfig().reactiveRedisTemplate(factory);
+        var active = session();
+        active.setStatus(InterviewStatus.IN_PROGRESS);
+        active.setCurrentQuestionIndex(-1);
+        objects.opsForValue().set("chat:interview:s", active).block();
+        var called = reactor.core.publisher.Sinks.<Void>empty();
+        var accepted = reactor.core.publisher.Sinks.<Void>empty();
+        var client = mock(ApiServerClient.class);
+        when(client.saveInterviewPayload(anyString())).thenReturn(Mono.defer(() -> {
+            called.tryEmitEmpty();
+            return accepted.asMono();
+        }));
+        var meters = new io.micrometer.core.instrument.simple.SimpleMeterRegistry();
+        var delivery = new InterviewResultDeliveryService(outbox, client, new InterviewResultRetryPolicy(), meters);
+        var tts = mock(InterviewTtsScheduler.class);
+        var service = new com.person.repit.interview.service.impl.ChatInterviewServiceImpl(
+                mock(AiQuestionClient.class), objects, delivery, lock,
+                new com.person.repit.common.metrics.RepitMetrics(meters),
+                mock(com.person.repit.tts.service.PersonaTtsProfileResolver.class), tts);
+        var request = new com.person.repit.interview.dto.request.ChatInterviewPrepareRequest();
+        request.setSessionId("s");
+        request.setInterviewId(1L);
+        request.setUserId(2L);
+        request.setStatus(InterviewStatus.IN_PROGRESS);
+        request.setMode(com.person.repit.interview.type.InterviewMode.SOLO);
+        var question = new com.person.repit.interview.dto.request.ChatInterviewPrepareRequest.Question();
+        question.setId(17L);
+        question.setPersonaId(1L);
+        question.setQuestion("질문");
+        request.setQuestions(List.of(question));
+        var closing = (status == InterviewStatus.COMPLETED
+                ? service.completeInterview("s") : service.quitInterview("s")).toFuture();
+        try {
+            called.asMono().block(Duration.ofSeconds(5));
+            StepVerifier.create(service.prepareInterview(request)).expectError(AlreadyUsingException.class)
+                    .verify(Duration.ofSeconds(5));
+            accepted.tryEmitEmpty();
+            assertThat(closing.orTimeout(5, java.util.concurrent.TimeUnit.SECONDS).join().getStatus()).isEqualTo(status);
+            String receipt = redis.opsForValue().get("chat:interview:result:s").block();
+            StepVerifier.create(service.prepareInterview(request))
+                    .expectError(com.person.repit.interview.exception.InterviewSessionAlreadyExistsException.class)
+                    .verify(Duration.ofSeconds(5));
+            assertThat(objects.hasKey("chat:interview:s").block()).isFalse();
+            assertThat(redis.opsForValue().get("chat:interview:result:s").block()).isEqualTo(receipt);
+            verify(tts).cancelSession("s");
+            verifyNoMoreInteractions(tts);
+        } finally {
+            accepted.tryEmitEmpty();
+            closing.cancel(true);
+        }
+    }
+
     private ChatInterviewSession session() {
         return ChatInterviewSession.builder().sessionId("s").interviewId(1L).userId(2L)
                 .status(InterviewStatus.COMPLETED).createdAt(LocalDateTime.of(2026, 10, 15, 10, 0))

src/test/java/com/person/repit/interview/service/impl/ChatInterviewIdentityTest.java

@@ -38,8 +38,13 @@ class ChatInterviewIdentityTest {
     @BeforeEach
     void setUp() {
         when(redis.opsForValue()).thenReturn(values);
-        service = new ChatInterviewServiceImpl(ai, redis, mock(InterviewResultDeliveryService.class),
-                mock(InterviewOperationLock.class), new RepitMetrics(new SimpleMeterRegistry()),
+        var delivery = mock(InterviewResultDeliveryService.class);
+        when(delivery.ensureNotFinalized(anyString())).thenReturn(Mono.empty());
+        var lock = mock(InterviewOperationLock.class);
+        when(lock.execute(anyString(), any())).thenAnswer(call ->
+                ((java.util.function.Supplier<Mono<?>>) call.getArgument(1)).get());
+        service = new ChatInterviewServiceImpl(ai, redis, delivery,
+                lock, new RepitMetrics(new SimpleMeterRegistry()),
                 mock(PersonaTtsProfileResolver.class), tts);
     }
 

src/test/java/com/person/repit/interview/service/impl/ChatInterviewServiceImplTest.java

@@ -81,6 +81,7 @@ void setUp() {
                 ((java.util.function.Supplier<Mono<?>>) invocation.getArgument(1)).get());
         InterviewResultDeliveryService delivery = mock(InterviewResultDeliveryService.class);
         when(delivery.resume(anyString())).thenReturn(Mono.empty());
+        when(delivery.ensureNotFinalized(anyString())).thenReturn(Mono.empty());
         service = new ChatInterviewServiceImpl(
                 aiQuestionClient,
                 redisTemplate,