task-12: extend coze result buffering beyond poll path

- bufferCozeRowsOrMerge: unified DONE result persist entry point; buffers to transient storage when toggle on, falls back to immediate chunk merge on failure (result never lost), no-op on empty input - submit/retry synchronous immediate DONE results now buffer too (previously always merged immediately); submit path persists a DONE scope state row to carry the buffer pointer (persistImmediateCozeDoneState) - poll path drops the batchTotal>1 gate: single-batch tasks also buffer - CozeCandidate/CozeBatchContext records package-private for test access
2026-08-29 16:46:29 +08:00
parent 59ee3c8154
commit 4d91146256
2 changed files with 552 additions and 36 deletions
@@ -2616,6 +2616,17 @@ public class SimilarAsinTaskService {
                if (!emptyResultMessage.isBlank()) {
                    throw new IllegalStateException(emptyResultMessage);
                }
+                // Task 12：同步 immediate DONE 结果也走缓冲（原立即 merge），由 finalize/assemble 前
+                // 一次性 flush 合并到 chunk，减少 chunk payload 频繁读写。先落一条 DONE state 承载
+                // 缓冲 pointer；缓冲关闭/失败/重复时回退立即 merge，结果不丢失。
+                if (isCozeResultBufferEnabled()) {
+                    TaskScopeStateEntity doneState = persistImmediateCozeDoneState(task, result, job, batchRows,
+                            batchScopeKey, batchScopeHash, batchIndex, batchTotal, submit.credentialName());
+                    if (doneState != null) {
+                        bufferCozeRowsOrMerge(doneState, readCozeBatchContext(doneState), cozeRows, task, allRowsByBaseId);
+                        return false;
+                    }
+                }
                mergeCozeRowsIntoChunk(task, null, null, cozeRows, allRowsByBaseId);
                return false;
            }
