추출 부분값을 INCOMPLETE 로 수용하고 유저 입력으로 완성한다

src/main/kotlin/com/depromeet/piki/item/domain/ItemSnapshot.kt

@@ -62,7 +62,7 @@ class ItemSnapshot(
     var currency: String? = currency
         protected set
 
-    // 이 버전의 추출 생애주기. PENDING(대기)→PROCESSING(추출 중)→READY(완료)/FAILED(실패).
+    // 이 버전의 추출 생애주기. PENDING(대기)→PROCESSING(추출 중)→READY(완료)/INCOMPLETE(일부만 채움)/FAILED(실패).
     // 상태는 되돌리지 않는다 — 수기 수정은 이 행을 고치지 않고 MANUAL 새 버전을 쌓는다(#825 결정 4).
     @Enumerated(EnumType.STRING)
     @Column(name = "status", nullable = false, length = 16)
@@ -140,11 +140,18 @@ class ItemSnapshot(
         status = ItemStatus.FAILED
     }
 
-    // PROCESSING → READY. 백그라운드 파싱이 성공해 추출 결과(snapshot)를 채우며 전이한다.
+    // PROCESSING → READY / INCOMPLETE / FAILED. 백그라운드 파싱이 끝나 추출 결과(snapshot)를 채우며 전이한다.
     // 전이 가능 상태가 아닌데 호출되면 워커가 잘못된 버전을 집은 코드 버그이므로 check(500).
     // extractedAt 은 전이 시점의 now() — Wish.delete() 등 도메인이 시간을 만드는 프로젝트 관례를 따른다.
-    fun markReady(snapshot: ProductSnapshot) {
-        check(status == ItemStatus.PROCESSING) { "PROCESSING 이 아닌 snapshot(status=$status)은 READY 로 전이할 수 없다" }
+    //
+    // 결과는 **추출이 무엇을 건졌는지**로만 갈린다 (#944):
+    //   - 세 필드(name·price·imageUrl)를 다 얻음 → READY
+    //   - 일부만 얻음 → INCOMPLETE. 사용자가 나머지를 채워 완성한다. 사진에 가격이 없는 것은 정상 입력이라,
+    //     여기서 실패로 끝내면 "쇼핑몰 화면을 캡처한 것"만 통과하는 계약이 된다.
+    //   - 하나도 못 얻음 → FAILED. 사용자에게 무엇을 채우라 할 근거조차 없다.
+    // 반환값은 확정된 상태다 — 호출부(서비스)가 이 값으로 발행할 이벤트를, 워커가 로그·메트릭을 가른다.
+    fun markExtracted(snapshot: ProductSnapshot): ItemStatus {
+        check(status == ItemStatus.PROCESSING) { "PROCESSING 이 아닌 snapshot(status=$status)은 추출 결과로 전이할 수 없다" }
         apply(
             name = snapshot.name,
             price = snapshot.price,
@@ -153,12 +160,22 @@ class ItemSnapshot(
         )
         // 출처(#825 결정 4) — 어느 기계가 뽑았는지를 버전에 박는다. 구버전 extractor 응답(method 없음)은 null(미기록).
         this.source = ItemSnapshotSource.fromWireMethod(snapshot.extractionMethod)
-        // 추출 결과와 추출시각을 채운 뒤 불변식을 검사한다 — READY 가 보장하는 네 필드(name·price·imageUrl·extractedAt)를
-        // 한 자리에서 확정하려고 set 을 검사 앞에 둔다. 추출이 이름을 못 얻었으면 READY 부적격 —
-        // 워커가 이 예외를 받아 FAILED 로 흡수한다(PROCESSING 방치 방지).
+        // 건진 값이 없으면 추출시각도 남기지 않는다 — 추출한 값이 없는데 "언제 추출했나"는 의미가 없다.
+        if (hasNoExtractedValue()) {
+            status = ItemStatus.FAILED
+            return status
+        }
+        // 추출 결과와 추출시각을 채운 뒤 불변식을 검사한다 — 각 상태가 보장하는 필드를 한 자리에서 확정하려고
+        // set 을 검사 앞에 둔다.
         this.extractedAt = LocalDateTime.now()
+        if (!hasAllReadyFields()) {
+            requireIncompleteInvariant()
+            status = ItemStatus.INCOMPLETE
+            return status
+        }
         requireReadyInvariant()
         status = ItemStatus.READY
+        return status
     }
 
     // PROCESSING → FAILED. 파싱 실패(상품 아님·신뢰 불가·타임아웃)를 동기 400 대신 상태로 남긴다.
@@ -169,8 +186,12 @@ class ItemSnapshot(
 
     // 파싱이 끝나 추출 결과가 채워진 버전인지. 토너먼트 출전·목록 노출처럼 "완성된 버전만" 요구하는 게이트에서 쓴다.
     // PROCESSING(파싱 중)·FAILED(실패)는 false — 이름·가격이 비어 출전에 부적합하다.
+    // INCOMPLETE 도 false 다 — 사용자가 나머지를 채우기 전까지는 같은 이유로 부적합하다(#944).
     fun isReady(): Boolean = status == ItemStatus.READY
 
+    // 파싱은 끝났으나 사용자 입력을 기다리는 버전인지. 클라이언트가 "나머지를 채워 주세요" 화면으로 유도하는 근거다.
+    fun isIncomplete(): Boolean = status == ItemStatus.INCOMPLETE
+
     // 추출이 실패로 종결된 버전인지. 새로고침의 FAILED 차단(수기 수정 유도) 등 상태 분기에서 쓴다.
     fun isFailed(): Boolean = status == ItemStatus.FAILED
 
@@ -190,6 +211,28 @@ class ItemSnapshot(
         requireNotNull(extractedAt) { "READY snapshot 은 extractedAt 이 있어야 한다" }
     }
 
+    // INCOMPLETE 불변식 — 사용자가 나머지를 채워 완성할 수 있는 버전이라, 추출이 최소 하나는 건졌고 그 시각이 남아 있어야
+    // 한다. 하나도 못 건졌으면 FAILED 로 끝냈어야 하는 행이므로 여기 닿으면 markExtracted 의 분기가 깨진 코드 버그다.
+    private fun requireIncompleteInvariant() {
+        require(!hasNoExtractedValue()) { "INCOMPLETE snapshot 은 추출값이 최소 하나 있어야 한다" }
+        requireNotNull(extractedAt) { "INCOMPLETE snapshot 은 extractedAt 이 있어야 한다" }
+    }
+
+    // READY 세 필드를 다 채웠는지 (extractedAt 은 전이가 직접 채우므로 여기서 보지 않는다).
+    private fun hasAllReadyFields(): Boolean {
+        if (name.isNullOrBlank()) return false
+        price ?: return false
+        imageUrl ?: return false
+        return true
+    }
+
+    // 추출값을 하나도 못 얻었는지 — 사용자에게 무엇을 채우라 할 근거조차 없는 상태. currency 는 READY 필수가 아니라
+    // 단독으로는 "건졌다"의 근거가 되지 못하므로 세지 않는다.
+    private fun hasNoExtractedValue(): Boolean {
+        val extracted = listOfNotNull(name?.takeIf { it.isNotBlank() }, price, imageUrl)
+        return extracted.isEmpty()
+    }
+
     private fun validate(
         name: String?,
         price: Int?,

src/main/kotlin/com/depromeet/piki/item/domain/ItemStatus.kt

@@ -13,6 +13,12 @@ enum class ItemStatus {
     // 프로세스가 죽어 반납조차 못 한 채 여기 갇힌 행은 recover 가 재실행으로 되살린다(execution at-least-once, #461).
     PROCESSING,
 
+    // 파싱은 끝났지만 READY 세 필드(name·price·imageUrl) 중 일부만 채워진 상태. 사용자가 나머지를 채워야 쓸 수 있다.
+    // 사진에는 가격이 찍혀 있지 않은 것이 정상이라, "하나라도 비면 등록 거부" 대신 채운 만큼 내려주고 나머지를 사용자에게
+    // 맡긴다(#944). READY 취급을 받지 못하므로 토너먼트 출전 등 "완성된 버전만" 요구하는 게이트에서는 걸러진다.
+    // 사용자가 빈 필드를 채우면 수기 수정(manual)이 이 버전을 base 로 새 READY 버전을 쌓는다.
+    INCOMPLETE,
+
     // 파싱 완료. 추출된 상품 정보가 채워졌다.
     READY,
 

src/main/kotlin/com/depromeet/piki/item/event/ItemParsingIncomplete.kt

@@ -0,0 +1,9 @@
+package com.depromeet.piki.item.event
+
+// 아이템 파싱이 일부 필드만 채우고 끝났다 — 도메인 사실. 실패가 아니라 "사용자가 나머지를 채우면 완성되는 상태"라
+// 완료·실패와 별개 사실로 둔다(#944). 소비자(알림)가 "나머지는 직접 채워주세요" 로 유도하는 근거다.
+data class ItemParsingIncomplete(
+    val itemId: Long,
+    // ItemParsingCompleted 와 같은 이유(#576) — 라우팅은 버전 단위가 정확하다.
+    val snapshotId: Long,
+)

src/main/kotlin/com/depromeet/piki/item/service/AsyncImageParsingWorker.kt

@@ -5,6 +5,7 @@ import com.depromeet.piki.common.exception.ErrorCategory
 import com.depromeet.piki.common.exception.HttpMappable
 import com.depromeet.piki.common.storage.ImageStorage
 import com.depromeet.piki.image.service.ImageSnapshotExtractor
+import com.depromeet.piki.item.domain.ItemStatus
 import com.depromeet.piki.product.service.ProductSnapshot
 import io.micrometer.core.instrument.MeterRegistry
 import io.micrometer.observation.Observation
@@ -19,9 +20,10 @@ import org.springframework.stereotype.Component
 // 위임하고, 이 워커는 상태 전이·재시도 정책·raw 회수만 진다.
 // 외부 호출은 트랜잭션 바깥에서 끝내고, 상태 전이 영속화만 ItemParsingService(@Transactional)에 위임한다.
 //
-// 결과는 셋으로 갈린다(AsyncItemParsingWorker 와 동일한 execution at-least-once 정책, #461):
+// 결과는 넷으로 갈린다(AsyncItemParsingWorker 와 동일한 execution at-least-once 정책, #461):
 //   - 성공 → READY. 파싱이 끝났으니 raw 원본을 회수(delete)한다.
-//   - 확정 실패(상품 아님·추출값 신뢰 불가·READY 전이 거부) → 즉시 FAILED + raw 회수. 다시 해도 결과가 같다.
+//   - 부분 성공(일부 필드만 채움) → INCOMPLETE + raw 회수. 사용자가 나머지를 채워 완성한다(#944).
+//   - 확정 실패(상품 아님·추출값 신뢰 불가·값 0개) → 즉시 FAILED + raw 회수. 다시 해도 결과가 같다.
 //   - 일시 외부 오류(원격 추출 서비스 5xx·연결 실패 등 RETRYABLE) → 소유권 반납(release, PROCESSING→PENDING). raw 는 보존하고 다음 tick 이 다시 집는다.
 @Component
 class AsyncImageParsingWorker(
@@ -80,23 +82,18 @@ class AsyncImageParsingWorker(
     ) {
         val elapsedMs = (System.nanoTime() - started) / 1_000_000
         // 일시 DB 오류(데드락·lock timeout)면 추출 재실행 없이 전이 write 만 짧게 재시도한다(TransitionRetry).
-        runCatchingException { transitionRetry.execute { itemParsingService.markReady(snapshotId, snapshot, attempt) } }
-            .onSuccess { applied ->
-                // 좀비 폐기(소유권 상실)면 전이가 스킵된다 — 결과를 성공으로 세지 않고, **특히 raw 를 지우지 않는다**.
+        runCatchingException { transitionRetry.execute { itemParsingService.markExtracted(snapshotId, snapshot, attempt) } }
+            .onSuccess { status ->
+                // 좀비 폐기(소유권 상실)면 전이가 스킵된다 — 결과를 세지 않고, **특히 raw 를 지우지 않는다**.
                 // 재클레임된 새 시도가 바로 그 원본으로 재실행해야 하므로, 여기서 지우면 되살릴 입력을 잃는다.
-                if (!applied) {
-                    log.info("item {} 이미지 좀비 결과 — 전이·raw 회수 생략 (attempt={})", itemId, attempt)
-                    return@onSuccess
-                }
-                // 링크 워커와 같은 구조화 결과 라인 — 로그 기반 결과 분포·알림이 이미지 경로도 같은 모집단으로 세게 한다(#902).
-                log.info(
-                    "item.parse.result item={} type=image result={} reason={} latency={}ms",
-                    itemId,
-                    ItemParsingMetrics.RESULT_READY,
-                    ItemParsingMetrics.REASON_NONE,
-                    elapsedMs,
-                )
-                ItemParsingMetrics.record(meterRegistry, ItemParsingMetrics.RESULT_READY, ItemParsingMetrics.REASON_NONE)
+                val settled =
+                    status ?: run {
+                        log.info("item {} 이미지 좀비 결과 — 전이·raw 회수 생략 (attempt={})", itemId, attempt)
+                        return@onSuccess
+                    }
+                recordOutcome(itemId, snapshot, settled, elapsedMs)
+                // 셋 다 종결이라 raw 를 회수한다 — INCOMPLETE 도 재파싱하지 않는다(파싱 기회는 단번). 사용자가 채울
+                // 화면이 쓰는 이미지는 추출이 올린 결과물(imageUrl)이지 raw 원본이 아니다.
                 deleteRawQuietly(imageKey)
             }
             .onFailure { e ->
@@ -120,6 +117,47 @@ class AsyncImageParsingWorker(
             }
     }
 
+    // 종결 결과를 원장(로그·메트릭)에 남긴다. 링크 워커와 같은 구조화 결과 라인이라 이미지 경로도 같은 모집단으로
+    // 세어진다(#902). INCOMPLETE 만 missing 을 덧붙이는 이유는 링크 워커 recordOutcome 주석과 같다(#944).
+    private fun recordOutcome(
+        itemId: Long,
+        snapshot: ProductSnapshot,
+        status: ItemStatus,
+        elapsedMs: Long,
+    ) {
+        val result =
+            when (status) {
+                ItemStatus.READY -> ItemParsingMetrics.RESULT_READY
+                ItemStatus.INCOMPLETE -> ItemParsingMetrics.RESULT_INCOMPLETE
+                ItemStatus.FAILED -> ItemParsingMetrics.RESULT_FAILED
+                ItemStatus.PENDING, ItemStatus.PROCESSING -> error("추출 전이가 만들 수 없는 상태 $status")
+            }
+        val reason =
+            when (status) {
+                ItemStatus.FAILED -> ItemParsingMetrics.REASON_EXTRACT_QUALITY
+                else -> ItemParsingMetrics.REASON_NONE
+            }
+        if (status == ItemStatus.INCOMPLETE) {
+            log.info(
+                "item.parse.result item={} type=image result={} reason={} latency={}ms missing={}",
+                itemId,
+                result,
+                reason,
+                elapsedMs,
+                ItemParsingMetrics.missingFieldsOf(snapshot),
+            )
+        } else {
+            log.info(
+                "item.parse.result item={} type=image result={} reason={} latency={}ms",
+                itemId,
+                result,
+                reason,
+                elapsedMs,
+            )
+        }
+        ItemParsingMetrics.record(meterRegistry, result, reason)
+    }
+
     // 파싱 실패는 두 갈래다 — 일시 오류는 소유권을 반납해 다시 집히게 하고, 확정 실패만 즉시 종결한다.
     // 판정은 ErrorCategory 가 쥔다: RETRYABLE(원격 추출 서비스 5xx·연결 실패 등)만 반납하고,
     // 그 외(비-HttpMappable 예외 포함)는 즉시 FAILED — 재시도해도 같은 결과인 코드 버그성 예외라 되살리지 않는다