feat(ai): S7b 瞬态全文载体 + SSE chunk 送达 + 决定即清

分支①第二步:新增独立瞬态加密表 muse_ai_generated_fulltext(V36:fulltext_content 经
EncryptTypeHandler AES 加密、expires_at TTL 30min、content_hash/length 供脱敏、决定即删),
仿 MuseAiRuntimePayloadStore 范式建 Store(write/read/purge/cleanupExpired)。执行器拿到成功
RuntimeResult 后把 fullOutput 写瞬态 Store(不在终态 finally purge——出向正文须存活到用户决定);
投影在 done 之前发一条非终态 chunk 引用行(payload 只含 {contentRef:taskId,sequenceNo},绝不带
正文,不落 outbox、不受终态唯一索引);SSE chunkData 遇 contentRef 回瞬态 Store 取正文填 content,
缺失(purge/过期)→空串,兼容旧内联行。采纳(merge facade)/放弃(reject 两路)即 purge。

主权红线(四重亲验):完整正文只存于执行线程内存/瞬态加密表/SSE 传输/客户端,永不进 chunk payload
(只放 ref)、永不进 muse_ai_suggestion/muse_content_block 任何长期列;content_snapshot.content 仍
写脱敏摘要(S7a 口径未动);DO toString 排除正文、Store 日志只算 sha256/length。

验证:ai 模块 528 用例全绿(新增 Store 6 + 投影/流/建议/合并各扩展),BUILD SUCCESS(-am 防 stale)。
真 PG 一次真生成端到端(chunk 携全文/决定后 purge/重连回取)并入 M1 harness,本步未造新 live IT(避
内容过滤器)。DDL:V35 由 S6 占,本表 V36,当前 V34→V36 临时空号,S6 落 V35 填平(Flyway 允许)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiGeneratedFulltextStore.java

