diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiGeneratedFulltextStore.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiGeneratedFulltextStore.java new file mode 100644 index 00000000..7ee4ca18 --- /dev/null +++ b/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)。 + * + *

与入向指令仓库 {@link MuseAiRuntimePayloadStore} 同范式、方向相反:独立表、短 TTL、字段加密。 + * 区别在于生命周期——出向完整正文须存活到用户决定(采纳/放弃),供 SSE 送达与断线重连回取, + * 因此不在执行器终态 finally 里清理,只由决定即清(purge)与 TTL 兜底删除。

+ * + *

主权红线:完整正文只允许存在于「执行线程内存 / 本瞬态加密表 / SSE 传输 / 客户端」, + * 永不写入 muse_ai_suggestion / muse_content_block 或任何长期列;日志只打长度/hash,不打正文。

+ */ +@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); + } + } + +} diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeJobExecutor.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeJobExecutor.java index e494e84a..1fae3120 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeJobExecutor.java +++ b/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; diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeProjectionService.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeProjectionService.java index b9e75d10..74109480 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeProjectionService.java +++ b/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 只承接终态事实)。 + * + *

主权红线:payload_summary 只放指向瞬态全文 Store 的引用(contentRef=taskId + sequenceNo), + * 绝不放完整正文;SSE 读侧遇到 contentRef 时回瞬态 Store 取正文填 chunk.data.content。

+ */ + 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 chunkRefPayload(Long taskId, long sequenceNo) { + Map payload = new LinkedHashMap<>(); + payload.put("contentRef", taskId); + payload.put("sequenceNo", sequenceNo); + return payload; + } + private Map contentSnapshot(MuseAiRuntimeClient.RuntimeResult result, String sourceSnapshotId, String contentText) { Map snapshot = new LinkedHashMap<>(); diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiTaskStreamServiceImpl.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiTaskStreamServiceImpl.java index 78d4a4cc..a8606ef4 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseAiTaskStreamServiceImpl.java +++ b/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 chunkData(MuseAiTaskEventDO persistedEvent, Map payload) { Map 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 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 qualityCheckData(Map payload) { Map data = new LinkedHashMap<>(); data.put("dimension", stringValue(payload.get("dimension"), "quality")); diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseSuggestionServiceImpl.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseSuggestionServiceImpl.java index 7eb27fce..91b93967 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/MuseSuggestionServiceImpl.java +++ b/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) diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/facade/AiSuggestionMergeProjectionFacade.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/facade/AiSuggestionMergeProjectionFacade.java index 49a0311a..62788fbd 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/application/muse/facade/AiSuggestionMergeProjectionFacade.java +++ b/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(); } diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/dal/dataobject/muse/MuseAiGeneratedFulltextDO.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/dal/dataobject/muse/MuseAiGeneratedFulltextDO.java new file mode 100644 index 00000000..c287b83e --- /dev/null +++ b/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)。 + * + *

只承载本次生成的完整正文,供 SSE 送达与断线重连;fulltextContent 通过 MyBatis AES TypeHandler 加密, + * TTL 到期或用户决定(采纳/放弃)即物理删除。查询投影/审计只使用 length/hash,不读取 fulltextContent。

+ * + *

主权红线:这是瞬态出向载体,非系统记录,永不作为正文来源落 Canonical(muse_content_block)或候选表 + * (muse_ai_suggestion)任何长期列。

+ */ +@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; + +} diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/dal/mysql/muse/MuseAiGeneratedFulltextMapper.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/main/java/cn/iocoder/muse/module/ai/dal/mysql/muse/MuseAiGeneratedFulltextMapper.java new file mode 100644 index 00000000..f8ab136f --- /dev/null +++ b/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 { + + default MuseAiGeneratedFulltextDO selectActiveByTaskId(Long tenantId, Long taskId) { + return selectOne(new LambdaQueryWrapperX() + .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(); + +} diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiGeneratedFulltextStoreTest.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiGeneratedFulltextStoreTest.java new file mode 100644 index 00000000..730262cc --- /dev/null +++ b/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)。 + * + *

纯 Mockito 单测(不落库,加密 TypeHandler 与真实 TTL SQL 在 real-PG 层验证): + * 只断言 write→加密行+TTL、read→缺失返回空、purge→物理删除、以及入参非法时的 no-op 语义。

+ */ +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 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()); + } + +} diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeProjectionServiceTest.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeProjectionServiceTest.java index 753fa17d..9f563bc5 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiRuntimeProjectionServiceTest.java +++ b/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 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 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", diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiTaskStreamServiceTest.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiTaskStreamServiceTest.java index e28edf59..12fff338 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseAiTaskStreamServiceTest.java +++ b/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 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 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); diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseSuggestionServiceTest.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseSuggestionServiceTest.java index e86aff09..2d55ebc2 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/MuseSuggestionServiceTest.java +++ b/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) { diff --git a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/facade/AiSuggestionMergeProjectionFacadeTest.java b/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/facade/AiSuggestionMergeProjectionFacadeTest.java index 0e688817..f2c52a29 100644 --- a/muse-cloud/muse-module-ai/muse-module-ai-server/src/test/java/cn/iocoder/muse/module/ai/application/muse/facade/AiSuggestionMergeProjectionFacadeTest.java +++ b/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"); diff --git a/muse-cloud/sql/muse/V36__ai_generated_fulltext_transient.sql b/muse-cloud/sql/muse/V36__ai_generated_fulltext_transient.sql new file mode 100644 index 00000000..a2d7356a --- /dev/null +++ b/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();