diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/configs/SecurityConfig.kt b/src/main/kotlin/kr/co/vividnext/sodalive/configs/SecurityConfig.kt
index b55e35f5..f1de5e02 100644
--- a/src/main/kotlin/kr/co/vividnext/sodalive/configs/SecurityConfig.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/configs/SecurityConfig.kt
@@ -157,6 +157,13 @@ class SecurityConfig(
                 JwtFilter::class.java
             )
             .authorizeRequests()
+            .antMatchers("/internal/api/subtitle-jobs/**")
+            .access(
+                "hasRole('BOT') and " +
+                    "principal instanceof T(kr.co.vividnext.sodalive.member.MemberAdapter) and " +
+                    "principal.member.role == T(kr.co.vividnext.sodalive.member.MemberRole).BOT and " +
+                    "principal.member.isActive"
+            )
             .antMatchers("/readyz").permitAll()
             .requestMatchers(EndpointRequest.toAnyEndpoint()).permitAll()
             .antMatchers("/member/check/email").permitAll()
diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/content/AudioContentService.kt b/src/main/kotlin/kr/co/vividnext/sodalive/content/AudioContentService.kt
index 2b18766e..5be729a1 100644
--- a/src/main/kotlin/kr/co/vividnext/sodalive/content/AudioContentService.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/content/AudioContentService.kt
@@ -43,15 +43,20 @@ import kr.co.vividnext.sodalive.utils.generateFileName
 import kr.co.vividnext.sodalive.v2.creatorname.application.CreatorLocalizedNameService
 import kr.co.vividnext.sodalive.v2.home.following.application.HomeFollowingNewsPublishService
 import kr.co.vividnext.sodalive.v2.recommendation.application.CreatorContentViewHistoryService
+import kr.co.vividnext.sodalive.v2.subtitle.application.SubtitleJobService
 import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Value
 import org.springframework.cache.annotation.Cacheable
 import org.springframework.context.ApplicationEventPublisher
 import org.springframework.data.repository.findByIdOrNull
 import org.springframework.stereotype.Service
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.TransactionDefinition
+import org.springframework.transaction.annotation.Propagation
 import org.springframework.transaction.annotation.Transactional
 import org.springframework.transaction.support.TransactionSynchronization
 import org.springframework.transaction.support.TransactionSynchronizationManager
+import org.springframework.transaction.support.TransactionTemplate
 import org.springframework.web.multipart.MultipartFile
 import java.text.SimpleDateFormat
 import java.time.LocalDateTime