@@ -0,0 +1,95 @@
+package cn.iocoder.muse.module.ai.application.muse;
+
+import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiGeneratedFulltextDO;
+import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiGeneratedFulltextMapper;
+import jakarta.annotation.Resource;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.StringUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.HexFormat;
+
+/**
+ * AI 瞬态出向完整正文仓库（S7b）。
+ *
+ * <p>与入向指令仓库 {@link MuseAiRuntimePayloadStore} 同范式、方向相反：独立表、短 TTL、字段加密。
+ * 区别在于生命周期——出向完整正文须存活到用户决定（采纳/放弃），供 SSE 送达与断线重连回取，
+ * 因此<b>不在执行器终态 finally 里清理</b>，只由决定即清（purge）与 TTL 兜底删除。</p>
+ *
+ * <p><b>主权红线</b>：完整正文只允许存在于「执行线程内存 / 本瞬态加密表 / SSE 传输 / 客户端」，
+ * 永不写入 muse_ai_suggestion / muse_content_block 或任何长期列；日志只打长度/hash，不打正文。</p>
+ */
+@Service
+public class MuseAiGeneratedFulltextStore {
+
+    private static final Duration DEFAULT_TTL = Duration.ofMinutes(30);
+
+    @Resource
+    private MuseAiGeneratedFulltextMapper generatedFulltextMapper;
+
+    /**
+     * 写入本次生成的完整正文（key = taskId）。同一 taskId 覆写前先物理删除旧行，保证幂等且不撞 uk。
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void write(Long tenantId, Long taskId, String fullText) {
+        if (tenantId == null || taskId == null || !StringUtils.hasText(fullText)) {
+            return;
+        }
+        cleanupExpired();
+        // 幂等覆写：同一 taskId 可能因异常重放重入，先物理删除旧行再插入，避免 uk(tenant_id, task_id) 冲突。
+        generatedFulltextMapper.deleteByTaskId(tenantId, taskId);
+        MuseAiGeneratedFulltextDO row = new MuseAiGeneratedFulltextDO();
+        row.setTenantId(tenantId);
+        row.setTaskId(taskId);
+        row.setFulltextContent(fullText);
+        row.setContentLength(fullText.length());
+        row.setContentHash(sha256(fullText));
+        row.setExpiresAt(LocalDateTime.now().plus(DEFAULT_TTL));
+        generatedFulltextMapper.insert(row);
+    }
+
+    /**
+     * 回取完整正文；TTL 过期/已 purge/不存在均返回 null（SSE 侧据此填空串，用户已决定，无害）。
+     */
+    @Transactional(readOnly = true)
+    public String read(Long tenantId, Long taskId) {
+        if (tenantId == null || taskId == null) {
+            return null;
+        }
+        MuseAiGeneratedFulltextDO row = generatedFulltextMapper.selectActiveByTaskId(tenantId, taskId);
+        return row == null ? null : row.getFulltextContent();
+    }
+
+    /**
+     * 决定即清：用户采纳/放弃后物理删除该 taskId 的加密正文，避免长期留表。
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void purge(Long tenantId, Long taskId) {
+        if (tenantId != null && taskId != null) {
+            generatedFulltextMapper.deleteByTaskId(tenantId, taskId);
+        }
+    }
+
+    /**
+     * 兜底清理 TTL 过期与软删残留，防止旧逻辑遗留的加密正文无限期保留。
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void cleanupExpired() {
+        generatedFulltextMapper.cleanupResidual();
+    }
+
+    private String sha256(String value) {
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            return HexFormat.of().formatHex(digest.digest(value.getBytes(StandardCharsets.UTF_8)));
+        } catch (NoSuchAlgorithmException ex) {
+            throw new IllegalStateException("JDK 缺少 SHA-256 摘要算法", ex);
+        }
+    }
+
+}

muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeJobExecutor.java

@@ -44,6 +44,8 @@ public class MuseAiRuntimeJobExecutor {
     @Resource
     private MuseAiRuntimePayloadStore runtimePayloadStore;
     @Resource
+    private MuseAiGeneratedFulltextStore generatedFulltextStore;
+    @Resource
     private KnowledgeRetrievalFacade knowledgeRetrievalFacade;
 
     public MuseAiRuntimeClient.RuntimeResponse executeAiTaskJob(Long taskId, Long jobPkId) {
@@ -90,10 +92,28 @@ public class MuseAiRuntimeJobExecutor {
         }
         response = normalizeRuntimeResponse(job, command, response);
         runtimeCallRecorder.recordFinished(call, response);
+        // S7b：拿到成功 RuntimeResult 后，把 provider 完整正文（仅内存字段 fullOutput）写入瞬态全文 Store（key=taskId），
+        // 供 SSE 送达与断线重连回取。此写入落瞬态加密表（TTL+决定即清），永不进候选/正文长期列。
+        writeGeneratedFulltextIfPresent(task, job, response);
         runtimeProjectionService.applyRuntimeResponse(task, job, response, command);
         return response;
     }
 
+    private void writeGeneratedFulltextIfPresent(MuseAiGenerationDO task, MuseAiJobDO job,
+                                                 MuseAiRuntimeClient.RuntimeResponse response) {
+        // 仅真实生成任务（task != null）需要出向全文送达；agent test 影子候选不走用户 SSE，不写瞬态全文。
+        if (task == null || response == null || response.result() == null) {
+            return;
+        }
+        String fullOutput = response.result().fullOutput();
+        if (!StringUtils.hasText(fullOutput)) {
+            return;
+        }
+        // 关键：出向全文须存活到用户决定（采纳/放弃），故这里写入后不在终态 finally 里 purge；
+        // 现有 runtimePayloadStore.remove（入向指令）保持不动，两个 Store 生命周期互不影响。
+        generatedFulltextStore.write(runtimeTenantId(job), task.getId(), fullOutput);
+    }
+
     private boolean requiresRuntimePayload(MuseAiRuntimeClient.RuntimeCommand command) {
         if (command == null || command.inputSummary() == null) {
             return false;

muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeProjectionService.java

@@ -41,6 +41,7 @@ public class MuseAiRuntimeProjectionService {
     private static final String STATUS_RUNNING = "running";
     private static final String STATUS_STREAMING = "streaming";
     private static final String SUGGESTION_PENDING = "pending";
+    private static final String EVENT_CHUNK = "chunk";
     private static final String EVENT_DONE = "done";
     private static final String EVENT_ERROR = "error";
 
@@ -106,6 +107,11 @@ public class MuseAiRuntimeProjectionService {
         job.setFinishedAt(succeeded || !retryable ? LocalDateTime.now() : null);
         jobMapper.updateById(job);
         if (task != null && (succeeded || !retryable)) {
+            // S7b：done 之前先发一条非终态 chunk 引用行（sequence_no 排在 done 前），供 SSE 送达完整正文。
+            // 仅当 provider 完整正文（fullOutput，仅内存）非空时发；正文本体已由 executor 写入瞬态全文 Store，此行只带引用。
+            if (succeeded && result != null && StringUtils.hasText(result.fullOutput())) {
+                appendGeneratedFulltextChunkEvent(task, job);
+            }
             appendTaskEvent(task, job, succeeded ? EVENT_DONE : EVENT_ERROR, terminalOrRetryStatus,
                     succeeded ? donePayload(result, suggestionId, qualityScores) : failureSummary(failure),
                     succeeded ? null : errorCode(failure), succeeded ? null : errorMessage(failure));
@@ -324,6 +330,43 @@ public class MuseAiRuntimeProjectionService {
                 || (EVENT_ERROR.equals(eventType) && STATUS_FAILED.equals(taskStatus));
     }
 
+    /**
+     * 追加一条 chunk 引用行（S7b）。这是 {@link #appendTaskEvent} 终态门禁之外的显式放行路径：
+     * chunk 非终态，不受 uk_muse_ai_task_event_terminal 约束，也不建 Events outbox（outbox 只承接终态事实）。
+     *
+     * <p><b>主权红线</b>：payload_summary 只放指向瞬态全文 Store 的引用（contentRef=taskId + sequenceNo），
+     * 绝不放完整正文；SSE 读侧遇到 contentRef 时回瞬态 Store 取正文填 chunk.data.content。</p>
+     */
+    private void appendGeneratedFulltextChunkEvent(MuseAiGenerationDO task, MuseAiJobDO job) {
+        // 终态已存在则不再补 chunk（防御：正常时序 chunk 早于 done，此处必为 null）。
+        if (taskEventMapper.selectTerminalByTaskId(String.valueOf(task.getId())) != null) {
+            return;
+        }
+        MuseAiTaskEventDO latest = taskEventMapper.selectLatestByTaskId(String.valueOf(task.getId()));
+        long sequenceNo = latest == null || latest.getSequenceNo() == null ? 1L : latest.getSequenceNo() + 1;
+        MuseAiTaskEventDO event = new MuseAiTaskEventDO();
+        event.setTaskId(String.valueOf(task.getId()));
+        event.setSequenceNo(sequenceNo);
+        event.setEventType(EVENT_CHUNK);
+        event.setTaskStatus(STATUS_STREAMING);
+        event.setOwnerUserId(task.getOwnerUserId());
+        event.setActorUserId(job.getActorUserId());
+        event.setJobId(String.valueOf(job.getId()));
+        event.setCorrelationId(job.getCorrelationId());
+        event.setPayloadSummary(JsonUtils.toJsonString(chunkRefPayload(task.getId(), sequenceNo)));
+        event.setEmittedAt(LocalDateTime.now());
+        event.setTenantId(task.getTenantId());
+        taskEventMapper.insert(event);
+    }
+
+    /** chunk 引用载荷：只含指向瞬态全文 Store 的引用，永不含完整正文。 */
+    private Map<String, Object> chunkRefPayload(Long taskId, long sequenceNo) {
+        Map<String, Object> payload = new LinkedHashMap<>();
+        payload.put("contentRef", taskId);
+        payload.put("sequenceNo", sequenceNo);
+        return payload;
+    }
+
     private Map<String, Object> contentSnapshot(MuseAiRuntimeClient.RuntimeResult result, String sourceSnapshotId,
                                                 String contentText) {
         Map<String, Object> snapshot = new LinkedHashMap<>();

muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiTaskStreamServiceImpl.java

@@ -50,6 +50,8 @@ public class MuseAiTaskStreamServiceImpl implements MuseAiTaskStreamService {
     private MuseAiGenerationMapper generationMapper;
     @Resource
     private MuseAiTaskEventMapper taskEventMapper;
+    @Resource
+    private MuseAiGeneratedFulltextStore generatedFulltextStore;
     @Resource(name = MUSE_AI_TASK_STREAM_EXECUTOR)
     private AsyncTaskExecutor taskStreamExecutor;
 
@@ -278,12 +280,29 @@ public class MuseAiTaskStreamServiceImpl implements MuseAiTaskStreamService {
 
     private Map<String, Object> chunkData(MuseAiTaskEventDO persistedEvent, Map<String, Object> payload) {
         Map<String, Object> data = new LinkedHashMap<>();
-        data.put("content", stringValue(payload.get("content"), ""));
+        data.put("content", resolveChunkContent(persistedEvent, payload));
         // chunk.data.sequenceNo 是 OpenAPI 已声明字段；其他事件只能通过 SSE id 表达顺序。
         data.put("sequenceNo", persistedEvent.getSequenceNo());
         return data;
     }
 
+    /**
+     * 解析 chunk 正文（S7b）：payload 带 contentRef 时回瞬态全文 Store 取完整正文（正文不落事件表）；
+     * Store 缺失（已 purge/TTL 过期）→ 空串（用户已决定，无害）。无 contentRef 时按内联 content 透出（兼容旧行）。
+     */
+    private String resolveChunkContent(MuseAiTaskEventDO persistedEvent, Map<String, Object> payload) {
+        Object contentRef = payload.get("contentRef");
+        if (contentRef == null) {
+            return stringValue(payload.get("content"), "");
+        }
+        Long taskId = longValue(contentRef, null);
+        if (taskId == null) {
+            return "";
+        }
+        String fullText = generatedFulltextStore.read(persistedEvent.getTenantId(), taskId);
+        return fullText == null ? "" : fullText;
+    }
+
     private Map<String, Object> qualityCheckData(Map<String, Object> payload) {
         Map<String, Object> data = new LinkedHashMap<>();
         data.put("dimension", stringValue(payload.get("dimension"), "quality"));

muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseSuggestionServiceImpl.java

@@ -45,6 +45,8 @@ public class MuseSuggestionServiceImpl implements MuseSuggestionService {
     private MuseAiAuditService auditService;
     @Resource
     private MuseContentWorkOwnerFacade workOwnerFacade;
+    @Resource
+    private MuseAiGeneratedFulltextStore generatedFulltextStore;
 
     @Override
     @Transactional(readOnly = true)
@@ -86,6 +88,8 @@ public class MuseSuggestionServiceImpl implements MuseSuggestionService {
         MuseAiSuggestionDO suggestion = requireSuggestion(suggestionId);
         if (STATUS_REJECTED.equals(suggestion.getStatus())) {
             commandService.recordSucceeded(envelope, rejectResponseSnapshot(suggestionId));
+            // S7b 决定即清（幂等）：已拒态再次命中也确保瞬态出向全文被清除。
+            generatedFulltextStore.purge(TenantContextHolder.getRequiredTenantId(), suggestion.getGenerationId());
             return;
         }
 
@@ -111,6 +115,8 @@ public class MuseSuggestionServiceImpl implements MuseSuggestionService {
         update.setStatus(STATUS_REJECTED);
         update.setDecisionArchiveId(decision.getId());
         suggestionMapper.updateById(update);
+        // S7b 决定即清：用户已放弃，物理清除该 taskId 的瞬态出向全文加密载荷。
+        generatedFulltextStore.purge(TenantContextHolder.getRequiredTenantId(), suggestion.getGenerationId());
         commandService.recordSucceeded(envelope, rejectResponseSnapshot(suggestionId));
         auditService.record(MuseAiAuditService.AuditCreateReq.builder()
                 .operationId(OPERATION_REJECT)

muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/facade/AiSuggestionMergeProjectionFacade.java

@@ -4,6 +4,7 @@ import cn.iocoder.muse.framework.common.util.json.JsonUtils;
 import cn.iocoder.muse.framework.tenant.core.context.TenantContextHolder;
 import cn.iocoder.muse.module.ai.application.muse.MuseAiAuditService;
 import cn.iocoder.muse.module.ai.application.muse.MuseAiCommandService;
+import cn.iocoder.muse.module.ai.application.muse.MuseAiGeneratedFulltextStore;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiCommandDO;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiSuggestionDO;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiSuggestionDecisionDO;
@@ -63,6 +64,8 @@ public class AiSuggestionMergeProjectionFacade implements ContentAiSuggestionFac
     private MuseAiCommandService commandService;
     @Resource
     private MuseAiAuditService auditService;
+    @Resource
+    private MuseAiGeneratedFulltextStore generatedFulltextStore;
 
     @Override
     public SuggestionLookupResult getSuggestion(Long userId, Long workId, Long blockId, Long suggestionId) {
@@ -207,6 +210,8 @@ public class AiSuggestionMergeProjectionFacade implements ContentAiSuggestionFac
                 .responseSummary(responseSnapshot)
                 .status("succeeded")
                 .build());
+        // S7b 决定即清：用户已采纳，该 taskId 的瞬态出向全文使命完成，立即物理清除加密载荷。
+        generatedFulltextStore.purge(TenantContextHolder.getRequiredTenantId(), suggestion.getGenerationId());
         return AcceptedDecisionResult.acceptedResult();
     }
 

muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/dal/dataobject/muse/MuseAiGeneratedFulltextDO.java

@@ -0,0 +1,44 @@
+package cn.iocoder.muse.module.ai.dal.dataobject.muse;
+
+import cn.iocoder.muse.framework.mybatis.core.type.EncryptTypeHandler;
+import cn.iocoder.muse.framework.tenant.core.db.TenantBaseDO;
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+import java.time.LocalDateTime;
+
+/**
+ * Muse AI 瞬态出向完整正文 DO（S7b）。
+ *
+ * <p>只承载本次生成的完整正文，供 SSE 送达与断线重连；fulltextContent 通过 MyBatis AES TypeHandler 加密，
+ * TTL 到期或用户决定（采纳/放弃）即物理删除。查询投影/审计只使用 length/hash，不读取 fulltextContent。</p>
+ *
+ * <p><b>主权红线</b>：这是瞬态出向载体，非系统记录，永不作为正文来源落 Canonical（muse_content_block）或候选表
+ * （muse_ai_suggestion）任何长期列。</p>
+ */
+@TableName(value = "muse_ai_generated_fulltext", autoResultMap = true)
+@Data
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true, exclude = "fulltextContent")
+public class MuseAiGeneratedFulltextDO extends TenantBaseDO {
+
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
+    /** 生成任务标识（= muse_ai_generation 主键，与 task_event.task_id 同源）。 */
+    private Long taskId;
+
+    /** provider 完整正文，加密存储；仅瞬态出向，永不落任何长期列。 */
+    @TableField(typeHandler = EncryptTypeHandler.class)
+    private String fulltextContent;
+
+    private Integer contentLength;
+    private String contentHash;
+    private LocalDateTime expiresAt;
+
+}

muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/dal/mysql/muse/MuseAiGeneratedFulltextMapper.java

@@ -0,0 +1,43 @@
+package cn.iocoder.muse.module.ai.dal.mysql.muse;
+
+import cn.iocoder.muse.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.muse.framework.mybatis.core.query.LambdaQueryWrapperX;
+import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiGeneratedFulltextDO;
+import org.apache.ibatis.annotations.Delete;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.time.LocalDateTime;
+
+/**
+ * Muse AI 瞬态出向完整正文 Mapper（S7b）。
+ */
+@Mapper
+public interface MuseAiGeneratedFulltextMapper extends BaseMapperX<MuseAiGeneratedFulltextDO> {
+
+    default MuseAiGeneratedFulltextDO selectActiveByTaskId(Long tenantId, Long taskId) {
+        return selectOne(new LambdaQueryWrapperX<MuseAiGeneratedFulltextDO>()
+                .eq(MuseAiGeneratedFulltextDO::getTenantId, tenantId)
+                .eq(MuseAiGeneratedFulltextDO::getTaskId, taskId)
+                // TTL 过期即视为不存在，与 V36 cleanup 语义一致，避免 SSE 回取到过期正文。
+                .gt(MuseAiGeneratedFulltextDO::getExpiresAt, LocalDateTime.now())
+                // active 查询与 V36 partial index 条件一致，软删/兜底残留一律不可读。
+                .eq(MuseAiGeneratedFulltextDO::getDeleted, false)
+                .last("LIMIT 1"));
+    }
+
+    @Delete("""
+            DELETE FROM muse_ai_generated_fulltext
+             WHERE tenant_id = #{tenantId}
+               AND task_id = #{taskId}
+            """)
+    int deleteByTaskId(@Param("tenantId") Long tenantId, @Param("taskId") Long taskId);
+
+    @Delete("""
+            DELETE FROM muse_ai_generated_fulltext
+             WHERE expires_at <= CURRENT_TIMESTAMP
+                OR deleted = TRUE
+            """)
+    int cleanupResidual();
+
+}

muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiGeneratedFulltextStoreTest.java

@@ -0,0 +1,98 @@
+package cn.iocoder.muse.module.ai.application.muse;
+
+import cn.iocoder.muse.framework.test.core.ut.BaseMockitoUnitTest;
+import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiGeneratedFulltextDO;
+import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiGeneratedFulltextMapper;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+
+import java.time.LocalDateTime;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * 瞬态出向完整正文 Store 测试（S7b）。
+ *
+ * <p>纯 Mockito 单测（不落库，加密 TypeHandler 与真实 TTL SQL 在 real-PG 层验证）：
+ * 只断言 write→加密行+TTL、read→缺失返回空、purge→物理删除、以及入参非法时的 no-op 语义。</p>
+ */
+class MuseAiGeneratedFulltextStoreTest extends BaseMockitoUnitTest {
+
+    private static final String FULL_TEXT = "这是一段完整的 AI 生成正文，长度远超 60 字脱敏摘要阈值，只应存在于瞬态加密表与 SSE 传输中。";
+
+    @InjectMocks
+    private MuseAiGeneratedFulltextStore store;
+    @Mock
+    private MuseAiGeneratedFulltextMapper generatedFulltextMapper;
+
+    @Test
+    void should_writeEncryptedRowWithTtl_when_fullTextPresent() {
+        store.write(100L, 3001L, FULL_TEXT);
+
+        ArgumentCaptor<MuseAiGeneratedFulltextDO> captor = ArgumentCaptor.forClass(MuseAiGeneratedFulltextDO.class);
+        // 幂等覆写：cleanup + 先删同 key 旧行 + 插入。
+        verify(generatedFulltextMapper).cleanupResidual();
+        verify(generatedFulltextMapper).deleteByTaskId(100L, 3001L);
+        verify(generatedFulltextMapper).insert(captor.capture());
+        MuseAiGeneratedFulltextDO row = captor.getValue();
+        assertEquals(Long.valueOf(100L), row.getTenantId());
+        assertEquals(Long.valueOf(3001L), row.getTaskId());
+        assertEquals(FULL_TEXT, row.getFulltextContent());
+        assertEquals(Integer.valueOf(FULL_TEXT.length()), row.getContentLength());
+        assertEquals(64, row.getContentHash().length());
+        // TTL 必须落在未来（决定即清之前存活，供 SSE 送达与断线重连）。
+        assertTrue(row.getExpiresAt().isAfter(LocalDateTime.now()));
+    }
+
+    @Test
+    void should_noop_when_writeArgsInvalid() {
+        store.write(null, 3001L, FULL_TEXT);
+        store.write(100L, null, FULL_TEXT);
+        store.write(100L, 3001L, "   ");
+
+        verify(generatedFulltextMapper, never()).insert(any(MuseAiGeneratedFulltextDO.class));
+        verify(generatedFulltextMapper, never()).deleteByTaskId(any(), any());
+    }
+
+    @Test
+    void should_readPlainText_when_activeRowExists() {
+        MuseAiGeneratedFulltextDO row = new MuseAiGeneratedFulltextDO();
+        row.setFulltextContent(FULL_TEXT);
+        when(generatedFulltextMapper.selectActiveByTaskId(100L, 3001L)).thenReturn(row);
+
+        assertEquals(FULL_TEXT, store.read(100L, 3001L));
+    }
+
+    @Test
+    void should_returnNull_when_rowMissingOrExpiredOrArgsNull() {
+        // TTL 过期/已 purge → mapper active 查询返回 null → read 返回 null（SSE 侧据此填空串）。
+        when(generatedFulltextMapper.selectActiveByTaskId(100L, 3001L)).thenReturn(null);
+        assertNull(store.read(100L, 3001L));
+        assertNull(store.read(null, 3001L));
+        assertNull(store.read(100L, null));
+    }
+
+    @Test
+    void should_physicallyDeleteRow_when_purge() {
+        store.purge(100L, 3001L);
+
+        verify(generatedFulltextMapper).deleteByTaskId(100L, 3001L);
+    }
+
+    @Test
+    void should_noop_when_purgeArgsNull() {
+        store.purge(null, 3001L);
+        store.purge(100L, null);
+
+        verify(generatedFulltextMapper, never()).deleteByTaskId(any(), any());
+    }
+
+}

muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeProjectionServiceTest.java

@@ -7,6 +7,7 @@ import cn.iocoder.muse.module.ai.application.muse.facade.AccountUsageFacade;
 import cn.iocoder.muse.module.ai.application.muse.facade.MuseAiRuntimeClient;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiGenerationDO;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiJobDO;
+import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiSuggestionDO;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiTaskEventDO;
 import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiGenerationMapper;
 import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiJobMapper;
@@ -23,9 +24,12 @@ import java.util.List;
 import java.util.Map;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
@@ -34,6 +38,10 @@ import static org.mockito.Mockito.when;
  */
 class MuseAiRuntimeProjectionServiceTest extends BaseMockitoUnitTest {
 
+    /** provider 完整正文样例：长度显著大于 60 字脱敏摘要，用于断言正文不落 chunk payload。 */
+    private static final String FULL_OUTPUT = "这是一段远超 60 字摘要阈值的 provider 完整正文，仅应经瞬态全文 Store 与 SSE 送达，"
+            + "绝不写入 chunk 事件 payload、候选表 content_snapshot 或任何长期列。";
+
     @InjectMocks
     private MuseAiRuntimeProjectionService projectionService;
     @Mock
@@ -114,6 +122,44 @@ class MuseAiRuntimeProjectionServiceTest extends BaseMockitoUnitTest {
         assertEquals("completed", eventCaptor.getValue().getTaskStatus());
     }
 
+    @Test
+    void should_emitChunkRefEventBeforeDone_when_fullOutputPresent() {
+        TenantContextHolder.setTenantId(100L);
+        MuseAiGenerationDO task = task("running");
+        MuseAiJobDO job = job("running");
+        when(jobMapper.selectByIdForUpdate(100L, 4001L)).thenReturn(job("running"));
+        when(candidateReviewService.review(any(), any(), any()))
+                .thenReturn(new MuseAiCandidateReviewService.CandidateReview(true, "oc-x", "sc-x", List.of(), Map.of()));
+        // 第一次 selectLatestByTaskId（chunk append）无历史→chunk seq=1；第二次（done append）已见 chunk→done seq=2。
+        MuseAiTaskEventDO chunkSeqOne = new MuseAiTaskEventDO();
+        chunkSeqOne.setSequenceNo(1L);
+        when(taskEventMapper.selectLatestByTaskId("3001")).thenReturn(null).thenReturn(chunkSeqOne);
+
+        projectionService.applyRuntimeResponse(task, job, runtimeSuccessWithFullOutput(FULL_OUTPUT), command());
+
+        ArgumentCaptor<MuseAiTaskEventDO> eventCaptor = ArgumentCaptor.forClass(MuseAiTaskEventDO.class);
+        verify(taskEventMapper, times(2)).insert(eventCaptor.capture());
+        MuseAiTaskEventDO chunk = eventCaptor.getAllValues().get(0);
+        MuseAiTaskEventDO done = eventCaptor.getAllValues().get(1);
+        // chunk 引用行排在 done 之前（sequence_no 更小）。
+        assertEquals("chunk", chunk.getEventType());
+        assertEquals("done", done.getEventType());
+        assertEquals(Long.valueOf(1L), chunk.getSequenceNo());
+        assertEquals(Long.valueOf(2L), done.getSequenceNo());
+        // 主权红线：chunk payload 只含指向瞬态 Store 的引用（contentRef=taskId），绝不含完整正文。
+        assertTrue(chunk.getPayloadSummary().contains("\"contentRef\":3001"));
+        assertFalse(chunk.getPayloadSummary().contains(FULL_OUTPUT));
+        // 终态 outbox 只由 done 触发；chunk 非终态，不建 outbox。
+        verify(eventPublishOutboxService, times(1)).createForTerminalEvent(any());
+        verify(eventPublishOutboxService).createForTerminalEvent(done);
+        // 主权红线：候选表 content_snapshot.content 仍是脱敏摘要（outputSummary.summary），绝不落完整正文。
+        ArgumentCaptor<MuseAiSuggestionDO> suggestionCaptor = ArgumentCaptor.forClass(MuseAiSuggestionDO.class);
+        verify(suggestionMapper).insert(suggestionCaptor.capture());
+        String contentSnapshot = suggestionCaptor.getValue().getContentSnapshot();
+        assertTrue(contentSnapshot.contains("\"content\":\"done\""));
+        assertFalse(contentSnapshot.contains(FULL_OUTPUT));
+    }
+
     @Test
     void should_releaseReservedQuota_when_runtimeFailureIsTerminal() {
         TenantContextHolder.setTenantId(100L);
@@ -194,6 +240,13 @@ class MuseAiRuntimeProjectionServiceTest extends BaseMockitoUnitTest {
                 "stop", Instant.parse("2026-05-30T10:00:00Z")), null);
     }
 
+    private static MuseAiRuntimeClient.RuntimeResponse runtimeSuccessWithFullOutput(String fullOutput) {
+        // fullOutput 仅内存字段（S7a）：落库仍用 outputSummary 摘要，chunk 引用行只带指向瞬态 Store 的 ref。
+        return new MuseAiRuntimeClient.RuntimeResponse(new MuseAiRuntimeClient.RuntimeResult("corr-queued",
+                "success", Map.of("summary", "done"), Map.of("promptTokens", 1), 1L, "provider-1",
+                "stop", Instant.parse("2026-05-30T10:00:00Z"), fullOutput), null);
+    }
+
     private static MuseAiRuntimeClient.RuntimeResponse runtimeFailure(boolean retryable) {
         return new MuseAiRuntimeClient.RuntimeResponse(null, new MuseAiRuntimeClient.RuntimeFailure("corr-queued",
                 retryable ? "network_timeout" : "runtime_unavailable", null, "AI_NEW_API_UNAVAILABLE",

muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiTaskStreamServiceTest.java

@@ -50,6 +50,8 @@ class MuseAiTaskStreamServiceTest extends BaseMockitoUnitTest {
     private MuseAiGenerationMapper generationMapper;
     @Mock
     private MuseAiTaskEventMapper taskEventMapper;
+    @Mock
+    private MuseAiGeneratedFulltextStore generatedFulltextStore;
 
     @BeforeEach
     void injectTaskTimeout() {
@@ -117,6 +119,39 @@ class MuseAiTaskStreamServiceTest extends BaseMockitoUnitTest {
         assertEquals(Map.of("taskId", 3001L, "suggestionId", 6001L, "summary", "ok"), events.get(2).data());
     }
 
+    @Test
+    void should_fillChunkContentFromTransientStore_when_chunkHasContentRef() {
+        String fullText = "完整正文，长度远超 60 字脱敏摘要阈值，来自瞬态全文 Store，经 SSE 送达用户。";
+        TenantContextHolder.setTenantId(100L);
+        when(generationMapper.selectByTaskId(3001L)).thenReturn(task(3001L, 1001L, 100L));
+        when(taskEventMapper.selectListByTaskIdOrderBySequence("3001")).thenReturn(List.of(
+                event(1L, "chunk", "streaming", Map.of("contentRef", 3001L, "sequenceNo", 1L), null, null),
+                event(2L, "done", "completed", Map.of("suggestionId", 6001L), null, null)));
+        when(generatedFulltextStore.read(100L, 3001L)).thenReturn(fullText);
+
+        List<MuseAiTaskStreamService.StreamEvent> events = streamService.buildReplayEvents(1001L, 3001L);
+
+        // chunk 引用行回瞬态 Store 取完整正文填 chunk.data.content（长度远大于摘要）。
+        assertEquals(List.of("chunk", "done"),
+                events.stream().map(MuseAiTaskStreamService.StreamEvent::event).toList());
+        assertEquals(Map.of("content", fullText, "sequenceNo", 1L), events.get(0).data());
+    }
+
+    @Test
+    void should_fillEmptyChunkContent_when_transientStoreMissing() {
+        TenantContextHolder.setTenantId(100L);
+        when(generationMapper.selectByTaskId(3001L)).thenReturn(task(3001L, 1001L, 100L));
+        when(taskEventMapper.selectListByTaskIdOrderBySequence("3001")).thenReturn(List.of(
+                event(1L, "chunk", "streaming", Map.of("contentRef", 3001L, "sequenceNo", 1L), null, null),
+                event(2L, "done", "completed", Map.of("suggestionId", 6001L), null, null)));
+        when(generatedFulltextStore.read(100L, 3001L)).thenReturn(null);
+
+        List<MuseAiTaskStreamService.StreamEvent> events = streamService.buildReplayEvents(1001L, 3001L);
+
+        // Store 缺失（已 purge/TTL 过期）→ content 空串，用户已决定，无害；不伪造正文、不报错。
+        assertEquals(Map.of("content", "", "sequenceNo", 1L), events.get(0).data());
+    }
+
     @Test
     void should_stopReplayAtFirstTerminalEvent_when_doneAndErrorBothPersisted() {
         TenantContextHolder.setTenantId(100L);

muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseSuggestionServiceTest.java

@@ -52,6 +52,8 @@ class MuseSuggestionServiceTest extends BaseMockitoUnitTest {
     private MuseAiAuditService auditService;
     @Mock
     private MuseContentWorkOwnerFacade workOwnerFacade;
+    @Mock
+    private MuseAiGeneratedFulltextStore generatedFulltextStore;
 
     @AfterEach
     void clearTenantContext() {
@@ -179,6 +181,8 @@ class MuseSuggestionServiceTest extends BaseMockitoUnitTest {
                         && "rejected".equals(updated.getStatus())
                         && Long.valueOf(8001L).equals(updated.getDecisionArchiveId())));
         verify(commandService).recordSucceeded(any(), argThat(snapshot -> snapshot.contains("\"suggestionId\":6001")));
+        // S7b 决定即清：放弃成功后 purge 该 taskId（=generationId 3001）的瞬态出向全文。
+        verify(generatedFulltextStore).purge(100L, 3001L);
     }
 
     @Test
@@ -242,6 +246,8 @@ class MuseSuggestionServiceTest extends BaseMockitoUnitTest {
         verify(suggestionMapper, never()).updateById(any(MuseAiSuggestionDO.class));
         verify(decisionMapper, never()).insert(any(MuseAiSuggestionDecisionDO.class));
         verify(commandService).recordSucceeded(any(), argThat(snapshot -> snapshot.contains("\"status\":\"rejected\"")));
+        // S7b 决定即清（幂等）：已拒态再次命中也确保 purge。
+        verify(generatedFulltextStore).purge(100L, 3001L);
     }
 
     private static SuggestionRejectReqVO rejectRequest(String commandId) {

muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/facade/AiSuggestionMergeProjectionFacadeTest.java

@@ -5,6 +5,7 @@ import cn.iocoder.muse.framework.test.core.ut.BaseMockitoUnitTest;
 import cn.iocoder.muse.framework.tenant.core.context.TenantContextHolder;
 import cn.iocoder.muse.module.ai.application.muse.MuseAiAuditService;
 import cn.iocoder.muse.module.ai.application.muse.MuseAiCommandService;
+import cn.iocoder.muse.module.ai.application.muse.MuseAiGeneratedFulltextStore;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiCommandDO;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiSuggestionDO;
 import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiSuggestionDecisionDO;
@@ -51,6 +52,8 @@ class AiSuggestionMergeProjectionFacadeTest extends BaseMockitoUnitTest {
     private MuseAiCommandService commandService;
     @Mock
     private MuseAiAuditService auditService;
+    @Mock
+    private MuseAiGeneratedFulltextStore generatedFulltextStore;
 
     @AfterEach
     void clearTenantContext() {
@@ -148,6 +151,8 @@ class AiSuggestionMergeProjectionFacadeTest extends BaseMockitoUnitTest {
         verify(commandService).recordSucceeded(any(), argThat(snapshot -> snapshot.contains("\"status\":\"accepted\"")));
         verify(auditService).record(argThat(audit -> "markSuggestionAccepted".equals(audit.getOperationId())
                 && "succeeded".equals(audit.getStatus())));
+        // S7b 决定即清：采纳成功后 purge 该 taskId（=generationId 3001）的瞬态出向全文。
+        verify(generatedFulltextStore).purge(100L, 3001L);
     }
 
     @Test
@@ -193,6 +198,8 @@ class AiSuggestionMergeProjectionFacadeTest extends BaseMockitoUnitTest {
         suggestion.setId(501L);
         suggestion.setWorkId(101L);
         suggestion.setBlockId(301L);
+        // generationId = taskId：决定即清 purge 以此为 key。
+        suggestion.setGenerationId(3001L);
         suggestion.setStatus("pending");
         suggestion.setSourceRevision(7);
         suggestion.setSourceStatus("active");

muse-cloud/sql/muse/V36__ai_generated_fulltext_transient.sql

@@ -0,0 +1,34 @@
+-- S7b 瞬态出向完整正文载体。
+-- 该表只承载「本次 AI 生成的完整正文」出向 SSE 送达 + 断线重连的临时载荷；fulltext_content 经 MyBatis
+-- EncryptTypeHandler 加密写入，TTL 到期或用户决定（采纳/放弃）即物理删除；audit / suggestion / content_block 一律
+-- 只保存 length/hash 或脱敏摘要，不读取、不投影本表正文。
+--
+-- 主权红线：这是瞬态出向载体，非系统记录，永不作为正文来源落 Canonical（muse_content_block）或候选表
+-- （muse_ai_suggestion）任何长期列；完整正文只允许存在于「执行线程内存 / 本瞬态加密表（TTL+决定即清）/ SSE 传输 / 客户端」。
+-- 版本协调：当前最大 V34；S6 的 runtime_batch_id 迁移预留 V35，本瞬态表取 V36（勿撞号）。
+
+CREATE TABLE muse_ai_generated_fulltext (
+    id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+    task_id BIGINT NOT NULL,
+    fulltext_content TEXT NOT NULL,
+    content_length INT NOT NULL,
+    content_hash CHAR(64) NOT NULL,
+    expires_at TIMESTAMP NOT NULL,
+    creator VARCHAR(64) NOT NULL DEFAULT '',
+    create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updater VARCHAR(64) NOT NULL DEFAULT '',
+    update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    deleted BOOLEAN NOT NULL DEFAULT FALSE,
+    tenant_id BIGINT NOT NULL DEFAULT 0,
+    CONSTRAINT uk_muse_ai_generated_fulltext_task UNIQUE (tenant_id, task_id)
+);
+
+CREATE INDEX idx_muse_ai_generated_fulltext_active
+    ON muse_ai_generated_fulltext(tenant_id, task_id, expires_at)
+    WHERE deleted = FALSE;
+CREATE INDEX idx_muse_ai_generated_fulltext_cleanup
+    ON muse_ai_generated_fulltext(expires_at)
+    WHERE deleted = FALSE;
+CREATE TRIGGER trg_muse_ai_generated_fulltext_updated_at
+    BEFORE UPDATE ON muse_ai_generated_fulltext
+    FOR EACH ROW EXECUTE FUNCTION update_updated_at_column();