From d562152789b5021c4f1ffde0b003ce5046cd7961 Mon Sep 17 00:00:00 2001 From: hywznn Date: Thu, 6 Aug 2026 18:43:25 +0900 Subject: [PATCH 1/3] =?UTF-8?q?feat(airun):=20AI=20=EC=8B=A4=ED=96=89=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=20SSE=20=EA=B5=AC=EB=8F=85=20=EC=B6=94?= =?UTF-8?q?=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AiRun 상태 변화를 개인정보 없는 공개 이벤트로 변환하고 사용자·사업장별 SSE 구독, 재연결, heartbeat 및 연결 제한을 적용합니다. 기존 GET 상세 조회는 최종 상태의 기준으로 유지합니다. --- .../server/airun/api/AiRunController.java | 85 ++++++- .../airun/api/AiRunEventStreamBroker.java | 219 ++++++++++++++++++ .../airun/api/AiRunEventStreamProperties.java | 59 +++++ .../airun/application/AiRunPublicEvent.java | 75 ++++++ .../application/AiRunPublicEventType.java | 11 + .../airun/application/AiRunService.java | 20 +- .../application/error/AiRunErrorCode.java | 2 + .../port/AiRunPublicEventPublisher.java | 9 + .../server/common/config/CorsConfig.java | 5 +- src/main/resources/application.yaml | 7 + 10 files changed, 486 insertions(+), 6 deletions(-) create mode 100644 src/main/java/com/fowoco/server/airun/api/AiRunEventStreamBroker.java create mode 100644 src/main/java/com/fowoco/server/airun/api/AiRunEventStreamProperties.java create mode 100644 src/main/java/com/fowoco/server/airun/application/AiRunPublicEvent.java create mode 100644 src/main/java/com/fowoco/server/airun/application/AiRunPublicEventType.java create mode 100644 src/main/java/com/fowoco/server/airun/application/port/AiRunPublicEventPublisher.java diff --git a/src/main/java/com/fowoco/server/airun/api/AiRunController.java b/src/main/java/com/fowoco/server/airun/api/AiRunController.java index 59615d0b..b0738ee5 100644 --- a/src/main/java/com/fowoco/server/airun/api/AiRunController.java +++ b/src/main/java/com/fowoco/server/airun/api/AiRunController.java @@ -2,20 +2,29 @@ import com.fowoco.server.airun.application.AiCandidateDecisionCommand; import com.fowoco.server.airun.application.AiCandidateDecisionService; +import com.fowoco.server.airun.application.AiRunPublicEvent; import com.fowoco.server.airun.application.AiRunService; import com.fowoco.server.auth.application.ActorContext; import com.fowoco.server.auth.application.port.ActorContextProvider; +import com.fowoco.server.common.error.ApiException; import com.fowoco.server.common.web.RequestMetadata; import io.swagger.v3.oas.annotations.Operation; import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.media.Content; +import io.swagger.v3.oas.annotations.media.Schema; import io.swagger.v3.oas.annotations.responses.ApiResponse; import io.swagger.v3.oas.annotations.responses.ApiResponses; import io.swagger.v3.oas.annotations.security.SecurityRequirement; import io.swagger.v3.oas.annotations.tags.Tag; -import jakarta.validation.Valid; import jakarta.servlet.http.HttpServletRequest; +import jakarta.validation.Valid; +import java.io.IOException; import java.net.URI; +import java.util.Map; import java.util.UUID; +import org.springframework.http.CacheControl; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.security.access.prepost.PreAuthorize; @@ -26,6 +35,7 @@ import org.springframework.web.bind.annotation.RequestHeader; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import org.springframework.web.servlet.support.ServletUriComponentsBuilder; @Tag(name = "AI Run", description = "자연어 업무 분석 실행·질문·답변") @@ -37,15 +47,18 @@ public class AiRunController { private final AiRunService aiRunService; private final AiCandidateDecisionService candidateDecisionService; private final ActorContextProvider actorContextProvider; + private final AiRunEventStreamBroker eventStreamBroker; public AiRunController( AiRunService aiRunService, AiCandidateDecisionService candidateDecisionService, - ActorContextProvider actorContextProvider + ActorContextProvider actorContextProvider, + AiRunEventStreamBroker eventStreamBroker ) { this.aiRunService = aiRunService; this.candidateDecisionService = candidateDecisionService; this.actorContextProvider = actorContextProvider; + this.eventStreamBroker = eventStreamBroker; } @Operation( @@ -93,6 +106,74 @@ public AiRunResponse findById(@PathVariable UUID aiRunId) { return AiRunResponse.from(aiRunService.requireRun(aiRunId, actor())); } + @Operation( + operationId = "subscribeAiRunEvents", + summary = "AI 분석 공개 상태 SSE 구독", + description = "Server DB의 AiRun 상태를 화면용 단방향 event로 제공합니다. " + + "연결이 끊겨도 실행은 계속되며 Client는 기존 상세 조회 API로 최종 상태를 확인합니다. " + + "브라우저 기본 EventSource는 Authorization header를 넣을 수 없으므로 fetch 기반 SSE Client를 사용합니다." + ) + @ApiResponses({ + @ApiResponse( + responseCode = "200", + description = "상태 event stream", + content = @Content( + mediaType = MediaType.TEXT_EVENT_STREAM_VALUE, + schema = @Schema(implementation = AiRunPublicEvent.class) + ) + ), + @ApiResponse(responseCode = "400", ref = "#/components/responses/BadRequest"), + @ApiResponse(responseCode = "404", ref = "#/components/responses/NotFound"), + @ApiResponse(responseCode = "429", description = "사용자·AiRun별 연결 수 제한 초과") + }) + @PreAuthorize("hasAnyRole('ADMIN', 'HR', 'VIEWER')") + @GetMapping(path = "/{aiRunId}/events", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public ResponseEntity subscribeEvents( + @PathVariable UUID aiRunId, + @Parameter(description = "마지막으로 처리한 SSE event id") + @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId + ) { + try { + ActorContext actor = actor(); + SseEmitter emitter = eventStreamBroker.subscribe( + actor, + aiRunService.requireRun(aiRunId, actor), + lastEventId + ); + return streamResponse(HttpStatus.OK, emitter); + } catch (ApiException exception) { + return streamResponse(exception.errorCode().status(), errorEmitter(exception)); + } + } + + private ResponseEntity streamResponse( + HttpStatus status, + SseEmitter emitter + ) { + return ResponseEntity.status(status) + .cacheControl(CacheControl.noCache()) + .header("X-Accel-Buffering", "no") + .header(HttpHeaders.CONNECTION, "keep-alive") + .contentType(MediaType.TEXT_EVENT_STREAM) + .body(emitter); + } + + private SseEmitter errorEmitter(ApiException exception) { + SseEmitter emitter = new SseEmitter(1_000L); + try { + emitter.send(SseEmitter.event() + .name("ERROR") + .data(Map.of( + "code", exception.errorCode().code(), + "message", exception.getMessage() + ))); + emitter.complete(); + } catch (IOException sendFailure) { + emitter.completeWithError(sendFailure); + } + return emitter; + } + @Operation(operationId = "answerAiRunQuestions", summary = "누락 Slot 답변 제출") @ApiResponses({ @ApiResponse(responseCode = "202", description = "답변 저장 및 새 분석 시도"), diff --git a/src/main/java/com/fowoco/server/airun/api/AiRunEventStreamBroker.java b/src/main/java/com/fowoco/server/airun/api/AiRunEventStreamBroker.java new file mode 100644 index 00000000..59bab53c --- /dev/null +++ b/src/main/java/com/fowoco/server/airun/api/AiRunEventStreamBroker.java @@ -0,0 +1,219 @@ +package com.fowoco.server.airun.api; + +import com.fowoco.server.airun.application.AiRunPublicEvent; +import com.fowoco.server.airun.application.AiRunResult; +import com.fowoco.server.airun.application.error.AiRunErrorCode; +import com.fowoco.server.airun.application.port.AiRunPublicEventPublisher; +import com.fowoco.server.auth.application.ActorContext; +import com.fowoco.server.common.error.ApiException; +import java.io.IOException; +import java.time.Clock; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +@Component +@EnableConfigurationProperties(AiRunEventStreamProperties.class) +public class AiRunEventStreamBroker implements AiRunPublicEventPublisher { + + private final AiRunEventStreamProperties properties; + private final Clock clock; + private final Map histories = new ConcurrentHashMap<>(); + private final Map> subscriptions = + new ConcurrentHashMap<>(); + + public AiRunEventStreamBroker(AiRunEventStreamProperties properties, Clock clock) { + this.properties = properties; + this.clock = clock; + } + + @Override + public void publish(UUID companyId, AiRunResult run) { + RunKey runKey = new RunKey(companyId, run.aiRunId()); + AiRunPublicEvent event = AiRunPublicEvent.from(run); + histories.computeIfAbsent(runKey, ignored -> new EventHistory()) + .record(event, clock.instant(), properties.historySize()); + + subscriptions.forEach((key, sessions) -> { + if (key.companyId().equals(companyId) && key.aiRunId().equals(run.aiRunId())) { + sessions.forEach(session -> send(session, key, event)); + } + }); + } + + public SseEmitter subscribe(ActorContext actor, AiRunResult current, String lastEventIdHeader) { + Long lastEventId = parseLastEventId(lastEventIdHeader); + RunKey runKey = new RunKey(actor.companyId(), current.aiRunId()); + SubscriptionKey subscriptionKey = new SubscriptionKey( + actor.companyId(), + actor.actorId(), + current.aiRunId() + ); + CopyOnWriteArrayList sessions = subscriptions.computeIfAbsent( + subscriptionKey, + ignored -> new CopyOnWriteArrayList<>() + ); + StreamSession session; + synchronized (sessions) { + if (sessions.size() >= properties.maxConnectionsPerUserRun()) { + throw new ApiException(AiRunErrorCode.AI_RUN_SSE_CONNECTION_LIMIT); + } + SseEmitter emitter = new SseEmitter(properties.timeout().toMillis()); + session = new StreamSession(emitter, lastEventId == null ? -1L : lastEventId); + sessions.add(session); + } + registerCallbacks(subscriptionKey, session); + + List replay = histories.getOrDefault(runKey, EventHistory.EMPTY) + .after(lastEventId == null ? -1L : lastEventId); + if (replay.isEmpty() && (lastEventId == null || current.version() > lastEventId)) { + replay = List.of(AiRunPublicEvent.from(current)); + } + replay.forEach(event -> send(session, subscriptionKey, event)); + + AiRunPublicEvent currentEvent = AiRunPublicEvent.from(current); + if (currentEvent.terminal() && currentEvent.eventId() <= session.lastSentEventId()) { + complete(subscriptionKey, session); + } + return session.emitter(); + } + + @Scheduled(fixedDelayString = "${app.ai-run.sse.heartbeat-interval:15s}") + void heartbeatAndCleanup() { + subscriptions.forEach((key, sessions) -> sessions.forEach(session -> { + try { + session.emitter().send(SseEmitter.event().comment("heartbeat")); + } catch (IOException | IllegalStateException exception) { + complete(key, session); + } + })); + + Instant cutoff = clock.instant().minus(properties.historyRetention()); + histories.entrySet().removeIf(entry -> entry.getValue().lastPublishedAt().isBefore(cutoff)); + } + + private Long parseLastEventId(String value) { + if (value == null || value.isBlank()) { + return null; + } + try { + long parsed = Long.parseLong(value.strip()); + if (parsed < 0) { + throw new NumberFormatException("negative event id"); + } + return parsed; + } catch (NumberFormatException exception) { + throw new ApiException(AiRunErrorCode.AI_RUN_INVALID_LAST_EVENT_ID); + } + } + + private void send(StreamSession session, SubscriptionKey key, AiRunPublicEvent event) { + try { + boolean sent = session.send(event); + if (sent && event.terminal()) { + complete(key, session); + } + } catch (IOException | IllegalStateException exception) { + complete(key, session); + } + } + + private void registerCallbacks(SubscriptionKey key, StreamSession session) { + session.emitter().onCompletion(() -> remove(key, session)); + session.emitter().onTimeout(() -> complete(key, session)); + session.emitter().onError(ignored -> remove(key, session)); + } + + private void complete(SubscriptionKey key, StreamSession session) { + remove(key, session); + try { + session.emitter().complete(); + } catch (IllegalStateException ignored) { + // 이미 완료된 연결입니다. + } + } + + private void remove(SubscriptionKey key, StreamSession session) { + CopyOnWriteArrayList sessions = subscriptions.get(key); + if (sessions == null) { + return; + } + sessions.remove(session); + if (sessions.isEmpty()) { + subscriptions.remove(key, sessions); + } + } + + private record RunKey(UUID companyId, UUID aiRunId) { + } + + private record SubscriptionKey(UUID companyId, UUID actorId, UUID aiRunId) { + } + + private static final class StreamSession { + private final SseEmitter emitter; + private long lastSentEventId; + + private StreamSession(SseEmitter emitter, long lastSentEventId) { + this.emitter = emitter; + this.lastSentEventId = lastSentEventId; + } + + private synchronized boolean send(AiRunPublicEvent event) throws IOException { + if (event.eventId() <= lastSentEventId) { + return false; + } + emitter.send(SseEmitter.event() + .id(Long.toString(event.eventId())) + .name(event.type().name()) + .data(event)); + lastSentEventId = event.eventId(); + return true; + } + + private SseEmitter emitter() { + return emitter; + } + + private synchronized long lastSentEventId() { + return lastSentEventId; + } + } + + private static final class EventHistory { + private static final EventHistory EMPTY = new EventHistory(); + + private final List events = new ArrayList<>(); + private Instant lastPublishedAt = Instant.EPOCH; + + private synchronized void record(AiRunPublicEvent event, Instant publishedAt, int limit) { + if (events.stream().noneMatch(existing -> existing.eventId() == event.eventId())) { + events.add(event); + events.sort(Comparator.comparingLong(AiRunPublicEvent::eventId)); + while (events.size() > limit) { + events.remove(0); + } + } + lastPublishedAt = publishedAt; + } + + private synchronized List after(long eventId) { + return events.stream() + .filter(event -> event.eventId() > eventId) + .toList(); + } + + private synchronized Instant lastPublishedAt() { + return lastPublishedAt; + } + } +} diff --git a/src/main/java/com/fowoco/server/airun/api/AiRunEventStreamProperties.java b/src/main/java/com/fowoco/server/airun/api/AiRunEventStreamProperties.java new file mode 100644 index 00000000..302e481e --- /dev/null +++ b/src/main/java/com/fowoco/server/airun/api/AiRunEventStreamProperties.java @@ -0,0 +1,59 @@ +package com.fowoco.server.airun.api; + +import java.time.Duration; +import org.springframework.boot.context.properties.ConfigurationProperties; + +@ConfigurationProperties(prefix = "app.ai-run.sse") +public final class AiRunEventStreamProperties { + + private final Duration timeout; + private final Duration historyRetention; + private final int historySize; + private final int maxConnectionsPerUserRun; + + public AiRunEventStreamProperties( + Duration timeout, + Duration historyRetention, + int historySize, + int maxConnectionsPerUserRun + ) { + this.timeout = requireDuration(timeout, "timeout", Duration.ofSeconds(1), Duration.ofMinutes(30)); + this.historyRetention = requireDuration( + historyRetention, + "historyRetention", + Duration.ofMinutes(1), + Duration.ofHours(24) + ); + if (historySize < 1 || historySize > 100) { + throw new IllegalArgumentException("AI Run SSE historySize must be between 1 and 100"); + } + if (maxConnectionsPerUserRun < 1 || maxConnectionsPerUserRun > 10) { + throw new IllegalArgumentException("AI Run SSE connection limit must be between 1 and 10"); + } + this.historySize = historySize; + this.maxConnectionsPerUserRun = maxConnectionsPerUserRun; + } + + public Duration timeout() { + return timeout; + } + + public Duration historyRetention() { + return historyRetention; + } + + public int historySize() { + return historySize; + } + + public int maxConnectionsPerUserRun() { + return maxConnectionsPerUserRun; + } + + private Duration requireDuration(Duration value, String name, Duration minimum, Duration maximum) { + if (value == null || value.compareTo(minimum) < 0 || value.compareTo(maximum) > 0) { + throw new IllegalArgumentException("AI Run SSE " + name + " is outside the allowed range"); + } + return value; + } +} diff --git a/src/main/java/com/fowoco/server/airun/application/AiRunPublicEvent.java b/src/main/java/com/fowoco/server/airun/application/AiRunPublicEvent.java new file mode 100644 index 00000000..ac0b1dbd --- /dev/null +++ b/src/main/java/com/fowoco/server/airun/application/AiRunPublicEvent.java @@ -0,0 +1,75 @@ +package com.fowoco.server.airun.application; + +import com.fowoco.server.aiintegration.application.model.AiAnalysisOutcome; +import com.fowoco.server.airun.domain.AiRunStatus; +import java.time.Instant; +import java.util.Objects; +import java.util.UUID; + +/** + * Client에 공개해도 되는 실행 상태만 담습니다. 발화문·Prompt·Provider 원문은 포함하지 않습니다. + */ +public record AiRunPublicEvent( + long eventId, + UUID aiRunId, + AiRunPublicEventType type, + AiRunStatus status, + AiAnalysisOutcome analysisOutcome, + int attemptCount, + long version, + Instant occurredAt +) { + public AiRunPublicEvent { + if (eventId < 0 || version < 0 || attemptCount < 0) { + throw new IllegalArgumentException("AI Run public event numbers must not be negative"); + } + Objects.requireNonNull(aiRunId, "aiRunId must not be null"); + Objects.requireNonNull(type, "type must not be null"); + Objects.requireNonNull(status, "status must not be null"); + Objects.requireNonNull(occurredAt, "occurredAt must not be null"); + } + + public static AiRunPublicEvent from(AiRunResult run) { + return new AiRunPublicEvent( + run.version(), + run.aiRunId(), + typeOf(run), + run.status(), + run.analysisOutcome(), + run.attemptCount(), + run.version(), + run.updatedAt() + ); + } + + public boolean terminal() { + return type == AiRunPublicEventType.NEEDS_INFO + || type == AiRunPublicEventType.REVIEW_REQUIRED + || type == AiRunPublicEventType.COMPLETED + || type == AiRunPublicEventType.FAILED; + } + + private static AiRunPublicEventType typeOf(AiRunResult run) { + if (run.status() == AiRunStatus.QUEUED) { + return AiRunPublicEventType.RUN_QUEUED; + } + if (run.status() == AiRunStatus.RUNNING) { + return run.attemptCount() <= 1 + ? AiRunPublicEventType.RUN_STARTED + : AiRunPublicEventType.SLOT_CHECKING; + } + if (run.status() == AiRunStatus.FAILED) { + return AiRunPublicEventType.FAILED; + } + if (run.analysisOutcome() == AiAnalysisOutcome.CONTEXT_REQUIRED) { + return AiRunPublicEventType.SLOT_CHECKING; + } + if (run.analysisOutcome() == AiAnalysisOutcome.NEEDS_INFO) { + return AiRunPublicEventType.NEEDS_INFO; + } + if (run.analysisOutcome() == AiAnalysisOutcome.REVIEW_REQUIRED) { + return AiRunPublicEventType.REVIEW_REQUIRED; + } + return AiRunPublicEventType.COMPLETED; + } +} diff --git a/src/main/java/com/fowoco/server/airun/application/AiRunPublicEventType.java b/src/main/java/com/fowoco/server/airun/application/AiRunPublicEventType.java new file mode 100644 index 00000000..705953b1 --- /dev/null +++ b/src/main/java/com/fowoco/server/airun/application/AiRunPublicEventType.java @@ -0,0 +1,11 @@ +package com.fowoco.server.airun.application; + +public enum AiRunPublicEventType { + RUN_QUEUED, + RUN_STARTED, + SLOT_CHECKING, + NEEDS_INFO, + REVIEW_REQUIRED, + COMPLETED, + FAILED +} diff --git a/src/main/java/com/fowoco/server/airun/application/AiRunService.java b/src/main/java/com/fowoco/server/airun/application/AiRunService.java index 38dba653..15bfeff2 100644 --- a/src/main/java/com/fowoco/server/airun/application/AiRunService.java +++ b/src/main/java/com/fowoco/server/airun/application/AiRunService.java @@ -14,6 +14,7 @@ import com.fowoco.server.airun.application.error.AiRunErrorCode; import com.fowoco.server.airun.application.port.AiAttemptStarter; import com.fowoco.server.airun.application.port.AiRunRepository; +import com.fowoco.server.airun.application.port.AiRunPublicEventPublisher; import com.fowoco.server.airun.application.port.AiRunRepository.ExecutionState; import com.fowoco.server.auth.application.ActorAuthorizer; import com.fowoco.server.auth.application.ActorContext; @@ -70,6 +71,7 @@ public class AiRunService implements AiAttemptStarter { private final Clock clock; private final TransactionTemplate transactionTemplate; private final AuditEventRepository auditEventRepository; + private final AiRunPublicEventPublisher publicEventPublisher; public AiRunService( ActorAuthorizer actorAuthorizer, @@ -81,7 +83,8 @@ public AiRunService( UuidGenerator uuidGenerator, Clock clock, TransactionTemplate transactionTemplate, - AuditEventRepository auditEventRepository + AuditEventRepository auditEventRepository, + AiRunPublicEventPublisher publicEventPublisher ) { this.actorAuthorizer = actorAuthorizer; this.tenantDatabaseContext = tenantDatabaseContext; @@ -93,6 +96,7 @@ public AiRunService( this.clock = clock; this.transactionTemplate = transactionTemplate; this.auditEventRepository = auditEventRepository; + this.publicEventPublisher = publicEventPublisher; } public AiRunResult createAndExecute( @@ -113,6 +117,7 @@ public AiRunResult createAndExecute( actor, metadata ); + publishCurrent(creation.aiRunId(), creation.companyId()); if (creation.newlyCreated()) { executePlan(creation); } @@ -160,6 +165,7 @@ public AiRunResult answerAndExecute( ); return state; }); + publishCurrent(started.aiRunId(), started.companyId()); executeOne(started, request( started.requestId(), attemptId, @@ -178,7 +184,7 @@ public UUID startAttempt( AnalysisInput analysisInput ) { UUID attemptId = uuidGenerator.generate(); - inTenant(companyId, () -> repository.startContinuationAttempt( + ExecutionState started = inTenant(companyId, () -> repository.startContinuationAttempt( requestId, attemptId, phase, @@ -186,6 +192,7 @@ public UUID startAttempt( analysisInput, clock.instant() )); + publishCurrent(started.aiRunId(), started.companyId()); return attemptId; } @@ -345,6 +352,7 @@ private void saveSuccess( repository.markAttemptSucceeded(aiRunId, companyId, attemptId, response, clock.instant()); return null; }); + publishCurrent(aiRunId, companyId); } private void markLatestFailed(ExecutionState fallback, RuntimeException failure) { @@ -361,6 +369,7 @@ private void markLatestFailed(ExecutionState fallback, RuntimeException failure) ); return null; }); + publishCurrent(latest.aiRunId(), latest.companyId()); } private ExecutionState requireExecution(UUID aiRunId, ActorContext actor) { @@ -479,6 +488,13 @@ private T inTenant(UUID companyId, Supplier action) { }); } + private void publishCurrent(UUID aiRunId, UUID companyId) { + AiRunResult current = inTenant(companyId, () -> repository + .findByIdAndCompanyId(aiRunId, companyId) + .orElseThrow(() -> new IllegalStateException("AI Run for public event was not found"))); + publicEventPublisher.publish(companyId, current); + } + private void appendAudit( UUID aiRunId, ActorContext actor, diff --git a/src/main/java/com/fowoco/server/airun/application/error/AiRunErrorCode.java b/src/main/java/com/fowoco/server/airun/application/error/AiRunErrorCode.java index bdcbf18a..c8920888 100644 --- a/src/main/java/com/fowoco/server/airun/application/error/AiRunErrorCode.java +++ b/src/main/java/com/fowoco/server/airun/application/error/AiRunErrorCode.java @@ -14,6 +14,8 @@ public enum AiRunErrorCode implements ApiErrorCode { AI_RUN_INVALID_INSTRUCTION(HttpStatus.BAD_REQUEST, "업무 요청 문장을 확인해 주세요."), AI_RUN_INVALID_IDEMPOTENCY_KEY(HttpStatus.BAD_REQUEST, "Idempotency-Key를 확인해 주세요."), AI_RUN_INVALID_ANSWER(HttpStatus.BAD_REQUEST, "추가 답변의 항목과 값을 확인해 주세요."), + AI_RUN_INVALID_LAST_EVENT_ID(HttpStatus.BAD_REQUEST, "Last-Event-ID를 확인해 주세요."), + AI_RUN_SSE_CONNECTION_LIMIT(HttpStatus.TOO_MANY_REQUESTS, "이 AI 분석에 연결할 수 있는 실시간 구독 수를 초과했습니다."), AI_RUN_INVALID_DECISION(HttpStatus.BAD_REQUEST, "AI 업무 후보 결정값을 확인해 주세요."); private final HttpStatus status; diff --git a/src/main/java/com/fowoco/server/airun/application/port/AiRunPublicEventPublisher.java b/src/main/java/com/fowoco/server/airun/application/port/AiRunPublicEventPublisher.java new file mode 100644 index 00000000..e13e4e94 --- /dev/null +++ b/src/main/java/com/fowoco/server/airun/application/port/AiRunPublicEventPublisher.java @@ -0,0 +1,9 @@ +package com.fowoco.server.airun.application.port; + +import com.fowoco.server.airun.application.AiRunResult; +import java.util.UUID; + +public interface AiRunPublicEventPublisher { + + void publish(UUID companyId, AiRunResult run); +} diff --git a/src/main/java/com/fowoco/server/common/config/CorsConfig.java b/src/main/java/com/fowoco/server/common/config/CorsConfig.java index 899109f7..2fc2d434 100644 --- a/src/main/java/com/fowoco/server/common/config/CorsConfig.java +++ b/src/main/java/com/fowoco/server/common/config/CorsConfig.java @@ -22,9 +22,10 @@ public CorsConfigurationSource corsConfigurationSource( "Authorization", "Content-Type", "X-Request-Id", - "Idempotency-Key" + "Idempotency-Key", + "Last-Event-ID" )); - configuration.setExposedHeaders(List.of("X-Request-Id")); + configuration.setExposedHeaders(List.of("X-Request-Id", "X-Accel-Buffering")); configuration.setAllowCredentials(true); configuration.setMaxAge(3600L); diff --git a/src/main/resources/application.yaml b/src/main/resources/application.yaml index daa9e7e6..1009d401 100644 --- a/src/main/resources/application.yaml +++ b/src/main/resources/application.yaml @@ -56,6 +56,13 @@ app: max-concurrent-calls: ${AI_RUNTIME_MAX_CONCURRENT_CALLS:8} circuit-breaker-failure-threshold: ${AI_RUNTIME_CIRCUIT_BREAKER_FAILURE_THRESHOLD:5} circuit-breaker-open-duration: ${AI_RUNTIME_CIRCUIT_BREAKER_OPEN_DURATION:30s} + ai-run: + sse: + timeout: ${AI_RUN_SSE_TIMEOUT:5m} + heartbeat-interval: ${AI_RUN_SSE_HEARTBEAT_INTERVAL:15s} + history-retention: ${AI_RUN_SSE_HISTORY_RETENTION:10m} + history-size: ${AI_RUN_SSE_HISTORY_SIZE:20} + max-connections-per-user-run: ${AI_RUN_SSE_MAX_CONNECTIONS:2} reliability: outbox: enabled: ${OUTBOX_ENABLED:true} From 41a85763e8c5fb79baccbd0ae59001a1774765e4 Mon Sep 17 00:00:00 2001 From: hywznn Date: Thu, 6 Aug 2026 18:43:25 +0900 Subject: [PATCH 2/3] =?UTF-8?q?test(airun):=20SSE=20=EB=B3=B4=EC=95=88?= =?UTF-8?q?=EA=B3=BC=20=EC=9E=AC=EC=97=B0=EA=B2=B0=20=EA=B3=84=EC=95=BD=20?= =?UTF-8?q?=EA=B2=80=EC=A6=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 이벤트 순서와 민감정보 미노출, Last-Event-ID 재연결, 사업장 격리, 연결 제한, CORS 및 OpenAPI 계약을 자동 테스트로 고정합니다. --- .env.example | 7 ++ .../server/airun/AiRunApiIntegrationTest.java | 67 +++++++++++++++++++ .../airun/api/AiRunEventStreamBrokerTest.java | 61 +++++++++++++++++ .../airun/api/AiRunOpenApiContractTest.java | 66 ++++++++++++++++++ 4 files changed, 201 insertions(+) create mode 100644 src/test/java/com/fowoco/server/airun/api/AiRunEventStreamBrokerTest.java create mode 100644 src/test/java/com/fowoco/server/airun/api/AiRunOpenApiContractTest.java diff --git a/.env.example b/.env.example index 3e55ea84..83feba03 100644 --- a/.env.example +++ b/.env.example @@ -31,6 +31,13 @@ AI_RUNTIME_MAX_CONCURRENT_CALLS=8 AI_RUNTIME_CIRCUIT_BREAKER_FAILURE_THRESHOLD=5 AI_RUNTIME_CIRCUIT_BREAKER_OPEN_DURATION=30s +# Client용 AiRun 상태 SSE입니다. 실행 기준은 SSE가 아니라 DB와 GET /api/v1/ai-runs/{id}입니다. +AI_RUN_SSE_TIMEOUT=5m +AI_RUN_SSE_HEARTBEAT_INTERVAL=15s +AI_RUN_SSE_HISTORY_RETENTION=10m +AI_RUN_SSE_HISTORY_SIZE=20 +AI_RUN_SSE_MAX_CONNECTIONS=2 + # Transactional Outbox worker 설정입니다. # 일반 실행에서는 켜 두며, 운영 장애 조사 중 자동 처리를 멈춰야 할 때만 false로 둡니다. OUTBOX_ENABLED=true diff --git a/src/test/java/com/fowoco/server/airun/AiRunApiIntegrationTest.java b/src/test/java/com/fowoco/server/airun/AiRunApiIntegrationTest.java index eb12ecb7..67d1998b 100644 --- a/src/test/java/com/fowoco/server/airun/AiRunApiIntegrationTest.java +++ b/src/test/java/com/fowoco/server/airun/AiRunApiIntegrationTest.java @@ -160,6 +160,59 @@ void createsQueriesAnswersAndFinishesOneWorkerAnalysis() throws Exception { )).containsExactly("AI_RUN_CREATED", "AI_RUN_ANSWERS_SUBMITTED"); } + @Test + void replaysSafeOrderedSseEventsAndEnforcesTenantScope() throws Exception { + String tokenA = login(HR_A_EMAIL); + String tokenB = login(HR_B_EMAIL); + HttpResponse created = post( + "/api/v1/ai-runs", + """ + {"instruction":"응웬반A 체류연장 준비해줘"} + """, + tokenA, + "airun-sse-001" + ); + UUID aiRunId = UUID.fromString(JsonPath.read(created.body(), "$.ai_run_id")); + long currentVersion = JsonPath.read(created.body(), "$.version").longValue(); + + HttpResponse stream = getEvents(aiRunId, tokenA, null); + assertThat(stream.statusCode()).isEqualTo(200); + assertThat(stream.headers().firstValue(HttpHeaders.CONTENT_TYPE).orElseThrow()) + .startsWith("text/event-stream"); + assertThat(stream.body()) + .contains("id:0", "event:RUN_STARTED", "event:SLOT_CHECKING", "event:NEEDS_INFO") + .doesNotContain("응웬반A 체류연장 준비해줘", "analysis_input", "prompt", "provider"); + + HttpResponse alreadyConsumed = getEvents(aiRunId, tokenA, Long.toString(currentVersion)); + assertThat(alreadyConsumed.statusCode()).isEqualTo(200); + assertThat(alreadyConsumed.body()).isEmpty(); + + assertThat(getEvents(aiRunId, tokenB, null).statusCode()).isEqualTo(404); + assertThat(getEvents(aiRunId, tokenA, "not-a-number").statusCode()).isEqualTo(400); + } + + @Test + void allowsLastEventIdHeaderInCorsPreflight() throws Exception { + HttpRequest request = HttpRequest.newBuilder( + uri("/api/v1/ai-runs/" + UUID.randomUUID() + "/events") + ) + .header("Origin", "http://localhost:5173") + .header("Access-Control-Request-Method", "GET") + .header("Access-Control-Request-Headers", "authorization,last-event-id") + .method("OPTIONS", HttpRequest.BodyPublishers.noBody()) + .build(); + + HttpResponse response = httpClient.send( + request, + HttpResponse.BodyHandlers.ofString() + ); + + assertThat(response.statusCode()).isEqualTo(200); + assertThat(response.headers().firstValue("Access-Control-Allow-Headers").orElseThrow()) + .containsIgnoringCase("authorization") + .containsIgnoringCase("last-event-id"); + } + @Test void idempotencyAndCompanyIsolationAreEnforced() throws Exception { String tokenA = login(HR_A_EMAIL); @@ -486,6 +539,20 @@ private HttpResponse get(String path, String token) throws Exception { return httpClient.send(request, HttpResponse.BodyHandlers.ofString()); } + private HttpResponse getEvents(UUID aiRunId, String token, String lastEventId) + throws Exception { + HttpRequest.Builder builder = HttpRequest.newBuilder( + uri("/api/v1/ai-runs/" + aiRunId + "/events") + ) + .header(HttpHeaders.AUTHORIZATION, "Bearer " + token) + .header(HttpHeaders.ACCEPT, "text/event-stream") + .GET(); + if (lastEventId != null) { + builder.header("Last-Event-ID", lastEventId); + } + return httpClient.send(builder.build(), HttpResponse.BodyHandlers.ofString()); + } + private HttpResponse post( String path, String body, diff --git a/src/test/java/com/fowoco/server/airun/api/AiRunEventStreamBrokerTest.java b/src/test/java/com/fowoco/server/airun/api/AiRunEventStreamBrokerTest.java new file mode 100644 index 00000000..2067b796 --- /dev/null +++ b/src/test/java/com/fowoco/server/airun/api/AiRunEventStreamBrokerTest.java @@ -0,0 +1,61 @@ +package com.fowoco.server.airun.api; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import com.fowoco.server.airun.application.AiRunResult; +import com.fowoco.server.airun.application.error.AiRunErrorCode; +import com.fowoco.server.airun.domain.AiRunStatus; +import com.fowoco.server.auth.application.ActorContext; +import com.fowoco.server.auth.domain.UserRole; +import com.fowoco.server.common.error.ApiException; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.List; +import java.util.Set; +import java.util.UUID; +import org.junit.jupiter.api.Test; + +class AiRunEventStreamBrokerTest { + + @Test + void limitsConnectionsForTheSameUserAndRun() { + AiRunEventStreamBroker broker = new AiRunEventStreamBroker( + new AiRunEventStreamProperties( + Duration.ofMinutes(1), + Duration.ofMinutes(10), + 20, + 1 + ), + Clock.fixed(Instant.parse("2026-08-06T00:00:00Z"), ZoneOffset.UTC) + ); + UUID actorId = UUID.fromString("10000000-0000-0000-0000-000000000001"); + UUID companyId = UUID.fromString("20000000-0000-0000-0000-000000000001"); + UUID aiRunId = UUID.fromString("30000000-0000-0000-0000-000000000001"); + ActorContext actor = new ActorContext(actorId, companyId, Set.of(UserRole.HR)); + AiRunResult running = new AiRunResult( + aiRunId, + UUID.fromString("40000000-0000-0000-0000-000000000001"), + "원문은 event에 포함하지 않음", + AiRunStatus.RUNNING, + null, + null, + null, + 1, + 0, + List.of(), + List.of(), + Instant.parse("2026-08-06T00:00:00Z"), + Instant.parse("2026-08-06T00:00:00Z") + ); + + broker.subscribe(actor, running, null); + + assertThatThrownBy(() -> broker.subscribe(actor, running, null)) + .isInstanceOfSatisfying(ApiException.class, exception -> + org.assertj.core.api.Assertions.assertThat(exception.errorCode()) + .isEqualTo(AiRunErrorCode.AI_RUN_SSE_CONNECTION_LIMIT) + ); + } +} diff --git a/src/test/java/com/fowoco/server/airun/api/AiRunOpenApiContractTest.java b/src/test/java/com/fowoco/server/airun/api/AiRunOpenApiContractTest.java new file mode 100644 index 00000000..eceb310e --- /dev/null +++ b/src/test/java/com/fowoco/server/airun/api/AiRunOpenApiContractTest.java @@ -0,0 +1,66 @@ +package com.fowoco.server.airun.api; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.server.LocalServerPort; +import org.springframework.test.context.ActiveProfiles; + +@ActiveProfiles("test") +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT) +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +class AiRunOpenApiContractTest { + + @LocalServerPort + private int port; + + private final HttpClient httpClient = HttpClient.newHttpClient(); + private final ObjectMapper objectMapper = new ObjectMapper(); + private JsonNode openApi; + + @BeforeAll + void loadOpenApi() throws Exception { + HttpResponse response = httpClient.send( + HttpRequest.newBuilder() + .uri(URI.create("http://localhost:" + port + "/v3/api-docs")) + .GET() + .build(), + HttpResponse.BodyHandlers.ofString() + ); + + assertThat(response.statusCode()).isEqualTo(200); + openApi = objectMapper.readTree(response.body()); + } + + @Test + void publishesAiRunEventStreamContract() { + JsonNode operation = openApi.at( + "/paths/~1api~1v1~1ai-runs~1{aiRunId}~1events/get" + ); + + assertThat(operation.path("operationId").asText()).isEqualTo("subscribeAiRunEvents"); + assertThat(operation.at("/security/0/bearerAuth").isArray()).isTrue(); + assertThat(operation.at("/responses/200/content/text~1event-stream").isMissingNode()) + .isFalse(); + assertThat(operation.at("/responses/400/$ref").asText()) + .isEqualTo("#/components/responses/BadRequest"); + assertThat(operation.at("/responses/404/$ref").asText()) + .isEqualTo("#/components/responses/NotFound"); + assertThat(operation.path("responses").has("429")).isTrue(); + assertThat(operation.path("parameters")) + .anySatisfy(parameter -> { + assertThat(parameter.path("name").asText()).isEqualTo("Last-Event-ID"); + assertThat(parameter.path("in").asText()).isEqualTo("header"); + assertThat(parameter.path("required").asBoolean()).isFalse(); + }); + } +} From a7636ac60d8f7a47f3144607081aac0ee7b05187 Mon Sep 17 00:00:00 2001 From: hywznn Date: Thu, 6 Aug 2026 20:15:09 +0900 Subject: [PATCH 3/3] =?UTF-8?q?fix(airun):=20=EC=B5=9C=EC=B4=88=20?= =?UTF-8?q?=EB=B6=84=EC=84=9D=EC=9D=84=20=EB=B9=84=EB=8F=99=EA=B8=B0?= =?UTF-8?q?=EB=A1=9C=20=EC=8B=A4=ED=96=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AiRun을 QUEUED 상태로 저장한 뒤 202 응답을 먼저 반환합니다. 별도 실행기에서 RUNNING 전환과 Runtime 호출을 수행해 최초 SSE 상태를 실시간으로 구독할 수 있게 합니다. --- .../server/airun/api/AiRunController.java | 2 +- .../airun/application/AiRunService.java | 31 ++++- .../application/port/AiRunRepository.java | 2 + .../execution/AiRunExecutionConfig.java | 23 ++++ .../persistence/JdbcAiRunRepository.java | 33 ++++- .../server/airun/AiRunApiIntegrationTest.java | 128 +++++++++++++----- 6 files changed, 181 insertions(+), 38 deletions(-) create mode 100644 src/main/java/com/fowoco/server/airun/infrastructure/execution/AiRunExecutionConfig.java diff --git a/src/main/java/com/fowoco/server/airun/api/AiRunController.java b/src/main/java/com/fowoco/server/airun/api/AiRunController.java index b0738ee5..00ad16c5 100644 --- a/src/main/java/com/fowoco/server/airun/api/AiRunController.java +++ b/src/main/java/com/fowoco/server/airun/api/AiRunController.java @@ -82,7 +82,7 @@ public ResponseEntity create( @Valid @RequestBody CreateAiRunRequest request, HttpServletRequest servletRequest ) { - AiRunResponse response = AiRunResponse.from(aiRunService.createAndExecute( + AiRunResponse response = AiRunResponse.from(aiRunService.createAndSchedule( request.instruction(), idempotencyKey, actor(), diff --git a/src/main/java/com/fowoco/server/airun/application/AiRunService.java b/src/main/java/com/fowoco/server/airun/application/AiRunService.java index 15bfeff2..d1747d2f 100644 --- a/src/main/java/com/fowoco/server/airun/application/AiRunService.java +++ b/src/main/java/com/fowoco/server/airun/application/AiRunService.java @@ -42,8 +42,10 @@ import java.util.Map; import java.util.Objects; import java.util.UUID; +import java.util.concurrent.Executor; import java.util.function.Supplier; import java.util.regex.Pattern; +import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.dao.DataIntegrityViolationException; import org.springframework.stereotype.Service; import org.springframework.transaction.support.TransactionTemplate; @@ -72,6 +74,7 @@ public class AiRunService implements AiAttemptStarter { private final TransactionTemplate transactionTemplate; private final AuditEventRepository auditEventRepository; private final AiRunPublicEventPublisher publicEventPublisher; + private final Executor aiRunTaskExecutor; public AiRunService( ActorAuthorizer actorAuthorizer, @@ -84,7 +87,8 @@ public AiRunService( Clock clock, TransactionTemplate transactionTemplate, AuditEventRepository auditEventRepository, - AiRunPublicEventPublisher publicEventPublisher + AiRunPublicEventPublisher publicEventPublisher, + @Qualifier("aiRunTaskExecutor") Executor aiRunTaskExecutor ) { this.actorAuthorizer = actorAuthorizer; this.tenantDatabaseContext = tenantDatabaseContext; @@ -97,9 +101,10 @@ public AiRunService( this.transactionTemplate = transactionTemplate; this.auditEventRepository = auditEventRepository; this.publicEventPublisher = publicEventPublisher; + this.aiRunTaskExecutor = aiRunTaskExecutor; } - public AiRunResult createAndExecute( + public AiRunResult createAndSchedule( String instruction, String idempotencyKey, ActorContext actor, @@ -118,10 +123,11 @@ public AiRunResult createAndExecute( metadata ); publishCurrent(creation.aiRunId(), creation.companyId()); + AiRunResult accepted = requireRun(creation.aiRunId(), actor); if (creation.newlyCreated()) { - executePlan(creation); + schedulePlan(creation); } - return requireRun(creation.aiRunId(), actor); + return accepted; } public AiRunResult requireRun(UUID aiRunId, ActorContext actor) { @@ -300,6 +306,12 @@ private void executePlan(AiRunCreation creation) { .findExecutionState(creation.aiRunId(), creation.companyId()) .orElseThrow(() -> new IllegalStateException("created AI Run has no attempt"))); try { + initial = inTenant(creation.companyId(), () -> repository.startInitialAttempt( + creation.aiRunId(), + creation.companyId(), + clock.instant() + )); + publishCurrent(initial.aiRunId(), initial.companyId()); AiAnalysisResponse planResponse = runtimeClient.analyze( creation.request(), AiRuntimeCallContext.withoutTrace() @@ -330,6 +342,17 @@ private void executePlan(AiRunCreation creation) { } } + private void schedulePlan(AiRunCreation creation) { + try { + aiRunTaskExecutor.execute(() -> executePlan(creation)); + } catch (RuntimeException schedulingFailure) { + ExecutionState queued = inTenant(creation.companyId(), () -> repository + .findExecutionState(creation.aiRunId(), creation.companyId()) + .orElseThrow(() -> new IllegalStateException("queued AI Run has no attempt"))); + markLatestFailed(queued, schedulingFailure); + } + } + private void executeOne(ExecutionState state, AiAnalysisRequest request) { try { AiAnalysisResponse response = runtimeClient.analyze( diff --git a/src/main/java/com/fowoco/server/airun/application/port/AiRunRepository.java b/src/main/java/com/fowoco/server/airun/application/port/AiRunRepository.java index cd7490a2..3e81c203 100644 --- a/src/main/java/com/fowoco/server/airun/application/port/AiRunRepository.java +++ b/src/main/java/com/fowoco/server/airun/application/port/AiRunRepository.java @@ -15,6 +15,8 @@ public interface AiRunRepository { void insertPlan(PlanRun run); + ExecutionState startInitialAttempt(UUID aiRunId, UUID companyId, Instant startedAt); + Optional findByIdAndCompanyId(UUID aiRunId, UUID companyId); Optional findExecutionState(UUID aiRunId, UUID companyId); diff --git a/src/main/java/com/fowoco/server/airun/infrastructure/execution/AiRunExecutionConfig.java b/src/main/java/com/fowoco/server/airun/infrastructure/execution/AiRunExecutionConfig.java new file mode 100644 index 00000000..9baea495 --- /dev/null +++ b/src/main/java/com/fowoco/server/airun/infrastructure/execution/AiRunExecutionConfig.java @@ -0,0 +1,23 @@ +package com.fowoco.server.airun.infrastructure.execution; + +import java.util.concurrent.Executor; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; + +@Configuration +public class AiRunExecutionConfig { + + @Bean(name = "aiRunTaskExecutor") + public Executor aiRunTaskExecutor() { + ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor(); + executor.setCorePoolSize(1); + executor.setMaxPoolSize(4); + executor.setQueueCapacity(100); + executor.setThreadNamePrefix("ai-run-"); + executor.setWaitForTasksToCompleteOnShutdown(true); + executor.setAwaitTerminationSeconds(15); + executor.initialize(); + return executor; + } +} diff --git a/src/main/java/com/fowoco/server/airun/infrastructure/persistence/JdbcAiRunRepository.java b/src/main/java/com/fowoco/server/airun/infrastructure/persistence/JdbcAiRunRepository.java index 2dbf2614..5e28db7a 100644 --- a/src/main/java/com/fowoco/server/airun/infrastructure/persistence/JdbcAiRunRepository.java +++ b/src/main/java/com/fowoco/server/airun/infrastructure/persistence/JdbcAiRunRepository.java @@ -74,7 +74,7 @@ INSERT INTO ai_run ( ai_run_id, company_id, requested_by, request_id, instruction, instruction_hash, idempotency_key_hash, status, attempt_count, created_at, updated_at, version - ) VALUES (?, ?, ?, ?, ?, ?, ?, 'RUNNING', 1, ?, ?, 0) + ) VALUES (?, ?, ?, ?, ?, ?, ?, 'QUEUED', 1, ?, ?, 0) """, run.aiRunId(), run.companyId(), @@ -99,6 +99,37 @@ INSERT INTO ai_run ( )); } + @Override + @Transactional + public ExecutionState startInitialAttempt(UUID aiRunId, UUID companyId, Instant startedAt) { + int updated = jdbcTemplate.update( + """ + UPDATE ai_run + SET status = 'RUNNING', updated_at = ?, version = version + 1 + WHERE ai_run_id = ? AND company_id = ? AND status = 'QUEUED' + """, + timestamp(startedAt), + aiRunId, + companyId + ); + if (updated != 1) { + throw new IllegalStateException("queued AI Run was not found"); + } + jdbcTemplate.update( + """ + UPDATE ai_attempt + SET started_at = ? + WHERE ai_run_id = ? AND company_id = ? AND sequence_no = 1 + AND status = 'RUNNING' + """, + timestamp(startedAt), + aiRunId, + companyId + ); + return findExecutionState(aiRunId, companyId) + .orElseThrow(() -> new IllegalStateException("started AI Run has no attempt")); + } + @Override @Transactional(readOnly = true) public Optional findByIdAndCompanyId(UUID aiRunId, UUID companyId) { diff --git a/src/test/java/com/fowoco/server/airun/AiRunApiIntegrationTest.java b/src/test/java/com/fowoco/server/airun/AiRunApiIntegrationTest.java index 67d1998b..c552f1e1 100644 --- a/src/test/java/com/fowoco/server/airun/AiRunApiIntegrationTest.java +++ b/src/test/java/com/fowoco/server/airun/AiRunApiIntegrationTest.java @@ -14,15 +14,19 @@ import com.fowoco.server.aiintegration.application.model.AiRuntimeVersions; import com.fowoco.server.aiintegration.application.port.AiRuntimeClient; import com.jayway.jsonpath.JsonPath; +import java.io.InputStream; import java.math.BigDecimal; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; import java.time.LocalDate; import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -117,21 +121,21 @@ void createsQueriesAnswersAndFinishesOneWorkerAnalysis() throws Exception { assertThat(created.statusCode()).isEqualTo(202); UUID aiRunId = UUID.fromString(JsonPath.read(created.body(), "$.ai_run_id")); - assertThat(JsonPath.read(created.body(), "$.analysis_outcome")) - .isEqualTo("NEEDS_INFO"); - assertThat(JsonPath.read(created.body(), "$.detected_intent")) - .isEqualTo("EXPIRY_RENEWAL"); + assertThat(JsonPath.read(created.body(), "$.status")).isEqualTo("QUEUED"); + assertThat(JsonPath.read(created.body(), "$.analysis_outcome")).isNull(); assertThat(JsonPath.read(created.body(), "$.instruction")) .isEqualTo("응웬반A 체류연장 준비해줘"); - assertThat(JsonPath.>read(created.body(), "$.questions[*].slot_key")) - .containsExactly("due_at"); assertThat(JsonPath.read(created.body(), "$.attempt_count").intValue()) - .isEqualTo(2); - long version = JsonPath.read(created.body(), "$.version").longValue(); + .isEqualTo(1); - HttpResponse detail = get("/api/v1/ai-runs/" + aiRunId, token); + HttpResponse detail = awaitRun(aiRunId, token, "NEEDS_INFO", 2); assertThat(detail.statusCode()).isEqualTo(200); assertThat(JsonPath.read(detail.body(), "$.status")).isEqualTo("SUCCEEDED"); + assertThat(JsonPath.read(detail.body(), "$.detected_intent")) + .isEqualTo("EXPIRY_RENEWAL"); + assertThat(JsonPath.>read(detail.body(), "$.questions[*].slot_key")) + .containsExactly("due_at"); + long version = JsonPath.read(detail.body(), "$.version").longValue(); HttpResponse answered = post( "/api/v1/ai-runs/" + aiRunId + "/answers", @@ -162,27 +166,54 @@ void createsQueriesAnswersAndFinishesOneWorkerAnalysis() throws Exception { @Test void replaysSafeOrderedSseEventsAndEnforcesTenantScope() throws Exception { + CountDownLatch planStarted = new CountDownLatch(1); + CountDownLatch releasePlan = new CountDownLatch(1); + reset(runtimeClient); + runtimeCalls.set(0); + when(runtimeClient.analyze(any(), any())).thenAnswer(invocation -> { + int call = runtimeCalls.incrementAndGet(); + if (call == 1) { + planStarted.countDown(); + if (!releasePlan.await(5, TimeUnit.SECONDS)) { + throw new IllegalStateException("test did not release delayed PLAN call"); + } + } + return scriptedResponse(invocation.getArgument(0), call); + }); String tokenA = login(HR_A_EMAIL); String tokenB = login(HR_B_EMAIL); - HttpResponse created = post( - "/api/v1/ai-runs", - """ - {"instruction":"응웬반A 체류연장 준비해줘"} - """, - tokenA, - "airun-sse-001" - ); - UUID aiRunId = UUID.fromString(JsonPath.read(created.body(), "$.ai_run_id")); - long currentVersion = JsonPath.read(created.body(), "$.version").longValue(); - - HttpResponse stream = getEvents(aiRunId, tokenA, null); - assertThat(stream.statusCode()).isEqualTo(200); - assertThat(stream.headers().firstValue(HttpHeaders.CONTENT_TYPE).orElseThrow()) - .startsWith("text/event-stream"); - assertThat(stream.body()) - .contains("id:0", "event:RUN_STARTED", "event:SLOT_CHECKING", "event:NEEDS_INFO") + UUID aiRunId; + String streamBody; + try { + HttpResponse created = post( + "/api/v1/ai-runs", + """ + {"instruction":"응웬반A 체류연장 준비해줘"} + """, + tokenA, + "airun-sse-001" + ); + assertThat(created.statusCode()).isEqualTo(202); + assertThat(JsonPath.read(created.body(), "$.status")).isEqualTo("QUEUED"); + aiRunId = UUID.fromString(JsonPath.read(created.body(), "$.ai_run_id")); + assertThat(planStarted.await(5, TimeUnit.SECONDS)).isTrue(); + + HttpResponse stream = openEventStream(aiRunId, tokenA); + assertThat(stream.statusCode()).isEqualTo(200); + assertThat(stream.headers().firstValue(HttpHeaders.CONTENT_TYPE).orElseThrow()) + .startsWith("text/event-stream"); + releasePlan.countDown(); + streamBody = new String(stream.body().readAllBytes(), StandardCharsets.UTF_8); + } finally { + releasePlan.countDown(); + } + assertThat(streamBody) + .contains("event:RUN_QUEUED", "event:RUN_STARTED", "event:SLOT_CHECKING", "event:NEEDS_INFO") .doesNotContain("응웬반A 체류연장 준비해줘", "analysis_input", "prompt", "provider"); + HttpResponse detail = awaitRun(aiRunId, tokenA, "NEEDS_INFO", 2); + long currentVersion = JsonPath.read(detail.body(), "$.version").longValue(); + HttpResponse alreadyConsumed = getEvents(aiRunId, tokenA, Long.toString(currentVersion)); assertThat(alreadyConsumed.statusCode()).isEqualTo(200); assertThat(alreadyConsumed.body()).isEmpty(); @@ -237,6 +268,7 @@ void idempotencyAndCompanyIsolationAreEnforced() throws Exception { assertThat(repeated.statusCode()).isEqualTo(202); assertThat(JsonPath.read(repeated.body(), "$.ai_run_id")) .isEqualTo(aiRunId.toString()); + awaitRun(aiRunId, tokenA, "NEEDS_INFO", 2); assertThat(runtimeCalls).hasValue(2); assertThat(get("/api/v1/ai-runs/" + aiRunId, tokenB).statusCode()).isEqualTo(404); @@ -270,16 +302,15 @@ void acceptedCandidateCreatesOneCaseAndThreeTasksIdempotently() throws Exception "airun-decision-run" ); assertThat(reviewed.statusCode()).isEqualTo(202); - assertThat(JsonPath.read(reviewed.body(), "$.analysis_outcome")) - .isEqualTo("REVIEW_REQUIRED"); - assertThat(runtimeCalls).hasValue(2); UUID aiRunId = UUID.fromString(JsonPath.read(reviewed.body(), "$.ai_run_id")); + HttpResponse detail = awaitRun(aiRunId, tokenA, "REVIEW_REQUIRED", 2); + assertThat(runtimeCalls).hasValue(2); UUID candidateId = UUID.fromString(JsonPath.read( - reviewed.body(), + detail.body(), "$.candidates[0].candidate_id" )); - long expectedVersion = JsonPath.read(reviewed.body(), "$.version").longValue(); - assertThat(JsonPath.read(reviewed.body(), "$.detected_intent")) + long expectedVersion = JsonPath.read(detail.body(), "$.version").longValue(); + assertThat(JsonPath.read(detail.body(), "$.detected_intent")) .isEqualTo("EXPIRY_RENEWAL"); String decisionBody = """ @@ -539,6 +570,39 @@ private HttpResponse get(String path, String token) throws Exception { return httpClient.send(request, HttpResponse.BodyHandlers.ofString()); } + private HttpResponse awaitRun( + UUID aiRunId, + String token, + String expectedOutcome, + int expectedAttemptCount + ) throws Exception { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + HttpResponse latest = null; + while (System.nanoTime() < deadline) { + latest = get("/api/v1/ai-runs/" + aiRunId, token); + String outcome = JsonPath.read(latest.body(), "$.analysis_outcome"); + int attemptCount = JsonPath.read(latest.body(), "$.attempt_count").intValue(); + if (expectedOutcome.equals(outcome) && attemptCount == expectedAttemptCount) { + return latest; + } + Thread.sleep(25); + } + throw new AssertionError("AI Run did not reach expected state: " + + (latest == null ? "no response" : latest.body())); + } + + private HttpResponse openEventStream(UUID aiRunId, String token) + throws Exception { + HttpRequest request = HttpRequest.newBuilder( + uri("/api/v1/ai-runs/" + aiRunId + "/events") + ) + .header(HttpHeaders.AUTHORIZATION, "Bearer " + token) + .header(HttpHeaders.ACCEPT, "text/event-stream") + .GET() + .build(); + return httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream()); + } + private HttpResponse getEvents(UUID aiRunId, String token, String lastEventId) throws Exception { HttpRequest.Builder builder = HttpRequest.newBuilder(