@@ -93,6 +98,8 @@ class AudioContentService(
     private val contentThemeTranslationRepository: ContentThemeTranslationRepository,
     private val homeFollowingNewsPublishService: HomeFollowingNewsPublishService,
     private val memberContentPreferenceService: MemberContentPreferenceService,
+    private val subtitleJobService: SubtitleJobService,
+    transactionManager: PlatformTransactionManager,
 
     @Value("\${cloud.aws.s3.content-bucket}")
     private val audioContentBucket: String,
@@ -104,6 +111,9 @@ class AudioContentService(
     private val coverImageHost: String
 ) {
     private val log = LoggerFactory.getLogger(javaClass)
+    private val registrationTransaction = TransactionTemplate(transactionManager).apply {
+        propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW
+    }
 
     @Transactional
     fun audioContentLike(request: PutAudioContentLikeRequest, member: Member): PutAudioContentLikeResponse {
@@ -220,7 +230,7 @@ class AudioContentService(
         pinContent?.isActive = false
     }
 
-    @Transactional
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     fun createAudioContent(
         contentFile: MultipartFile?,
         coverImage: MultipartFile?,
@@ -245,7 +255,7 @@ class AudioContentService(
         return createAudioContent(contentFile, coverImage, request, releaseDate, member)
     }
 
-    @Transactional
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     fun createAudioContent(
         contentFile: MultipartFile?,
         coverImage: MultipartFile?,
@@ -260,115 +270,118 @@ class AudioContentService(
             throw SodaException(messageKey = "content.error.content_required")
         }
 
-        validatePreviewTime(request.previewStartTime, request.previewEndTime)
+        val registration = requireNotNull(
+            registrationTransaction.execute {
+                validatePreviewTime(request.previewStartTime, request.previewEndTime)
 
-        // 테마 체크
-        val theme = themeQueryRepository.findThemeByIdAndActive(id = request.themeId)
-            ?: throw SodaException(messageKey = "content.error.invalid_theme")
+                // 테마 체크
+                val theme = themeQueryRepository.findThemeByIdAndActive(id = request.themeId)
+                    ?: throw SodaException(messageKey = "content.error.invalid_theme")
 
-        if ((request.themeId == 12L || request.themeId == 13L || request.themeId == 14L) && request.price < 5) {
-            throw SodaException(messageKey = "content.error.alarm_theme_price_min")
-        }
+                if ((request.themeId == 12L || request.themeId == 13L || request.themeId == 14L) && request.price < 5) {
+                    throw SodaException(messageKey = "content.error.alarm_theme_price_min")
+                }
 
-        if (request.price in 1..4) throw SodaException(messageKey = "content.error.minimum_price")
+                if (request.price in 1..4) throw SodaException(messageKey = "content.error.minimum_price")
 
-        val isFullDetailVisible = if (request.price >= 50) {
-            request.isFullDetailVisible
-        } else {
-            true
-        }
+                val isFullDetailVisible = if (request.price >= 50) {
+                    request.isFullDetailVisible
+                } else {
+                    true
+                }
 
-        val purchaseOption = if (request.themeId == 12L || request.themeId == 13L || request.themeId == 14L) {
-            PurchaseOption.BUY_ONLY
-        } else {
-            request.purchaseOption
-        }
+                val purchaseOption = if (request.themeId == 12L || request.themeId == 13L || request.themeId == 14L) {
+                    PurchaseOption.BUY_ONLY
+                } else {
+                    request.purchaseOption
+                }
 
-        val isOnlyRental = if (
-            (request.limited != null && request.limited > 0) ||
-            purchaseOption == PurchaseOption.BUY_ONLY
-        ) {
-            false
-        } else if (request.purchaseOption == PurchaseOption.RENT_ONLY) {
-            true
-        } else {
-            request.isOnlyRental
-        }
+                val isOnlyRental = if (
+                    (request.limited != null && request.limited > 0) ||
+                    purchaseOption == PurchaseOption.BUY_ONLY
+                ) {
+                    false
+                } else if (request.purchaseOption == PurchaseOption.RENT_ONLY) {
+                    true
+                } else {
+                    request.isOnlyRental
+                }
 
-        // DB에 값 추가
-        val audioContent = AudioContent(
-            title = request.title.trim(),
-            detail = request.detail.trim(),
-            languageCode = request.languageCode,
-            price = if (request.price > 0) {
-                request.price
-            } else {
-                0
-            },
-            releaseDate = releaseDate,
-            limited = request.limited,
-            remaining = request.limited,
-            isAdult = request.isAdult,
-            purchaseOption = purchaseOption,
-            isGeneratePreview = request.isGeneratePreview,
-            isOnlyRental = isOnlyRental,
-            isPointAvailable = request.isPointAvailable,
-            isCommentAvailable = request.isCommentAvailable,
-            isFullDetailVisible = isFullDetailVisible
-        )
-        audioContent.theme = theme
-        audioContent.member = member
-        repository.save(audioContent)
-
-        // 태그 분리, #추가, 등록
-        if (request.tags.isNotBlank()) {
-            val tags = request.tags
-                .replace("#", " #")
-                .split(" ")
-                .asSequence()
-                .map { it.trim() }
-                .filter { it.isNotBlank() }
-                .map {
-                    val tag = if (!it.startsWith("#")) {
-                        "#$it"
+                // DB에 값 추가
+                val audioContent = AudioContent(
+                    title = request.title.trim(),
+                    detail = request.detail.trim(),
+                    languageCode = request.languageCode,
+                    price = if (request.price > 0) {
+                        request.price
                     } else {
-                        it
-                    }
+                        0
+                    },
+                    releaseDate = releaseDate,
+                    limited = request.limited,
+                    remaining = request.limited,
+                    isAdult = request.isAdult,
+                    purchaseOption = purchaseOption,
+                    isGeneratePreview = request.isGeneratePreview,
+                    isOnlyRental = isOnlyRental,
+                    isPointAvailable = request.isPointAvailable,
+                    isCommentAvailable = request.isCommentAvailable,
+                    isFullDetailVisible = isFullDetailVisible
+                )
+                audioContent.theme = theme
+                audioContent.member = member
+                repository.save(audioContent)
 
-                    val hashTag = hashTagRepository.findByTag(tag)
-                        ?: hashTagRepository.save(HashTag(tag))
+                // 태그 분리, #추가, 등록
+                if (request.tags.isNotBlank()) {
+                    val tags = request.tags
+                        .replace("#", " #")
+                        .split(" ")
+                        .asSequence()
+                        .map { it.trim() }
+                        .filter { it.isNotBlank() }
+                        .map {
+                            val tag = if (!it.startsWith("#")) {
+                                "#$it"
+                            } else {
+                                it
+                            }
 
-                    val audioContentHashTag = AudioContentHashTag()
-                    audioContentHashTag.audioContent = audioContent
-                    audioContentHashTag.hashTag = hashTag
+                            val hashTag = hashTagRepository.findByTag(tag)
+                                ?: hashTagRepository.save(HashTag(tag))
 
-                    audioContentHashTag
-                }.toList()
+                            val audioContentHashTag = AudioContentHashTag()
+                            audioContentHashTag.audioContent = audioContent
+                            audioContentHashTag.hashTag = hashTag
 
-            audioContent.audioContentHashTags.addAll(tags)
-        }
+                            audioContentHashTag
+                        }.toList()
 
-        var metadata = ObjectMetadata()
-        metadata.contentLength = coverImage.size
+                    audioContent.audioContentHashTags.addAll(tags)
+                }
 
-        // 커버 이미지 파일명 생성
-        val coverImageFileName = generateFileName(prefix = "${audioContent.id}-cover")
+                val metadata = ObjectMetadata()
+                metadata.contentLength = coverImage.size
 
-        // 커버 이미지 업로드
-        val coverImagePath = s3Uploader.upload(
-            inputStream = coverImage.inputStream,
-            bucket = coverImageBucket,
-            filePath = "audio_content_cover/${audioContent.id}/$coverImageFileName",
-            metadata = metadata
+                // 커버 이미지 파일명 생성
+                val coverImageFileName = generateFileName(prefix = "${audioContent.id}-cover")
+
+                // 커버 이미지 업로드
+                val coverImagePath = s3Uploader.upload(
+                    inputStream = coverImage.inputStream,
+                    bucket = coverImageBucket,
+                    filePath = "audio_content_cover/${audioContent.id}/$coverImageFileName",
+                    metadata = metadata
+                )
+
+                audioContent.coverImage = coverImagePath
+                audioContent to subtitleJobService.register(audioContent, contentFile.originalFilename)
+            }
         )
-
-        audioContent.coverImage = coverImagePath
-
-        // 콘텐츠 파일명 생성
-        val contentFileName = generateFileName(prefix = "${audioContent.id}-content")
+        val (audioContent, subtitleJob) = registration
 
         // 콘텐츠 파일 업로드
-        metadata = ObjectMetadata()
+        val metadata = ObjectMetadata()
         metadata.contentLength = contentFile.size
         metadata.addUserMetadata(
             "generate_preview",
@@ -384,38 +397,45 @@ class AudioContentService(
             metadata.addUserMetadata("preview_end_time", request.previewEndTime)
         }
 
-        val contentPath = s3Uploader.upload(
-            inputStream = contentFile.inputStream,
-            bucket = audioContentBucket,
-            filePath = "input/${audioContent.id}/$contentFileName",
-            metadata = metadata
-        )
-
-        audioContent.content = contentPath
-
-        // 언어 코드가 지정되지 않은 경우, 파파고 언어 감지 API를 통해 비동기로 언어를 식별한다.
-        if (audioContent.languageCode.isNullOrBlank()) {
-            val papagoQuery = listOf(
-                request.title.trim(),
-                request.detail.trim(),
-                request.tags.trim()
-            )
-                .filter { it.isNotBlank() }
-                .joinToString(" ")
-
-            applicationEventPublisher.publishEvent(
-                LanguageDetectEvent(
-                    id = audioContent.id!!,
-                    query = papagoQuery
+        try {
+            contentFile.inputStream.use { inputStream ->
+                s3Uploader.upload(
+                    inputStream = inputStream,
+                    bucket = audioContentBucket,
+                    filePath = subtitleJob.sourceKey,
+                    metadata = metadata
                 )
-            )
-        } else {
-            applicationEventPublisher.publishEvent(
-                LanguageTranslationEvent(
-                    id = audioContent.id!!,
-                    targetType = LanguageTranslationTargetType.CONTENT
+            }
+        } catch (exception: Exception) {
+            subtitleJobService.recordUploadError(subtitleJob.jobId)
+            throw exception
+        }
+
+        registrationTransaction.executeWithoutResult {
+            // 언어 코드가 지정되지 않은 경우, 파파고 언어 감지 API를 통해 비동기로 언어를 식별한다.
+            if (audioContent.languageCode.isNullOrBlank()) {
+                val papagoQuery = listOf(
+                    request.title.trim(),
+                    request.detail.trim(),
+                    request.tags.trim()
                 )
-            )
+                    .filter { it.isNotBlank() }
+                    .joinToString(" ")
+
+                applicationEventPublisher.publishEvent(
+                    LanguageDetectEvent(
+                        id = audioContent.id!!,
+                        query = papagoQuery
+                    )
+                )
+            } else {
+                applicationEventPublisher.publishEvent(
+                    LanguageTranslationEvent(
+                        id = audioContent.id!!,
+                        targetType = LanguageTranslationTargetType.CONTENT
+                    )
+                )
+            }
         }
 
         return CreateAudioContentResponse(contentId = audioContent.id!!)
@@ -481,8 +501,9 @@ class AudioContentService(
 
     @Transactional
     fun uploadComplete(contentId: Long, content: String, duration: String) {
-        val keyFileName = content.split("/").last()
-        if (!keyFileName.startsWith(contentId.toString())) throw SodaException(messageKey = "common.error.invalid_request")
+        val registeredOutputMatches = subtitleJobService.matchesOutput(contentId, content)
+        val validOutput = registeredOutputMatches ?: content.substringAfterLast("/").startsWith(contentId.toString())
+        if (!validOutput) throw SodaException(messageKey = "common.error.invalid_request")
 
         val audioContent = repository.findByIdOrNull(contentId)
             ?: throw SodaException(messageKey = "common.error.invalid_request")
@@ -978,7 +999,19 @@ class AudioContentService(
             nextContent = nextContent,
             buyerList = buyerList,
             isAvailableUsePoint = audioContent.isPointAvailable && audioContent.price > 0,
-            translated = translated
+            translated = translated,
+            subtitle = if (
+                (isExistsAudioContent || creatorId == member.id || audioContent.price <= 0) &&
+                (audioContent.releaseDate == null || audioContent.releaseDate!! <= LocalDateTime.now() || creatorId == member.id)
+            ) {
+                subtitleJobService.subtitle(
+                    audioContent,
+                    langContext.lang,
+                    1000 * 60 * 60 * ((audioContent.duration ?: "00:00:00").split(":")[0].toLong() + 2)
+                )
+            } else {
+                null
+            }
         )
     }
 
@@ -1302,16 +1335,25 @@ class AudioContentService(
             contentId = audioContent.id!!
         )
 
-        val contentUrl = if (isExistsAudioContent) {
-            audioContentCloudFront.generateSignedURL(
-                resourcePath = audioContent.content!!,
-                expirationTime = 1000 * 60 * 60 * (audioContent.duration!!.split(":")[0].toLong() + 2)
+        return if (isExistsAudioContent) {
+            val expirationTime = 1000 * 60 * 60 * (audioContent.duration!!.split(":")[0].toLong() + 2)
+            GenerateUrlResponse(
+                contentUrl = audioContentCloudFront.generateSignedURL(
+                    resourcePath = audioContent.content!!,
+                    expirationTime = expirationTime
+                ),
+                subtitle = if (
+                    audioContent.releaseDate == null || audioContent.releaseDate!! <= LocalDateTime.now() ||
+                    audioContent.member!!.id == member.id
+                ) {
+                    subtitleJobService.subtitle(audioContent, langContext.lang, expirationTime)
+                } else {
+                    null
+                }
             )
         } else {
-            ""
+            GenerateUrlResponse("")
         }
-
-        return GenerateUrlResponse(contentUrl)
     }
 
     @Transactional(readOnly = true)
diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/content/GenerateUrlResponse.kt b/src/main/kotlin/kr/co/vividnext/sodalive/content/GenerateUrlResponse.kt
index 9a51cfea..381e2dba 100644
--- a/src/main/kotlin/kr/co/vividnext/sodalive/content/GenerateUrlResponse.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/content/GenerateUrlResponse.kt
@@ -1,5 +1,8 @@
 package kr.co.vividnext.sodalive.content
 
+import kr.co.vividnext.sodalive.v2.subtitle.dto.ContentSubtitleResponse
+
 data class GenerateUrlResponse(
-    val contentUrl: String
+    val contentUrl: String,
+    val subtitle: ContentSubtitleResponse? = null
 )
diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/content/GetAudioContentDetailResponse.kt b/src/main/kotlin/kr/co/vividnext/sodalive/content/GetAudioContentDetailResponse.kt
index 69f83294..8dba2e79 100644
--- a/src/main/kotlin/kr/co/vividnext/sodalive/content/GetAudioContentDetailResponse.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/content/GetAudioContentDetailResponse.kt
@@ -4,6 +4,7 @@ import com.querydsl.core.annotations.QueryProjection
 import kr.co.vividnext.sodalive.content.comment.GetAudioContentCommentListItem
 import kr.co.vividnext.sodalive.content.order.OrderType
 import kr.co.vividnext.sodalive.content.translation.TranslatedContent
+import kr.co.vividnext.sodalive.v2.subtitle.dto.ContentSubtitleResponse
 
 data class GetAudioContentDetailResponse(
     val contentId: Long,
@@ -42,7 +43,8 @@ data class GetAudioContentDetailResponse(
     val nextContent: OtherContentResponse?,
     val buyerList: List<ContentBuyer>,
     val isAvailableUsePoint: Boolean,
-    val translated: TranslatedContent?
+    val translated: TranslatedContent?,
+    val subtitle: ContentSubtitleResponse? = null
 )
 
 data class OtherContentResponse @QueryProjection constructor(
diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentFacade.kt b/src/main/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentFacade.kt
index e4fbe1b2..30221531 100644
--- a/src/main/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentFacade.kt
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentFacade.kt
@@ -19,6 +19,7 @@ import kr.co.vividnext.sodalive.v2.api.admin.aicharacter.error.AiCharacterAdminA
 import org.springframework.data.domain.PageRequest
 import org.springframework.http.HttpStatus
 import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Propagation
 import org.springframework.transaction.annotation.Transactional
 import org.springframework.web.multipart.MultipartFile
 import java.time.DateTimeException
@@ -158,7 +159,7 @@ class AiCharacterAdminAudioContentFacade(
         )
     }
 
-    @Transactional
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     fun create(
         characterId: Long,
         coverImage: MultipartFile,
diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleExceptionHandler.kt b/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleExceptionHandler.kt
new file mode 100644
index 00000000..5dc5b0e2
--- /dev/null
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleExceptionHandler.kt
@@ -0,0 +1,25 @@
+package kr.co.vividnext.sodalive.v2.subtitle.adapter.web
+
+import org.springframework.core.Ordered
+import org.springframework.core.annotation.Order
+import org.springframework.http.ResponseEntity
+import org.springframework.http.converter.HttpMessageNotReadableException
+import org.springframework.web.bind.annotation.ExceptionHandler
+import org.springframework.web.bind.annotation.RestControllerAdvice
+import org.springframework.web.server.ResponseStatusException
+
+@Order(Ordered.HIGHEST_PRECEDENCE)
+@RestControllerAdvice(assignableTypes = [SubtitleJobController::class])
+class SubtitleExceptionHandler {
+    @ExceptionHandler(ResponseStatusException::class)
+    fun handle(exception: ResponseStatusException): ResponseEntity<Map<String, String>> =
+        ResponseEntity.status(exception.status).body(mapOf("error" to "Subtitle request rejected"))
+
+    @ExceptionHandler(HttpMessageNotReadableException::class)
+    fun handleUnreadableBody(): ResponseEntity<Map<String, String>> =
+        ResponseEntity.unprocessableEntity().body(mapOf("error" to "Subtitle request rejected"))
+
+    @ExceptionHandler(Exception::class)
+    fun handleUnexpectedException(): ResponseEntity<Map<String, String>> =
+        ResponseEntity.internalServerError().body(mapOf("error" to "Subtitle request failed"))
+}
diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleJobController.kt b/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleJobController.kt
new file mode 100644
index 00000000..a4908eb8
--- /dev/null
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleJobController.kt
@@ -0,0 +1,32 @@
+package kr.co.vividnext.sodalive.v2.subtitle.adapter.web
+
+import kr.co.vividnext.sodalive.v2.subtitle.application.SubtitleJobService
+import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.PostMapping
+import org.springframework.web.bind.annotation.RequestBody
+import org.springframework.web.bind.annotation.RequestMapping
+import org.springframework.web.bind.annotation.RequestParam
+import org.springframework.web.bind.annotation.RestController
+
+@RestController
+@RequestMapping("/internal/api/subtitle-jobs")
+class SubtitleJobController(private val service: SubtitleJobService, private val parser: SubtitleRequestParser) {
+    @GetMapping("/resolve")
+    fun resolve(@RequestParam bucket: String, @RequestParam key: String) = service.resolve(bucket, key)
+
+    @GetMapping("/{jobId}")
+    fun get(@PathVariable jobId: String) = service.get(jobId)
+
+    @PostMapping("/{jobId}/started", consumes = ["application/json"])
+    fun started(@PathVariable jobId: String, @RequestBody body: ByteArray): Map<String, String> {
+        service.started(jobId, parser.started(body))
+        return emptyMap()
+    }
+
+    @PostMapping("/callback", consumes = ["application/json"])
+    fun callback(@RequestBody body: ByteArray): Map<String, String> {
+        service.callback(parser.callback(body), body)
+        return emptyMap()
+    }
+}
diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleRequestParser.kt b/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleRequestParser.kt
new file mode 100644
index 00000000..4125f55b
--- /dev/null
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/adapter/web/SubtitleRequestParser.kt
@@ -0,0 +1,72 @@
+package kr.co.vividnext.sodalive.v2.subtitle.adapter.web
+
+import com.fasterxml.jackson.core.JsonParser
+import com.fasterxml.jackson.core.JsonProcessingException
+import com.fasterxml.jackson.databind.DeserializationFeature
+import com.fasterxml.jackson.databind.JsonNode
+import com.fasterxml.jackson.databind.ObjectMapper
+import kr.co.vividnext.sodalive.v2.subtitle.dto.SubtitleCallbackRequest
+import kr.co.vividnext.sodalive.v2.subtitle.dto.SubtitleResultRequest
+import kr.co.vividnext.sodalive.v2.subtitle.dto.SubtitleStartedRequest
+import org.springframework.http.HttpStatus
+import org.springframework.stereotype.Component
+import org.springframework.web.server.ResponseStatusException
+
+@Component
+class SubtitleRequestParser(objectMapper: ObjectMapper) {
+    private val mapper = objectMapper.copy()
+        .enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION)
+        .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+
+    fun started(body: ByteArray): SubtitleStartedRequest {
+        val node = parse(body)
+        fields(node, setOf("transcribeJobName", "status"))
+        return SubtitleStartedRequest(text(node, "transcribeJobName"), text(node, "status"))
+    }
+
+    fun callback(body: ByteArray): SubtitleCallbackRequest {
+        val node = parse(body)
+        fields(
+            node,
+            setOf("eventId", "jobId", "contentId", "transcribeJobName", "status", "subtitles", "errorCode", "errorMessage")
+        )
+        val contentId = node.path("contentId")
+        if (!contentId.isIntegralNumber || !contentId.canConvertToLong() || contentId.longValue() <= 0) throw invalid()
+        val subtitles = node.path("subtitles")
+        if (!subtitles.isArray || subtitles.size() > 2) throw invalid()
+        return SubtitleCallbackRequest(
+            eventId = text(node, "eventId"),
+            jobId = text(node, "jobId"),
+            contentId = contentId.longValue(),
+            transcribeJobName = text(node, "transcribeJobName"),
+            status = text(node, "status"),
+            subtitles = subtitles.map {
+                fields(it, setOf("language", "format", "s3Bucket", "s3Key"))
+                SubtitleResultRequest(text(it, "language"), text(it, "format"), text(it, "s3Bucket"), text(it, "s3Key"))
+            },
+            errorCode = nullableText(node, "errorCode"),
+            errorMessage = nullableText(node, "errorMessage")
+        )
+    }
+
+    private fun parse(body: ByteArray): JsonNode = try {
+        mapper.readTree(body) ?: throw invalid()
+    } catch (_: JsonProcessingException) {
+        throw invalid()
+    }
+
+    private fun fields(node: JsonNode, expected: Set<String>) {
+        if (!node.isObject || node.fieldNames().asSequence().toSet() != expected) throw invalid()
+    }
+
+    private fun text(node: JsonNode, field: String): String {
+        val value = node.path(field)
+        if (!value.isTextual || value.textValue().isBlank()) throw invalid()
+        return value.textValue()
+    }
+
+    private fun nullableText(node: JsonNode, field: String): String? =
+        if (node.path(field).isNull) null else text(node, field)
+
+    private fun invalid() = ResponseStatusException(HttpStatus.UNPROCESSABLE_ENTITY, "Invalid subtitle request")
+}
diff --git a/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/application/SubtitleJobService.kt b/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/application/SubtitleJobService.kt
new file mode 100644
index 00000000..c12255b6
--- /dev/null
+++ b/src/main/kotlin/kr/co/vividnext/sodalive/v2/subtitle/application/SubtitleJobService.kt
@@ -0,0 +1,219 @@
+package kr.co.vividnext.sodalive.v2.subtitle.application
+
+import kr.co.vividnext.sodalive.aws.cloudfront.AudioContentCloudFront
+import kr.co.vividnext.sodalive.content.AudioContent
+import kr.co.vividnext.sodalive.i18n.Lang
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.Subtitle
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleCallbackReceipt
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleCallbackReceiptRepository
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleJob
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleJobRepository
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleRepository
+import kr.co.vividnext.sodalive.v2.subtitle.domain.SubtitleJobStatus
+import kr.co.vividnext.sodalive.v2.subtitle.dto.ContentSubtitleResponse
+import kr.co.vividnext.sodalive.v2.subtitle.dto.SubtitleCallbackRequest
+import kr.co.vividnext.sodalive.v2.subtitle.dto.SubtitleJobResponse
+import kr.co.vividnext.sodalive.v2.subtitle.dto.SubtitleResultRequest
+import kr.co.vividnext.sodalive.v2.subtitle.dto.SubtitleSourceResponse
+import kr.co.vividnext.sodalive.v2.subtitle.dto.SubtitleStartedRequest
+import org.springframework.beans.factory.annotation.Value
+import org.springframework.dao.DataIntegrityViolationException
+import org.springframework.data.repository.findByIdOrNull
+import org.springframework.http.HttpStatus
+import org.springframework.stereotype.Service
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.TransactionDefinition
+import org.springframework.transaction.annotation.Propagation
+import org.springframework.transaction.annotation.Transactional
+import org.springframework.transaction.support.TransactionTemplate
+import org.springframework.web.server.ResponseStatusException
+import java.security.MessageDigest
+import java.time.LocalDateTime
+import java.time.ZoneOffset
+import java.time.temporal.ChronoUnit
+import java.util.Locale
+import java.util.UUID
+
+@Service
+class SubtitleJobService(
+    private val jobs: SubtitleJobRepository,
+    private val receipts: SubtitleCallbackReceiptRepository,
+    private val subtitles: SubtitleRepository,
+    private val cloudFront: AudioContentCloudFront,
+    transactionManager: PlatformTransactionManager,
+    @Value("\${cloud.aws.s3.content-bucket}") private val bucket: String
+) {
+    private val transaction = TransactionTemplate(transactionManager).apply {
+        propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW
+        isolationLevel = TransactionDefinition.ISOLATION_READ_COMMITTED
+    }
+
+    @Transactional(propagation = Propagation.MANDATORY)
+    fun register(content: AudioContent, originalFilename: String?): SubtitleJob {
+        val contentId = requireNotNull(content.id)
+        val sourceName = UUID.randomUUID().toString()
+        val extension = originalFilename?.replace('\\', '/')?.substringAfterLast('/')
+            ?.substringAfterLast('.', "")?.takeIf { it.matches(Regex("[A-Za-z0-9]{1,16}")) }
+            ?.lowercase(Locale.ROOT)
+        val sourceFileName = if (extension == null) sourceName else "$sourceName.$extension"
+        val job = jobs.save(
+            SubtitleJob(
+                jobId = UUID.randomUUID().toString(),
+                contentId = contentId,
+                sourceBucket = bucket,
+                sourceKey = "input/$contentId/$sourceFileName",
+                outputKey = "output/$contentId/$sourceName"
+            )
+        )
+        content.latestSubtitleJobId = job.jobId
+        content.content = job.sourceKey
+        return job
+    }
+
+    @Transactional(readOnly = true)
+    fun resolve(bucket: String, key: String): SubtitleJobResponse =
+        response(jobs.findBySourceBucketAndSourceKey(bucket, key) ?: throw missing())
+
+    @Transactional(readOnly = true)
+    fun get(jobId: String): SubtitleJobResponse = response(jobs.findByIdOrNull(jobId) ?: throw missing())
+
+    fun started(jobId: String, request: SubtitleStartedRequest) {
+        if (request.transcribeJobName != "subtitle-$jobId" || request.status != "PROCESSING") throw invalid()
+        transaction.executeWithoutResult {
+            val job = jobs.findForUpdate(jobId) ?: throw missing()
+            if (job.status == SubtitleJobStatus.PENDING) {
+                job.status = SubtitleJobStatus.PROCESSING
+                job.updatedAt = now()
+            }
+        }
+    }
+
+    fun callback(request: SubtitleCallbackRequest, rawBody: ByteArray) {
+        if (request.eventId.isBlank() || request.eventId.length > 255) throw invalid()
+        val hash = MessageDigest.getInstance("SHA-256").digest(rawBody).joinToString("") { "%02x".format(it) }
+        try {
+            transaction.executeWithoutResult {
+                val existing = receipts.findByIdOrNull(request.eventId)
+                if (existing != null) {
+                    replay(existing, hash)
+                    return@executeWithoutResult
+                }
+                val job = jobs.findForUpdate(request.jobId) ?: throw missing()
+                val afterLock = receipts.findByIdOrNull(request.eventId)
+                if (afterLock != null) {
+                    replay(afterLock, hash)
+                    return@executeWithoutResult
+                }
+                val status = validate(job, request)
+                val existingResults = subtitles.findAllByJobId(job.jobId).map {
+                    SubtitleResultRequest(it.language, it.format, it.s3Bucket, it.s3Key)
+                }.sortedBy { it.language }
+                if (job.status.terminal && (
+                    job.status != status || existingResults != request.subtitles.sortedBy { it.language } ||
+                        job.errorCode != request.errorCode || job.errorMessage != request.errorMessage
+                    )
+                ) {
+                    throw ResponseStatusException(HttpStatus.CONFLICT, "Subtitle job already finished")
+                }
+
+                if (!job.status.terminal) {
+                    subtitles.saveAll(
+                        request.subtitles.map {
+                            Subtitle(job.jobId, it.language, it.format, it.s3Bucket, it.s3Key)
+                        }
+                    )
+                    job.status = status
+                    job.errorCode = request.errorCode
+                    job.errorMessage = request.errorMessage
+                    job.updatedAt = now()
+                }
+                receipts.saveAndFlush(SubtitleCallbackReceipt(request.eventId, job.jobId, hash))
+            }
+        } catch (exception: DataIntegrityViolationException) {
+            transaction.executeWithoutResult {
+                val receipt = receipts.findByIdOrNull(request.eventId) ?: throw exception
+                replay(receipt, hash)
+            }
+        }
+    }
+
+    fun recordUploadError(jobId: String) {
+        transaction.executeWithoutResult {
+            val job = jobs.findForUpdate(jobId) ?: throw missing()
+            val recordedAt = now()
+            job.sourceUploadErrorCode = "SOURCE_UPLOAD_UNCERTAIN"
+            job.sourceUploadErrorAt = recordedAt
+            job.updatedAt = recordedAt
+        }
+    }
+
+    @Transactional(readOnly = true)
+    fun matchesOutput(contentId: Long, outputKey: String): Boolean? =
+        jobs.findFirstByContentIdOrderByCreatedAtAsc(contentId)?.let { job ->
+            outputKey.substringBeforeLast('.', "") == job.outputKey &&
+                outputKey.substringAfterLast('.', "") in setOf("aac", "m4a", "mp3")
+        }
+
+    @Transactional(readOnly = true)
+    fun subtitle(content: AudioContent, lang: Lang, expirationTime: Long): ContentSubtitleResponse? {
+        val jobId = content.latestSubtitleJobId ?: return null
+        val job = jobs.findByIdOrNull(jobId) ?: return null
+        if (job.contentId != content.id || !job.status.terminal || job.status == SubtitleJobStatus.FAILED) return null
+        val language = when (lang) {
+            Lang.KO -> "ko-KR"
+            Lang.JA -> "ja-JP"
+            else -> return null
+        }
+        val subtitle = subtitles.findByJobIdAndLanguage(jobId, language) ?: return null
+        return ContentSubtitleResponse(
+            subtitle.language,
+            subtitle.format,
+            cloudFront.generateSignedURL(subtitle.s3Key, expirationTime)
+        )
+    }
+
+    private fun validate(job: SubtitleJob, request: SubtitleCallbackRequest): SubtitleJobStatus {
+        if (request.contentId != job.contentId || request.transcribeJobName != "subtitle-${job.jobId}") throw invalid()
+        if (request.subtitles.map { it.language }.distinct().size != request.subtitles.size) throw invalid()
+        request.subtitles.forEach {
+            val language = when (it.language) {
+                "ko-KR" -> "ko"
+                "ja-JP" -> "ja"
+                else -> throw invalid()
+            }
+            if (it.format != "SRT" || it.s3Bucket != bucket ||
+                it.s3Key != "subtitles/${job.contentId}/${job.jobId}/$language.srt"
+            ) {
+                throw invalid()
+            }
+        }
+        val status = when (request.subtitles.size) {
+            2 -> SubtitleJobStatus.COMPLETED
+            1 -> SubtitleJobStatus.PARTIAL_COMPLETED
+            0 -> SubtitleJobStatus.FAILED
+            else -> throw invalid()
+        }
+        if (request.status != status.name) throw invalid()
+        if (status == SubtitleJobStatus.COMPLETED) {
+            if (request.errorCode != null || request.errorMessage != null) throw invalid()
+        } else if (request.errorCode.isNullOrBlank() || request.errorMessage.isNullOrBlank() || request.errorCode.length > 255) {
+            throw invalid()
+        }
+        return status
+    }
+
+    private fun replay(receipt: SubtitleCallbackReceipt, hash: String) {
+        if (receipt.bodyHash != hash) throw invalid()
+    }
+
+    private fun response(job: SubtitleJob) = SubtitleJobResponse(
+        job.jobId,
+        job.contentId,
+        job.status,
+        SubtitleSourceResponse(job.sourceBucket, job.sourceKey)
+    )
+
+    private fun now() = LocalDateTime.now(ZoneOffset.UTC).truncatedTo(ChronoUnit.SECONDS)
+    private fun missing() = ResponseStatusException(HttpStatus.NOT_FOUND, "Subtitle job not found")
+    private fun invalid() = ResponseStatusException(HttpStatus.UNPROCESSABLE_ENTITY, "Invalid subtitle request")
+}
diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/content/AudioContentServiceTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/content/AudioContentServiceTest.kt
index b04bc186..5704bf6f 100644
--- a/src/test/kotlin/kr/co/vividnext/sodalive/content/AudioContentServiceTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/content/AudioContentServiceTest.kt
@@ -27,6 +27,7 @@ import kr.co.vividnext.sodalive.member.contentpreference.MemberContentPreference
 import kr.co.vividnext.sodalive.support.originalLocalizedNameService
 import kr.co.vividnext.sodalive.v2.home.following.application.HomeFollowingNewsPublishService
 import kr.co.vividnext.sodalive.v2.recommendation.application.CreatorContentViewHistoryService
+import kr.co.vividnext.sodalive.v2.subtitle.application.SubtitleJobService
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertFalse
 import org.junit.jupiter.api.Assertions.assertNull
@@ -40,6 +41,7 @@ import org.mockito.Mockito
 import org.springframework.boot.test.system.CapturedOutput
 import org.springframework.boot.test.system.OutputCaptureExtension
 import org.springframework.context.ApplicationEventPublisher
+import org.springframework.transaction.PlatformTransactionManager
 import java.time.LocalDateTime
 import java.util.Optional
 
@@ -66,6 +68,7 @@ class AudioContentServiceTest {
     private lateinit var contentThemeTranslationRepository: ContentThemeTranslationRepository
     private lateinit var homeFollowingNewsPublishService: HomeFollowingNewsPublishService
     private lateinit var memberContentPreferenceService: MemberContentPreferenceService
+    private lateinit var subtitleJobService: SubtitleJobService
 
     private lateinit var service: AudioContentService
 
@@ -92,6 +95,8 @@ class AudioContentServiceTest {
         contentThemeTranslationRepository = Mockito.mock(ContentThemeTranslationRepository::class.java)
         homeFollowingNewsPublishService = Mockito.mock(HomeFollowingNewsPublishService::class.java)
         memberContentPreferenceService = Mockito.mock(MemberContentPreferenceService::class.java)
+        subtitleJobService = Mockito.mock(SubtitleJobService::class.java)
+        Mockito.`when`(subtitleJobService.matchesOutput(Mockito.anyLong(), Mockito.anyString())).thenReturn(null)
 
         service = AudioContentService(
             repository = repository,
@@ -119,6 +124,8 @@ class AudioContentServiceTest {
             contentThemeTranslationRepository = contentThemeTranslationRepository,
             homeFollowingNewsPublishService = homeFollowingNewsPublishService,
             memberContentPreferenceService = memberContentPreferenceService,
+            subtitleJobService = subtitleJobService,
+            transactionManager = Mockito.mock(PlatformTransactionManager::class.java),
             audioContentBucket = "audio-bucket",
             coverImageBucket = "cover-bucket",
             coverImageHost = "https://cdn.test"
diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentControllerTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentControllerTest.kt
index 7b146d31..4193c4fd 100644
--- a/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentControllerTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentControllerTest.kt
@@ -50,7 +50,10 @@ import org.springframework.test.web.servlet.result.MockMvcResultMatchers.content
 import org.springframework.test.web.servlet.result.MockMvcResultMatchers.header
 import org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath
 import org.springframework.test.web.servlet.result.MockMvcResultMatchers.status
+import org.springframework.transaction.TransactionDefinition
+import org.springframework.transaction.annotation.Propagation
 import org.springframework.transaction.annotation.Transactional
+import org.springframework.transaction.support.TransactionTemplate
 import java.net.URL
 import java.time.Instant
 import java.time.LocalDateTime
@@ -66,6 +69,7 @@ class AiCharacterAdminAudioContentControllerTest @Autowired constructor(
     private val chatCharacterService: ChatCharacterService,
     private val audioContentService: AudioContentService,
     private val creatorAdminContentService: CreatorAdminContentService,
+    private val transactionTemplate: TransactionTemplate,
     private val entityManager: EntityManager
 ) {
     @MockBean
@@ -79,6 +83,7 @@ class AiCharacterAdminAudioContentControllerTest @Autowired constructor(
 
     private lateinit var originalAudioContentServicePublisher: ApplicationEventPublisher
     private lateinit var originalCreatorAdminContentServicePublisher: ApplicationEventPublisher
+    private var originalContentIds: Set<Long> = emptySet()
 
     @BeforeEach
     fun replaceActualServicePublishers() {
@@ -88,15 +93,46 @@ class AiCharacterAdminAudioContentControllerTest @Autowired constructor(
             creatorAdminContentService,
             applicationEventPublisher
         )
+        originalContentIds = entityManager.createQuery("select c.id from AudioContent c", java.lang.Long::class.java)
+            .resultList.map { it.toLong() }.toSet()
     }
 
     @AfterEach
     fun restoreActualServicePublishers() {
         replacePublisher(audioContentService, originalAudioContentServicePublisher)
         replacePublisher(creatorAdminContentService, originalCreatorAdminContentServicePublisher)
+        deleteCommittedAudioContents()
+    }
+
+    private fun deleteCommittedAudioContents() {
+        TransactionTemplate(transactionTemplate.transactionManager!!).apply {
+            propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW
+        }.executeWithoutResult {
+            val created = entityManager.createQuery("select c from AudioContent c", AudioContent::class.java)
+                .resultList.filter { it.id !in originalContentIds }
+            created.forEach { content ->
+                content.latestSubtitleJobId = null
+                entityManager.flush()
+                val jobIds = entityManager.createQuery(
+                    "select j.jobId from SubtitleJob j where j.contentId = :id",
+                    String::class.java
+                )
+                    .setParameter("id", content.id).resultList
+                if (jobIds.isNotEmpty()) {
+                    entityManager.createQuery("delete from Subtitle where jobId in :ids")
+                        .setParameter("ids", jobIds).executeUpdate()
+                    entityManager.createQuery("delete from SubtitleCallbackReceipt where jobId in :ids")
+                        .setParameter("ids", jobIds).executeUpdate()
+                    entityManager.createQuery("delete from SubtitleJob where jobId in :ids")
+                        .setParameter("ids", jobIds).executeUpdate()
+                }
+                entityManager.remove(content)
+            }
+        }
     }
 
     @Test
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     @DisplayName("생성은 레거시 multipart request로 target 소유 콘텐츠 ID를 반환한다")
     fun shouldCreateOwnedContentWithUtcMultipartRequest() {
         val character = chatCharacterService.createChatCharacterWithDetails(
@@ -105,9 +141,12 @@ class AiCharacterAdminAudioContentControllerTest @Autowired constructor(
             description = "description",
             systemPrompt = "prompt"
         )
-        val theme = AudioContentTheme(theme = "create-theme", image = "theme.png")
-        entityManager.persist(theme)
-        entityManager.flush()
+        val theme = transactionTemplate.execute {
+            val newTheme = AudioContentTheme(theme = "create-theme", image = "theme.png")
+            entityManager.persist(newTheme)
+            entityManager.flush()
+            newTheme
+        }!!
         Mockito.`when`(amazonS3Client.getUrl(Mockito.anyString(), Mockito.anyString()))
             .thenReturn(URL("https://s3.example.com/test"))
 
@@ -1006,12 +1045,14 @@ class AiCharacterAdminAudioContentControllerTest @Autowired constructor(
     }
 
     private fun registerMysqlDateFunctions() {
-        entityManager.createNativeQuery(
-            "CREATE ALIAS IF NOT EXISTS DATE_FORMAT FOR 'kr.co.vividnext.sodalive.support.H2MysqlDateFunctions.dateFormat'"
-        ).executeUpdate()
-        entityManager.createNativeQuery(
-            "CREATE ALIAS IF NOT EXISTS CONVERT_TZ FOR 'kr.co.vividnext.sodalive.support.H2MysqlDateFunctions.convertTz'"
-        ).executeUpdate()
+        transactionTemplate.executeWithoutResult {
+            entityManager.createNativeQuery(
+                "CREATE ALIAS IF NOT EXISTS DATE_FORMAT FOR 'kr.co.vividnext.sodalive.support.H2MysqlDateFunctions.dateFormat'"
+            ).executeUpdate()
+            entityManager.createNativeQuery(
+                "CREATE ALIAS IF NOT EXISTS CONVERT_TZ FOR 'kr.co.vividnext.sodalive.support.H2MysqlDateFunctions.convertTz'"
+            ).executeUpdate()
+        }
     }
 
     private fun adminAuthentication() = authentication(
diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentCreateTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentCreateTest.kt
index ee40bf3e..c2024f85 100644
--- a/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentCreateTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentCreateTest.kt
@@ -14,6 +14,7 @@ import kr.co.vividnext.sodalive.member.MemberRole
 import kr.co.vividnext.sodalive.support.EmbeddedRedisInitializer
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.Assertions.assertNull
 import org.junit.jupiter.api.Assertions.assertSame
 import org.junit.jupiter.api.Assertions.assertTrue
@@ -46,6 +47,7 @@ import org.springframework.test.web.servlet.request.MockMvcRequestBuilders.multi
 import org.springframework.test.web.servlet.result.MockMvcResultMatchers.header
 import org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath
 import org.springframework.test.web.servlet.result.MockMvcResultMatchers.status
+import org.springframework.transaction.TransactionDefinition
 import org.springframework.transaction.annotation.Propagation
 import org.springframework.transaction.annotation.Transactional
 import org.springframework.transaction.support.TransactionTemplate
@@ -72,15 +74,42 @@ class AiCharacterAdminAudioContentCreateTest @Autowired constructor(
     private lateinit var applicationEventPublisher: ApplicationEventPublisher
 
     private lateinit var originalAudioContentServicePublisher: ApplicationEventPublisher
+    private var originalContentIds: Set<Long> = emptySet()
 
     @BeforeEach
     fun replaceActualServicePublisher() {
         originalAudioContentServicePublisher = replaceAudioContentServicePublisher(applicationEventPublisher)
+        originalContentIds = entityManager.createQuery("select c.id from AudioContent c", java.lang.Long::class.java)
+            .resultList.map { it.toLong() }.toSet()
     }
 
     @AfterEach
     fun restoreActualServicePublisher() {
         replaceAudioContentServicePublisher(originalAudioContentServicePublisher)
+        TransactionTemplate(transactionTemplate.transactionManager!!).apply {
+            propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW
+        }.executeWithoutResult {
+            val created = entityManager.createQuery("select c from AudioContent c", AudioContent::class.java)
+                .resultList.filter { it.id !in originalContentIds }
+            created.forEach { content ->
+                content.latestSubtitleJobId = null
+                entityManager.flush()
+                val jobIds = entityManager.createQuery(
+                    "select j.jobId from SubtitleJob j where j.contentId = :id",
+                    String::class.java
+                )
+                    .setParameter("id", content.id).resultList
+                if (jobIds.isNotEmpty()) {
+                    entityManager.createQuery("delete from Subtitle where jobId in :ids")
+                        .setParameter("ids", jobIds).executeUpdate()
+                    entityManager.createQuery("delete from SubtitleCallbackReceipt where jobId in :ids")
+                        .setParameter("ids", jobIds).executeUpdate()
+                    entityManager.createQuery("delete from SubtitleJob where jobId in :ids")
+                        .setParameter("ids", jobIds).executeUpdate()
+                }
+                entityManager.remove(content)
+            }
+        }
     }
 
     @ParameterizedTest
@@ -319,6 +348,7 @@ class AiCharacterAdminAudioContentCreateTest @Autowired constructor(
     }
 
     @Test
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     @DisplayName("생성은 존재하지 않는 themeId를 업로드 전 invalid request로 거부한다")
     fun shouldRejectMissingThemeBeforeUpload() {
         val character = chatCharacterService.createChatCharacterWithDetails(
@@ -360,6 +390,7 @@ class AiCharacterAdminAudioContentCreateTest @Autowired constructor(
         ],
         delimiter = '|'
     )
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     @DisplayName("생성은 잘못된 preview 시간을 업로드 전 기존 오류로 거부한다")
     fun shouldRejectInvalidPreviewTimeBeforeUpload(previewCase: String, language: String, message: String) {
         val character = chatCharacterService.createChatCharacterWithDetails(
@@ -539,8 +570,8 @@ class AiCharacterAdminAudioContentCreateTest @Autowired constructor(
 
     @Test
     @Transactional(propagation = Propagation.NOT_SUPPORTED)
-    @DisplayName("생성은 audio 업로드 실패 시 DB와 이벤트를 남기지 않고 cover 업로드 호출만 남긴다")
-    fun shouldRollBackDatabaseAndEventWhenAudioUploadFails() {
+    @DisplayName("audio 업로드 실패는 이미 commit된 등록을 보존하고 업로드 오류를 기록하며 이벤트는 발행하지 않는다")
+    fun shouldKeepCommittedRegistrationAndRecordErrorWhenAudioUploadFails() {
         val character = chatCharacterService.createChatCharacterWithDetails(
             characterUUID = "v2-audio-create-audio-failure-character",
             name = "v2-audio-create-audio-failure-character",
@@ -568,13 +599,21 @@ class AiCharacterAdminAudioContentCreateTest @Autowired constructor(
 
         Mockito.verify(amazonS3Client, Mockito.times(2)).putObject(Mockito.any(PutObjectRequest::class.java))
         Mockito.verifyNoInteractions(applicationEventPublisher)
-        assertNull(findAudioContentByTitle("audio"))
+        val content = findAudioContentByTitle("audio")!!
+        assertNotNull(content.latestSubtitleJobId)
+        assertEquals(
+            "SOURCE_UPLOAD_UNCERTAIN",
+            entityManager.createQuery(
+                "select j.sourceUploadErrorCode from SubtitleJob j where j.jobId = :id",
+                String::class.java
+            ).setParameter("id", content.latestSubtitleJobId).singleResult
+        )
     }
 
     @Test
     @Transactional(propagation = Propagation.NOT_SUPPORTED)
-    @DisplayName("생성은 이벤트 발행 실패 시 DB를 롤백하고 두 S3 업로드 호출만 남긴다")
-    fun shouldRollBackDatabaseWhenCreateEventPublishFails() {
+    @DisplayName("생성 후 이벤트 실패는 이미 commit된 원본/job binding을 롤백하지 않는다")
+    fun shouldKeepCommittedRegistrationWhenCreateEventPublishFails() {
         val character = chatCharacterService.createChatCharacterWithDetails(
             characterUUID = "v2-audio-create-event-failure-character",
             name = "v2-audio-create-event-failure-character",
@@ -607,10 +646,11 @@ class AiCharacterAdminAudioContentCreateTest @Autowired constructor(
 
         Mockito.verify(amazonS3Client, Mockito.times(2)).putObject(Mockito.any(PutObjectRequest::class.java))
         Mockito.verify(applicationEventPublisher).publishEvent(Mockito.any(Any::class.java))
-        assertNull(findAudioContentByTitle("audio"))
+        assertNotNull(findAudioContentByTitle("audio")!!.latestSubtitleJobId)
     }
 
     @Test
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     @DisplayName("생성은 timezone 없이 UTC releaseDate를 저장하고 data.contentId만 반환한다")
     fun shouldCreateWithUtcReleaseDateAndReturnContentId() {
         val character = chatCharacterService.createChatCharacterWithDetails(
@@ -660,6 +700,7 @@ class AiCharacterAdminAudioContentCreateTest @Autowired constructor(
     }
 
     @ParameterizedTest
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     @ValueSource(strings = ["", ",\"releaseDate\":null"])
     @DisplayName("생성은 생략 또는 null releaseDate를 허용한다")
     fun shouldAllowOmittedOrNullReleaseDate(releaseDateField: String) {
@@ -693,12 +734,12 @@ class AiCharacterAdminAudioContentCreateTest @Autowired constructor(
             .andExpect(jsonPath("$.data.contentId").isNumber)
     }
 
-    private fun saveTheme(): AudioContentTheme {
+    private fun saveTheme(): AudioContentTheme = transactionTemplate.execute {
         val theme = AudioContentTheme(theme = "create-test-theme", image = "theme.png")
         entityManager.persist(theme)
         entityManager.flush()
-        return theme
-    }
+        theme
+    }!!
 
     private fun requestPart(
         themeId: Long,
diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentQueryTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentQueryTest.kt
index 35645d33..117a9357 100644
--- a/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentQueryTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/AiCharacterAdminAudioContentQueryTest.kt
@@ -182,7 +182,8 @@ class AiCharacterAdminAudioContentQueryTest @Autowired constructor(
                 "nextContent",
                 "buyerList",
                 "isAvailableUsePoint",
-                "translated"
+                "translated",
+                "subtitle"
             ),
             data.fieldNames().asSequence().toSet()
         )
diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/LegacyCreatorAdminAudioContentCharacterizationTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/LegacyCreatorAdminAudioContentCharacterizationTest.kt
index 69b2af34..7a72e73d 100644
--- a/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/LegacyCreatorAdminAudioContentCharacterizationTest.kt
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/v2/api/admin/aicharacter/content/LegacyCreatorAdminAudioContentCharacterizationTest.kt
@@ -35,7 +35,9 @@ import org.springframework.context.ApplicationEventPublisher
 import org.springframework.data.domain.PageRequest
 import org.springframework.mock.web.MockMultipartFile
 import org.springframework.test.context.ContextConfiguration
+import org.springframework.transaction.annotation.Propagation
 import org.springframework.transaction.annotation.Transactional
+import org.springframework.transaction.support.TransactionTemplate
 import java.net.URL
 import java.time.LocalDateTime
 import javax.persistence.EntityManager
@@ -48,7 +50,8 @@ class LegacyCreatorAdminAudioContentCharacterizationTest @Autowired constructor(
     private val repository: CreatorAdminContentRepository,
     private val hashTagRepository: HashTagRepository,
     private val contentPriceChangeLogRepository: ContentPriceChangeLogRepository,
-    private val entityManager: EntityManager
+    private val entityManager: EntityManager,
+    private val transactionTemplate: TransactionTemplate
 ) {
     private lateinit var audioContentCloudFront: AudioContentCloudFront
 
@@ -81,12 +84,16 @@ class LegacyCreatorAdminAudioContentCharacterizationTest @Autowired constructor(
     }
 
     @Test
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     @DisplayName("기존 콘텐츠 생성은 표지와 오디오를 업로드하고 처리 대기 콘텐츠를 저장한다")
     fun shouldCreateContentThroughLegacyUploadPipeline() {
-        val owner = saveMember("legacy-create-owner")
-        val theme = AudioContentTheme(theme = "legacy-create-theme", image = "theme.png")
-        entityManager.persist(theme)
-        entityManager.flush()
+        val (owner, theme) = transactionTemplate.execute {
+            val owner = saveMember("legacy-create-owner")
+            val theme = AudioContentTheme(theme = "legacy-create-theme", image = "theme.png")
+            entityManager.persist(theme)
+            entityManager.flush()
+            owner to theme
+        }!!
 
         val response = audioContentService.createAudioContent(
             contentFile = MockMultipartFile("contentFile", "audio.mp3", "audio/mpeg", byteArrayOf(1, 2, 3)),
@@ -96,9 +103,7 @@ class LegacyCreatorAdminAudioContentCharacterizationTest @Autowired constructor(
             """.trimIndent(),
             member = owner
         )
-        entityManager.flush()
-
-        val content = entityManager.find(AudioContent::class.java, response.contentId)
+        val content = transactionTemplate.execute { entityManager.find(AudioContent::class.java, response.contentId) }!!
         assertEquals("legacy created audio", content.title)
         assertEquals(owner.id, content.member!!.id)
         assertEquals(theme.id, content.theme!!.id)
@@ -240,7 +245,7 @@ class LegacyCreatorAdminAudioContentCharacterizationTest @Autowired constructor(
         }
     }
 
-    private fun registerMysqlDateFunctions() {
+    private fun registerMysqlDateFunctions() = transactionTemplate.executeWithoutResult {
         entityManager.createNativeQuery(
             "CREATE ALIAS IF NOT EXISTS DATE_FORMAT FOR 'kr.co.vividnext.sodalive.support.H2MysqlDateFunctions.dateFormat'"
         ).executeUpdate()
diff --git a/src/test/kotlin/kr/co/vividnext/sodalive/v2/subtitle/SubtitleHttpIntegrationTest.kt b/src/test/kotlin/kr/co/vividnext/sodalive/v2/subtitle/SubtitleHttpIntegrationTest.kt
new file mode 100644
index 00000000..cb1fefb4
--- /dev/null
+++ b/src/test/kotlin/kr/co/vividnext/sodalive/v2/subtitle/SubtitleHttpIntegrationTest.kt
@@ -0,0 +1,905 @@
+package kr.co.vividnext.sodalive.v2.subtitle
+
+import com.amazonaws.services.s3.AmazonS3Client
+import com.amazonaws.services.s3.model.PutObjectRequest
+import com.amazonaws.services.s3.model.PutObjectResult
+import com.fasterxml.jackson.databind.ObjectMapper
+import kr.co.vividnext.sodalive.aws.cloudfront.AudioContentCloudFront
+import kr.co.vividnext.sodalive.content.AudioContent
+import kr.co.vividnext.sodalive.content.AudioContentService
+import kr.co.vividnext.sodalive.content.CreateAudioContentRequest
+import kr.co.vividnext.sodalive.content.order.Order
+import kr.co.vividnext.sodalive.content.order.OrderType
+import kr.co.vividnext.sodalive.content.theme.AudioContentTheme
+import kr.co.vividnext.sodalive.member.Member
+import kr.co.vividnext.sodalive.member.MemberRepository
+import kr.co.vividnext.sodalive.member.MemberRole
+import kr.co.vividnext.sodalive.support.EmbeddedRedisInitializer
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleCallbackReceipt
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleCallbackReceiptRepository
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleJob
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleJobRepository
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.persistence.SubtitleRepository
+import kr.co.vividnext.sodalive.v2.subtitle.adapter.web.SubtitleRequestParser
+import kr.co.vividnext.sodalive.v2.subtitle.application.SubtitleJobService
+import kr.co.vividnext.sodalive.v2.subtitle.domain.SubtitleJobStatus
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Assertions.assertNull
+import org.junit.jupiter.api.Assertions.assertThrows
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.DisplayName
+import org.junit.jupiter.api.Test
+import org.mockito.Mockito
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.boot.test.mock.mockito.MockBean
+import org.springframework.boot.test.mock.mockito.SpyBean
+import org.springframework.boot.test.web.client.TestRestTemplate
+import org.springframework.context.ApplicationEventPublisher
+import org.springframework.dao.DataIntegrityViolationException
+import org.springframework.data.repository.findByIdOrNull
+import org.springframework.http.HttpEntity
+import org.springframework.http.HttpHeaders
+import org.springframework.http.HttpMethod
+import org.springframework.http.MediaType
+import org.springframework.mock.web.MockMultipartFile
+import org.springframework.security.crypto.password.PasswordEncoder
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.DynamicPropertyRegistry
+import org.springframework.test.context.DynamicPropertySource
+import org.springframework.test.util.AopTestUtils
+import org.springframework.test.util.ReflectionTestUtils
+import org.springframework.transaction.PlatformTransactionManager
+import org.springframework.transaction.TransactionDefinition
+import org.springframework.transaction.support.TransactionSynchronizationManager
+import org.springframework.transaction.support.TransactionTemplate
+import java.net.URL
+import java.security.MessageDigest
+import java.time.LocalDateTime
+import java.util.Base64
+import java.util.UUID
+import java.util.concurrent.Callable
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import javax.persistence.EntityManager
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@ContextConfiguration(initializers = [EmbeddedRedisInitializer::class])
+class SubtitleHttpIntegrationTest @Autowired constructor(
+    private val http: TestRestTemplate,
+    private val mapper: ObjectMapper,
+    private val members: MemberRepository,
+    private val passwordEncoder: PasswordEncoder,
+    private val jobs: SubtitleJobRepository,
+    private val receipts: SubtitleCallbackReceiptRepository,
+    private val subtitles: SubtitleRepository,
+    private val service: SubtitleJobService,
+    private val contentService: AudioContentService,
+    private val parser: SubtitleRequestParser,
+    private val entityManager: EntityManager,
+    transactionManager: PlatformTransactionManager
+) {
+    companion object {
+        @JvmStatic
+        @DynamicPropertySource
+        fun properties(registry: DynamicPropertyRegistry) {
+            registry.add("jwt.secret") { Base64.getEncoder().encodeToString(ByteArray(64) { 7 }) }
+            registry.add("cloud.aws.s3.content-bucket") { "subtitle-audio" }
+            registry.add("cloud.aws.s3.bucket") { "subtitle-cover" }
+            registry.add("spring.jpa.properties.hibernate.show_sql") { "false" }
+        }
+    }
+
+    @MockBean
+    private lateinit var s3: AmazonS3Client
+
+    @MockBean
+    private lateinit var cloudFront: AudioContentCloudFront
+
+    @SpyBean
+    private lateinit var receiptWriter: SubtitleCallbackReceiptRepository
+
+    private val transaction = TransactionTemplate(transactionManager).apply {
+        propagationBehavior = TransactionDefinition.PROPAGATION_REQUIRES_NEW
+    }
+    private lateinit var originalPublisher: ApplicationEventPublisher
+    private lateinit var publisher: ApplicationEventPublisher
+
+    @BeforeEach
+    fun setupExternalBoundaries() {
+        val target = AopTestUtils.getTargetObject<AudioContentService>(contentService)
+        originalPublisher = ReflectionTestUtils.getField(target, "applicationEventPublisher") as ApplicationEventPublisher
+        publisher = Mockito.mock(ApplicationEventPublisher::class.java)
+        ReflectionTestUtils.setField(target, "applicationEventPublisher", publisher)
+        Mockito.`when`(s3.getUrl(Mockito.anyString(), Mockito.anyString())).thenReturn(URL("https://example.test/object"))
+        Mockito.`when`(s3.putObject(Mockito.any(PutObjectRequest::class.java))).thenReturn(PutObjectResult())
+        Mockito.`when`(cloudFront.generateSignedURL(Mockito.anyString(), Mockito.anyLong()))
+            .thenAnswer { "https://signed.example.test/${it.arguments[0]}" }
+    }
+
+    @AfterEach
+    fun restorePublisher() {
+        ReflectionTestUtils.setField(
+            AopTestUtils.getTargetObject<AudioContentService>(contentService),
+            "applicationEventPublisher",
+            originalPublisher
+        )
+    }
+
+    @Test
+    @DisplayName("실제 EMAIL 로그인 JWT라도 BOT 외 역할은 내부 자막 경로에 접근할 수 없다")
+    fun shouldRejectActualNonBotLoginJwt() {
+        listOf(MemberRole.ADMIN, MemberRole.USER, MemberRole.CREATOR).forEach { role ->
+            val token = login(role)
+            assertEquals(403, get("/resolve?bucket=audio&key=unknown", token).statusCodeValue)
+            assertEquals(403, post("/callback", token, "{}").statusCodeValue)
+            assertEquals(403, get("/${UUID.randomUUID()}", token).statusCodeValue)
+            assertEquals(403, post("/${UUID.randomUUID()}/started", token, "{}").statusCodeValue)
+        }
+        assertEquals(
+            401,
+            http.getForEntity("/internal/api/subtitle-jobs/resolve?bucket=a&key=b", String::class.java).statusCodeValue
+        )
+    }
+
+    @Test
+    @DisplayName("활성 EMAIL BOT의 실제 JWT는 네 API를 허용하며 조회 응답은 envelope 없는 정확한 계약이다")
+    fun shouldUseRealBotJwtAndStoreCallbackBeforeStartedWithExactByteReplay() {
+        val token = login(MemberRole.BOT)
+        val job = fixture()
+        val lookup = get("/resolve?bucket=${job.sourceBucket}&key=${job.sourceKey}", token)
+        assertEquals(200, lookup.statusCodeValue)
+        val body = mapper.readTree(lookup.body)
+        assertEquals(setOf("jobId", "contentId", "status", "source"), body.fieldNames().asSequence().toSet())
+        assertTrue(body.path("contentId").isIntegralNumber)
+        assertEquals(job.contentId, body.path("contentId").longValue())
+        assertEquals(setOf("bucket", "key", "versionId"), body.path("source").fieldNames().asSequence().toSet())
+        assertTrue(body.path("source").path("versionId").isNull)
+        assertEquals("PENDING", body.path("status").textValue())
+        assertEquals(lookup.body, get("/${job.jobId}", token).body)
+        val raw = callback(job, eventId = "non-UUID/event-1")
+        assertEquals(200, post("/callback", token, raw).statusCodeValue)
+        assertEquals(200, post("/${job.jobId}/started", token, started(job)).statusCodeValue)
+        assertEquals(200, post("/callback", token, raw).statusCodeValue)
+        assertEquals(422, post("/callback", token, "$raw ").statusCodeValue)
+        assertEquals(422, post("/callback", token, raw.replace(job.jobId, UUID.randomUUID().toString())).statusCodeValue)
+        val resultIds = subtitles.findAllByJobId(job.jobId).map { it.id }.sortedBy { it }
+        assertEquals(200, post("/callback", token, raw.replace("non-UUID/event-1", "same-result-new-event")).statusCodeValue)
+        assertEquals(resultIds, subtitles.findAllByJobId(job.jobId).map { it.id }.sortedBy { it })
+        assertEquals(409, post("/callback", token, callback(job, "new-result", emptyList())).statusCodeValue)
+        val stored = jobs.findByIdOrNull(job.jobId)!!
+        assertEquals(SubtitleJobStatus.COMPLETED, stored.status)
+        assertEquals(
+            "subtitles/${job.contentId}/${job.jobId}/ko.srt",
+            subtitles.findByJobIdAndLanguage(job.jobId, "ko-KR")!!.s3Key
+        )
+        assertEquals(
+            "subtitles/${job.contentId}/${job.jobId}/ja.srt",
+            subtitles.findByJobIdAndLanguage(job.jobId, "ja-JP")!!.s3Key
+        )
+        assertEquals(0, stored.updatedAt.nano)
+        val receipt = receipts.findByIdOrNull("non-UUID/event-1")!!
+        assertEquals(
+            MessageDigest.getInstance("SHA-256").digest(raw.toByteArray()).joinToString("") { "%02x".format(it) },
+            receipt.bodyHash
+        )
+        assertEquals(0, receipt.createdAt.nano)
+        assertEquals(
+            job.jobId,
+            transaction.execute { entityManager.find(AudioContent::class.java, job.contentId).latestSubtitleJobId }
+        )
+    }
+
+    @Test
+    @DisplayName("started는 PENDING만 전이하며 partial/failed는 정확한 언어 결과만 저장한다")
+    fun shouldStartMonotonicallyAndAcceptPartialAndFailedResults() {
+        val token = login(MemberRole.BOT)
+        val partial = fixture()
+        assertEquals(200, post("/${partial.jobId}/started", token, started(partial)).statusCodeValue)
+        val processing = jobs.findByIdOrNull(partial.jobId)!!
+        assertEquals(SubtitleJobStatus.PROCESSING, processing.status)
+        assertEquals(200, post("/${partial.jobId}/started", token, started(partial)).statusCodeValue)
+        assertEquals(processing.updatedAt, jobs.findByIdOrNull(partial.jobId)!!.updatedAt)
+        assertEquals(200, post("/callback", token, callback(partial, languages = listOf("ja-JP"))).statusCodeValue)
+        val storedPartial = jobs.findByIdOrNull(partial.jobId)!!
+        assertEquals(SubtitleJobStatus.PARTIAL_COMPLETED, storedPartial.status)
+        assertNull(subtitles.findByJobIdAndLanguage(partial.jobId, "ko-KR"))
+        assertNotNull(subtitles.findByJobIdAndLanguage(partial.jobId, "ja-JP"))
+        val failed = fixture()
+        assertEquals(200, post("/callback", token, callback(failed, languages = emptyList())).statusCodeValue)
+        assertEquals(SubtitleJobStatus.FAILED, jobs.findByIdOrNull(failed.jobId)!!.status)
+        assertEquals(200, post("/${failed.jobId}/started", token, started(failed)).statusCodeValue)
+        assertEquals(SubtitleJobStatus.FAILED, jobs.findByIdOrNull(failed.jobId)!!.status)
+    }
+
+    @Test
+    @DisplayName("unknown source/job은 생성 없이 404이며 DB role 변경 및 비활성화는 기존 JWT도 차단한다")
+    fun shouldReturn404WithoutDiscoveryAndCheckCurrentDatabaseRole() {
+        val token = login(MemberRole.BOT)
+        val count = jobs.count()
+        val unknown = UUID.randomUUID().toString()
+        assertEquals(404, get("/resolve?bucket=wrong&key=unknown", token).statusCodeValue)
+        assertEquals(404, get("/$unknown", token).statusCodeValue)
+        val unknownStarted = """{"transcribeJobName":"subtitle-$unknown","status":"PROCESSING"}"""
+        assertEquals(
+            404,
+            post("/$unknown/started", token, unknownStarted).statusCodeValue
+        )
+        val job = fixture()
+        assertEquals(404, post("/callback", token, callback(job).replace(job.jobId, unknown)).statusCodeValue)
+        assertEquals(count + 1, jobs.count())
+        val member = members.findAll().single { it.email == lastLoginEmail }
+        member.role = MemberRole.ADMIN
+        members.saveAndFlush(member)
+        assertEquals(403, get("/${job.jobId}", token).statusCodeValue)
+        member.role = MemberRole.BOT
+        member.isActive = false
+        members.saveAndFlush(member)
+        assertEquals(403, post("/callback", token, callback(job)).statusCodeValue)
+    }
+
+    @Test
+    @DisplayName("callback JSON은 숫자 coercion, 누락 null, 추가/중복 필드와 잘못된 binding을 원자적으로 거부한다")
+    fun shouldRejectMalformedContractsWithoutPersistingAnything() {
+        val token = login(MemberRole.BOT)
+        val job = fixture()
+        val valid = callback(job, "strict-event")
+        val invalidBodies = listOf(
+            valid.replace("\"contentId\":${job.contentId}", "\"contentId\":\"${job.contentId}\""),
+            valid.replace("\"contentId\":${job.contentId}", "\"contentId\":${job.contentId}.0"),
+            valid.replace("\"contentId\":${job.contentId}", "\"contentId\":9223372036854775808"),
+            valid.replace(",\"errorMessage\":null", ""),
+            valid.dropLast(1) + ",\"unknown\":null}",
+            valid.dropLast(1) + ",\"status\":\"COMPLETED\"}",
+            valid.replace("\"language\":\"ko-KR\"", "\"language\":\"ko-KR\",\"language\":\"ko-KR\""),
+            valid.replace("\"format\":\"SRT\"", "\"format\":\"SRT\",\"extra\":true"),
+            valid.replace("subtitle-${job.jobId}", "wrong-name"),
+            valid.replace("\"contentId\":${job.contentId}", "\"contentId\":${job.contentId + 1}"),
+            valid.replace(job.sourceBucket, "other-bucket"),
+            valid.replace("/${job.jobId}/ko.srt", "/${job.jobId}/../ko.srt"),
+            valid.replace("ja-JP", "ko-KR"),
+            valid.replace("\"SRT\"", "\"VTT\""),
+            valid.replace("ko-KR", "en-US"),
+            valid.replace("\"COMPLETED\"", "\"PARTIAL_COMPLETED\""),
+            valid.replace("\"errorCode\":null", "\"errorCode\":\"unexpected\""),
+            valid.replace("strict-event", ""),
+            "$valid {}",
+            callback(job, languages = emptyList()).replace("\"errorMessage\":\"pipeline failed\"", "\"errorMessage\":null")
+        )
+        invalidBodies.forEachIndexed { index, raw ->
+            assertEquals(422, post("/callback", token, raw).statusCodeValue, "invalid case $index")
+        }
+        listOf(
+            started(job).replace("PROCESSING", "COMPLETED"),
+            started(job).replace("subtitle-${job.jobId}", "wrong"),
+            started(job).dropLast(1) + ",\"extra\":true}",
+            started(job).dropLast(1) + ",\"status\":\"PROCESSING\"}"
+        ).forEach { assertEquals(422, post("/${job.jobId}/started", token, it).statusCodeValue) }
+        val stored = jobs.findByIdOrNull(job.jobId)!!
+        assertEquals(SubtitleJobStatus.PENDING, stored.status)
+        assertTrue(subtitles.findAllByJobId(job.jobId).isEmpty())
+        assertNull(receipts.findByIdOrNull("strict-event"))
+    }
+
+    @Test
+    @DisplayName("빈 callback과 started body는 422이며 job, receipt, 결과를 변경하지 않는다")
+    fun shouldRejectEmptyCallbackAndStartedBodiesWithoutPersistingAnything() {
+        val token = login(MemberRole.BOT)
+        val job = fixture()
+        val counts = listOf(jobs.count(), receipts.count(), subtitles.count())
+        val responses = listOf("/callback", "/${job.jobId}/started").map { post(it, token, "") }
+        assertEquals(listOf(422, 422), responses.map { it.statusCodeValue })
+        assertEquals(counts, listOf(jobs.count(), receipts.count(), subtitles.count()))
+        val stored = jobs.findByIdOrNull(job.jobId)!!
+        assertEquals(SubtitleJobStatus.PENDING, stored.status)
+        assertEquals(job.updatedAt, stored.updatedAt)
+        assertNull(stored.errorCode)
+        assertNull(stored.errorMessage)
+        assertTrue(subtitles.findAllByJobId(job.jobId).isEmpty())
+    }
+
+    @Test
+    @DisplayName("HTTP callback의 receipt 쓰기 실패는 일반화된 500이며 상태, 오류, 결과와 receipt를 rollback한다")
+    fun shouldReturn500AndRollbackWhenReceiptInsertFailsOverHttp() {
+        val token = login(MemberRole.BOT)
+        val job = fixture()
+        Mockito.doThrow(DataIntegrityViolationException("receipt write failed"))
+            .`when`(receiptWriter).saveAndFlush(
+                Mockito.any(SubtitleCallbackReceipt::class.java) ?: SubtitleCallbackReceipt("unused", "unused", "unused")
+            )
+        val response = post("/callback", token, callback(job, "http-receipt-write-failure", listOf("ko-KR")))
+        assertEquals(500, response.statusCodeValue)
+        assertEquals(mapper.readTree("""{"error":"Subtitle request failed"}"""), mapper.readTree(response.body))
+        val stored = jobs.findByIdOrNull(job.jobId)!!
+        assertEquals(SubtitleJobStatus.PENDING, stored.status)
+        assertEquals(job.updatedAt, stored.updatedAt)
+        assertNull(stored.errorCode)
+        assertNull(stored.errorMessage)
+        assertTrue(subtitles.findAllByJobId(job.jobId).isEmpty())
+        assertNull(receipts.findByIdOrNull("http-receipt-write-failure"))
+    }
+
+    @Test
+    @DisplayName("동시 callback은 같은 event를 중복 반영하지 않으며 다른 job 간 unique 경합은 rollback 후 판정한다")
+    fun shouldSerializeConcurrentCallbacksAndRollbackLosingJob() {
+        val token = login(MemberRole.BOT)
+        val first = fixture()
+        val sameBody = callback(first, "concurrent-same")
+        assertEquals(
+            listOf(200, 200),
+            race({ post("/callback", token, sameBody).statusCodeValue }, { post("/callback", token, sameBody).statusCodeValue })
+        )
+        val left = fixture()
+        val right = fixture()
+        val statuses = race(
+            { post("/callback", token, callback(left, "concurrent-cross-job")).statusCodeValue },
+            { post("/callback", token, callback(right, "concurrent-cross-job")).statusCodeValue }
+        )
+        assertEquals(listOf(200, 422), statuses)
+        val winner = receipts.findByIdOrNull("concurrent-cross-job")!!.jobId
+        val loser = if (left.jobId == winner) right else left
+        assertEquals(SubtitleJobStatus.PENDING, jobs.findByIdOrNull(loser.jobId)!!.status)
+        assertTrue(subtitles.findAllByJobId(loser.jobId).isEmpty())
+    }
+
+    @Test
+    @DisplayName("receipt 쓰기 실패 시 이미 flush된 결과 행도 상태와 함께 rollback된다")
+    fun shouldRollbackResultsAndStatusWhenReceiptInsertFails() {
+        val job = fixture()
+        val raw = callback(job, "receipt-write-failure").toByteArray()
+        Mockito.doThrow(DataIntegrityViolationException("receipt write failed"))
+            .`when`(receiptWriter).saveAndFlush(
+                Mockito.any(SubtitleCallbackReceipt::class.java) ?: SubtitleCallbackReceipt("unused", "unused", "unused")
+            )
+        assertThrows(DataIntegrityViolationException::class.java) {
+            service.callback(parser.callback(raw), raw)
+        }
+        assertEquals(SubtitleJobStatus.PENDING, jobs.findByIdOrNull(job.jobId)!!.status)
+        assertTrue(subtitles.findAllByJobId(job.jobId).isEmpty())
+        assertNull(receipts.findByIdOrNull("receipt-write-failure"))
+    }
+
+    @Test
+    @DisplayName("이전 job의 결과는 저장하지만 content의 최신 job 선택값을 변경하지 않는다")
+    fun shouldKeepLatestPointerWhenOlderJobCompletes() {
+        val token = login(MemberRole.BOT)
+        val older = fixture()
+        val latest = transaction.execute {
+            service.register(entityManager.find(AudioContent::class.java, older.contentId), "later.mp3")
+        }!!
+        assertEquals(200, post("/callback", token, callback(older)).statusCodeValue)
+        assertEquals(
+            latest.jobId,
+            transaction.execute { entityManager.find(AudioContent::class.java, older.contentId).latestSubtitleJobId }
+        )
+        assertEquals(SubtitleJobStatus.PENDING, jobs.findByIdOrNull(latest.jobId)!!.status)
+        assertEquals(SubtitleJobStatus.COMPLETED, jobs.findByIdOrNull(older.jobId)!!.status)
+    }
+
+    @Test
+    @DisplayName("외부 transaction이 있어도 S3 원본 업로드 전에 content/source/PENDING이 별도 HTTP에서 보인다")
+    fun shouldCommitRegistrationBeforeUploadAndNeverRevertEarlyMediaConvertOutput() {
+        val token = login(MemberRole.BOT)
+        val owner = members.saveAndFlush(Member(password = "unused", nickname = "upload-owner-${UUID.randomUUID()}"))
+        val themeId = transaction.execute {
+            AudioContentTheme(theme = "upload theme", image = "theme.png").also { entityManager.persist(it) }.id!!
+        }!!
+        var uploadedJob: SubtitleJob? = null
+        Mockito.`when`(s3.putObject(Mockito.any(PutObjectRequest::class.java))).thenAnswer { invocation ->
+            val request = invocation.arguments[0] as PutObjectRequest
+            if (request.key.startsWith("input/")) {
+                assertFalse(TransactionSynchronizationManager.isActualTransactionActive())
+                val response = get("/resolve?bucket=${request.bucketName}&key=${request.key}", token)
+                assertEquals(200, response.statusCodeValue)
+                assertEquals("PENDING", mapper.readTree(response.body).path("status").textValue())
+                val jobId = mapper.readTree(response.body).path("jobId").textValue()
+                uploadedJob = jobs.findByIdOrNull(jobId)!!
+                val sourceFileName = request.key.substringAfterLast('/')
+                assertTrue(sourceFileName.endsWith(".mp3"))
+                assertEquals(
+                    UUID.fromString(sourceFileName.substringBeforeLast('.')).toString(),
+                    sourceFileName.substringBeforeLast('.')
+                )
+                val content = transaction.execute { entityManager.find(AudioContent::class.java, uploadedJob!!.contentId) }!!
+                assertEquals(request.key, content.content)
+                assertNotNull(content.coverImage)
+                assertEquals(jobId, content.latestSubtitleJobId)
+                assertEquals(
+                    1,
+                    transaction.execute { entityManager.find(AudioContent::class.java, content.id).audioContentHashTags.size }
+                )
+                contentService.uploadComplete(content.id!!, "${uploadedJob!!.outputKey}.aac", "00:03:00")
+            }
+            PutObjectResult()
+        }
+        val request = CreateAudioContentRequest(
+            title = "new uploaded",
+            detail = "detail",
+            tags = "#tag",
+            price = 100,
+            themeId = themeId,
+            languageCode = "ko"
+        )
+        transaction.executeWithoutResult {
+            contentService.createAudioContent(
+                MockMultipartFile("contentFile", "C:\\unsafe\\nested/audio.MP3", "audio/mpeg", byteArrayOf(1)),
+                MockMultipartFile("coverImage", "cover.png", "image/png", byteArrayOf(2)),
+                request,
+                LocalDateTime.now().minusMinutes(1),
+                owner
+            )
+        }
+        val job = uploadedJob!!
+        val stored = transaction.execute { entityManager.find(AudioContent::class.java, job.contentId) }!!
+        assertEquals("${job.outputKey}.aac", stored.content)
+        assertEquals("00:03:00", stored.duration)
+        assertTrue(stored.isActive)
+        assertTrue(service.matchesOutput(job.contentId, "${job.outputKey}.aac")!!)
+        assertTrue(service.matchesOutput(job.contentId, "${job.outputKey}.m4a")!!)
+        assertTrue(service.matchesOutput(job.contentId, "${job.outputKey}.mp3")!!)
+        listOf(
+            "${job.outputKey}-other.aac",
+            "${job.outputKey}.txt",
+            "output/999/${job.outputKey.substringAfterLast('/')}.aac",
+            "${job.outputKey}.aac/extra"
+        ).forEach {
+            assertFalse(service.matchesOutput(job.contentId, it)!!)
+        }
+    }
+
+    @Test
+    @DisplayName("putObject 수락 후 getUrl 실패도 원본 오류만 기록하고 조기 완료 callback을 terminal에서 회귀시키지 않는다")
+    fun shouldKeepAcceptedSourceAndTerminalResultWhenUploadReturnFails() {
+        val token = login(MemberRole.BOT)
+        val owner = members.saveAndFlush(Member(password = "unused", nickname = "uncertain-${UUID.randomUUID()}"))
+        val themeId = transaction.execute {
+            AudioContentTheme(theme = "uncertain theme", image = "theme.png").also { entityManager.persist(it) }.id!!
+        }!!
+        var uploadedJob: SubtitleJob? = null
+        Mockito.`when`(s3.putObject(Mockito.any(PutObjectRequest::class.java))).thenAnswer { invocation ->
+            val request = invocation.arguments[0] as PutObjectRequest
+            if (request.key.startsWith("input/")) {
+                uploadedJob = jobs.findBySourceBucketAndSourceKey(request.bucketName, request.key)!!
+                assertEquals(200, post("/callback", token, callback(uploadedJob!!)).statusCodeValue)
+            }
+            PutObjectResult()
+        }
+        Mockito.`when`(s3.getUrl(Mockito.anyString(), Mockito.anyString())).thenAnswer { invocation ->
+            if ((invocation.arguments[1] as String).startsWith("input/")) throw IllegalStateException("upload return failed")
+            URL("https://example.test/cover")
+        }
+        assertThrows(IllegalStateException::class.java) {
+            contentService.createAudioContent(
+                MockMultipartFile("contentFile", byteArrayOf(1)),
+                MockMultipartFile("coverImage", byteArrayOf(2)),
+                CreateAudioContentRequest(title = "uncertain", detail = "detail", tags = "", price = 0, themeId = themeId),
+                LocalDateTime.now(),
+                owner
+            )
+        }
+        val job = jobs.findByIdOrNull(uploadedJob!!.jobId)!!
+        assertEquals(SubtitleJobStatus.COMPLETED, job.status)
+        assertEquals("SOURCE_UPLOAD_UNCERTAIN", job.sourceUploadErrorCode)
+        assertEquals(0, job.sourceUploadErrorAt!!.nano)
+        assertEquals(2, subtitles.findAllByJobId(job.jobId).size)
+        assertNotNull(transaction.execute { entityManager.find(AudioContent::class.java, job.contentId) })
+    }
+
+    @Test
+    @DisplayName("원본 filename의 안전한 확장자만 소문자로 보존하고 canonical UUID 이외 경로는 source로 전달하지 않는다")
+    fun shouldPreserveOnlySafeActualSourceExtensionsWithoutClaimingConversion() {
+        val filenames = listOf(
+            "C:\\client\\dir/file.MP3" to "mp3",
+            "../../voice.WAV" to "wav",
+            "voice.FLAC" to "flac",
+            "voice.aac" to "aac",
+            "voice.opus" to "opus",
+            "voice" to null,
+            "voice.bad-extension" to null,
+            null to null
+        )
+        filenames.forEach { (filename, extension) ->
+            val job = fixture(filename)
+            val sourceFile = job.sourceKey.substringAfterLast('/')
+            val uuid = if (extension == null) sourceFile else sourceFile.removeSuffix(".$extension")
+            assertEquals(UUID.fromString(uuid).toString(), uuid)
+            if (extension != null) assertTrue(sourceFile.endsWith(".$extension"))
+            assertEquals("input/${job.contentId}/$sourceFile", job.sourceKey)
+            assertEquals("output/${job.contentId}/$uuid", job.outputKey)
+        }
+    }
+
+    @Test
+    @DisplayName("콘텐츠 상세는 preview에 자막을 숨기고 소유자/구매자/무료 원본 권한과 공개 시점 및 ko/ja를 적용한다")
+    fun shouldExposeLanguageSubtitleOnlyWithReleasedOriginalAccessInExistingDetail() {
+        val bot = login(MemberRole.BOT)
+        val ownerToken = login(MemberRole.CREATOR)
+        val owner = members.findByEmail(lastLoginEmail)!!
+        val viewerToken = login(MemberRole.USER)
+        val viewer = members.findByEmail(lastLoginEmail)!!
+        val job = fixture(owner = owner)
+        transaction.executeWithoutResult {
+            entityManager.find(AudioContent::class.java, job.contentId).apply {
+                price = 100
+                content = "${job.outputKey}.aac"
+                duration = "00:03:00"
+                coverImage = "cover/test.png"
+                isActive = true
+                isCommentAvailable = false
+                releaseDate = LocalDateTime.now().minusMinutes(1)
+            }
+        }
+        assertEquals(200, post("/callback", bot, callback(job)).statusCodeValue)
+        fun detail(token: String, language: String) = http.exchange(
+            "/audio-content/${job.contentId}?timezone=UTC",
+            HttpMethod.GET,
+            HttpEntity<Void>(headers(token).apply { set(HttpHeaders.ACCEPT_LANGUAGE, language) }),
+            String::class.java
+        ).let {
+            assertEquals(200, it.statusCodeValue)
+            val response = mapper.readTree(it.body)
+            assertTrue(response.path("success").booleanValue(), response.path("message").asText())
+            response.path("data")
+        }
+        assertTrue(detail(viewerToken, "ko").path("subtitle").isNull)
+        val ownerKo = detail(ownerToken, "ko").path("subtitle")
+        assertEquals(setOf("language", "format", "url"), ownerKo.fieldNames().asSequence().toSet())
+        assertEquals("ko-KR", ownerKo.path("language").textValue())
+        assertEquals("SRT", ownerKo.path("format").textValue())
+        assertEquals(
+            "https://signed.example.test/subtitles/${job.contentId}/${job.jobId}/ko.srt",
+            ownerKo.path("url").textValue()
+        )
+        assertEquals("ja-JP", detail(ownerToken, "ja").path("subtitle").path("language").textValue())
+        assertTrue(detail(ownerToken, "en").path("subtitle").isNull)
+        transaction.executeWithoutResult {
+            val content = entityManager.find(AudioContent::class.java, job.contentId)
+            entityManager.persist(
+                Order(type = OrderType.KEEP).apply {
+                    member = viewer
+                    creator = owner
+                    audioContent = content
+                }
+            )
+        }
+        assertEquals("ko-KR", detail(viewerToken, "ko").path("subtitle").path("language").textValue())
+        transaction.executeWithoutResult {
+            entityManager.find(AudioContent::class.java, job.contentId).releaseDate = LocalDateTime.now().plusDays(1)
+        }
+        assertTrue(detail(viewerToken, "ko").path("subtitle").isNull)
+        assertEquals("ko-KR", detail(ownerToken, "ko").path("subtitle").path("language").textValue())
+        transaction.executeWithoutResult {
+            entityManager.find(AudioContent::class.java, job.contentId).apply {
+                price = 0
+                releaseDate = LocalDateTime.now().minusDays(1)
+            }
+        }
+        val freeViewer = login(MemberRole.USER)
+        assertEquals("ja-JP", detail(freeViewer, "ja").path("subtitle").path("language").textValue())
+    }
+
+    @Test
+    @DisplayName("자막 조회는 최신 job의 실제 언어 행만 선택하고 미완료/없는 언어/content 불일치는 null이다")
+    fun shouldReadOnlyPersistedLanguageOfMatchingLatestCompletedJob() {
+        val token = login(MemberRole.BOT)
+        val job = fixture()
+        val content = transaction.execute { entityManager.find(AudioContent::class.java, job.contentId) }!!
+        assertNull(service.subtitle(content, kr.co.vividnext.sodalive.i18n.Lang.KO, 1000))
+        assertEquals(200, post("/callback", token, callback(job, languages = listOf("ja-JP"))).statusCodeValue)
+        assertNull(service.subtitle(content, kr.co.vividnext.sodalive.i18n.Lang.KO, 1000))
+        assertEquals("ja-JP", service.subtitle(content, kr.co.vividnext.sodalive.i18n.Lang.JA, 1000)!!.language)
+        val different = fixture()
+        content.latestSubtitleJobId = different.jobId
+        assertNull(service.subtitle(content, kr.co.vividnext.sodalive.i18n.Lang.JA, 1000))
+        content.latestSubtitleJobId = null
+        assertNull(service.subtitle(content, kr.co.vividnext.sodalive.i18n.Lang.JA, 1000))
+    }
+
+    @Test
+    @DisplayName("구매자의 generate-url은 ko/ja 자막과 기존 음원 URL 및 상세와 같은 자막 스키마를 반환한다")
+    fun shouldIncludeRequestedSubtitleInGenerateUrl() {
+        val bot = login(MemberRole.BOT)
+        val owner = members.saveAndFlush(Member(password = "unused", nickname = "generate-owner-${UUID.randomUUID()}"))
+        val token = login(MemberRole.USER)
+        val buyer = members.findByEmail(lastLoginEmail)!!
+        val job = fixture(owner = owner)
+        transaction.executeWithoutResult {
+            val content = entityManager.find(AudioContent::class.java, job.contentId).apply {
+                price = 100
+                this.content = "${job.outputKey}.aac"
+                duration = "02:03:00"
+                coverImage = "cover/test.png"
+                isActive = true
+                isCommentAvailable = false
+                releaseDate = LocalDateTime.now().minusDays(1)
+            }
+            entityManager.persist(
+                Order(type = OrderType.KEEP).apply {
+                    member = buyer
+                    creator = owner
+                    audioContent = content
+                }
+            )
+        }
+        assertEquals(200, post("/callback", bot, callback(job)).statusCodeValue)
+        Mockito.`when`(cloudFront.generateSignedURL(Mockito.anyString(), Mockito.anyLong()))
+            .thenAnswer { "https://signed.example.test/${it.arguments[0]}?expiresIn=${it.arguments[1]}" }
+        listOf("ko" to "ko-KR", "ja" to "ja-JP").forEach { (language, expectedLanguage) ->
+            val data = generateUrl(job.contentId, token, language)
+            assertTrue(data.has("subtitle"))
+            assertEquals(setOf("contentUrl", "subtitle"), data.fieldNames().asSequence().toSet())
+            assertEquals(
+                "https://signed.example.test/${job.outputKey}.aac?expiresIn=14400000",
+                data.path("contentUrl").textValue()
+            )
+            val subtitle = data.path("subtitle")
+            assertEquals(setOf("language", "format", "url"), subtitle.fieldNames().asSequence().toSet())
+            assertEquals(expectedLanguage, subtitle.path("language").textValue())
+            assertEquals("SRT", subtitle.path("format").textValue())
+            assertEquals(
+                "https://signed.example.test/subtitles/${job.contentId}/${job.jobId}/$language.srt?expiresIn=14400000",
+                subtitle.path("url").textValue()
+            )
+            val detail = http.exchange(
+                "/audio-content/${job.contentId}?timezone=UTC",
+                HttpMethod.GET,
+                HttpEntity<Void>(headers(token).apply { set(HttpHeaders.ACCEPT_LANGUAGE, language) }),
+                String::class.java
+            )
+            assertEquals(200, detail.statusCodeValue)
+            val detailBody = mapper.readTree(detail.body)
+            assertTrue(detailBody.path("success").booleanValue())
+            assertEquals(detailBody.path("data").path("subtitle"), subtitle)
+        }
+    }
+
+    @Test
+    @DisplayName("generate-url은 pending/미등록/없는 언어/미지원 언어의 자막을 명시적 null로 반환하고 음원 URL은 유지한다")
+    fun shouldReturnNullSubtitleInGenerateUrlWhenUnavailable() {
+        val bot = login(MemberRole.BOT)
+        val owner = members.saveAndFlush(Member(password = "unused", nickname = "unavailable-owner-${UUID.randomUUID()}"))
+        val token = login(MemberRole.USER)
+        val buyer = members.findByEmail(lastLoginEmail)!!
+        val job = fixture(owner = owner)
+        transaction.executeWithoutResult {
+            val content = entityManager.find(AudioContent::class.java, job.contentId).apply {
+                this.content = "${job.outputKey}.aac"
+                duration = "00:03:00"
+                releaseDate = LocalDateTime.now().minusDays(1)
+            }
+            entityManager.persist(
+                Order(type = OrderType.KEEP).apply {
+                    member = buyer
+                    creator = owner
+                    audioContent = content
+                }
+            )
+        }
+        val pending = generateUrl(job.contentId, token, "ko")
+        assertTrue(pending.has("subtitle"))
+        assertTrue(pending.path("subtitle").isNull)
+        assertEquals("https://signed.example.test/${job.outputKey}.aac", pending.path("contentUrl").textValue())
+        transaction.executeWithoutResult {
+            entityManager.find(AudioContent::class.java, job.contentId).latestSubtitleJobId = null
+        }
+        val missing = generateUrl(job.contentId, token, "ko")
+        assertTrue(missing.has("subtitle"))
+        assertTrue(missing.path("subtitle").isNull)
+        assertEquals(pending.path("contentUrl"), missing.path("contentUrl"))
+        transaction.executeWithoutResult {
+            entityManager.find(AudioContent::class.java, job.contentId).latestSubtitleJobId = job.jobId
+        }
+        assertEquals(200, post("/callback", bot, callback(job, languages = listOf("ja-JP"))).statusCodeValue)
+        listOf("ko", "en").forEach { language ->
+            val unavailable = generateUrl(job.contentId, token, language)
+            assertTrue(unavailable.has("subtitle"))
+            assertTrue(unavailable.path("subtitle").isNull)
+            assertEquals(pending.path("contentUrl"), unavailable.path("contentUrl"))
+        }
+        val available = generateUrl(job.contentId, token, "ja")
+        assertEquals("ja-JP", available.path("subtitle").path("language").textValue())
+        assertEquals("SRT", available.path("subtitle").path("format").textValue())
+        assertEquals(
+            "https://signed.example.test/subtitles/${job.contentId}/${job.jobId}/ja.srt",
+            available.path("subtitle").path("url").textValue()
+        )
+        assertEquals(pending.path("contentUrl"), available.path("contentUrl"))
+    }
+
+    @Test
+    @DisplayName("generate-url은 소유자/무료도 활성 구매가 필요하며 예약 공개는 자막에만 상세의 공개 제한을 적용한다")
+    fun shouldPreserveGenerateUrlPurchaseAndReleaseRestrictions() {
+        val bot = login(MemberRole.BOT)
+        val ownerToken = login(MemberRole.CREATOR)
+        val owner = members.findByEmail(lastLoginEmail)!!
+        val viewerToken = login(MemberRole.USER)
+        val viewer = members.findByEmail(lastLoginEmail)!!
+        val job = fixture(owner = owner)
+        transaction.executeWithoutResult {
+            entityManager.find(AudioContent::class.java, job.contentId).apply {
+                price = 100
+                content = "${job.outputKey}.aac"
+                duration = "00:03:00"
+                releaseDate = LocalDateTime.now().minusDays(1)
+            }
+        }
+        assertEquals(200, post("/callback", bot, callback(job)).statusCodeValue)
+        listOf(viewerToken, ownerToken).forEach { token ->
+            val denied = generateUrl(job.contentId, token, "ko")
+            assertTrue(denied.has("subtitle"))
+            assertTrue(denied.path("subtitle").isNull)
+            assertEquals("", denied.path("contentUrl").textValue())
+        }
+        transaction.executeWithoutResult { entityManager.find(AudioContent::class.java, job.contentId).price = 0 }
+        listOf(viewerToken, ownerToken).forEach { token ->
+            val free = generateUrl(job.contentId, token, "ko")
+            assertTrue(free.has("subtitle"))
+            assertTrue(free.path("subtitle").isNull)
+            assertEquals("", free.path("contentUrl").textValue())
+        }
+        val expiredOrderId = transaction.execute {
+            val content = entityManager.find(AudioContent::class.java, job.contentId)
+            content.price = 100
+            entityManager.persist(
+                Order(type = OrderType.KEEP, isActive = false).apply {
+                    member = viewer
+                    creator = owner
+                    audioContent = content
+                }
+            )
+            Order(type = OrderType.RENTAL).apply {
+                member = viewer
+                creator = owner
+                audioContent = content
+                entityManager.persist(this)
+                entityManager.flush()
+                endDate = LocalDateTime.now().minusDays(1)
+            }.id!!
+        }!!
+        val expired = generateUrl(job.contentId, viewerToken, "ko")
+        assertTrue(expired.has("subtitle"))
+        assertTrue(expired.path("subtitle").isNull)
+        assertEquals("", expired.path("contentUrl").textValue())
+        transaction.executeWithoutResult {
+            entityManager.find(Order::class.java, expiredOrderId).endDate = LocalDateTime.now().plusDays(1)
+            val content = entityManager.find(AudioContent::class.java, job.contentId)
+            content.releaseDate = LocalDateTime.now().plusDays(1)
+            entityManager.persist(
+                Order(type = OrderType.KEEP).apply {
+                    member = owner
+                    creator = owner
+                    audioContent = content
+                }
+            )
+        }
+        val scheduled = generateUrl(job.contentId, viewerToken, "ko")
+        assertTrue(scheduled.has("subtitle"))
+        assertTrue(scheduled.path("subtitle").isNull)
+        assertEquals("https://signed.example.test/${job.outputKey}.aac", scheduled.path("contentUrl").textValue())
+        val purchasedOwner = generateUrl(job.contentId, ownerToken, "ko")
+        assertEquals(scheduled.path("contentUrl"), purchasedOwner.path("contentUrl"))
+        assertEquals("ko-KR", purchasedOwner.path("subtitle").path("language").textValue())
+        transaction.executeWithoutResult { entityManager.find(AudioContent::class.java, job.contentId).releaseDate = null }
+        assertEquals("ko-KR", generateUrl(job.contentId, viewerToken, "ko").path("subtitle").path("language").textValue())
+        transaction.executeWithoutResult {
+            entityManager.find(AudioContent::class.java, job.contentId).releaseDate = LocalDateTime.now().minusDays(1)
+        }
+        val released = generateUrl(job.contentId, viewerToken, "ko")
+        assertEquals(scheduled.path("contentUrl"), released.path("contentUrl"))
+        assertEquals("ko-KR", released.path("subtitle").path("language").textValue())
+        assertEquals(
+            401,
+            http.getForEntity("/audio-content/${job.contentId}/generate-url", String::class.java).statusCodeValue
+        )
+    }
+
+    private fun generateUrl(contentId: Long, token: String, language: String) = http.exchange(
+        "/audio-content/$contentId/generate-url",
+        HttpMethod.GET,
+        HttpEntity<Void>(headers(token).apply { set(HttpHeaders.ACCEPT_LANGUAGE, language) }),
+        String::class.java
+    ).let {
+        assertEquals(200, it.statusCodeValue)
+        val body = mapper.readTree(it.body)
+        assertTrue(body.path("success").booleanValue())
+        body.path("data")
+    }
+
+    private var lastLoginEmail: String? = null
+
+    private fun login(role: MemberRole): String {
+        val email = "subtitle-${UUID.randomUUID()}@example.test"
+        lastLoginEmail = email
+        members.saveAndFlush(
+            Member(email = email, password = passwordEncoder.encode("test-password"), nickname = email, role = role)
+        )
+        val response = http.postForEntity(
+            "/member/login",
+            HttpEntity(
+                """{"email":"$email","password":"test-password","isAdmin":false,"isCreator":false}""",
+                HttpHeaders().apply { contentType = MediaType.APPLICATION_JSON }
+            ),
+            String::class.java
+        )
+        assertEquals(200, response.statusCodeValue)
+        val body = mapper.readTree(response.body)
+        assertTrue(body.path("success").booleanValue())
+        assertFalse(body.path("data").path("isNew").booleanValue())
+        return body.path("data").path("token").textValue().also { assertTrue(it.isNotBlank()) }
+    }
+
+    private fun headers(token: String) = HttpHeaders().apply {
+        setBearerAuth(token)
+        contentType = MediaType.APPLICATION_JSON
+    }
+
+    private fun get(path: String, token: String) = http.exchange(
+        "/internal/api/subtitle-jobs$path",
+        HttpMethod.GET,
+        HttpEntity<Void>(headers(token)),
+        String::class.java
+    )
+
+    private fun post(path: String, token: String, raw: String) = http.exchange(
+        "/internal/api/subtitle-jobs$path",
+        HttpMethod.POST,
+        HttpEntity(raw.toByteArray(), headers(token)),
+        String::class.java
+    )
+
+    private fun started(job: SubtitleJob) = """{"transcribeJobName":"subtitle-${job.jobId}","status":"PROCESSING"}"""
+
+    private fun callback(
+        job: SubtitleJob,
+        eventId: String = UUID.randomUUID().toString(),
+        languages: List<String> = listOf(
+            "ko-KR",
+            "ja-JP"
+        )
+    ): String {
+        val subtitles = languages.joinToString(",") {
+            val short = if (it == "ko-KR") "ko" else "ja"
+            """{"language":"$it","format":"SRT","s3Bucket":"${job.sourceBucket}",""" +
+                """"s3Key":"subtitles/${job.contentId}/${job.jobId}/$short.srt"}"""
+        }
+        val status = when (languages.size) { 2 -> "COMPLETED"; 1 -> "PARTIAL_COMPLETED"; else -> "FAILED" }
+        val error = if (languages.size == 2) "null" else "\"PIPELINE_FAILED\""
+        val message = if (languages.size == 2) "null" else "\"pipeline failed\""
+        return """{"eventId":"$eventId","jobId":"${job.jobId}","contentId":${job.contentId},""" +
+            """"transcribeJobName":"subtitle-${job.jobId}","status":"$status","subtitles":[$subtitles],""" +
+            """"errorCode":$error,"errorMessage":$message}"""
+    }
+
+    private fun fixture(originalFilename: String? = "fixture.mp3", owner: Member? = null): SubtitleJob = transaction.execute {
+        val creator = owner ?: Member(password = "unused", nickname = "fixture-${UUID.randomUUID()}").also {
+            entityManager.persist(
+                it
+            )
+        }
+        val theme = AudioContentTheme(theme = "subtitle theme", image = "theme.png")
+        entityManager.persist(theme)
+        val content = AudioContent(title = "subtitle content", detail = "detail", languageCode = "ko").apply {
+            member = creator
+            this.theme = theme
+        }
+        entityManager.persist(content)
+        service.register(content, originalFilename)
+    }!!
+
+    private fun race(first: () -> Int, second: () -> Int): List<Int> {
+        val pool = Executors.newFixedThreadPool(2)
+        val gate = CountDownLatch(1)
+        return try {
+            val futures = listOf(first, second).map { action -> pool.submit(Callable { gate.await(); action() }) }
+            gate.countDown()
+            futures.map { it.get(30, TimeUnit.SECONDS) }.sorted()
+        } finally {
+            pool.shutdownNow()
+        }
+    }
+}