()
+ .eq(UnifiedEventDO::getTenantId, tenantId)
+ .eq(UnifiedEventDO::getOwnerUserId, ownerUserId)
+ .eq(UnifiedEventDO::getPublishStatus, EventsPublishServiceImpl.PUBLISH_STATUS_ACCEPTED)
+ .eq(UnifiedEventDO::getDeleted, false)
+ .le(UnifiedEventDO::getVisibleFrom, now)
+ // 空 cursor 首连只冻结当前最大可见 sequence，不 replay 历史事件。
+ .orderByDesc(UnifiedEventDO::getSequenceNo)
+ .last("LIMIT 1"));
+ return latest == null ? 0L : latest.getSequenceNo();
+ }
+
+ /**
+ * 首次写入统一事件；并发重复 commandId 或 source tuple 时不覆盖既有事件，只由服务层回查幂等结果。
+ *
+ * 注意：这里不写 sequence_no 字段，必须由 V16 DDL 的 PostgreSQL sequence/default 分配。
+ */
+ @Insert("""
+ INSERT INTO muse_unified_event(command_id, event_id, event_type, owner_user_id,
+ source_owner, source_type, source_id, source_revision,
+ resource_type, resource_id, payload_summary,
+ publish_status, publish_error_code, visible_from, emitted_at, tenant_id)
+ VALUES (#{commandId}, #{eventId}, #{eventType}, #{ownerUserId},
+ #{sourceOwner}, #{sourceType}, #{sourceId}, #{sourceRevision},
+ #{resourceType}, #{resourceId},
+ #{payloadSummary,typeHandler=cn.iocoder.muse.module.events.dal.type.JsonbStringTypeHandler},
+ #{publishStatus}, #{publishErrorCode}, #{visibleFrom}, #{emittedAt}, #{tenantId})
+ ON CONFLICT DO NOTHING
+ """)
+ int insertIgnore(UnifiedEventDO event);
+
+}
diff --git a/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/dal/type/JsonbStringTypeHandler.java b/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/dal/type/JsonbStringTypeHandler.java
new file mode 100644
index 00000000..893ab291
--- /dev/null
+++ b/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/dal/type/JsonbStringTypeHandler.java
@@ -0,0 +1,48 @@
+package cn.iocoder.muse.module.events.dal.type;
+
+import org.apache.ibatis.type.BaseTypeHandler;
+import org.apache.ibatis.type.JdbcType;
+
+import java.sql.CallableStatement;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Types;
+
+/**
+ * Events 模块本地 PostgreSQL JSONB 字符串 TypeHandler。
+ *
+ * 统一事件只保存已经脱敏后的 JSON 摘要；这里用 JDBC OTHER 交给 PostgreSQL JSONB 解析，
+ * 避免把 JSON 字符串再次序列化成字符串字面量。
+ */
+public class JsonbStringTypeHandler extends BaseTypeHandler {
+
+ @Override
+ public void setNonNullParameter(PreparedStatement ps, int i, String parameter, JdbcType jdbcType)
+ throws SQLException {
+ ps.setObject(i, parameter, Types.OTHER);
+ }
+
+ @Override
+ public String getNullableResult(ResultSet rs, String columnName) throws SQLException {
+ return readJson(rs.getObject(columnName));
+ }
+
+ @Override
+ public String getNullableResult(ResultSet rs, int columnIndex) throws SQLException {
+ return readJson(rs.getObject(columnIndex));
+ }
+
+ @Override
+ public String getNullableResult(CallableStatement cs, int columnIndex) throws SQLException {
+ return readJson(cs.getObject(columnIndex));
+ }
+
+ private String readJson(Object value) {
+ if (value == null) {
+ return null;
+ }
+ return value.toString();
+ }
+
+}
diff --git a/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/domain/EventsCursor.java b/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/domain/EventsCursor.java
new file mode 100644
index 00000000..f6332285
--- /dev/null
+++ b/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/domain/EventsCursor.java
@@ -0,0 +1,63 @@
+package cn.iocoder.muse.module.events.domain;
+
+import org.springframework.util.StringUtils;
+
+/**
+ * Events SSE Last-Event-ID 游标。
+ *
+ * 游标解析必须 fail-closed：非法输入只返回安全错误码，不能把原始 cursor 或解析异常透出到浏览器。
+ */
+public record EventsCursor(Long sequenceNo, boolean valid, String errorCode, String errorMessage) {
+
+ private static final String PREFIX = "muse:";
+ public static final String ERROR_CURSOR_INVALID = "EVENTS_CURSOR_INVALID";
+
+ /**
+ * 解析 Events SSE 游标。
+ *
+ * @param rawCursor 浏览器传入的 lastEventId；空值表示首连，不 replay 历史
+ * @return 解析结果；非法 cursor 由调用方发送 OpenAPI error event
+ */
+ public static EventsCursor parse(String rawCursor) {
+ if (!StringUtils.hasText(rawCursor)) {
+ return new EventsCursor(null, true, null, null);
+ }
+ String cursor = rawCursor.trim();
+ if (!cursor.startsWith(PREFIX)) {
+ return invalid();
+ }
+ String sequenceText = cursor.substring(PREFIX.length());
+ if (!StringUtils.hasText(sequenceText)) {
+ return invalid();
+ }
+ try {
+ long sequenceNo = Long.parseLong(sequenceText);
+ if (sequenceNo < 0L) {
+ return invalid();
+ }
+ return new EventsCursor(sequenceNo, true, null, null);
+ } catch (NumberFormatException ex) {
+ return invalid();
+ }
+ }
+
+ /**
+ * SSE id 使用全局统一格式，避免客户端把裸数字和其他 stream 的游标混用。
+ */
+ public static String toSseId(Long sequenceNo) {
+ return PREFIX + sequenceNo;
+ }
+
+ public boolean isInitialConnection() {
+ return valid && sequenceNo == null;
+ }
+
+ public boolean isValid() {
+ return valid;
+ }
+
+ private static EventsCursor invalid() {
+ return new EventsCursor(null, false, ERROR_CURSOR_INVALID, "Invalid events cursor");
+ }
+
+}
diff --git a/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/domain/EventsPayloadSanitizer.java b/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/domain/EventsPayloadSanitizer.java
new file mode 100644
index 00000000..71c01b4c
--- /dev/null
+++ b/muse-cloud/muse-module-events/muse-module-events-server/src/main/java/cn/iocoder/muse/module/events/domain/EventsPayloadSanitizer.java
@@ -0,0 +1,96 @@
+package cn.iocoder.muse.module.events.domain;
+
+import cn.iocoder.muse.framework.common.util.json.JsonUtils;
+import com.fasterxml.jackson.core.type.TypeReference;
+import lombok.Value;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+
+import java.util.*;
+import java.util.regex.Pattern;
+
+/**
+ * Events payload 脱敏器。
+ *
+ * source owner 可能来自外部模型、知识库或支付/授权链路；这里递归检查 key 与 value，
+ * 发现授权头、bearer token、secret 或 provider raw body 后，事件会被拒绝且只保存脱敏摘要。
+ */
+@Component
+public class EventsPayloadSanitizer {
+
+ private static final String REDACTED = "[REDACTED]";
+
+ private static final Pattern SECRET_KEY_PATTERN = Pattern.compile(
+ "(?i)(authorization|bearer|token|access[-_]?token|refresh[-_]?token|api[-_]?key|secret|password|credential|provider.*raw.*body|raw.*body)");
+
+ private static final Pattern SECRET_VALUE_PATTERN = Pattern.compile(
+ "(?i)(bearer\\s+[a-z0-9._\\-]+|authorization\\s*[:=]|api[-_]?key\\s*[:=]|secret\\s*[:=]|token\\s*[:=])");
+
+ public SanitizedPayload sanitize(Map payloadSummary) {
+ Map source = payloadSummary == null ? Map.of() : payloadSummary;
+ Detection detection = new Detection();
+ Object sanitized = sanitizeValue(source, detection);
+ String json = JsonUtils.toJsonString(sanitized);
+ return new SanitizedPayload(json, detection.containsSecret);
+ }
+
+ @SuppressWarnings("unchecked")
+ private Object sanitizeValue(Object value, Detection detection) {
+ if (value instanceof Map map) {
+ Map sanitizedMap = new LinkedHashMap<>();
+ map.forEach((key, childValue) -> {
+ String normalizedKey = String.valueOf(key);
+ if (isSensitiveKey(normalizedKey)) {
+ detection.containsSecret = true;
+ sanitizedMap.put(normalizedKey, REDACTED);
+ return;
+ }
+ sanitizedMap.put(normalizedKey, sanitizeValue(childValue, detection));
+ });
+ return sanitizedMap;
+ }
+ if (value instanceof Collection collection) {
+ List