@@ -2851,31 +2862,23 @@ public class SimilarAsinTaskService {
            if (!failureMessage.isBlank()) {
                cozeRows = cozeClient.markRowsFailed(batchRows, failureMessage);
            }
-            // P0-3：仅"DONE 且 batchTotal>1 且 feature toggle 开启"时缓冲 cozeRows 到 transient storage，
-            // 由 finalize 阶段一次性合并到 chunk。失败 batch / 单 batch 任务保留原立即 merge 路径。
-            boolean buffered = false;
-            if (failureMessage.isBlank()
-                    && isCozeResultBufferEnabled()
-                    && context.batchTotal() != null && context.batchTotal() > 1
-                    && cozeRows != null && !cozeRows.isEmpty()) {
-                CozeBatchContext bufferedContext = bufferCozeResultForFlush(state, context, cozeRows);
-                if (bufferedContext != null
-                        && bufferedContext.resultPayloadPointer() != null
-                        && !bufferedContext.resultPayloadPointer().isBlank()) {
-                    context = bufferedContext;
-                    buffered = true;
+            // Task 12：DONE 结果统一走 bufferCozeRowsOrMerge（原 P0-3 仅 poll 且 batchTotal>1 缓冲；
+            // 现单 batch、submit/retry 同步 immediate 结果也缓冲），缓冲失败回退立即 merge，
+            // flush 在 finalize/assemble 前一次性合并到 chunk，结果不丢失。
+            // 失败行（markRowsFailed）保持立即 merge 语义不变。
+            if (failureMessage.isBlank()) {
+                if (cozeRows != null && !cozeRows.isEmpty()) {
+                    FileTaskEntity pollTask = taskForPoll(state.getTaskId());
+                    bufferCozeRowsOrMerge(state, context, cozeRows, pollTask,
+                            pollTask == null ? Map.of() : allRowsByBaseIdForPoll(pollTask));
                }
-            }
-            if (!buffered) {
-                FileTaskEntity task = taskForPoll(state.getTaskId());
-                if (task != null) {
-                    Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(task);
+            } else {
+                FileTaskEntity pollTask = taskForPoll(state.getTaskId());
+                if (pollTask != null) {
+                    Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId = allRowsByBaseIdForPoll(pollTask);
                    try {
-                        mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId);
+                        mergeCozeRowsIntoChunk(pollTask, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId);
                    } catch (Exception mergeEx) {
-                        if (failureMessage.isBlank()) {
-                            throw mergeEx;
-                        }
                        log.warn("[similar-asin] coze failed result merge failed, mark state terminal anyway taskId={} stateId={} executeId={} err={}",
                                state.getTaskId(), state.getId(), state.getCozeExecuteId(),
                                firstNonBlank(mergeEx.getMessage(), mergeEx.getClass().getSimpleName()));
@@ -3406,7 +3409,9 @@ public class SimilarAsinTaskService {
                if (!emptyResultMessage.isBlank()) {
                    throw new IllegalStateException(emptyResultMessage);
                }
-                mergeCozeRowsIntoChunk(task, null, null, cozeRows, allRowsByBaseId);
+                // Task 12：retry 同步 immediate DONE 结果也走缓冲（原立即 merge），
+                // 缓冲失败回退立即 merge；flush 在 finalize 时一次性完成。
+                bufferCozeRowsOrMerge(state, context, cozeRows, task, allRowsByBaseId);
                markCozeStateTerminal(state, COZE_STATUS_DONE, null);
                maybeFinalizeCozeJobLocked(state.getTaskId(), context);
                return;
@@ -4048,6 +4053,86 @@ public class SimilarAsinTaskService {
        return properties.isCozeResultBufferEnabled();
    }

+    /**
+     * Task 12：统一 Coze DONE 结果落库入口。
+     * 缓冲开关开启时把 cozeRows 写入 transient storage（pointer 存进 state.stateJson），
+     * 由 flushBufferedCozeResults 在 finalize/assemble 前一次性合并到 chunk；
+     * 缓冲失败（存储异常 / state 更新失败 / 开关关闭）回退立即 merge，结果不丢失。
+     * 空 rows / 空 state / 空 context 直接返回，不产生任何写入。
+     */
+    private void bufferCozeRowsOrMerge(TaskScopeStateEntity state,
+                                       CozeBatchContext context,
+                                       List<SimilarAsinResultRowDto> cozeRows,
+                                       FileTaskEntity task,
+                                       Map<String, List<SimilarAsinParsedRowVo>> allRowsByBaseId) {
+        if (state == null || context == null || cozeRows == null || cozeRows.isEmpty()) {
+            return;
+        }
+        if (!isCozeResultBufferEnabled()) {
+            mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId);
+            return;
+        }
+        CozeBatchContext bufferedContext = bufferCozeResultForFlush(state, context, cozeRows);
+        if (bufferedContext == null
+                || bufferedContext.resultPayloadPointer() == null
+                || bufferedContext.resultPayloadPointer().isBlank()) {
+            // 缓冲失败：回退立即 merge，避免结果悬挂在 transient storage 之外。
+            mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId);
+        }
+    }
+
+    /**
+     * Task 12：为 submit 同步 immediate DONE 结果落一条 DONE state 承载缓冲 pointer。
+     * 与 saveCozeBatchState（SUBMITTED 异步）不同，该 state 直接以 DONE 终态插入，
+     * 不会被 countPendingCozeStates 扫描；缓冲失败/重复插入时返回 null，调用方回退立即 merge。
+     */
+    private TaskScopeStateEntity persistImmediateCozeDoneState(FileTaskEntity task,
+                                                               FileResultEntity result,
+                                                               TaskFileJobEntity job,
+                                                               List<SimilarAsinResultRowDto> batchRows,
+                                                               String batchScopeKey,
+                                                               String batchScopeHash,
+                                                               int batchIndex,
+                                                               int batchTotal,
+                                                               String credentialName) {
+        LocalDateTime now = LocalDateTime.now();
+        CozeBatchContext context = new CozeBatchContext(
+                job.getId(),
+                result.getId(),
+                null,
+                null,
+                batchIndex,
+                batchTotal,
+                currentInstanceId(),
+                0,
+                credentialName,
+                null
+        );
+        TaskScopeStateEntity state = new TaskScopeStateEntity();
+        state.setTaskId(task.getId());
+        state.setModuleType(MODULE_TYPE);
+        state.setScopeKey(batchScopeKey);
+        state.setScopeHash(batchScopeHash);
+        state.setStateJson(writeJson(context, "serialize immediate coze done state context failed"));
+        state.setCozeStatus(COZE_STATUS_DONE);
+        state.setCozeSubmittedAt(now);
+        state.setCozeCompletedAt(now);
+        state.setCozeAttemptCount(0);
+        state.setChunkTotal(batchTotal);
+        state.setReceivedChunkCount(batchIndex);
+        state.setCompleted(1);
+        state.setCreatedAt(now);
+        state.setUpdatedAt(now);
+        try {
+            taskScopeStateMapper.insert(state);
+            return state;
+        } catch (DuplicateKeyException ex) {
+            log.info("[similar-asin] duplicate immediate done coze state ignored taskId={} scope={}",
+                    task.getId(), batchScopeKey);
+            return null;
+        }
+    }
+
    /**
     * P0-3：在 finalize 之前一次性把所有 DONE/FAILED state 上缓冲的 cozeRows
     * 按 chunkScopeHash 分组合并到 chunk。把每个 batch 的"loadSubmittedChunks +
@@ -4291,9 +4376,9 @@ public class SimilarAsinTaskService {
        return "coze:task:" + taskId + ":rows:" + DigestUtil.sha256Hex(rowKeys.toString());
    }

-    private record CozeCandidate(String chunkScopeHash,
-                                 Integer chunkIndex,
-                                 SimilarAsinResultRowDto row) {
+    record CozeCandidate(String chunkScopeHash,
+                         Integer chunkIndex,
+                         SimilarAsinResultRowDto row) {
    }

    private record PythonUploadProgress(int current, int total, String unit) {
@@ -6789,16 +6874,16 @@ public class SimilarAsinTaskService {
                                      boolean terminal) {
    }

-    private record CozeBatchContext(Long jobId,
-                                     Long resultId,
-                                     String chunkScopeHash,
-                                     Integer chunkIndex,
-                                     Integer batchIndex,
-                                     Integer batchTotal,
-                                     String ownerInstanceId,
-                                     Integer submitRetryCount,
-                                     String credentialName,
-                                     String resultPayloadPointer) {
+    record CozeBatchContext(Long jobId,
+                            Long resultId,
+                            String chunkScopeHash,
+                            Integer chunkIndex,
+                            Integer batchIndex,
+                            Integer batchTotal,
+                            String ownerInstanceId,
+                            Integer submitRetryCount,
+                            String credentialName,
+                            String resultPayloadPointer) {
    }

    private static class SourceRowsBuilder {
@@ -0,0 +1,431 @@
+package com.nanri.aiimage.modules.similarasin.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.nanri.aiimage.config.SimilarAsinProperties;
+import com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService;
+import com.nanri.aiimage.modules.similarasin.client.SimilarAsinCozeClient;
+import com.nanri.aiimage.modules.similarasin.mapper.SimilarAsinFilterConditionMapper;
+import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto;
+import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder;
+import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
+import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
+import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
+import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper;
+import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper;
+import com.nanri.aiimage.modules.task.model.entity.FileResultEntity;
+import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
+import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
+import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
+import com.nanri.aiimage.modules.task.service.TaskFileJobService;
+import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.Spy;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Task 12：扩展 Coze 结果缓冲覆盖范围，减少频繁读写完整 chunk payload。
+ * P0-3 缓冲只覆盖"poll DONE 且 batchTotal>1"；Task 12 扩展为：
+ * 1) poll DONE 结果去掉 batchTotal 限制，单 batch 也走缓冲；
+ * 2) retry 提交同步 immediate DONE 结果也走缓冲（原立即 merge）；
+ * 3) 统一走 bufferCozeRowsOrMerge：缓冲失败回退立即 merge，结果不丢失。
+ * flushBufferedCozeResults 在 finalize 前一次性合并，全任务收敛为一次 chunk 读写。
+ */
+@ExtendWith(MockitoExtension.class)
+class SimilarAsinTaskServiceCozeBufferScopeTest {
+
+    private static final AtomicLong NEXT_ID = new AtomicLong(71000);
+    private static final String MODULE = SimilarAsinTaskService.MODULE_TYPE;
+    private static final String CREDENTIAL = "cred-1";
+
+    @Mock private LocalFileStorageService localFileStorageService;
+    @Mock private com.nanri.aiimage.modules.file.service.oss.OssStorageService ossStorageService;
+    @Mock private com.nanri.aiimage.config.StorageProperties storageProperties;
+    @Mock private FileTaskMapper fileTaskMapper;
+    @Mock private FileResultMapper fileResultMapper;
+    @Mock private TaskScopeStateMapper taskScopeStateMapper;
+    @Mock private TaskChunkMapper taskChunkMapper;
+    @Mock private SimilarAsinFilterConditionMapper filterConditionMapper;
+    @Spy private ObjectMapper objectMapper = new ObjectMapper();
+    @Mock private SimilarAsinCozeClient cozeClient;
+    @Mock private SimilarAsinTaskCacheService taskCacheService;
+    @Mock private SimilarAsinProperties properties;
+    @Mock private TaskFileJobService taskFileJobService;
+    @Mock private com.nanri.aiimage.modules.task.service.TaskDistributedLockService taskDistributedLockService;
+    @Mock private com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService taskProgressSnapshotService;
+    @Mock private TransientPayloadStorageService transientPayloadStorageService;
+    @Mock private org.springframework.transaction.PlatformTransactionManager transactionManager;
+    @Mock private com.nanri.aiimage.common.service.DistributedJobLockService distributedJobLockService;
+    @Mock private com.nanri.aiimage.config.InstanceMetadata instanceMetadata;
+    @Mock private com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService cozeCredentialPoolService;
+    @Mock private SimilarAsinImageEmbedder imageEmbedder;
+    @Mock private SimilarAsinImagePrefetchService imagePrefetchService;
+
+    @InjectMocks private SimilarAsinTaskService service;
+
+    @BeforeAll
+    static void initializeMybatisMetadata() {
+        MapperBuilderAssistant assistant = new MapperBuilderAssistant(new MybatisConfiguration(), "");
+        TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class);
+        TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class);
+        TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class);
+    }
+
+    @BeforeEach
+    void setUp() throws Exception {
+        lenient().when(properties.isBoundedResultAssemblyEnabled()).thenReturn(true);
+        lenient().when(properties.getImageCacheMaxBytes()).thenReturn(256L * 1024L * 1024L);
+        lenient().when(properties.getResultFileTimeoutMinutes()).thenReturn(90);
+        lenient().when(properties.getParseResponsePreviewLimit()).thenReturn(100);
+        lenient().when(properties.getMaxSourceFileBytes()).thenReturn(50L * 1024L * 1024L);
+        lenient().when(properties.getMaxParseRows()).thenReturn(50000);
+        lenient().when(properties.getMaxFieldLength()).thenReturn(2000);
+        lenient().when(properties.getMaxWorkbookZipEntries()).thenReturn(20000);
+        lenient().when(properties.getMaxWorkbookUncompressedBytes()).thenReturn(512L * 1024L * 1024L);
+        lenient().when(properties.getCozeBatchSize()).thenReturn(5);
+        lenient().when(properties.getCozeTextOnlyBatchSize()).thenReturn(10);
+        lenient().when(properties.isCozeResultBufferEnabled()).thenReturn(true);
+        lenient().when(properties.getCozePollTimeoutMillis()).thenReturn(30_000);
+        lenient().when(properties.getDbJobTouchIntervalMillis()).thenReturn(2_000L);
+        lenient().when(properties.getDbTaskTouchIntervalMillis()).thenReturn(2_000L);
+        lenient().when(properties.getCozeSubmitLockWaitMillis()).thenReturn(1_000L);
+        lenient().when(properties.getCozeSubmitLockRetryDelayMillis()).thenReturn(100L);
+        lenient().when(properties.getCozeSubmitMinIntervalMillis()).thenReturn(0L);
+        lenient().when(properties.getCozeSubmitMaxRetryCount()).thenReturn(3);
+        lenient().when(properties.getCozeFlushPendingMinutes()).thenReturn(10);
+        lenient().when(cozeClient.configuredCredentialCount()).thenReturn(1);
+        lenient().when(cozeClient.nextCredential()).thenReturn(new SimilarAsinCozeClient.CozeCredentialRef(
+                CREDENTIAL, "wf-1", "token-1", 4));
+        lenient().when(cozeClient.credentialByName(CREDENTIAL)).thenReturn(new SimilarAsinCozeClient.CozeCredentialRef(
+                CREDENTIAL, "wf-1", "token-1", 4));
+        lenient().when(cozeCredentialPoolService.borrow(eq(MODULE), any())).thenReturn(
+                mock(CozeCredentialPoolService.BorrowedCredential.class));
+        lenient().when(distributedJobLockService.tryLock(anyString(), any())).thenReturn(
+                mock(com.nanri.aiimage.common.service.DistributedJobLockService.LockHandle.class));
+        lenient().doAnswer(invocation -> {
+            FileTaskEntity task = invocation.getArgument(0);
+            task.setId(NEXT_ID.incrementAndGet());
+            return 1;
+        }).when(fileTaskMapper).insert(any(FileTaskEntity.class));
+        lenient().when(fileResultMapper.insert(any(FileResultEntity.class))).thenReturn(1);
+        lenient().when(taskScopeStateMapper.insert(any(TaskScopeStateEntity.class))).thenReturn(1);
+        lenient().when(taskScopeStateMapper.update(any(), any())).thenReturn(1);
+        lenient().when(taskScopeStateMapper.selectList(any())).thenReturn(List.of());
+        lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of());
+        lenient().when(fileResultMapper.selectList(any())).thenReturn(List.of());
+        lenient().when(transientPayloadStorageService.storeParsedPayloadEntry(
+                        eq(MODULE), any(), anyString(), anyString(), anyString(), eq(true)))
+                .thenAnswer(invocation -> "rustfs:coze-result/" + NEXT_ID.incrementAndGet());
+    }
+
+    @AfterEach
+    void shutdown() {
+        service.shutdownAssembleExecutor();
+    }
+
+    private static SimilarAsinResultRowDto row(String rowToken, String id, String asin, String country, String title) {
+        SimilarAsinResultRowDto r = new SimilarAsinResultRowDto();
+        r.setRowToken(rowToken);
+        r.setId(id);
+        r.setAsin(asin);
+        r.setCountry(country);
+        r.setTitle(title);
+        r.setMainUrl("https://img.example.com/" + asin + ".jpg");
+        return r;
+    }
+
+    private static TaskChunkEntity chunk(Long id, String scopeHash, Integer chunkIndex, String payloadJson) {
+        TaskChunkEntity chunk = new TaskChunkEntity();
+        chunk.setId(id);
+        chunk.setTaskId(7104L);
+        chunk.setModuleType(MODULE);
+        chunk.setScopeHash(scopeHash);
+        chunk.setChunkIndex(chunkIndex);
+        chunk.setPayloadJson(payloadJson);
+        chunk.setPayloadHash("h-" + id);
+        return chunk;
+    }
+
+    private static FileTaskEntity task() {
+        FileTaskEntity task = new FileTaskEntity();
+        task.setId(7104L);
+        task.setModuleType(MODULE);
+        task.setStatus("RUNNING");
+        task.setResultJson("{\"categorySwitch\":true}");
+        return task;
+    }
+
+    private static TaskScopeStateEntity state(FileTaskEntity task, long id, String status, int batchTotal) {
+        TaskScopeStateEntity state = new TaskScopeStateEntity();
+        state.setId(id);
+        state.setTaskId(task.getId());
+        state.setModuleType(MODULE);
+        state.setScopeHash("scope-" + id);
+        state.setCozeStatus(status);
+        state.setParsedPayloadJson("ptr:batch-" + id);
+        state.setStateJson("{\"jobId\":7101,\"resultId\":7201,\"chunkScopeHash\":null,\"chunkIndex\":null,"
+                + "\"batchIndex\":1,\"batchTotal\":" + batchTotal + ",\"ownerInstanceId\":\"test-instance\","
+                + "\"submitRetryCount\":0,\"credentialName\":\"" + CREDENTIAL + "\",\"resultPayloadPointer\":\"ptr:buffer-" + id + "\"}");
+        return state;
+    }
+
+    private static String rowsJson(List<SimilarAsinResultRowDto> rows) throws Exception {
+        return new ObjectMapper().writeValueAsString(rows);
+    }
+
+    private SimilarAsinTaskService.CozeBatchContext context(int batchTotal) {
+        return new SimilarAsinTaskService.CozeBatchContext(
+                7101L, 7201L, null, null, 1, batchTotal, "test-instance", 0, CREDENTIAL, null);
+    }
+
+    private void stubChunkMerge(String payloadJson) throws Exception {
+        TaskChunkEntity chunk = chunk(1L, "scope-1", 1, "ptr:chunk-1");
+        // loadSubmittedChunks 只保留非空 chunk，chunk payload 必须能解析出至少一行。
+        // 全部 lenient：缓冲成功路径不触达 merge，仅缓冲失败/flush 合并路径消费。
+        lenient().when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk));
+        lenient().when(transientPayloadStorageService.resolvePayload(anyString(), anyString()))
+                .thenAnswer(invocation -> {
+                    String pointer = invocation.getArgument(0);
+                    if (pointer != null && pointer.startsWith("ptr:chunk-")) {
+                        return payloadJson;
+                    }
+                    return "[]";
+                });
+        lenient().when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()))
+                .thenAnswer(invocation -> "ptr:stored-" + invocation.getArgument(3));
+        lenient().when(taskChunkMapper.selectOne(any())).thenReturn(chunk);
+        lenient().when(taskChunkMapper.update(any(), any())).thenReturn(1);
+    }
+
+    private static String chunkRowsJson() throws Exception {
+        // chunk-1 已含 r1 行：loadSubmittedChunks 只保留非空 chunk，且 rowKey 索引能命中缓冲行。
+        return rowsJson(List.of(row("r1", "1", "B0A0000001", "英国", "Title 1")));
+    }
+
+    @Test
+    void test_task_012_payload_chunk_normal_default_path() throws Exception {
+        // 正常输入：DONE 结果（batchTotal=1 单 batch）经 bufferCozeRowsOrMerge 走缓冲，
+        // 不立即写 chunk；缓冲失败回退立即 merge 结果不丢失。
+        FileTaskEntity task = task();
+        List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
+
+        Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge",
+                TaskScopeStateEntity.class,
+                SimilarAsinTaskService.CozeBatchContext.class,
+                List.class, FileTaskEntity.class, Map.class);
+        bufferOrMerge.setAccessible(true);
+        bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
+
+        verify(transientPayloadStorageService, times(1)).storeParsedPayloadEntry(
+                eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true));
+        verify(transientPayloadStorageService, never()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
+        verify(taskChunkMapper, never()).update(any(), any());
+    }
+
+    @Test
+    void test_task_012_payload_chunk_normal_multiple_items() throws Exception {
+        // 批量场景：多个 DONE state（单 batch）全部缓冲；flush 后按 chunk 分组一次合并
+        FileTaskEntity task = task();
+        stubChunkMerge(chunkRowsJson());
+        when(taskScopeStateMapper.selectList(any())).thenReturn(
+                List.of(state(task, 1L, "DONE", 1), state(task, 2L, "DONE", 1)));
+        when(fileTaskMapper.selectById(7104L)).thenReturn(task);
+        when(transientPayloadStorageService.resolvePayload(anyString(), anyString()))
+                .thenAnswer(invocation -> {
+                    String pointer = invocation.getArgument(0);
+                    if (pointer != null && pointer.startsWith("ptr:buffer-")) {
+                        return rowsJson(List.of(row("r1", "1", "B0A0000001", "英国", "Title 1")));
+                    }
+                    if (pointer != null && pointer.startsWith("ptr:chunk-")) {
+                        return chunkRowsJson();
+                    }
+                    return "[]";
+                });
+        when(transientPayloadStorageService.storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString()))
+                .thenAnswer(invocation -> "ptr:stored-" + invocation.getArgument(3));
+        when(taskChunkMapper.selectOne(any())).thenReturn(chunk(1L, "scope-1", 1, "ptr:chunk-1"));
+        when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk(1L, "scope-1", 1, "ptr:chunk-1")));
+        when(taskChunkMapper.update(any(), any())).thenReturn(1);
+
+        Method flush = SimilarAsinTaskService.class.getDeclaredMethod("flushBufferedCozeResults", Long.class);
+        flush.setAccessible(true);
+        flush.invoke(service, 7104L);
+
+        verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
+        verify(taskChunkMapper, atLeastOnce()).update(any(), any());
+        // pointer 清理：每个缓冲 state 都更新
+        verify(taskScopeStateMapper, atLeastOnce()).update(any(), any());
+    }
+
+    @Test
+    void test_task_012_payload_chunk_normal_repeated_operation_is_idempotent() throws Exception {
+        // 重复执行同一输入：缓冲写幂等（同一 state 不产生重复 buffer/merge）
+        FileTaskEntity task = task();
+        List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
+
+        Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge",
+                TaskScopeStateEntity.class,
+                SimilarAsinTaskService.CozeBatchContext.class,
+                List.class, FileTaskEntity.class, Map.class);
+        bufferOrMerge.setAccessible(true);
+        TaskScopeStateEntity state = state(task, 1L, "DONE", 2);
+        bufferOrMerge.invoke(service, state, context(2), rows, task, Map.of());
+        bufferOrMerge.invoke(service, state, context(2), rows, task, Map.of());
+
+        // 缓冲 2 次（每次重新写 pointer 是幂等语义：同一 state 覆盖写，无重复行）
+        verify(transientPayloadStorageService, times(2)).storeParsedPayloadEntry(
+                eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true));
+        verify(taskChunkMapper, never()).update(any(), any());
+    }
+
+    @Test
+    void test_task_012_payload_chunk_boundary_empty_input() throws Exception {
+        // 空输入：无行时缓冲与 merge 都不发生，不创建无效资源
+        FileTaskEntity task = task();
+        Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge",
+                TaskScopeStateEntity.class,
+                SimilarAsinTaskService.CozeBatchContext.class,
+                List.class, FileTaskEntity.class, Map.class);
+        bufferOrMerge.setAccessible(true);
+        bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), null, task, Map.of());
+        bufferOrMerge.invoke(service, state(task, 2L, "DONE", 1), context(1), List.of(), task, Map.of());
+
+        verify(transientPayloadStorageService, never()).storeParsedPayloadEntry(any(), any(), anyString(), anyString(), anyString(), eq(true));
+        verify(taskChunkMapper, never()).update(any(), any());
+    }
+
+    @Test
+    void test_task_012_payload_chunk_boundary_single_item() throws Exception {
+        // 单 batch（batchTotal=1）：原 P0-3 例外，现在也缓冲
+        FileTaskEntity task = task();
+        List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
+        Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge",
+                TaskScopeStateEntity.class,
+                SimilarAsinTaskService.CozeBatchContext.class,
+                List.class, FileTaskEntity.class, Map.class);
+        bufferOrMerge.setAccessible(true);
+        bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
+
+        verify(transientPayloadStorageService, times(1)).storeParsedPayloadEntry(
+                eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true));
+        verify(taskChunkMapper, never()).update(any(), any());
+    }
+
+    @Test
+    void test_task_012_payload_chunk_boundary_limit_and_overflow() throws Exception {
+        // 缓冲开关关闭：回退立即 merge，DONE 结果仍落 chunk 不丢失
+        FileTaskEntity task = task();
+        stubChunkMerge(chunkRowsJson());
+        when(properties.isCozeResultBufferEnabled()).thenReturn(false);
+        List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
+        Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge",
+                TaskScopeStateEntity.class,
+                SimilarAsinTaskService.CozeBatchContext.class,
+                List.class, FileTaskEntity.class, Map.class);
+        bufferOrMerge.setAccessible(true);
+        bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
+
+        verify(transientPayloadStorageService, never()).storeParsedPayloadEntry(any(), any(), anyString(), anyString(), anyString(), eq(true));
+        verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
+    }
+
+    @Test
+    void test_task_012_payload_chunk_invalid_input_rejected() throws Exception {
+        // 缓冲写失败（storeParsedPayloadEntry 抛异常）：回退立即 merge，结果不丢失
+        FileTaskEntity task = task();
+        stubChunkMerge(chunkRowsJson());
+        when(transientPayloadStorageService.storeParsedPayloadEntry(
+                        eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true)))
+                .thenThrow(new IllegalStateException("rustfs full"));
+        List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
+        Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferCozeRowsOrMerge",
+                TaskScopeStateEntity.class,
+                SimilarAsinTaskService.CozeBatchContext.class,
+                List.class, FileTaskEntity.class, Map.class);
+        bufferOrMerge.setAccessible(true);
+        bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
+
+        verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
+        verify(taskChunkMapper, atLeastOnce()).update(any(), any());
+    }
+
+    @Test
+    void test_task_012_payload_chunk_dependency_failure_releases_resources() throws Exception {
+        // flush 时 chunk 写失败：抛可识别业务异常且不清 pointer（保留待重试）；
+        // 依赖恢复后重试 flush 成功，chunk 合并一次、pointer 清理。
+        FileTaskEntity task = task();
+        stubChunkMerge(chunkRowsJson());
+        TaskScopeStateEntity s = state(task, 1L, "DONE", 1);
+        when(taskScopeStateMapper.selectList(any())).thenReturn(List.of(s));
+        when(fileTaskMapper.selectById(7104L)).thenReturn(task);
+        when(transientPayloadStorageService.resolvePayload(anyString(), anyString()))
+                .thenAnswer(invocation -> {
+                    String pointer = invocation.getArgument(0);
+                    if (pointer != null && pointer.startsWith("ptr:buffer-")) {
+                        return rowsJson(List.of(row("r1", "1", "B0A0000001", "英国", "Title 1")));
+                    }
+                    if (pointer != null && pointer.startsWith("ptr:chunk-")) {
+                        return chunkRowsJson();
+                    }
+                    return "[]";
+                });
+        java.util.concurrent.atomic.AtomicInteger storeCalls = new java.util.concurrent.atomic.AtomicInteger(0);
+        doAnswer(invocation -> {
+            if (storeCalls.incrementAndGet() == 1) {
+                throw new IllegalStateException("rustfs write failed");
+            }
+            return "ptr:stored-" + invocation.getArgument(3);
+        }).when(transientPayloadStorageService).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
+
+        Method flush = SimilarAsinTaskService.class.getDeclaredMethod("flushBufferedCozeResults", Long.class);
+        flush.setAccessible(true);
+        Exception ex = assertThrows(Exception.class, () -> {
+            try {
+                flush.invoke(service, 7104L);
+            } catch (java.lang.reflect.InvocationTargetException e) {
+                throw e.getCause();
+            }
+        });
+        assertTrue(ex.getMessage() != null && ex.getMessage().contains("刷新缓冲区"),
+                "flush 失败消息必须可识别, 实际: " + ex.getMessage());
+        // 失败分组不清 pointer：buffer 未被删除、stateJson 未更新，留待重试
+        verify(transientPayloadStorageService, never()).deletePayloadIfPresent(anyString());
+        verify(taskScopeStateMapper, never()).update(any(), any());
+
+        // 恢复后重试 flush：chunk 合并成功一次，pointer 清理
+        flush.invoke(service, 7104L);
+        assertEquals(2, storeCalls.get(), "恢复后重试应再次写 chunk");
+        verify(taskScopeStateMapper, atLeastOnce()).update(any(), any());
+    }
+}