feat: 新增设备 API 拉取采集及配套处理能力

- 接入 NDRServer 告警和流量日志，支持凭据与任务管理
- 支持分页拉取、窗口拆分、调度优先级及 WAL 持久化后推进断点
- 增加迟到告警规则重放，优化多条泛化规则选择和事件时间处理
- 补充回归测试、数据库升级脚本及部署文档

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/LogNormalProcessor.java

@@ -95,6 +95,7 @@ public class LogNormalProcessor {
     private final NormalizationShadowComparator shadowComparator;
     private final NormalizationShadowExecutor shadowExecutor;
     private final NormalizationRuleMetrics globalMetrics;
+    private final RuleSelectionEngine ruleSelectionEngine;
     private final Map<String, AtomicLong> shadowLogWindows = new ConcurrentHashMap<>();
 
     /**
@@ -111,6 +112,7 @@ public class LogNormalProcessor {
         String strSyslogTopic;
         LocalDateTime fallbackFirstReceivedAt;
         LocalDateTime initialReceivedAt;
+        LocalDateTime eventAt;
         boolean isSaveNonNormal = false;
         SyslogNonNormalMessage syslogNonNormalMessage = new SyslogNonNormalMessage();
         List<RuleSource> dmNormalizeRuleList;
@@ -209,6 +211,8 @@ public class LogNormalProcessor {
         this.shadowComparator = shadowComparator;
         this.shadowExecutor = shadowExecutor;
         this.globalMetrics = globalMetrics;
+        this.ruleSelectionEngine = new RuleSelectionEngine(
+                compiledRuleCache, optimizationProperties, globalMetrics);
     }
 
     private static CompiledRuleCache defaultCompiledRuleCache() {
@@ -230,10 +234,9 @@ public class LogNormalProcessor {
     /**
      * 处理一条日志，并显式传入 Kafka 链路已解析好的首次接收时间。
      *
-     * <p>该时间会同时写入标准化表的 created_at 和 log_time。created_at 表示日志首次进入系统，
-     * log_time 是现场标准化表的 RANGE 分区键。原始业务代码用本次处理时间覆盖 log_time；这里改为
-     * 稳定的首次接收时间，使 Raw、Retry 和 Replay 始终进入同一日分区。设备事件时间继续由
-     * collect_time_ts 保存，不与分区时间混用。</p>
+     * <p>标准化表的 {@code created_at} 使用首次接收时间，{@code log_time} 使用设备事件时间。
+     * 现场两张标准化表都按 {@code log_time} 做 RANGE 分区，因此 API 历史拉取会进入事件发生日分区；
+     * TCP/UDP 的 event_time 等于 receive_time，原有分区行为保持不变。</p>
      */
     public ProcessingOutcome process(String logMsg, String syslogUUID, String syslogTopic,
                                      LocalDateTime fallbackFirstReceivedAt) {
@@ -310,6 +313,7 @@ public class LogNormalProcessor {
                 ? kafkaConsumerProperties.getTopic() : input.getTopic();
         ctx.fallbackFirstReceivedAt = input.getReceivedAt();
         ctx.initialReceivedAt = input.getReceivedAt();
+        ctx.eventAt = input.getEventAt();
         ctx.strKafkaMessage = input.getFullMessage();
         ctx.strLogMsg = input.getSyslogBody();
         ctx.strDeviceInfo = input.getDeviceHeader();
@@ -354,6 +358,9 @@ public class LogNormalProcessor {
             ctx.initialReceivedAt = resolveInitialReceivedAt(
                     mapdev.get("receive_time"), ctx.fallbackFirstReceivedAt);
         }
+        if (ctx.eventAt == null) {
+            ctx.eventAt = resolveEventAt(mapdev.get("event_time"), ctx.initialReceivedAt);
+        }
         long deviceID = Long.parseLong(mapdev.get("device_id"));
         logger.debug("标准化开始: eventId={}, deviceId={}", ctx.strSyslogUUID, deviceID);
         ctx.strDataType = mapdev.get("data_type");
@@ -368,6 +375,14 @@ public class LogNormalProcessor {
                 // 因此只跳过达梦标准化并允许 ACK；系统依赖和规则执行异常仍由下方异常链处理。
                 return NormalizationResult.filtered();
             }
+            NormalizationOptimizationProperties.RuleSelection selectionConfig =
+                    optimizationProperties.getRuleSelection();
+            if (ctx.dmNormalizeRuleList.size() > 1 && selectionConfig.isEnabled()) {
+                return executeSelected(ctx, deviceID);
+            }
+            if (ctx.dmNormalizeRuleList.size() > 1 && selectionConfig.isShadowEnabled()) {
+                shadowSelect(ctx);
+            }
             if (optimizationProperties.getCompiledRule().isEnabled()) {
                 return executeCompiled(ctx, deviceID);
             }
@@ -395,6 +410,217 @@ public class LogNormalProcessor {
         return finish(ctx);
     }
 
+    /**
+     * 多规则唯一命中路径：所有规则只做无副作用解析，只有最终胜出规则进入映射及后续阶段。
+     */
+    private NormalizationResult executeSelected(ProcessContext ctx, long deviceId) {
+        List<RuleSelectionEngine.Candidate<ExecutionPlan>> candidates = collectCandidates(ctx);
+        RuleSelectionEngine.Decision<ExecutionPlan> decision =
+                ruleSelectionEngine.select(candidates, ctx.strLogMsg, false);
+        if (!decision.isSelected()) {
+            saveSelectionFailure(ctx, deviceId, "未命中规则",
+                    "同设备多规则均未形成可用归属候选");
+            // 规则均未命中是确定性的业务终态；非标保存成功后直接 ACK，避免 Retry
+            // 为同一个 syslog_uuid 重复写入随机 ID 的非标记录。保存失败仍会由上层异常链重试。
+            return NormalizationResult.filtered();
+        }
+
+        RuleSelectionEngine.Candidate<ExecutionPlan> selected = decision.getWinner();
+        BusinessResult prepared = null;
+        if (decision.isAmbiguous()) {
+            if (optimizationProperties.getRuleSelection().isStrictAmbiguity()) {
+                List<BusinessResult> results = new ArrayList<>();
+                Map<Long, BusinessResult> byRule = new LinkedHashMap<>();
+                for (RuleSelectionEngine.Candidate<ExecutionPlan> tied
+                        : decision.getTiedCandidates()) {
+                    BusinessResult result = evaluateBusiness(ctx, tied.getPayload());
+                    results.add(result);
+                    byRule.put(tied.getSource().getId(), result);
+                }
+                if (!businessResultsEquivalent(results)) {
+                    globalMetrics.ruleSelection("conflicting_tie");
+                    saveSelectionFailure(ctx, deviceId, "多规则归属歧义",
+                            "候选评分相同但业务结果不同, candidateRuleIds="
+                                    + candidateRuleIds(decision.getTiedCandidates()));
+                    return NormalizationResult.filtered();
+                }
+                globalMetrics.ruleSelection("equivalent_tie");
+                prepared = byRule.get(selected.getSource().getId());
+            }
+            if (optimizationProperties.getRuleSelection().isDetailLogEnabled()) {
+                logger.warn("同设备多规则归属并列: eventHash={}, candidateRuleIds={}, selectedRuleId={}, mode={}",
+                        shadowComparator.eventHash(ctx.strSyslogUUID),
+                        candidateRuleIds(decision.getTiedCandidates()),
+                        selected.getSource().getId(), decision.getMode());
+            }
+        }
+
+        if (prepared == null) prepared = evaluateBusiness(ctx, selected.getPayload());
+        if (prepared.filtered) {
+            ctx.filteredCount++;
+            return finish(ctx);
+        }
+        assemble(ctx, deviceId, selected.getSource(), prepared.destination);
+        return finish(ctx);
+    }
+
+    /** 关闭切换开关时仅做归属影子计算，不改变旧的多结果行为。 */
+    private void shadowSelect(ProcessContext ctx) {
+        try {
+            List<RuleSelectionEngine.Candidate<ExecutionPlan>> candidates = collectCandidates(ctx);
+            ruleSelectionEngine.select(candidates, ctx.strLogMsg, true);
+        } catch (Exception failure) {
+            // 影子链路不能影响生产结果。
+            logger.debug("规则归属影子计算失败: eventHash={}, error={}",
+                    shadowComparator.eventHash(ctx.strSyslogUUID),
+                    failure.getClass().getSimpleName());
+        }
+    }
+
+    private List<RuleSelectionEngine.Candidate<ExecutionPlan>> collectCandidates(ProcessContext ctx) {
+        List<RuleSelectionEngine.Candidate<ExecutionPlan>> candidates = new ArrayList<>();
+        boolean compiledEnabled = optimizationProperties.getCompiledRule().isEnabled();
+        for (RuleSource source : ctx.dmNormalizeRuleList) {
+            try {
+                ExecutionPlan plan;
+                if (compiledEnabled) {
+                    try {
+                        CompiledRule compiled = compiledRuleCache.get(source, ctx.executionMetrics);
+                        long parseStarted = System.nanoTime();
+                        ParsedMessage parsed = ctx.messageParseContext.parse(
+                                compiled, ctx.strLogMsg, ctx.executionMetrics);
+                        ctx.executionMetrics.addMessageParseNanos(System.nanoTime() - parseStarted);
+                        if (parsed == null || parsed.getFields().isEmpty()) continue;
+                        plan = ExecutionPlan.compiled(source, compiled, parsed);
+                    } catch (CompiledRuleCompiler.RuleCompilationException incompatibleRule) {
+                        plan = legacyCandidate(ctx, source);
+                    }
+                } else {
+                    plan = legacyCandidate(ctx, source);
+                }
+                if (plan != null && !plan.parsedFields.isEmpty()) {
+                    candidates.add(new RuleSelectionEngine.Candidate<>(
+                            source, plan.parsedFields, plan));
+                }
+            } catch (Exception candidateFailure) {
+                // 单个候选失败不是整条日志失败；全部失败后由唯一的非标出口处理。
+                logger.debug("规则候选解析失败: eventHash={}, ruleId={}, error={}",
+                        shadowComparator.eventHash(ctx.strSyslogUUID), source.getId(),
+                        candidateFailure.getClass().getSimpleName());
+            }
+        }
+        return candidates;
+    }
+
+    private ExecutionPlan legacyCandidate(ProcessContext ctx, RuleSource source) {
+        long parseStarted = System.nanoTime();
+        globalMetrics.ruleJsonParse();
+        LinkedHashMap<String, Object> parsed = getMessageToMap(
+                source.asLegacyMap(), ctx.strLogMsg, source.getDataType());
+        ctx.executionMetrics.addMessageParseNanos(System.nanoTime() - parseStarted);
+        return parsed == null || parsed.isEmpty() ? null
+                : ExecutionPlan.legacy(source, parsed);
+    }
+
+    private BusinessResult evaluateBusiness(ProcessContext ctx, ExecutionPlan plan) {
+        long mappingStarted = System.nanoTime();
+        Map<String, Object> destination;
+        if (plan.compiledRule != null) {
+            destination = plan.compiledRule.map(plan.compiledParsed);
+        } else {
+            globalMetrics.ruleJsonParse();
+            List<HashMap<String, Object>> destinationColumns =
+                    getRuleContentMappers(plan.source.asLegacyMap());
+            if (destinationColumns == null) {
+                throw new IllegalArgumentException("规则 mappers 无法解析: ruleId=" + plan.source.getId());
+            }
+            destination = getColumnMap(getNormalColumnList(
+                    new LinkedHashMap<>(plan.parsedFields), destinationColumns));
+        }
+        ctx.executionMetrics.addFieldMappingNanos(System.nanoTime() - mappingStarted);
+
+        long filterStarted = System.nanoTime();
+        boolean filtered;
+        if (plan.compiledRule != null) {
+            filtered = plan.compiledRule.isFiltered(destination);
+            if (!filtered) plan.compiledRule.complete(destination);
+        } else {
+            globalMetrics.ruleJsonParse();
+            filtered = logDataFilterService.evaluateFilterRule(
+                    plan.source.getRuleContent(), destination);
+            if (!filtered) {
+                globalMetrics.ruleJsonParse();
+                logDataCompleteService.processDataCompletion(
+                        plan.source.getRuleContent(), destination);
+            }
+        }
+        ctx.executionMetrics.addFilterCompleteNanos(System.nanoTime() - filterStarted);
+        return new BusinessResult(filtered, destination);
+    }
+
+    private static boolean businessResultsEquivalent(List<BusinessResult> results) {
+        if (results.isEmpty()) return true;
+        BusinessResult first = results.get(0);
+        for (int i = 1; i < results.size(); i++) {
+            BusinessResult current = results.get(i);
+            if (first.filtered != current.filtered) return false;
+            if (!first.filtered && !Objects.equals(first.destination, current.destination)) return false;
+        }
+        return true;
+    }
+
+    private static List<Long> candidateRuleIds(
+            List<? extends RuleSelectionEngine.Candidate<?>> candidates) {
+        List<Long> ids = new ArrayList<>();
+        for (RuleSelectionEngine.Candidate<?> candidate : candidates) {
+            ids.add(candidate.getSource().getId());
+        }
+        return ids;
+    }
+
+    private void saveSelectionFailure(ProcessContext ctx, long deviceId,
+                                      String reason, String detail) {
+        ctx.syslogNonNormalMessage.setReason(reason);
+        ctx.syslogNonNormalMessage.setReasonDetail(detail);
+        if (!ctx.shadowMode && !ctx.isSaveNonNormal) {
+            SaveNonNormalMessage(ctx, deviceId, DateTime.now());
+        }
+    }
+
+    private static final class ExecutionPlan {
+        private final RuleSource source;
+        private final CompiledRule compiledRule;
+        private final ParsedMessage compiledParsed;
+        private final Map<String, Object> parsedFields;
+
+        private ExecutionPlan(RuleSource source, CompiledRule compiledRule,
+                              ParsedMessage compiledParsed, Map<String, Object> parsedFields) {
+            this.source = source;
+            this.compiledRule = compiledRule;
+            this.compiledParsed = compiledParsed;
+            this.parsedFields = Collections.unmodifiableMap(new LinkedHashMap<>(parsedFields));
+        }
+
+        private static ExecutionPlan compiled(RuleSource source, CompiledRule rule,
+                                              ParsedMessage parsed) {
+            return new ExecutionPlan(source, rule, parsed, parsed.getFields());
+        }
+
+        private static ExecutionPlan legacy(RuleSource source, Map<String, Object> parsed) {
+            return new ExecutionPlan(source, null, null, parsed);
+        }
+    }
+
+    private static final class BusinessResult {
+        private final boolean filtered;
+        private final Map<String, Object> destination;
+
+        private BusinessResult(boolean filtered, Map<String, Object> destination) {
+            this.filtered = filtered;
+            this.destination = destination;
+        }
+    }
+
     private NormalizationResult executeCompiled(ProcessContext ctx, long deviceId) {
         for (RuleSource source : ctx.dmNormalizeRuleList) {
             CompiledRule rule;
@@ -561,6 +787,7 @@ public class LogNormalProcessor {
         shadow.strSyslogTopic = source.strSyslogTopic;
         shadow.fallbackFirstReceivedAt = source.fallbackFirstReceivedAt;
         shadow.initialReceivedAt = source.initialReceivedAt;
+        shadow.eventAt = source.eventAt;
         shadow.dmNormalizeRuleList = source.dmNormalizeRuleList;
         shadow.batchEnrichmentContext = source.batchEnrichmentContext;
         shadow.executionMetrics = new RuleExecutionMetrics();
@@ -793,11 +1020,9 @@ public class LogNormalProcessor {
             }
             //补全日志基础信息
             columnMap.put("device_id", deviceId);
-            // 现场两张标准化主表均按 log_time 做 RANGE 分区。原始业务代码会无条件用 DateTime.now()
-            // 覆盖 log_time，其实际意图是按系统接收/处理日期分区，而不是按设备事件日期分区。
-            // 这里把 log_time 和 created_at 都固定为最初 receive_time：既保持原来的分区口径，
-            // 又避免 Kafka 积压、Retry/Replay 或跨日重试改变分区。设备事件时间保留在 collect_time_ts。
-            applyInitialReceiveTime(columnMap, ctx.initialReceivedAt);
+            // log_time 是两张标准化表的 RANGE 分区键，必须使用稳定的设备事件时间；created_at
+            // 记录日志首次进入系统的时间。这样跨日 API 回补进入事件发生日分区，同时仍能计算接入延迟。
+            applyIngressTimes(columnMap, ctx.eventAt, ctx.initialReceivedAt);
             // 输入字符串只用于计算固定 128 位 UUID，最终 id 仍是与旧版相同长度的 36 字符 UUID。
             // 同一 event_id + 规则 + 版本在重试/回放时生成相同 id，与稳定 log_time 组成每日本地唯一键。
             columnMap.put("id", deterministicResultId(ctx.strSyslogUUID, normalizeRuleId, LEGACY_RULE_VERSION));
@@ -822,8 +1047,9 @@ public class LogNormalProcessor {
         }
     }
 
-    static void applyInitialReceiveTime(Map<String, Object> columnMap, LocalDateTime initialReceivedAt) {
-        columnMap.put("log_time", initialReceivedAt);
+    static void applyIngressTimes(Map<String, Object> columnMap, LocalDateTime eventAt,
+                                  LocalDateTime initialReceivedAt) {
+        columnMap.put("log_time", eventAt);
         columnMap.put("created_at", initialReceivedAt);
     }
 
@@ -902,7 +1128,9 @@ public class LogNormalProcessor {
             normalMessage.setCollectTaskId(collect_id);
             if (collectTask != null)
                 normalMessage.setCollectTaskName(collectTask.getTaskName());
-            this.messageService.saveMessage(normalMessage);
+            if (!this.messageService.saveMessage(normalMessage)) {
+                throw new IllegalStateException("非标日志服务返回保存失败");
+            }
             ctx.isSaveNonNormal = true;
         } catch (Exception e) {
             logger.error("SaveNonNormalMessage  失败：");
@@ -937,6 +1165,14 @@ public class LogNormalProcessor {
         return LocalDateTime.now(RECEIVE_TIME_ZONE);
     }
 
+    /** 新消息使用 event_time；历史 TCP/UDP 或旧 Kafka 消息缺失时回退到 receive_time。 */
+    static LocalDateTime resolveEventAt(String eventTime, LocalDateTime initialReceivedAt) {
+        if (eventTime == null || eventTime.trim().isEmpty()) {
+            return initialReceivedAt;
+        }
+        return IngressReceiveTimeUtil.parse(eventTime);
+    }
+
     public static void main(String[] args) {
         // 旧入口保留，仅用于本地测试
     }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/SysLogProcessor.java

@@ -5,13 +5,11 @@ import com.Modules.NormalData.compiled.BatchRuleSnapshot;
 import com.Modules.NormalData.compiled.RuleExecutionMetrics;
 import com.Modules.NormalData.compiled.PreparedNormalizationInput;
 import com.common.util.SyslogParser;
-import com.common.util.IngressReceiveTimeUtil;
 import com.config.AppProperties;
 import com.config.KafkaConsumerProperties;
 import com.common.service.DmStandardizedBatchWriter;
 import com.common.service.IngressStatsProjectionWriter;
 import com.common.entity.IngressStatsEvent;
-import com.influxdb.client.domain.WritePrecision;
 import com.influxdb.client.write.Point;
 import com.influx.raw.InfluxRawPointFactory;
 import com.influx.raw.PointTimeResolver;
@@ -41,20 +39,13 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-import java.time.Instant;
 import java.time.LocalDateTime;
-import java.time.ZoneId;
-import java.time.format.DateTimeFormatter;
 
 import com.common.util.Sm4Util;
 @Slf4j
 @Component
 public class SysLogProcessor {
 
-    private static final ZoneId RECEIVE_TIME_ZONE = ZoneId.of("Asia/Shanghai");
-    private static final DateTimeFormatter MEASUREMENT_DATE_FORMATTER =
-            DateTimeFormatter.ofPattern("yyyyMMdd");
-
     @Autowired
     @Qualifier("logNormalProcessorExecutor")
     private Executor taskExecutor;
@@ -363,36 +354,37 @@ public class SysLogProcessor {
         }
 
         String receiveTime = deviceInfo.get("receive_time");
-        ReceiveTimeParser.ParsedReceiveTime parsed = ReceiveTimeParser.parse(receiveTime);
+        ReceiveTimeParser.ParsedReceiveTime parsedReceiveTime = ReceiveTimeParser.parse(receiveTime);
+        String eventTime = effectiveEventTime(deviceInfo.get("event_time"), receiveTime);
+        ReceiveTimeParser.ParsedReceiveTime parsedEventTime = ReceiveTimeParser.parse(eventTime);
         boolean headerHasPointTime = deviceInfo.containsKey("point_time_ns");
         Long envelopePointTime = messageContext.getPointTimeNs();
         Long headerPointTime = pointTimeResolver.resolve(headerHasPointTime,
-                deviceInfo.get("point_time_ns"), parsed.getEpochMillis(), false);
+                deviceInfo.get("point_time_ns"), parsedEventTime.getEpochMillis(), false);
         if (envelopePointTime != null) {
-            PointTimeResolver.validate(envelopePointTime, parsed.getEpochMillis());
+            PointTimeResolver.validate(envelopePointTime, parsedEventTime.getEpochMillis());
             if (headerPointTime != null && !headerPointTime.equals(envelopePointTime)) {
                 throw new IllegalArgumentException("Retry/Replay point_time_ns 与原始 Kafka 载荷不一致");
             }
         }
         Long pointTimeNs = envelopePointTime != null ? envelopePointTime : headerPointTime;
         if (pointTimeNs == null && rawStorageProperties.getWriteSchema() == RawStorageProperties.WriteSchema.V2) {
-            pointTimeNs = pointTimeResolver.resolve(false, null, parsed.getEpochMillis(), true);
+            pointTimeNs = pointTimeResolver.resolve(false, null, parsedEventTime.getEpochMillis(), true);
         }
         messageContext.setPointTimeNs(pointTimeNs);
         String originalTopic = messageContext.getOriginalTopic();
         InfluxRawPointFactory.RawPoint rawPoint = rawPointFactory.create(eventId, decryptedMessage,
                 receiveTime, deviceInfo.get("device_id"), deviceInfo.get("device_collect_id"),
-                originalTopic, parsed, pointTimeNs);
-        ResolvedInfluxTime influxTime = new ResolvedInfluxTime(parsed.getEpochMillis(),
-                parsed.getMeasurementDate(), parsed.getLocalDateTime());
+                originalTopic, parsedEventTime, pointTimeNs);
         Long deviceId = optionalLong(deviceInfo.get("device_id"));
         Long collectId = optionalLong(deviceInfo.get("device_collect_id"));
         PreparedNormalizationInput normalizationInput = new PreparedNormalizationInput(
-                eventId, originalTopic, influxTime.getReceivedAt(), deviceId, collectId,
+                eventId, originalTopic, parsedReceiveTime.getLocalDateTime(),
+                parsedEventTime.getLocalDateTime(), deviceId, collectId,
                 deviceInfo.get("data_type"), SyslogParser.substringAfterFirstCloseBracket(decryptedMessage),
                 decryptedMessage, deviceHeader, deviceInfo);
         return new PreparedMessage(record, messageContext, decryptedMessage, eventId, rawPoint.getPoint(),
-                rawPoint.getBucket(), rawPoint.isV2(), influxTime.getReceivedAt(), deviceId, collectId,
+                rawPoint.getBucket(), rawPoint.isV2(), parsedReceiveTime.getLocalDateTime(), deviceId, collectId,
                 startTime, normalizationInput);
     }
 
@@ -631,7 +623,14 @@ public class SysLogProcessor {
                 body = SyslogParser.substringAfterFirstCloseBracket(message);
                 info = SyslogParser.parseKeyValuePairs(header);
             }
-            return new PreparedNormalizationInput(eventId, null, receivedAt, deviceId, collectId,
+            LocalDateTime eventAt = receivedAt;
+            if (!info.isEmpty()) {
+                String rawEventTime = effectiveEventTime(info.get("event_time"), info.get("receive_time"));
+                if (rawEventTime != null && !rawEventTime.trim().isEmpty()) {
+                    eventAt = ReceiveTimeParser.parse(rawEventTime).getLocalDateTime();
+                }
+            }
+            return new PreparedNormalizationInput(eventId, null, receivedAt, eventAt, deviceId, collectId,
                     info.get("data_type"), body,
                     message, header, info);
         }
@@ -764,26 +763,28 @@ public class SysLogProcessor {
         return partitions.toString();
     }
 
-    /**
-     * 为 InfluxDB 选择跨 Kafka Retry/Replay 稳定的时间戳。
-     * 优先使用接入端 receive_time；旧消息缺失或格式异常时，回退到首次 Kafka 接收时间。
-     */
-    static ResolvedInfluxTime resolveInfluxTime(String receiveTime, Instant ignoredFallbackTime) {
-        ReceiveTimeParser.ParsedReceiveTime parsed = ReceiveTimeParser.parse(receiveTime);
+    /** 新消息使用 event_time；历史 Kafka 消息缺失该字段时回退到 receive_time。 */
+    static ResolvedInfluxTime resolveInfluxTime(String eventTime, String receiveTime) {
+        ReceiveTimeParser.ParsedReceiveTime parsed = ReceiveTimeParser.parse(
+                effectiveEventTime(eventTime, receiveTime));
         return new ResolvedInfluxTime(
                 parsed.getEpochMillis(), parsed.getMeasurementDate(), parsed.getLocalDateTime());
     }
 
-    /** InfluxDB 点的时间戳和 measurement 日期必须由同一个接收时间计算。 */
+    static String effectiveEventTime(String eventTime, String receiveTime) {
+        return eventTime == null || eventTime.trim().isEmpty() ? receiveTime : eventTime;
+    }
+
+    /** InfluxDB 点的时间戳和 measurement 日期必须由同一个事件时间计算。 */
     static final class ResolvedInfluxTime {
         private final long timestampMs;
         private final String measurementDate;
-        private final LocalDateTime receivedAt;
+        private final LocalDateTime eventAt;
 
-        private ResolvedInfluxTime(long timestampMs, String measurementDate, LocalDateTime receivedAt) {
+        private ResolvedInfluxTime(long timestampMs, String measurementDate, LocalDateTime eventAt) {
             this.timestampMs = timestampMs;
             this.measurementDate = measurementDate;
-            this.receivedAt = receivedAt;
+            this.eventAt = eventAt;
         }
 
         long getTimestampMs() {
@@ -794,8 +795,8 @@ public class SysLogProcessor {
             return measurementDate;
         }
 
-        LocalDateTime getReceivedAt() {
-            return receivedAt;
+        LocalDateTime getEventAt() {
+            return eventAt;
         }
     }
 

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCache.java

@@ -28,13 +28,16 @@ public class CompiledRuleCache {
     private static final ObjectMapper DIAGNOSTIC_JSON = new ObjectMapper();
     private static final List<String> EXECUTION_RULE_FIELDS = Arrays.asList(
             "cropper_params", "kv_params", "sep", "regexp", "mappers",
-            "filters_params", "complete_columns");
+            "match_params", "filters_params", "complete_columns");
 
     private final CompiledRuleCompiler compiler;
     private final Cache<String, CompiledRule> compiled;
+    private final Cache<String, RuleMatchProfile> matchProfiles;
     private final Cache<String, FailureRecord> failures;
+    private final Cache<String, FailureRecord> matchFailures;
     private final Cache<String, Boolean> loggedFailures;
     private final Cache<Long, Set<String>> keysByRule;
+    private final Cache<Long, Set<String>> matchKeysByRule;
     private final NormalizationRuleMetrics metrics;
     private final boolean ruleDetailLogEnabled;
     private final LongAdder compileSuccess = new LongAdder();
@@ -58,6 +61,10 @@ public class CompiledRuleCache {
                 .maximumSize(config.getLocalLimit())
                 .expireAfterAccess(config.getExpireAfterAccessMinutes(), TimeUnit.MINUTES)
                 .build();
+        this.matchKeysByRule = Caffeine.newBuilder()
+                .maximumSize(config.getLocalLimit())
+                .expireAfterAccess(config.getExpireAfterAccessMinutes(), TimeUnit.MINUTES)
+                .build();
         this.compiled = Caffeine.newBuilder()
                 .maximumSize(config.getLocalLimit())
                 .expireAfterAccess(config.getExpireAfterAccessMinutes(), TimeUnit.MINUTES)
@@ -66,10 +73,23 @@ public class CompiledRuleCache {
                     if (key != null && value != null) unlink(((CompiledRule) value).getRuleId(), key.toString());
                 })
                 .build();
+        this.matchProfiles = Caffeine.newBuilder()
+                .maximumSize(config.getLocalLimit())
+                .expireAfterAccess(config.getExpireAfterAccessMinutes(), TimeUnit.MINUTES)
+                .removalListener((key, value, cause) -> {
+                    if (key != null && value != null) {
+                        unlink(matchKeysByRule, sourceRuleId(key.toString()), key.toString());
+                    }
+                })
+                .build();
         this.failures = Caffeine.newBuilder()
                 .maximumSize(Math.min(config.getLocalLimit(), 1_000))
                 .expireAfterWrite(config.getFailureCacheSeconds(), TimeUnit.SECONDS)
                 .build();
+        this.matchFailures = Caffeine.newBuilder()
+                .maximumSize(Math.min(config.getLocalLimit(), 1_000))
+                .expireAfterWrite(config.getFailureCacheSeconds(), TimeUnit.SECONDS)
+                .build();
         this.loggedFailures = Caffeine.newBuilder()
                 .maximumSize(Math.min(config.getLocalLimit(), 1_000))
                 .expireAfterAccess(config.getExpireAfterAccessMinutes(), TimeUnit.MINUTES)
@@ -133,15 +153,42 @@ public class CompiledRuleCache {
         });
     }
 
+    /** 获取只包含显式条件和自动字段指纹的缓存画像。 */
+    public RuleMatchProfile getMatchProfile(RuleSource source) {
+        String key = compiler.fingerprint(source);
+        RuleMatchProfile cached = matchProfiles.getIfPresent(key);
+        if (cached != null) return cached;
+        FailureRecord failed = matchFailures.getIfPresent(key);
+        if (failed != null) throw failed.toException(source.getId());
+        return matchProfiles.get(key, ignored -> {
+            FailureRecord recentFailure = matchFailures.getIfPresent(key);
+            if (recentFailure != null) throw recentFailure.toException(source.getId());
+            try {
+                RuleMatchProfile profile = compiler.compileMatchProfile(source);
+                matchKeysByRule.asMap().computeIfAbsent(source.getId(), ruleId ->
+                        Collections.newSetFromMap(new ConcurrentHashMap<>())).add(key);
+                return profile;
+            } catch (RuntimeException failure) {
+                matchFailures.put(key, new FailureRecord(failure));
+                throw failure;
+            }
+        });
+    }
+
     public void invalidateRule(long ruleId) {
         Set<String> keys = keysByRule.asMap().remove(ruleId);
         if (keys != null) compiled.invalidateAll(keys);
+        Set<String> matchKeys = matchKeysByRule.asMap().remove(ruleId);
+        if (matchKeys != null) matchProfiles.invalidateAll(matchKeys);
         for (String key : failures.asMap().keySet()) {
             if (key.startsWith(ruleId + ":")) failures.invalidate(key);
         }
         for (String key : loggedFailures.asMap().keySet()) {
             if (key.startsWith(ruleId + ":")) loggedFailures.invalidate(key);
         }
+        for (String key : matchFailures.asMap().keySet()) {
+            if (key.startsWith(ruleId + ":")) matchFailures.invalidate(key);
+        }
         metrics.compiledCacheGauge(compiled.estimatedSize(), evictions.sum());
     }
 
@@ -185,10 +232,24 @@ public class CompiledRuleCache {
     }
 
     private void unlink(long ruleId, String key) {
-        Set<String> keys = keysByRule.getIfPresent(ruleId);
+        unlink(keysByRule, ruleId, key);
+    }
+
+    private static void unlink(Cache<Long, Set<String>> index, long ruleId, String key) {
+        Set<String> keys = index.getIfPresent(ruleId);
         if (keys == null) return;
         keys.remove(key);
-        if (keys.isEmpty()) keysByRule.invalidate(ruleId);
+        if (keys.isEmpty()) index.invalidate(ruleId);
+    }
+
+    private static long sourceRuleId(String key) {
+        int colon = key.indexOf(':');
+        if (colon <= 0) return -1L;
+        try {
+            return Long.parseLong(key.substring(0, colon));
+        } catch (NumberFormatException ignored) {
+            return -1L;
+        }
     }
 
     private static final class FailureRecord {

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/CompiledRuleCompiler.java

@@ -47,6 +47,9 @@ public class CompiledRuleCompiler {
                 Pattern.compile(separator);
             }
 
+            // match_params 和自动 origin_field 指纹都是规则执行内容的一部分。
+            // 即使当前开关未启用唯一选择，也必须在规则保存/编译时完成结构校验。
+            compileMatchProfile(root, source);
             Map<String, List<CompiledMapper>> mappers = compileMappers(root.get("mappers"), source);
             List<CompiledFilterGroup> filters = compileFilters(root.get("filters_params"), source);
             List<CompiledCompletion> completions = compileCompletions(root.get("complete_columns"), source);
@@ -69,6 +72,29 @@ public class CompiledRuleCompiler {
         return source.getId() + ":" + source.getFingerprint();
     }
 
+    /**
+     * 只编译归属判定画像，不要求 mapper action、过滤和补全能够进入编译执行器。
+     * 历史规则回退旧执行器时仍可因此参与同一套唯一选择算法。
+     */
+    public RuleMatchProfile compileMatchProfile(RuleSource source) {
+        if (source == null) throw new IllegalArgumentException("规则源不能为空");
+        if (blank(source.getRuleContent())) throw invalid(source, "rule_content 不能为空");
+        try {
+            JsonNode root = objectMapper.readTree(source.getRuleContent());
+            if (root == null || !root.isObject()) throw invalid(source, "rule_content 必须是 JSON 对象");
+            return compileMatchProfile(root, source);
+        } catch (RuleCompilationException known) {
+            throw known;
+        } catch (Exception failure) {
+            throw new RuleCompilationException(source.getId(),
+                    "规则归属画像编译失败: " + failure.getClass().getSimpleName(), failure);
+        }
+    }
+
+    private RuleMatchProfile compileMatchProfile(JsonNode root, RuleSource source) {
+        return RuleMatchProfile.compile(root, root.get("mappers"), source, objectMapper);
+    }
+
     private CompiledCropper compileCropper(JsonNode node, RuleSource source) {
         if (node == null || node.isNull()) return null;
         if (!node.isObject()) throw invalid(source, "cropper_params 必须是对象");

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/PreparedNormalizationInput.java

@@ -10,6 +10,7 @@ public final class PreparedNormalizationInput {
     private final String eventId;
     private final String topic;
     private final LocalDateTime receivedAt;
+    private final LocalDateTime eventAt;
     private final Long deviceId;
     private final Long deviceCollectId;
     private final String dataType;
@@ -19,12 +20,14 @@ public final class PreparedNormalizationInput {
     private final Map<String, String> deviceInfo;
 
     public PreparedNormalizationInput(String eventId, String topic, LocalDateTime receivedAt,
+                                      LocalDateTime eventAt,
                                       Long deviceId, Long deviceCollectId, String dataType,
                                       String syslogBody, String fullMessage, String deviceHeader,
                                       Map<String, String> deviceInfo) {
         this.eventId = eventId;
         this.topic = topic;
         this.receivedAt = receivedAt;
+        this.eventAt = eventAt;
         this.deviceId = deviceId;
         this.deviceCollectId = deviceCollectId;
         this.dataType = dataType;
@@ -37,6 +40,7 @@ public final class PreparedNormalizationInput {
     public String getEventId() { return eventId; }
     public String getTopic() { return topic; }
     public LocalDateTime getReceivedAt() { return receivedAt; }
+    public LocalDateTime getEventAt() { return eventAt; }
     public Long getDeviceId() { return deviceId; }
     public Long getDeviceCollectId() { return deviceCollectId; }
     public String getDataType() { return dataType; }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/RuleMatchProfile.java

@@ -0,0 +1,292 @@
+package com.Modules.NormalData.compiled;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/**
+ * 一条规则用于“归属判定”的不可变画像。
+ *
+ * <p>画像只包含显式匹配计划和由 {@code mappers.origin_field} 生成的自动字段指纹，
+ * 不包含消息状态，也不会执行映射、过滤、补全或任何外部调用。</p>
+ */
+public final class RuleMatchProfile {
+    private final boolean explicitEnabled;
+    private final int priority;
+    private final List<MatchCondition> all;
+    private final List<MatchCondition> any;
+    private final List<MatchCondition> none;
+    private final Set<String> originFields;
+
+    private RuleMatchProfile(boolean explicitEnabled, int priority,
+                             List<MatchCondition> all, List<MatchCondition> any,
+                             List<MatchCondition> none, Set<String> originFields) {
+        this.explicitEnabled = explicitEnabled;
+        this.priority = priority;
+        this.all = immutable(all);
+        this.any = immutable(any);
+        this.none = immutable(none);
+        this.originFields = Collections.unmodifiableSet(new LinkedHashSet<>(originFields));
+    }
+
+    static RuleMatchProfile compile(JsonNode root, JsonNode mappers, RuleSource source,
+                                    ObjectMapper objectMapper) {
+        Set<String> origins = compileOrigins(mappers, source, objectMapper);
+        JsonNode params = root.get("match_params");
+        if (params == null || params.isNull()) {
+            return new RuleMatchProfile(false, 0, Collections.emptyList(),
+                    Collections.emptyList(), Collections.emptyList(), origins);
+        }
+        if (!params.isObject()) throw invalid(source, "match_params 必须是对象");
+
+        JsonNode enabledNode = params.get("enabled");
+        if (enabledNode != null && !enabledNode.isBoolean()) {
+            throw invalid(source, "match_params.enabled 必须是布尔值");
+        }
+        boolean enabled = enabledNode != null && enabledNode.booleanValue();
+        JsonNode priorityNode = params.get("priority");
+        if (priorityNode != null && (!priorityNode.isIntegralNumber()
+                || !priorityNode.canConvertToInt())) {
+            throw invalid(source, "match_params.priority 必须是整数");
+        }
+        int priority = priorityNode == null ? 0 : priorityNode.intValue();
+        List<MatchCondition> all = compileConditions(params.get("all"), "all", source, objectMapper);
+        List<MatchCondition> any = compileConditions(params.get("any"), "any", source, objectMapper);
+        List<MatchCondition> none = compileConditions(params.get("none"), "none", source, objectMapper);
+        if (enabled && all.isEmpty() && any.isEmpty() && none.isEmpty()) {
+            throw invalid(source, "match_params.enabled=true 时必须至少配置一个条件");
+        }
+        return new RuleMatchProfile(enabled, priority, all, any, none, origins);
+    }
+
+    public ExplicitEvaluation evaluate(Map<String, Object> parsedFields, String rawMessage) {
+        if (!explicitEnabled) return ExplicitEvaluation.notMatched();
+        int specificity = 0;
+        for (MatchCondition condition : all) {
+            if (!condition.matches(parsedFields, rawMessage)) return ExplicitEvaluation.notMatched();
+            specificity += condition.specificity();
+        }
+        if (!any.isEmpty()) {
+            boolean matched = false;
+            int anySpecificity = 0;
+            for (MatchCondition condition : any) {
+                if (condition.matches(parsedFields, rawMessage)) {
+                    matched = true;
+                    anySpecificity = Math.max(anySpecificity, condition.specificity());
+                }
+            }
+            if (!matched) return ExplicitEvaluation.notMatched();
+            specificity += anySpecificity;
+        }
+        for (MatchCondition condition : none) {
+            if (condition.matches(parsedFields, rawMessage)) return ExplicitEvaluation.notMatched();
+            // 成功排除一种格式同样提高规则的具体度。
+            specificity += condition.specificity();
+        }
+        return new ExplicitEvaluation(true, specificity);
+    }
+
+    public boolean isExplicitEnabled() { return explicitEnabled; }
+    public int getPriority() { return priority; }
+    public Set<String> getOriginFields() { return originFields; }
+
+    private static Set<String> compileOrigins(JsonNode node, RuleSource source,
+                                              ObjectMapper objectMapper) {
+        if (node != null && node.isTextual()) {
+            try {
+                node = objectMapper.readTree(node.textValue());
+            } catch (Exception invalidJson) {
+                throw invalid(source, "mappers 内嵌 JSON 无效");
+            }
+        }
+        if (node == null || !node.isArray()) throw invalid(source, "mappers 必须是数组");
+        Set<String> result = new LinkedHashSet<>();
+        for (JsonNode mapper : node) {
+            JsonNode origin = mapper.get("origin_field");
+            if (origin == null || !origin.isTextual() || blank(origin.textValue())) {
+                throw invalid(source, "mappers.origin_field 不能为空");
+            }
+            result.add(origin.textValue());
+        }
+        return result;
+    }
+
+    private static List<MatchCondition> compileConditions(JsonNode node, String group,
+                                                           RuleSource source,
+                                                           ObjectMapper objectMapper) {
+        if (node == null || node.isNull()) return Collections.emptyList();
+        if (!node.isArray()) throw invalid(source, "match_params." + group + " 必须是数组");
+        List<MatchCondition> conditions = new ArrayList<>();
+        int index = 0;
+        for (JsonNode condition : node) {
+            String path = "match_params." + group + "[" + index + "]";
+            conditions.add(MatchCondition.compile(condition, path, source, objectMapper));
+            index++;
+        }
+        return conditions;
+    }
+
+    private static <T> List<T> immutable(List<T> source) {
+        return Collections.unmodifiableList(new ArrayList<>(source));
+    }
+
+    private static boolean blank(String value) {
+        return value == null || value.trim().isEmpty();
+    }
+
+    private static CompiledRuleCompiler.RuleCompilationException invalid(RuleSource source,
+                                                                          String message) {
+        return new CompiledRuleCompiler.RuleCompilationException(source.getId(), message);
+    }
+
+    public static final class ExplicitEvaluation {
+        private static final ExplicitEvaluation NOT_MATCHED = new ExplicitEvaluation(false, 0);
+        private final boolean matched;
+        private final int specificity;
+
+        private ExplicitEvaluation(boolean matched, int specificity) {
+            this.matched = matched;
+            this.specificity = specificity;
+        }
+
+        static ExplicitEvaluation notMatched() { return NOT_MATCHED; }
+        public boolean isMatched() { return matched; }
+        public int getSpecificity() { return specificity; }
+    }
+
+    private enum Source { PARSED, RAW }
+
+    private enum Operation {
+        EXISTS, NOT_EXISTS, NOT_BLANK, EQ, NE, IN, CONTAINS, REGEXP
+    }
+
+    private static final class MatchCondition {
+        private final Source source;
+        private final String field;
+        private final Operation operation;
+        private final String value;
+        private final Set<String> values;
+        private final Pattern pattern;
+
+        private MatchCondition(Source source, String field, Operation operation, String value,
+                               Set<String> values, Pattern pattern) {
+            this.source = source;
+            this.field = field;
+            this.operation = operation;
+            this.value = value;
+            this.values = values;
+            this.pattern = pattern;
+        }
+
+        static MatchCondition compile(JsonNode node, String path, RuleSource rule,
+                                      ObjectMapper objectMapper) {
+            if (node == null || !node.isObject()) throw invalid(rule, path + " 必须是对象");
+            String sourceText = text(node.get("source"));
+            Source source;
+            try {
+                source = sourceText == null ? Source.PARSED
+                        : Source.valueOf(sourceText.trim().toUpperCase(Locale.ROOT));
+            } catch (IllegalArgumentException unsupported) {
+                throw invalid(rule, path + ".source 不支持: " + sourceText);
+            }
+            String opText = text(node.get("op"));
+            if (blank(opText)) throw invalid(rule, path + ".op 不能为空");
+            Operation operation;
+            try {
+                operation = Operation.valueOf(opText.trim().toUpperCase(Locale.ROOT));
+            } catch (IllegalArgumentException unsupported) {
+                throw invalid(rule, path + ".op 不支持: " + opText);
+            }
+            String field = text(node.get("field"));
+            if (source == Source.PARSED && blank(field)) {
+                throw invalid(rule, path + ".field 不能为空");
+            }
+            if (source == Source.RAW && !(operation == Operation.EQ || operation == Operation.NE
+                    || operation == Operation.NOT_BLANK || operation == Operation.CONTAINS
+                    || operation == Operation.REGEXP)) {
+                throw invalid(rule, path + ".op 不适用于 raw: " + opText);
+            }
+
+            JsonNode valueNode = node.get("value");
+            boolean needsSingleValue = operation == Operation.EQ || operation == Operation.NE
+                    || operation == Operation.CONTAINS || operation == Operation.REGEXP;
+            if (needsSingleValue && (valueNode == null || valueNode.isNull()
+                    || !valueNode.isValueNode())) {
+                throw invalid(rule, path + ".value 必须是单值");
+            }
+            String value = needsSingleValue ? valueNode.asText() : null;
+            Set<String> values = Collections.emptySet();
+            if (operation == Operation.IN) {
+                if (valueNode == null || !valueNode.isArray()) {
+                    throw invalid(rule, path + ".value 必须是数组");
+                }
+                Set<String> compiledValues = new LinkedHashSet<>();
+                for (JsonNode item : valueNode) {
+                    if (item == null || item.isContainerNode()) {
+                        throw invalid(rule, path + ".value 只能包含标量");
+                    }
+                    compiledValues.add(item.isNull() ? null : item.asText());
+                }
+                values = Collections.unmodifiableSet(compiledValues);
+            }
+            Pattern pattern = null;
+            if (operation == Operation.REGEXP) {
+                try {
+                    pattern = Pattern.compile(value);
+                } catch (RuntimeException invalidRegex) {
+                    throw invalid(rule, path + ".value 正则无效");
+                }
+            }
+            return new MatchCondition(source, field, operation, value, values, pattern);
+        }
+
+        boolean matches(Map<String, Object> parsedFields, String rawMessage) {
+            boolean present;
+            Object actual;
+            if (source == Source.RAW) {
+                present = rawMessage != null;
+                actual = rawMessage;
+            } else {
+                present = parsedFields != null && parsedFields.containsKey(field);
+                actual = present ? parsedFields.get(field) : null;
+            }
+            switch (operation) {
+                case EXISTS: return present;
+                case NOT_EXISTS: return !present;
+                case NOT_BLANK: return present && actual != null && !actual.toString().trim().isEmpty();
+                case EQ: return present && string(actual).equals(value);
+                case NE: return present && !string(actual).equals(value);
+                case IN: return present && values.contains(actual == null ? null : actual.toString());
+                case CONTAINS: return present && actual != null && actual.toString().contains(value);
+                case REGEXP: return present && actual != null && pattern.matcher(actual.toString()).find();
+                default: throw new IllegalStateException("未知显式匹配操作符: " + operation);
+            }
+        }
+
+        int specificity() {
+            switch (operation) {
+                case EQ:
+                case IN:
+                case REGEXP: return 3;
+                case CONTAINS: return 2;
+                default: return 1;
+            }
+        }
+
+        private static String string(Object value) {
+            return value == null ? "" : value.toString();
+        }
+
+        private static String text(JsonNode node) {
+            return node == null || node.isNull() ? null : node.asText();
+        }
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/compiled/RuleSelectionEngine.java

@@ -0,0 +1,263 @@
+package com.Modules.NormalData.compiled;
+
+import com.Modules.NormalData.metrics.NormalizationRuleMetrics;
+import com.config.NormalizationOptimizationProperties;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/** 同设备多规则的纯内存归属选择器。 */
+public final class RuleSelectionEngine {
+    private static final double EPSILON = 1.0e-12d;
+
+    private final CompiledRuleCache ruleCache;
+    private final NormalizationOptimizationProperties properties;
+    private final NormalizationRuleMetrics metrics;
+
+    public RuleSelectionEngine(CompiledRuleCache ruleCache,
+                               NormalizationOptimizationProperties properties,
+                               NormalizationRuleMetrics metrics) {
+        this.ruleCache = ruleCache;
+        this.properties = properties;
+        this.metrics = metrics;
+    }
+
+    public <T> Decision<T> select(List<Candidate<T>> parsedCandidates, String rawMessage,
+                                  boolean shadowOnly) {
+        NormalizationOptimizationProperties.RuleSelection config = properties.getRuleSelection();
+        List<Ranked<T>> usable = new ArrayList<>();
+        for (Candidate<T> candidate : parsedCandidates) {
+            try {
+                usable.add(new Ranked<>(candidate, ruleCache.getMatchProfile(candidate.getSource())));
+            } catch (RuntimeException invalidProfile) {
+                metrics.ruleSelection("invalid_match_config");
+            }
+        }
+
+        Decision<T> result = null;
+        if (config.isExplicitEnabled()) {
+            List<Ranked<T>> explicit = new ArrayList<>();
+            for (Ranked<T> ranked : usable) {
+                if (!ranked.profile.isExplicitEnabled()) continue;
+                RuleMatchProfile.ExplicitEvaluation evaluation = ranked.profile.evaluate(
+                        ranked.candidate.getParsedFields(), rawMessage);
+                if (evaluation.isMatched()) {
+                    ranked.explicitSpecificity = evaluation.getSpecificity();
+                    explicit.add(ranked);
+                }
+            }
+            if (!explicit.isEmpty()) result = choose(explicit, SelectionMode.EXPLICIT);
+        }
+
+        if (result == null && config.isAutomaticFallbackEnabled()) {
+            List<Ranked<T>> automatic = new ArrayList<>();
+            for (Ranked<T> ranked : usable) {
+                // 启用了显式条件但未命中的规则，严禁绕回字段评分。
+                if (!ranked.profile.isExplicitEnabled()) automatic.add(ranked);
+            }
+            if (!automatic.isEmpty()) result = choose(automatic, SelectionMode.AUTOMATIC);
+        }
+
+        if (result == null || !result.isSelected()) {
+            metrics.ruleSelection(shadowOnly ? "shadow_no_match" : "no_match");
+            metrics.ruleSelectionObservation(parsedCandidates.size(), 0.0d, 0.0d);
+            return Decision.none();
+        }
+
+        if (shadowOnly) {
+            metrics.ruleSelection("shadow_selected");
+        } else {
+            metrics.ruleSelection("selected");
+            metrics.ruleSelection(result.getMode() == SelectionMode.EXPLICIT
+                    ? "explicit_selected" : "automatic_selected");
+            if (result.isAmbiguous()) metrics.ruleSelection("ambiguous");
+        }
+        metrics.ruleSelectionObservation(parsedCandidates.size(),
+                parsedCandidates.isEmpty() ? 0.0d
+                        : (double) result.getRankedCandidateCount() / parsedCandidates.size(),
+                result.getScoreMargin());
+        return result;
+    }
+
+    private <T> Decision<T> choose(List<Ranked<T>> candidates, SelectionMode mode) {
+        applyAutomaticScores(candidates);
+        List<Ranked<T>> eligible = new ArrayList<>();
+        for (Ranked<T> ranked : candidates) {
+            // 显式条件已经证明格式归属，允许没有 mapper 指纹的完成型规则参与。
+            if (mode == SelectionMode.EXPLICIT || ranked.score.matchedCount > 0
+                    || (candidates.size() == 1 && ranked.profile.getOriginFields().isEmpty())) {
+                eligible.add(ranked);
+            }
+        }
+        if (eligible.isEmpty()) return Decision.none();
+
+        Comparator<Ranked<T>> comparator = mode == SelectionMode.EXPLICIT
+                ? RuleSelectionEngine::compareExplicit : RuleSelectionEngine::compareAutomatic;
+        eligible.sort(comparator);
+        Ranked<T> first = eligible.get(0);
+        List<Candidate<T>> tied = new ArrayList<>();
+        for (Ranked<T> ranked : eligible) {
+            if (!sameRank(first, ranked, mode)) break;
+            tied.add(ranked.candidate);
+        }
+        tied.sort(Comparator.comparingLong(candidate -> candidate.getSource().getId()));
+        Candidate<T> winner = tied.get(0);
+        double margin = eligible.size() < 2 ? first.score.ownershipScore
+                : Math.max(0.0d, first.score.ownershipScore
+                - eligible.get(1).score.ownershipScore);
+        return new Decision<>(winner, mode, tied.size() > 1, tied,
+                eligible.size(), margin);
+    }
+
+    private static <T> void applyAutomaticScores(List<Ranked<T>> candidates) {
+        Map<String, Integer> frequency = new HashMap<>();
+        for (Ranked<T> candidate : candidates) {
+            for (String field : candidate.profile.getOriginFields()) {
+                frequency.put(field, frequency.getOrDefault(field, 0) + 1);
+            }
+        }
+        for (Ranked<T> candidate : candidates) {
+            double totalWeight = 0.0d;
+            double matchedWeight = 0.0d;
+            int matchedCount = 0;
+            Set<String> parsed = candidate.candidate.getParsedFields().keySet();
+            for (String field : candidate.profile.getOriginFields()) {
+                double weight = 1.0d / frequency.get(field);
+                totalWeight += weight;
+                if (parsed.contains(field)) {
+                    matchedWeight += weight;
+                    matchedCount++;
+                }
+            }
+            double coverage = totalWeight == 0.0d ? 0.0d : matchedWeight / totalWeight;
+            candidate.score = new AutomaticScore(matchedWeight * coverage,
+                    matchedWeight, coverage, matchedCount);
+        }
+    }
+
+    private static int compareExplicit(Ranked<?> left, Ranked<?> right) {
+        int compared = Integer.compare(right.profile.getPriority(), left.profile.getPriority());
+        if (compared != 0) return compared;
+        compared = Integer.compare(right.explicitSpecificity, left.explicitSpecificity);
+        if (compared != 0) return compared;
+        compared = compareScore(left.score, right.score);
+        if (compared != 0) return compared;
+        return Long.compare(left.candidate.getSource().getId(), right.candidate.getSource().getId());
+    }
+
+    private static int compareAutomatic(Ranked<?> left, Ranked<?> right) {
+        int compared = compareScore(left.score, right.score);
+        if (compared != 0) return compared;
+        return Long.compare(left.candidate.getSource().getId(), right.candidate.getSource().getId());
+    }
+
+    private static int compareScore(AutomaticScore left, AutomaticScore right) {
+        int compared = Double.compare(right.ownershipScore, left.ownershipScore);
+        if (compared != 0) return compared;
+        compared = Double.compare(right.matchedWeight, left.matchedWeight);
+        if (compared != 0) return compared;
+        compared = Double.compare(right.coverage, left.coverage);
+        if (compared != 0) return compared;
+        return Integer.compare(right.matchedCount, left.matchedCount);
+    }
+
+    private static boolean sameRank(Ranked<?> left, Ranked<?> right, SelectionMode mode) {
+        if (mode == SelectionMode.EXPLICIT
+                && (left.profile.getPriority() != right.profile.getPriority()
+                || left.explicitSpecificity != right.explicitSpecificity)) {
+            return false;
+        }
+        return close(left.score.ownershipScore, right.score.ownershipScore)
+                && close(left.score.matchedWeight, right.score.matchedWeight)
+                && close(left.score.coverage, right.score.coverage)
+                && left.score.matchedCount == right.score.matchedCount;
+    }
+
+    private static boolean close(double left, double right) {
+        return Math.abs(left - right) <= EPSILON;
+    }
+
+    public enum SelectionMode { EXPLICIT, AUTOMATIC }
+
+    public static final class Candidate<T> {
+        private final RuleSource source;
+        private final Map<String, Object> parsedFields;
+        private final T payload;
+
+        public Candidate(RuleSource source, Map<String, Object> parsedFields, T payload) {
+            this.source = source;
+            this.parsedFields = Collections.unmodifiableMap(new LinkedHashMap<>(parsedFields));
+            this.payload = payload;
+        }
+
+        public RuleSource getSource() { return source; }
+        public Map<String, Object> getParsedFields() { return parsedFields; }
+        public T getPayload() { return payload; }
+    }
+
+    public static final class Decision<T> {
+        private static final Decision<?> NONE = new Decision<>(null, null, false,
+                Collections.emptyList(), 0, 0.0d);
+        private final Candidate<T> winner;
+        private final SelectionMode mode;
+        private final boolean ambiguous;
+        private final List<Candidate<T>> tiedCandidates;
+        private final int rankedCandidateCount;
+        private final double scoreMargin;
+
+        private Decision(Candidate<T> winner, SelectionMode mode, boolean ambiguous,
+                         List<Candidate<T>> tiedCandidates, int rankedCandidateCount,
+                         double scoreMargin) {
+            this.winner = winner;
+            this.mode = mode;
+            this.ambiguous = ambiguous;
+            this.tiedCandidates = Collections.unmodifiableList(new ArrayList<>(tiedCandidates));
+            this.rankedCandidateCount = rankedCandidateCount;
+            this.scoreMargin = scoreMargin;
+        }
+
+        @SuppressWarnings("unchecked")
+        static <T> Decision<T> none() { return (Decision<T>) NONE; }
+        public boolean isSelected() { return winner != null; }
+        public Candidate<T> getWinner() { return winner; }
+        public SelectionMode getMode() { return mode; }
+        public boolean isAmbiguous() { return ambiguous; }
+        public List<Candidate<T>> getTiedCandidates() { return tiedCandidates; }
+        public int getRankedCandidateCount() { return rankedCandidateCount; }
+        public double getScoreMargin() { return scoreMargin; }
+    }
+
+    private static final class Ranked<T> {
+        private final Candidate<T> candidate;
+        private final RuleMatchProfile profile;
+        private int explicitSpecificity;
+        private AutomaticScore score = AutomaticScore.ZERO;
+
+        private Ranked(Candidate<T> candidate, RuleMatchProfile profile) {
+            this.candidate = candidate;
+            this.profile = profile;
+        }
+    }
+
+    private static final class AutomaticScore {
+        private static final AutomaticScore ZERO = new AutomaticScore(0.0d, 0.0d, 0.0d, 0);
+        private final double ownershipScore;
+        private final double matchedWeight;
+        private final double coverage;
+        private final int matchedCount;
+
+        private AutomaticScore(double ownershipScore, double matchedWeight,
+                               double coverage, int matchedCount) {
+            this.ownershipScore = ownershipScore;
+            this.matchedWeight = matchedWeight;
+            this.coverage = coverage;
+            this.matchedCount = matchedCount;
+        }
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/Modules/NormalData/metrics/NormalizationRuleMetrics.java

@@ -5,6 +5,7 @@ import org.springframework.stereotype.Component;
 import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.DoubleAdder;
 import java.util.concurrent.atomic.LongAdder;
 
 /** 固定维度的进程内规则指标；不以 rule/device/event 作为长期标签。 */
@@ -18,7 +19,15 @@ public class NormalizationRuleMetrics {
             "collect_task_hit", "collect_task_load", "collect_task_failure");
     private final Counters invalidation = new Counters("local", "remote", "after_commit", "failure");
     private final Counters messageParse = new Counters("hit", "miss");
+    private final Counters ruleSelection = new Counters("selected", "explicit_selected",
+            "automatic_selected", "ambiguous", "equivalent_tie", "conflicting_tie",
+            "no_match", "invalid_match_config", "shadow_selected", "shadow_no_match");
     private final LongAdder ruleJsonParse = new LongAdder();
+    private final LongAdder selectionObservationCount = new LongAdder();
+    private final LongAdder selectionCandidateTotal = new LongAdder();
+    private final AtomicLong selectionCandidateMax = new AtomicLong();
+    private final DoubleAdder selectionMatchRateTotal = new DoubleAdder();
+    private final DoubleAdder selectionScoreMarginTotal = new DoubleAdder();
     private final LongAdder compileDurationTotalNanos = new LongAdder();
     private final LongAdder compileDurationCount = new LongAdder();
     private final AtomicLong compileDurationMaxNanos = new AtomicLong();
@@ -39,6 +48,15 @@ public class NormalizationRuleMetrics {
     }
     public void invalidation(String outcome) { invalidation.increment(outcome); }
     public void messageParse(String outcome) { messageParse.increment(outcome); }
+    public void ruleSelection(String outcome) { ruleSelection.increment(outcome); }
+    public void ruleSelectionObservation(int candidateCount, double matchRate, double scoreMargin) {
+        int candidates = Math.max(0, candidateCount);
+        selectionObservationCount.increment();
+        selectionCandidateTotal.add(candidates);
+        updateMax(selectionCandidateMax, candidates);
+        selectionMatchRateTotal.add(Math.max(0.0d, Math.min(1.0d, matchRate)));
+        selectionScoreMarginTotal.add(Math.max(0.0d, scoreMargin));
+    }
     public void ruleJsonParse() { ruleJsonParse.increment(); }
     public void compiledCacheGauge(long size, long evictions) {
         compiledCacheSize.set(Math.max(0L, size));
@@ -61,6 +79,15 @@ public class NormalizationRuleMetrics {
         result.put("enrichment_cache_get_total", enrichment.snapshot());
         result.put("rule_invalidation_total", invalidation.snapshot());
         result.put("message_parse_cache_get_total", messageParse.snapshot());
+        result.put("rule_selection_total", ruleSelection.snapshot());
+        result.put("rule_selection_observation_count", selectionObservationCount.sum());
+        result.put("rule_selection_candidate_count_total", selectionCandidateTotal.sum());
+        result.put("rule_selection_candidate_count_max", selectionCandidateMax.get());
+        long observations = selectionObservationCount.sum();
+        result.put("rule_selection_match_rate_avg", observations == 0L ? 0.0d
+                : selectionMatchRateTotal.sum() / observations);
+        result.put("rule_selection_score_margin_avg", observations == 0L ? 0.0d
+                : selectionScoreMarginTotal.sum() / observations);
         result.put("rule_json_parse_total", ruleJsonParse.sum());
         return result;
     }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/DmStandardizedBatchExecutor.java

@@ -32,6 +32,7 @@ public class DmStandardizedBatchExecutor {
     private final StatsChangeCaptureService statsChangeCaptureService;
     private final DirectStatsProjectionWriter directStatsProjectionWriter;
     private final DmDirectTransactionMetrics transactionMetrics;
+    private final LateAlarmBucketWriter lateAlarmBucketWriter;
 
     @org.springframework.beans.factory.annotation.Value("${stats.direct.transaction-log-enabled:false}")
     private boolean transactionLogEnabled;
@@ -46,7 +47,8 @@ public class DmStandardizedBatchExecutor {
                                        PlatformTransactionManager transactionManager,
                                        StatsChangeCaptureService statsChangeCaptureService,
                                        DirectStatsProjectionWriter directStatsProjectionWriter,
-                                       DmDirectTransactionMetrics transactionMetrics) {
+                                       DmDirectTransactionMetrics transactionMetrics,
+                                       LateAlarmBucketWriter lateAlarmBucketWriter) {
         // SqlSessionTemplate 本身线程安全。ExecutorType.BATCH 会把相同 SQL 的参数暂存在当前事务会话中，
         // 直到 flushStatements()/commit 时才真正交给 JDBC 驱动批量执行。
         this.batchSqlSession = new SqlSessionTemplate(sqlSessionFactory, ExecutorType.BATCH);
@@ -54,6 +56,7 @@ public class DmStandardizedBatchExecutor {
         this.statsChangeCaptureService = statsChangeCaptureService;
         this.directStatsProjectionWriter = directStatsProjectionWriter;
         this.transactionMetrics = transactionMetrics;
+        this.lateAlarmBucketWriter = lateAlarmBucketWriter;
     }
 
     /**
@@ -126,6 +129,12 @@ public class DmStandardizedBatchExecutor {
             batchSqlSession.flushStatements();
             timing.factFlushMs = elapsedMs(flushStarted);
 
+            // 与标准化告警事实共用同一事务写入事件时间桶。事实成功而通知失败时必须整体回滚，
+            // 否则迟到数据可能永久失去重放机会。
+            if (alarmTable) {
+                lateAlarmBucketWriter.record(rows);
+            }
+
             // DIRECT/SHADOW 模式在当前事务内按设备、规则聚合写统计；LEGACY 模式会直接返回。
             // 任一统计 SQL 失败都会回滚本事务的事实行，禁止产生无法定位的统计缺口。
             timing.statsResult = directStatsProjectionWriter.writeStandardizedRows(rows);

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/LateAlarmBucketWriter.java

@@ -0,0 +1,162 @@
+package com.common.service;
+
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.dao.CannotAcquireLockException;
+import org.springframework.dao.DeadlockLoserDataAccessException;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * 在标准化告警事实事务内登记事件时间桶。
+ *
+ * <p>该通知不是“告警一定迟到”的结论。通知与事实在同一事务提交；下游能领取通知时，事实必然
+ * 已经可见，再根据领取时已经完成的规则窗口决定是否重放。这里刻意不使用事实的
+ * {@code created_at} 判断提交先后，因为 Kafka 积压时接收时间可能早于实际入库时间。
+ */
+@Component
+public class LateAlarmBucketWriter {
+
+  private static final String UPSERT_SQL =
+      "MERGE INTO analysis_late_alarm_bucket t "
+          + "USING (SELECT ? device_id, ? bucket_start FROM dual) s "
+          + "ON (t.device_id=s.device_id AND t.bucket_start=s.bucket_start) "
+          + "WHEN MATCHED THEN UPDATE SET "
+          + "t.requested_generation=t.requested_generation+1, "
+          + "t.last_observed_at=SYSDATE, "
+          + "t.status=CASE WHEN t.status='RUNNING' THEN 'RUNNING' ELSE 'PENDING' END, "
+          + "t.updated_at=SYSDATE "
+          + "WHEN NOT MATCHED THEN INSERT "
+          + "(id,device_id,bucket_start,bucket_end,last_observed_at,requested_generation,"
+          + "claimed_generation,processed_generation,status,attempt_count,fencing_token,created_at,updated_at) "
+          + "VALUES (?,?,?,?,SYSDATE,1,0,0,'PENDING',0,0,SYSDATE,SYSDATE)";
+
+  private final JdbcTemplate jdbcTemplate;
+
+  @Value("${analysis.late-replay.outbox-enabled:true}")
+  private boolean enabled;
+
+  public LateAlarmBucketWriter(JdbcTemplate jdbcTemplate) {
+    this.jdbcTemplate = jdbcTemplate;
+  }
+
+  /**
+   * 每个设备每分钟只写一条通知。调用方必须已经开启标准化事实事务。
+   *
+   * @param rows 已确认写入 {@code syslog_normal_alarm} 的同构小批次
+   */
+  public void record(List<Map<String, Object>> rows) {
+    if (!enabled || rows == null || rows.isEmpty()) {
+      return;
+    }
+
+    Set<BucketKey> buckets = new LinkedHashSet<>();
+    for (Map<String, Object> row : rows) {
+      Integer deviceId = integerValue(row.get("device_id"));
+      LocalDateTime eventTime = timeValue(row.get("log_time"));
+      if (deviceId == null || eventTime == null) {
+        throw new IllegalArgumentException("迟到告警通知缺少 device_id 或 log_time");
+      }
+      buckets.add(new BucketKey(deviceId, eventTime.truncatedTo(ChronoUnit.MINUTES)));
+    }
+
+    for (BucketKey key : buckets) {
+      LocalDateTime bucketEnd = key.bucketStart.plusMinutes(1);
+      try {
+        jdbcTemplate.update(
+            UPSERT_SQL,
+            key.deviceId,
+            Timestamp.valueOf(key.bucketStart),
+            UUID.randomUUID().toString(),
+            key.deviceId,
+            Timestamp.valueOf(key.bucketStart),
+            Timestamp.valueOf(bucketEnd));
+      } catch (RuntimeException failure) {
+        // 多个 Kafka 分区可能同时首次写同一设备分钟，MERGE 的并发插入会竞争唯一键。
+        // 交给外层事实事务整体回滚并有限重试，不能只吞掉 outbox 失败。
+        if (isConcurrentWrite(failure)) {
+          throw new DirectStatsProjectionWriter.ConcurrentStatsWriteException(
+              "迟到告警分钟桶并发写入失败，可安全回滚后有限重试", failure);
+        }
+        throw failure;
+      }
+    }
+  }
+
+  private Integer integerValue(Object value) {
+    if (value instanceof Number) {
+      return ((Number) value).intValue();
+    }
+    if (value == null) {
+      return null;
+    }
+    return Integer.valueOf(value.toString());
+  }
+
+  private LocalDateTime timeValue(Object value) {
+    if (value instanceof LocalDateTime) {
+      return (LocalDateTime) value;
+    }
+    if (value instanceof Timestamp) {
+      return ((Timestamp) value).toLocalDateTime();
+    }
+    if (value instanceof java.util.Date) {
+      return new Timestamp(((java.util.Date) value).getTime()).toLocalDateTime();
+    }
+    return null;
+  }
+
+  private boolean isConcurrentWrite(Throwable failure) {
+    Throwable current = failure;
+    while (current != null) {
+      if (current instanceof DuplicateKeyException
+          || current instanceof CannotAcquireLockException
+          || current instanceof DeadlockLoserDataAccessException) {
+        return true;
+      }
+      String message = String.valueOf(current.getMessage()).toLowerCase(java.util.Locale.ROOT);
+      if (message.contains("unique constraint") || message.contains("duplicate key")) {
+        return true;
+      }
+      current = current.getCause();
+    }
+    return false;
+  }
+
+  private static final class BucketKey {
+    private final Integer deviceId;
+    private final LocalDateTime bucketStart;
+
+    private BucketKey(Integer deviceId, LocalDateTime bucketStart) {
+      this.deviceId = deviceId;
+      this.bucketStart = bucketStart;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+      if (this == other) {
+        return true;
+      }
+      if (!(other instanceof BucketKey)) {
+        return false;
+      }
+      BucketKey that = (BucketKey) other;
+      return Objects.equals(deviceId, that.deviceId)
+          && Objects.equals(bucketStart, that.bucketStart);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(deviceId, bucketStart);
+    }
+  }
+}

haobang-security-dm/syslog-consumer-rule/src/main/java/com/common/service/SyslogNormalDataService.java

@@ -30,6 +30,8 @@ public class SyslogNormalDataService {
     private StatsChangeCaptureService statsChangeCaptureService;
     @Autowired
     private DirectStatsProjectionWriter directStatsProjectionWriter;
+    @Autowired
+    private LateAlarmBucketWriter lateAlarmBucketWriter;
 
 
     /**
@@ -49,6 +51,9 @@ public class SyslogNormalDataService {
                     ? syslogNormalAlarmMapper.insertDynamic(dataMap)
                     : syslogNormalDataMapper.insertDynamic(dataMap);
             if (inserted > 0) {
+                if (alarmData && lateAlarmBucketWriter != null) {
+                    lateAlarmBucketWriter.record(java.util.Collections.singletonList(dataMap));
+                }
                 statsChangeCaptureService.captureStandardizedInsert(dataMap, alarmData);
                 directStatsProjectionWriter.writeStandardizedRow(dataMap);
             }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/config/NormalizationOptimizationProperties.java

@@ -5,17 +5,37 @@ import org.springframework.beans.factory.InitializingBean;
 import org.springframework.stereotype.Component;
 
 /**
- * 灰度控制和本地缓存边界。默认值必须与 application*.properties 保持一致。
+ * 标准化热路径的执行策略、灰度开关和缓存边界。
+ *
+ * <p>配置前缀为 {@code normalization}。本类中的初始值是配置缺失时的最后保护值，必须与
+ * {@code application*.properties} 保持一致。修改开关时需要同时考虑下面三类影响：</p>
+ *
+ * <ul>
+ *   <li>结果语义：是否启用多规则唯一归属、是否仍可能产生多条标准化结果；</li>
+ *   <li>执行方式：使用编译规则还是旧解释执行器、是否进行影子计算；</li>
+ *   <li>缓存一致性：规则源、消息解析和设备补全数据是否允许复用。</li>
+ * </ul>
+ *
+ * <p>{@link #afterPropertiesSet()} 在 Spring 完成属性绑定后校验组合配置，防止应用带着
+ * 无效缓存依赖或未知歧义策略启动。</p>
  */
 @Component
 @ConfigurationProperties(prefix = "normalization")
 public class NormalizationOptimizationProperties implements InitializingBean {
 
+    /** JetCache 总开关；远端规则/补全缓存依赖该开关。 */
     private final Toggle jetcache = new Toggle();
+    /** 规则 JSON 预编译和编译/旧执行器影子对比配置。 */
     private final CompiledRule compiledRule = new CompiledRule();
+    /** 单个 Kafka 批次内复用设备和采集任务补全结果。 */
     private final Toggle batchEnrichmentCache = new Toggle();
+    /** 同一条消息在解析签名相同时复用 crop + parse 结果。 */
     private final Toggle messageParseCache = new Toggle();
+    /** 同设备多条泛化规则的唯一归属策略；默认启用。 */
+    private final RuleSelection ruleSelection = new RuleSelection();
+    /** 设备关联规则列表的本地/远端缓存参数。 */
     private final RuleSourceCache ruleSourceCache = new RuleSourceCache();
+    /** 设备和采集任务基础信息的本地/远端缓存参数。 */
     private final EnrichmentCache enrichmentCache = new EnrichmentCache();
 
     public boolean isJetcacheEnabled() {
@@ -31,22 +51,32 @@ public class NormalizationOptimizationProperties implements InitializingBean {
         return jetcache;
     }
 
+    /** 对应 {@code normalization.compiled-rule.*}。 */
     public CompiledRule getCompiledRule() {
         return compiledRule;
     }
 
+    /** 对应 {@code normalization.batch-enrichment-cache.enabled}。 */
     public Toggle getBatchEnrichmentCache() {
         return batchEnrichmentCache;
     }
 
+    /** 对应 {@code normalization.message-parse-cache.enabled}。 */
     public Toggle getMessageParseCache() {
         return messageParseCache;
     }
 
+    /** 对应 {@code normalization.rule-selection.*}。 */
+    public RuleSelection getRuleSelection() {
+        return ruleSelection;
+    }
+
+    /** 对应 {@code normalization.rule-source-cache.*}。 */
     public RuleSourceCache getRuleSourceCache() {
         return ruleSourceCache;
     }
 
+    /** 对应 {@code normalization.enrichment-cache.*}。 */
     public EnrichmentCache getEnrichmentCache() {
         return enrichmentCache;
     }
@@ -62,10 +92,12 @@ public class NormalizationOptimizationProperties implements InitializingBean {
                     "normalization.batch-enrichment-cache.enabled=true 要求 normalization.jetcache.enabled=true");
         }
         compiledRule.validate();
+        ruleSelection.validate();
         ruleSourceCache.validate();
         enrichmentCache.validate();
     }
 
+    /** 只包含 {@code enabled} 的通用配置段。未显式赋值时默认关闭。 */
     public static class Toggle {
         private boolean enabled;
 
@@ -78,14 +110,28 @@ public class NormalizationOptimizationProperties implements InitializingBean {
         }
     }
 
+    /**
+     * 编译规则执行器配置。
+     *
+     * <p>{@code enabled=false} 时使用旧解释执行器；编译失败的单条规则仍会自动回退旧执行器，
+     * 不会因为缓存中存在不兼容历史规则而阻塞整条 Kafka 消息。</p>
+     */
     public static class CompiledRule {
+        /** 是否让编译规则执行器成为主路径。 */
         private boolean enabled;
+        /** 旧执行器为主路径时，是否抽样运行编译执行器并比较业务结果。 */
         private boolean shadowCompareEnabled = true;
+        /** 编译执行器影子对比抽样率，范围 0～1。 */
         private double shadowSampleRate = 0.001d;
+        /** JVM 内编译规则缓存最大条目数。 */
         private int localLimit = 10_000;
+        /** 编译规则缓存访问后过期时间，单位分钟。 */
         private int expireAfterAccessMinutes = 30;
+        /** 编译失败短缓存时间，避免每条日志反复编译同一条无效规则。 */
         private int failureCacheSeconds = 10;
+        /** 单次影子对比最大等待时间，单位毫秒。 */
         private int shadowTimeoutMs = 1_000;
+        /** 是否记录脱敏后的规则编译详情；不包含原始日志。 */
         private boolean ruleDetailLogEnabled = true;
 
         public boolean isEnabled() {
@@ -173,6 +219,77 @@ public class NormalizationOptimizationProperties implements InitializingBean {
         }
     }
 
+    /**
+     * 同设备多条泛化规则的唯一归属配置。
+     *
+     * <p>主流程是：对所有规则做无副作用试解析；先检查规则正文中的显式
+     * {@code match_params}；没有显式命中时，再对未启用显式配置的历史规则执行自动字段
+     * 指纹评分；最后只允许唯一胜出规则执行映射、过滤、补全和入库。</p>
+     *
+     * <p>{@code enabled} 继承自 {@link Toggle}，本配置段默认设为 {@code true}。关闭时恢复
+     * 历史“逐条完整执行所有设备规则”的行为，因此同一日志可能重新产生多条结果。</p>
+     */
+    public static class RuleSelection extends Toggle {
+        /**
+         * 主开关关闭时是否仍在后台计算新选择结果。影子计算只记录指标，不改变旧路径输出；
+         * 主开关开启时直接执行唯一选择，本值不再决定业务输出。
+         */
+        private boolean shadowEnabled = true;
+        /**
+         * 候选排名完全相同时的策略：{@code LOWEST_RULE_ID} 稳定选择最小规则 ID；
+         * {@code STRICT} 会复核并列规则的业务结果，冲突时只保存一条非标记录。
+         */
+        private String ambiguousFallback = "LOWEST_RULE_ID";
+        /**
+         * 是否输出低频歧义详情日志。日志只包含 event hash、候选规则 ID 和分数，不输出原文。
+         */
+        private boolean detailLogEnabled;
+        /** 是否启用 {@code rule_content.match_params} 显式格式指纹。 */
+        private boolean explicitEnabled = true;
+        /** 没有显式命中时，是否允许未配置显式条件的历史规则使用自动字段指纹评分。 */
+        private boolean automaticFallbackEnabled = true;
+
+        /** 默认启用唯一归属，确保配置文件缺失时仍不会退回多规则重复产出。 */
+        public RuleSelection() {
+            setEnabled(true);
+        }
+
+        /** 返回是否启用关闭主开关后的影子计算。 */
+        public boolean isShadowEnabled() { return shadowEnabled; }
+        public void setShadowEnabled(boolean shadowEnabled) { this.shadowEnabled = shadowEnabled; }
+        /** 返回并列候选兜底策略名称。 */
+        public String getAmbiguousFallback() { return ambiguousFallback; }
+        public void setAmbiguousFallback(String ambiguousFallback) {
+            this.ambiguousFallback = ambiguousFallback;
+        }
+        /** 返回是否记录脱敏后的规则选择详情。 */
+        public boolean isDetailLogEnabled() { return detailLogEnabled; }
+        public void setDetailLogEnabled(boolean detailLogEnabled) {
+            this.detailLogEnabled = detailLogEnabled;
+        }
+        /** 返回是否启用规则正文中的显式 {@code match_params}。 */
+        public boolean isExplicitEnabled() { return explicitEnabled; }
+        public void setExplicitEnabled(boolean explicitEnabled) { this.explicitEnabled = explicitEnabled; }
+        /** 返回是否启用历史规则自动字段指纹兜底。 */
+        public boolean isAutomaticFallbackEnabled() { return automaticFallbackEnabled; }
+        public void setAutomaticFallbackEnabled(boolean automaticFallbackEnabled) {
+            this.automaticFallbackEnabled = automaticFallbackEnabled;
+        }
+        /** 是否使用“业务结果冲突即转非标”的严格歧义策略。 */
+        public boolean isStrictAmbiguity() {
+            return "STRICT".equalsIgnoreCase(ambiguousFallback);
+        }
+
+        private void validate() {
+            if (!("LOWEST_RULE_ID".equalsIgnoreCase(ambiguousFallback)
+                    || "STRICT".equalsIgnoreCase(ambiguousFallback))) {
+                throw new IllegalArgumentException(
+                        "ruleSelection.ambiguousFallback 仅支持 LOWEST_RULE_ID 或 STRICT");
+            }
+        }
+    }
+
+    /** 设备关联规则快照的两级缓存 TTL 配置，单位均为秒。 */
     public static class RuleSourceCache extends Toggle {
         private int localTtlSeconds = 30;
         private int remoteTtlSeconds = 300;
@@ -191,6 +308,7 @@ public class NormalizationOptimizationProperties implements InitializingBean {
         }
     }
 
+    /** 设备和采集任务补全事实的两级缓存 TTL 配置，单位均为秒。 */
     public static class EnrichmentCache {
         private int localTtlSeconds = 60;
         private int remoteTtlSeconds = 3600;

haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/PointTimeResolver.java

@@ -14,7 +14,7 @@ public class PointTimeResolver {
         this.metrics = metrics;
     }
 
-    public Long resolve(boolean present, String rawValue, long receiveTimeMillis, boolean required) {
+    public Long resolve(boolean present, String rawValue, long eventTimeMillis, boolean required) {
         if (!present) {
             if (!required) return null;
             metrics.legacyMissing();
@@ -23,7 +23,7 @@ public class PointTimeResolver {
                 throw new IllegalArgumentException("旧 Kafka 消息缺少 point_time_ns，且 legacy fallback 已关闭");
             }
             try {
-                long assigned = allocator.allocate(receiveTimeMillis);
+                long assigned = allocator.allocate(eventTimeMillis);
                 metrics.legacyAssigned();
                 return assigned;
             } catch (RuntimeException failure) {
@@ -37,14 +37,14 @@ public class PointTimeResolver {
         final long value;
         try { value = Long.parseLong(rawValue); }
         catch (NumberFormatException invalid) { throw new IllegalArgumentException("point_time_ns 非数字或溢出", invalid); }
-        validate(value, receiveTimeMillis);
+        validate(value, eventTimeMillis);
         return value;
     }
 
-    public static void validate(long pointTimeNs, long receiveTimeMillis) {
+    public static void validate(long pointTimeNs, long eventTimeMillis) {
         if (pointTimeNs <= 0L) throw new IllegalArgumentException("point_time_ns 必须为正整数");
-        if (Math.floorDiv(pointTimeNs, PointTimeAllocator.NANOS_PER_MILLI) != receiveTimeMillis) {
-            throw new IllegalArgumentException("point_time_ns 毫秒部分与 receive_time 不一致");
+        if (Math.floorDiv(pointTimeNs, PointTimeAllocator.NANOS_PER_MILLI) != eventTimeMillis) {
+            throw new IllegalArgumentException("point_time_ns 毫秒部分与 event_time 不一致");
         }
     }
 }

haobang-security-dm/syslog-consumer-rule/src/main/java/com/influx/raw/RawStorageProperties.java

@@ -12,9 +12,9 @@ public class RawStorageProperties {
     public static final int UUID_BUCKET_COUNT = 4096;
 
     @Value("${influxdb.raw.write-schema:V1}") private String writeSchemaValue;
-    @Value("${influxdb.raw.read-mode:V2_ONLY}") private String readModeValue;
+    @Value("${influxdb.raw.read-mode:V1_ONLY}") private String readModeValue;
     @Value("${influxdb.raw.v1-bucket:${influxdb.bucket:yelangbucket}}") private String v1Bucket;
-    @Value("${influxdb.raw.v2-bucket:yelangbucket}") private String v2Bucket;
+    @Value("${influxdb.raw.v2-bucket:yelangbucket_v2}") private String v2Bucket;
     @Value("${influxdb.raw.uuid-bucket-count:4096}") private int uuidBucketCount;
     @Value("${influxdb.raw.legacy-point-time-fallback-enabled:true}") private boolean legacyFallbackEnabled;
     @Value("${influxdb.raw.query-max-rows:10000}") private int queryMaxRows;

haobang-security-dm/syslog-consumer-rule/src/main/resources/application-dev.properties

@@ -139,6 +139,12 @@ normalization.compiled-rule.shadow-timeout-ms=1000
 normalization.compiled-rule.rule-detail-log-enabled=true
 normalization.batch-enrichment-cache.enabled=true
 normalization.message-parse-cache.enabled=true
+normalization.rule-selection.enabled=true
+normalization.rule-selection.shadow-enabled=true
+normalization.rule-selection.ambiguous-fallback=LOWEST_RULE_ID
+normalization.rule-selection.detail-log-enabled=false
+normalization.rule-selection.explicit-enabled=true
+normalization.rule-selection.automatic-fallback-enabled=true
 normalization.rule-source-cache.local-ttl-seconds=30
 normalization.rule-source-cache.remote-ttl-seconds=300
 normalization.rule-source-cache.empty-ttl-seconds=10

haobang-security-dm/syslog-consumer-rule/src/main/resources/application-pre-nw.properties

@@ -200,6 +200,12 @@ normalization.compiled-rule.shadow-timeout-ms=1000
 normalization.compiled-rule.rule-detail-log-enabled=true
 normalization.batch-enrichment-cache.enabled=true
 normalization.message-parse-cache.enabled=true
+normalization.rule-selection.enabled=true
+normalization.rule-selection.shadow-enabled=true
+normalization.rule-selection.ambiguous-fallback=LOWEST_RULE_ID
+normalization.rule-selection.detail-log-enabled=false
+normalization.rule-selection.explicit-enabled=true
+normalization.rule-selection.automatic-fallback-enabled=true
 normalization.rule-source-cache.local-ttl-seconds=30
 normalization.rule-source-cache.remote-ttl-seconds=300
 normalization.rule-source-cache.empty-ttl-seconds=10

haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod-zc.properties

@@ -163,6 +163,12 @@ normalization.compiled-rule.shadow-timeout-ms=1000
 normalization.compiled-rule.rule-detail-log-enabled=true
 normalization.batch-enrichment-cache.enabled=true
 normalization.message-parse-cache.enabled=true
+normalization.rule-selection.enabled=true
+normalization.rule-selection.shadow-enabled=true
+normalization.rule-selection.ambiguous-fallback=LOWEST_RULE_ID
+normalization.rule-selection.detail-log-enabled=false
+normalization.rule-selection.explicit-enabled=true
+normalization.rule-selection.automatic-fallback-enabled=true
 normalization.rule-source-cache.local-ttl-seconds=30
 normalization.rule-source-cache.remote-ttl-seconds=300
 normalization.rule-source-cache.empty-ttl-seconds=10

haobang-security-dm/syslog-consumer-rule/src/main/resources/application-prod.properties

@@ -142,6 +142,12 @@ normalization.compiled-rule.shadow-timeout-ms=1000
 normalization.compiled-rule.rule-detail-log-enabled=true
 normalization.batch-enrichment-cache.enabled=true
 normalization.message-parse-cache.enabled=true
+normalization.rule-selection.enabled=true
+normalization.rule-selection.shadow-enabled=true
+normalization.rule-selection.ambiguous-fallback=LOWEST_RULE_ID
+normalization.rule-selection.detail-log-enabled=false
+normalization.rule-selection.explicit-enabled=true
+normalization.rule-selection.automatic-fallback-enabled=true
 normalization.rule-source-cache.local-ttl-seconds=30
 normalization.rule-source-cache.remote-ttl-seconds=300
 normalization.rule-source-cache.empty-ttl-seconds=10

haobang-security-dm/syslog-consumer-rule/src/main/resources/application-test.properties

@@ -101,6 +101,12 @@ normalization.compiled-rule.shadow-timeout-ms=1000
 normalization.compiled-rule.rule-detail-log-enabled=true
 normalization.batch-enrichment-cache.enabled=true
 normalization.message-parse-cache.enabled=true
+normalization.rule-selection.enabled=true
+normalization.rule-selection.shadow-enabled=true
+normalization.rule-selection.ambiguous-fallback=LOWEST_RULE_ID
+normalization.rule-selection.detail-log-enabled=false
+normalization.rule-selection.explicit-enabled=true
+normalization.rule-selection.automatic-fallback-enabled=true
 normalization.rule-source-cache.local-ttl-seconds=30
 normalization.rule-source-cache.remote-ttl-seconds=300
 normalization.rule-source-cache.empty-ttl-seconds=10

haobang-security-dm/syslog-consumer-rule/src/main/resources/application.properties

@@ -27,9 +27,9 @@ influxdb.retry.attempts=3
 influxdb.retry.delay=1000
 # 原始 Syslog V2 分阶段开关。兼容部署先保持 V1/V1_ONLY，切换时统一写 V2。
 influxdb.raw.write-schema=${INFLUXDB_RAW_WRITE_SCHEMA:V2}
-influxdb.raw.read-mode=${INFLUXDB_RAW_READ_MODE:V2_ONLY}
+influxdb.raw.read-mode=${INFLUXDB_RAW_READ_MODE:V1_V2}
 influxdb.raw.v1-bucket=${INFLUXDB_RAW_V1_BUCKET:${influxdb.bucket}}
-influxdb.raw.v2-bucket=${INFLUXDB_RAW_V2_BUCKET:yelangbucket}
+influxdb.raw.v2-bucket=${INFLUXDB_RAW_V2_BUCKET:yelangbucket_v2}
 influxdb.raw.uuid-bucket-count=${INFLUXDB_RAW_UUID_BUCKET_COUNT:4096}
 influxdb.raw.legacy-point-time-fallback-enabled=${INFLUXDB_RAW_LEGACY_POINT_TIME_FALLBACK_ENABLED:true}
 influxdb.raw.query-max-rows=${INFLUXDB_RAW_QUERY_MAX_ROWS:10000}
@@ -103,6 +103,9 @@ spring.kafka.listener.type=batch
 # 定时任务配置
 spring.task.scheduling.pool.size=10
 
+# 标准化告警事实与迟到重放通知使用同一个达梦事务。
+analysis.late-replay.outbox-enabled=${ANALYSIS_LATE_REPLAY_OUTBOX_ENABLED:true}
+
 # 日志配置
 logging.level.com.common.schedule=INFO
 logging.level.com.common.service=INFO
@@ -204,6 +207,18 @@ normalization.compiled-rule.shadow-timeout-ms=1000
 normalization.compiled-rule.rule-detail-log-enabled=true
 normalization.batch-enrichment-cache.enabled=true
 normalization.message-parse-cache.enabled=true
+# 同设备多规则唯一归属：默认开启，最终只允许一条胜出规则进入映射和入库。
+normalization.rule-selection.enabled=true
+# 仅在 enabled=false 时用于后台观察新选择结果，不改变旧路径的实际输出。
+normalization.rule-selection.shadow-enabled=true
+# 并列策略：LOWEST_RULE_ID=稳定选择最小规则ID；STRICT=业务结果冲突时转为一条非标记录。
+normalization.rule-selection.ambiguous-fallback=LOWEST_RULE_ID
+# 只记录 event hash、候选规则ID和分数等脱敏详情，不记录原始日志。
+normalization.rule-selection.detail-log-enabled=false
+# 优先使用 rule_content.match_params 中的显式格式条件。
+normalization.rule-selection.explicit-enabled=true
+# 没有显式命中时，允许未配置 match_params 的历史规则按自动字段指纹选优。
+normalization.rule-selection.automatic-fallback-enabled=true
 normalization.rule-source-cache.local-ttl-seconds=30
 normalization.rule-source-cache.remote-ttl-seconds=300
 normalization.rule-source-cache.empty-ttl-seconds=10

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorCompiledEquivalenceTest.java

@@ -59,6 +59,8 @@ class LogNormalProcessorCompiledEquivalenceTest {
 
         assertEquals(legacyResult.getOutcome(), compiledResult.getOutcome());
         assertEquals(legacyResult.getRows(), compiledResult.getRows());
+        assertEquals(1, compiledResult.getRows().size());
+        assertEquals(31L, compiledResult.getRows().get(0).get("normalize_rule_id"));
         assertEquals(1, metrics.getParseMisses());
         assertEquals(1, metrics.getParseHits());
         assertEquals(2, metrics.getCompiledMisses());
@@ -118,6 +120,8 @@ class LogNormalProcessorCompiledEquivalenceTest {
         properties.getCompiledRule().setShadowCompareEnabled(false);
         properties.getMessageParseCache().setEnabled(parseCache);
         properties.getBatchEnrichmentCache().setEnabled(compiled);
+        properties.getRuleSelection().setEnabled(true);
+        properties.getRuleSelection().setShadowEnabled(false);
         return properties;
     }
 

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorReceiveTimeTest.java

@@ -33,29 +33,46 @@ class LogNormalProcessorReceiveTimeTest {
     }
 
     @Test
-    void initialReceiveTimeSetsBothPartitionAndCreationTime() {
+    void eventTimeSetsPartitionAndReceiveTimeSetsCreationTime() {
+        LocalDateTime eventTime = LocalDateTime.of(2026, 8, 10, 20, 11, 24, 736_717_000);
         LocalDateTime receiveTime = LocalDateTime.of(2026, 8, 11, 3, 55, 57, 109_000_000);
         Map<String, Object> columns = new HashMap<>();
 
-        LogNormalProcessor.applyInitialReceiveTime(columns, receiveTime);
+        LogNormalProcessor.applyIngressTimes(columns, eventTime, receiveTime);
 
-        assertEquals(receiveTime, columns.get("log_time"));
+        assertEquals(eventTime, columns.get("log_time"));
         assertEquals(receiveTime, columns.get("created_at"));
     }
 
     @Test
-    void ruleProducedLogTimeCannotChangeReceiveDayPartition() {
+    void eventTimeOverridesRuleValueAndSelectsEventDayPartition() {
         LocalDateTime eventTime = LocalDateTime.of(2026, 8, 10, 20, 11, 24, 736_717_000);
         LocalDateTime receiveTime = LocalDateTime.of(2026, 8, 11, 12, 30, 44, 199_000_000);
         Map<String, Object> columns = new HashMap<>();
-        columns.put("log_time", eventTime);
+        columns.put("log_time", receiveTime);
 
-        LogNormalProcessor.applyInitialReceiveTime(columns, receiveTime);
+        LogNormalProcessor.applyIngressTimes(columns, eventTime, receiveTime);
 
-        assertEquals(receiveTime, columns.get("log_time"));
+        assertEquals(eventTime, columns.get("log_time"));
         assertEquals(receiveTime, columns.get("created_at"));
     }
 
+    @Test
+    void missingEventTimeFallsBackToReceiveTimeForLegacyMessages() {
+        LocalDateTime receiveTime = LocalDateTime.of(2026, 8, 11, 12, 30, 44, 199_000_000);
+
+        assertEquals(receiveTime, LogNormalProcessor.resolveEventAt(null, receiveTime));
+        assertEquals(receiveTime, LogNormalProcessor.resolveEventAt("", receiveTime));
+    }
+
+    @Test
+    void explicitEventTimeIsParsedForHistoricalApiPull() {
+        LocalDateTime receiveTime = LocalDateTime.of(2026, 9, 2, 9, 18, 38);
+
+        assertEquals(LocalDateTime.of(2026, 9, 1, 16, 17),
+                LogNormalProcessor.resolveEventAt("20260901161700000", receiveTime));
+    }
+
     @Test
     void deterministicResultIdIsStableAndKeepsUuidLength() {
         String first = LogNormalProcessor.deterministicResultId("20260811-event", 7L, "legacy");

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/LogNormalProcessorRuleSelectionTest.java

@@ -0,0 +1,236 @@
+package com.Modules.NormalData;
+
+import com.Modules.NormalData.compiled.BatchEnrichmentContext;
+import com.Modules.NormalData.compiled.BatchRuleSnapshot;
+import com.Modules.NormalData.compiled.CompiledRuleCache;
+import com.Modules.NormalData.compiled.CompiledRuleCompiler;
+import com.Modules.NormalData.compiled.RuleExecutionMetrics;
+import com.Modules.NormalData.compiled.RuleSource;
+import com.common.entity.SyslogNonNormalMessage;
+import com.common.mapper.DmColumnMapper;
+import com.common.mapper.DmNormalizeRuleMapper;
+import com.common.service.DeviceCollectTaskService;
+import com.common.service.DeviceDeviceService;
+import com.common.service.DmColumnService;
+import com.common.service.DmNormalizeRuleService;
+import com.common.service.LogDataCompleteService;
+import com.common.service.LogDataFilterService;
+import com.common.service.SyslogNonNormalMessageService;
+import com.common.service.SyslogNormalDataService;
+import com.config.AppProperties;
+import com.config.KafkaConsumerProperties;
+import com.config.NormalizationOptimizationProperties;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+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.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class LogNormalProcessorRuleSelectionTest {
+
+    @Test
+    void explicitRuleProducesTheOnlyRowAndOverridesHistoricalRule() {
+        RuleSource historical = jsonRule(10L, "historical", "", mapping("allow"), "");
+        RuleSource explicit = jsonRule(20L, "explicit",
+                "\"match_params\":{\"enabled\":true,\"priority\":100,"
+                        + "\"all\":[{\"field\":\"type\",\"op\":\"eq\","
+                        + "\"value\":\"firewall\"}]},",
+                mapping("deny"), "");
+        Fixture fixture = fixture(Arrays.asList(historical, explicit), config(false));
+
+        LogNormalProcessor.NormalizationResult result = fixture.prepare(
+                "{\"type\":\"firewall\",\"status\":\"x\"}");
+
+        assertEquals(LogNormalProcessor.ProcessingOutcome.SUCCESS, result.getOutcome());
+        assertEquals(1, result.getRows().size());
+        assertEquals(20L, result.getRows().get(0).get("normalize_rule_id"));
+        assertEquals("deny", result.getRows().get(0).get("action"));
+        verify(fixture.nonNormalService, never()).saveMessage(any());
+    }
+
+    @Test
+    void failedCandidateDoesNotCreateNonNormalWhenAnotherRuleWins() {
+        RuleSource failedRegex = new RuleSource(10L, "regex", "regex", "regex", "{"
+                + "\"regexp\":\"never=(\\\\S+)\","
+                + "\"mappers\":[{\"origin_field\":\"1\",\"dest_field\":\"value\","
+                + "\"action\":{\"type\":\"equal\",\"param\":{}}}]}", true, "0");
+        RuleSource valid = jsonRule(20L, "json", "", equalMapping(), "");
+        Fixture fixture = fixture(Arrays.asList(failedRegex, valid), config(true));
+
+        LogNormalProcessor.NormalizationResult result = fixture.prepare(
+                "{\"type\":\"firewall\",\"status\":\"ok\"}");
+
+        assertEquals(1, result.getRows().size());
+        assertEquals(20L, result.getRows().get(0).get("normalize_rule_id"));
+        verify(fixture.nonNormalService, never()).saveMessage(any());
+    }
+
+    @Test
+    void allCandidatesFailAndOnlyOneNonNormalRecordIsSaved() {
+        RuleSource first = regexRule(10L, "first", "first=(\\S+)");
+        RuleSource second = regexRule(20L, "second", "second=(\\S+)");
+        Fixture fixture = fixture(Arrays.asList(first, second), config(true));
+
+        LogNormalProcessor.NormalizationResult result = fixture.prepare("nothing matches");
+
+        assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome());
+        verify(fixture.nonNormalService).saveMessage(any(SyslogNonNormalMessage.class));
+    }
+
+    @Test
+    void nonNormalSaveFailureRemainsRetryable() {
+        Fixture fixture = fixture(Arrays.asList(
+                regexRule(10L, "first", "first=(\\S+)"),
+                regexRule(20L, "second", "second=(\\S+)")), config(true));
+        when(fixture.nonNormalService.saveMessage(any())).thenReturn(false);
+
+        assertThrows(RuntimeException.class, () -> fixture.prepare("nothing matches"));
+    }
+
+    @Test
+    void winnerFilterIsTerminalAndDoesNotTryTheRunnerUp() {
+        String filter = "\"filters_params\":{\"expressions\":[{"
+                + "\"filters_relation\":\"and\",\"expressions\":[{"
+                + "\"column_name\":\"action\",\"op\":\"=\",\"value\":\"deny\"}]}]},";
+        RuleSource historical = jsonRule(10L, "historical", "", equalMapping(), "");
+        RuleSource explicit = jsonRule(20L, "explicit",
+                "\"match_params\":{\"enabled\":true,\"all\":[{"
+                        + "\"field\":\"type\",\"op\":\"eq\","
+                        + "\"value\":\"firewall\"}]},",
+                equalMapping(), filter);
+        Fixture fixture = fixture(Arrays.asList(historical, explicit), config(true));
+
+        LogNormalProcessor.NormalizationResult result = fixture.prepare(
+                "{\"type\":\"firewall\",\"status\":\"deny\"}");
+
+        assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome());
+        assertTrue(result.getRows().isEmpty());
+        verify(fixture.nonNormalService, never()).saveMessage(any());
+    }
+
+    @Test
+    void strictAmbiguitySavesNonNormalWhenEqualCandidatesProduceDifferentBusinessResults() {
+        RuleSource allow = jsonRule(10L, "allow", "", mapping("allow"), "");
+        RuleSource deny = jsonRule(20L, "deny", "", mapping("deny"), "");
+        NormalizationOptimizationProperties properties = config(true);
+        properties.getRuleSelection().setAmbiguousFallback("STRICT");
+        Fixture fixture = fixture(Arrays.asList(allow, deny), properties);
+
+        LogNormalProcessor.NormalizationResult result = fixture.prepare(
+                "{\"type\":\"firewall\",\"status\":\"x\"}");
+
+        assertEquals(LogNormalProcessor.ProcessingOutcome.FILTERED, result.getOutcome());
+        assertTrue(result.getRows().isEmpty());
+        ArgumentCaptor<SyslogNonNormalMessage> captor =
+                ArgumentCaptor.forClass(SyslogNonNormalMessage.class);
+        verify(fixture.nonNormalService).saveMessage(captor.capture());
+        assertEquals("多规则归属歧义", captor.getValue().getReason());
+        assertTrue(captor.getValue().getReasonDetail().contains("10"));
+        assertTrue(captor.getValue().getReasonDetail().contains("20"));
+    }
+
+    @Test
+    void strictAmbiguityStillUsesLowestIdWhenBusinessResultsAreEquivalent() {
+        RuleSource highId = jsonRule(20L, "same-high", "", mapping("allow"), "");
+        RuleSource lowId = jsonRule(10L, "same-low", "", mapping("allow"), "");
+        NormalizationOptimizationProperties properties = config(true);
+        properties.getRuleSelection().setAmbiguousFallback("STRICT");
+        Fixture fixture = fixture(Arrays.asList(highId, lowId), properties);
+
+        LogNormalProcessor.NormalizationResult result = fixture.prepare(
+                "{\"type\":\"firewall\",\"status\":\"x\"}");
+
+        assertEquals(LogNormalProcessor.ProcessingOutcome.SUCCESS, result.getOutcome());
+        assertEquals(1, result.getRows().size());
+        assertEquals(10L, result.getRows().get(0).get("normalize_rule_id"));
+        verify(fixture.nonNormalService, never()).saveMessage(any());
+    }
+
+    private static NormalizationOptimizationProperties config(boolean compiled) {
+        NormalizationOptimizationProperties properties = new NormalizationOptimizationProperties();
+        properties.getCompiledRule().setEnabled(compiled);
+        properties.getCompiledRule().setShadowCompareEnabled(false);
+        properties.getMessageParseCache().setEnabled(true);
+        properties.getRuleSelection().setEnabled(true);
+        properties.getRuleSelection().setShadowEnabled(false);
+        return properties;
+    }
+
+    private static RuleSource jsonRule(long id, String name, String prefix,
+                                       String mapper, String extra) {
+        String content = "{" + prefix
+                + "\"cropper_params\":{\"head_key\":\"{\",\"head_offset\":0,"
+                + "\"tail_key\":\"\",\"tail_offset\":0},"
+                + "\"mappers\":[" + mapper + "]," + extra
+                + "\"complete_columns\":[]}";
+        return new RuleSource(id, name, name, "json", content, true, "0");
+    }
+
+    private static RuleSource regexRule(long id, String name, String regexp) {
+        String content = "{\"regexp\":\"" + regexp.replace("\\", "\\\\") + "\","
+                + "\"mappers\":[{\"origin_field\":\"1\",\"dest_field\":\"value\","
+                + "\"action\":{\"type\":\"equal\",\"param\":{}}}]}";
+        return new RuleSource(id, name, name, "regex", content, true, "0");
+    }
+
+    private static String equalMapping() {
+        return "{\"origin_field\":\"status\",\"dest_field\":\"action\","
+                + "\"action\":{\"type\":\"equal\",\"param\":{}}}";
+    }
+
+    private static String mapping(String value) {
+        return "{\"origin_field\":\"status\",\"dest_field\":\"action\","
+                + "\"action\":{\"type\":\"mapping\",\"param\":{\"mapping\":{"
+                + "\"x\":\"" + value + "\"}}}}";
+    }
+
+    private static Fixture fixture(List<RuleSource> rules,
+                                   NormalizationOptimizationProperties properties) {
+        Fixture fixture = new Fixture();
+        fixture.taskService = mock(DeviceCollectTaskService.class);
+        fixture.deviceService = mock(DeviceDeviceService.class);
+        fixture.nonNormalService = mock(SyslogNonNormalMessageService.class);
+        when(fixture.nonNormalService.saveMessage(any())).thenReturn(true);
+        CompiledRuleCompiler compiler = new CompiledRuleCompiler(new ObjectMapper());
+        fixture.processor = new LogNormalProcessor(mock(DmNormalizeRuleService.class),
+                mock(DmColumnService.class), mock(SyslogNormalDataService.class),
+                fixture.nonNormalService, new LogDataFilterService(), new LogDataCompleteService(),
+                fixture.taskService, fixture.deviceService, mock(DmColumnMapper.class),
+                mock(DmNormalizeRuleMapper.class), mock(KafkaConsumerProperties.class),
+                mock(AppProperties.class), new CompiledRuleCache(compiler, properties), properties);
+        Map<Long, List<RuleSource>> byDevice = new LinkedHashMap<>();
+        byDevice.put(7L, rules);
+        fixture.snapshot = new BatchRuleSnapshot(byDevice, System.currentTimeMillis());
+        return fixture;
+    }
+
+    private static final class Fixture {
+        private LogNormalProcessor processor;
+        private SyslogNonNormalMessageService nonNormalService;
+        private DeviceCollectTaskService taskService;
+        private DeviceDeviceService deviceService;
+        private BatchRuleSnapshot snapshot;
+
+        private LogNormalProcessor.NormalizationResult prepare(String body) {
+            String message = "[receive_time=20260821150000000 event_id=event-selection device_id=7 "
+                    + "device_name=fw vendor=vendor data_type=json device_collect_id=3]" + body;
+            return processor.prepare(message, "event-selection", "raw",
+                    LocalDateTime.of(2026, 8, 21, 15, 0), snapshot,
+                    new BatchEnrichmentContext(false, taskService, deviceService),
+                    new RuleExecutionMetrics());
+        }
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/SysLogProcessorTimeTest.java

@@ -1,31 +1,53 @@
 package com.Modules.NormalData;
 
+import com.influx.raw.PointTimeAllocator;
+import com.influx.raw.PointTimeResolver;
 import org.junit.jupiter.api.Test;
 
-import java.time.Instant;
-import java.time.LocalDateTime;
-
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 
 class SysLogProcessorTimeTest {
 
     @Test
-    void receiveTimeDeterminesBothPointTimestampAndMeasurementDate() {
+    void eventTimeDeterminesBothPointTimestampAndMeasurementDate() {
         SysLogProcessor.ResolvedInfluxTime resolved = SysLogProcessor.resolveInfluxTime(
-                "20260811035557109", Instant.parse("2026-08-09T00:00:00Z"));
+                "20260810035557109", "20260811035557109");
 
-        assertEquals(Instant.parse("2026-08-10T19:55:57.109Z").toEpochMilli(),
+        assertEquals(java.time.Instant.parse("2026-08-09T19:55:57.109Z").toEpochMilli(),
                 resolved.getTimestampMs());
-        assertEquals("20260811", resolved.getMeasurementDate());
-        assertEquals(LocalDateTime.of(2026, 8, 11, 3, 55, 57, 109_000_000),
-                resolved.getReceivedAt());
+        assertEquals("20260810", resolved.getMeasurementDate());
+        assertEquals(java.time.LocalDateTime.of(2026, 8, 10, 3, 55, 57, 109_000_000),
+                resolved.getEventAt());
     }
 
     @Test
-    void invalidReceiveTimeIsRejectedInsteadOfUsingConsumerTime() {
-        Instant firstReceivedAt = Instant.parse("2026-08-10T19:55:57.109Z");
+    void legacyMessageWithoutEventTimeFallsBackToReceiveTime() {
+        SysLogProcessor.ResolvedInfluxTime resolved = SysLogProcessor.resolveInfluxTime(
+                null, "20260811035557109");
 
-        org.junit.jupiter.api.Assertions.assertThrows(IllegalArgumentException.class,
-                () -> SysLogProcessor.resolveInfluxTime("invalid", firstReceivedAt));
+        assertEquals(java.time.Instant.parse("2026-08-10T19:55:57.109Z").toEpochMilli(),
+                resolved.getTimestampMs());
+        assertEquals("20260811", resolved.getMeasurementDate());
+    }
+
+    @Test
+    void invalidEventTimeIsRejectedInsteadOfUsingReceiveTime() {
+        assertThrows(IllegalArgumentException.class,
+                () -> SysLogProcessor.resolveInfluxTime("invalid", "20260811035557109"));
+    }
+
+    @Test
+    void historicalPointTimeIsValidatedAgainstEventTimeRatherThanNextDayReceiveTime() {
+        SysLogProcessor.ResolvedInfluxTime event = SysLogProcessor.resolveInfluxTime(
+                "20260901161700123", "20260902091800456");
+        long pointTimeNs = event.getTimestampMs() * PointTimeAllocator.NANOS_PER_MILLI + 789L;
+
+        assertDoesNotThrow(() -> PointTimeResolver.validate(pointTimeNs, event.getTimestampMs()));
+
+        long receiveTimeMs = java.time.Instant.parse("2026-09-02T01:18:00.456Z").toEpochMilli();
+        assertThrows(IllegalArgumentException.class,
+                () -> PointTimeResolver.validate(pointTimeNs, receiveTimeMs));
     }
 }

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/compiled/RuleSelectionEngineTest.java

@@ -0,0 +1,200 @@
+package com.Modules.NormalData.compiled;
+
+import com.Modules.NormalData.metrics.NormalizationRuleMetrics;
+import com.config.NormalizationOptimizationProperties;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+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.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class RuleSelectionEngineTest {
+    private NormalizationOptimizationProperties properties;
+    private NormalizationRuleMetrics metrics;
+    private CompiledRuleCache cache;
+    private RuleSelectionEngine engine;
+
+    @BeforeEach
+    void setUp() {
+        properties = new NormalizationOptimizationProperties();
+        properties.getRuleSelection().setEnabled(true);
+        properties.getRuleSelection().setShadowEnabled(false);
+        metrics = new NormalizationRuleMetrics();
+        cache = new CompiledRuleCache(new CompiledRuleCompiler(new ObjectMapper()), properties, metrics);
+        engine = new RuleSelectionEngine(cache, properties, metrics);
+    }
+
+    @Test
+    void explicitMatchWinsBeforeHistoricalAutomaticRule() {
+        RuleSource explicit = rule(20L, "title,client", "\"match_params\":{"
+                + "\"enabled\":true,\"priority\":10,"
+                + "\"all\":[{\"field\":\"title\",\"op\":\"eq\",\"value\":\"HFish\"}],"
+                + "\"any\":[{\"source\":\"raw\",\"op\":\"contains\",\"value\":\"threat\"}],"
+                + "\"none\":[{\"field\":\"honeypot_id\",\"op\":\"exists\"}]},");
+        RuleSource historical = rule(10L, "title,client", "");
+        Map<String, Object> fields = fields("title", "HFish", "client", "agent");
+
+        RuleSelectionEngine.Decision<String> result = engine.select(Arrays.asList(
+                candidate(historical, fields), candidate(explicit, fields)), "HFish threat", false);
+
+        assertTrue(result.isSelected());
+        assertEquals(20L, result.getWinner().getSource().getId());
+        assertEquals(RuleSelectionEngine.SelectionMode.EXPLICIT, result.getMode());
+    }
+
+    @Test
+    void failedExplicitRuleCannotFallBackToAutomaticScoring() {
+        RuleSource explicit = rule(1L, "title,special", "\"match_params\":{"
+                + "\"enabled\":true,\"priority\":999,"
+                + "\"all\":[{\"field\":\"title\",\"op\":\"eq\",\"value\":\"other\"}]},");
+        RuleSource historical = rule(2L, "title", "");
+        Map<String, Object> fields = fields("title", "actual", "special", "yes");
+
+        RuleSelectionEngine.Decision<String> result = engine.select(Arrays.asList(
+                candidate(explicit, fields), candidate(historical, fields)), "actual", false);
+
+        assertEquals(2L, result.getWinner().getSource().getId());
+        assertEquals(RuleSelectionEngine.SelectionMode.AUTOMATIC, result.getMode());
+    }
+
+    @Test
+    void explicitPriorityOnlyAppliesAfterConditionsMatch() {
+        RuleSource low = rule(10L, "kind", explicitEq("kind", "event", 10));
+        RuleSource high = rule(20L, "kind", explicitEq("kind", "event", 100));
+
+        RuleSelectionEngine.Decision<String> result = engine.select(Arrays.asList(
+                candidate(low, fields("kind", "event")),
+                candidate(high, fields("kind", "event"))), "event", false);
+
+        assertEquals(20L, result.getWinner().getSource().getId());
+        assertFalse(result.isAmbiguous());
+    }
+
+    @Test
+    void weightedOwnershipPreventsTinyCommonRuleFromBeatingSpecificRule() {
+        RuleSource small = rule(10L, "src_ip,time", "");
+        RuleSource large = rule(20L, "src_ip,time,u1,u2,u3,u4,u5,u6,u7,u8", "");
+        Map<String, Object> parsed = fields("src_ip", "10.0.0.1", "time", "now",
+                "u1", 1, "u2", 2, "u3", 3, "u4", 4, "u5", 5, "u6", 6, "u7", 7);
+
+        RuleSelectionEngine.Decision<String> result = engine.select(Arrays.asList(
+                candidate(small, parsed), candidate(large, parsed)), "raw", false);
+
+        assertEquals(20L, result.getWinner().getSource().getId());
+        assertEquals(RuleSelectionEngine.SelectionMode.AUTOMATIC, result.getMode());
+    }
+
+    @Test
+    void completelyEqualCandidatesUseLowestRuleIdAndExposeAmbiguity() {
+        RuleSource highId = rule(40L, "src_ip,type", "");
+        RuleSource lowId = rule(30L, "src_ip,type", "");
+        Map<String, Object> parsed = fields("src_ip", "10.0.0.1", "type", "deny");
+
+        RuleSelectionEngine.Decision<String> result = engine.select(Arrays.asList(
+                candidate(highId, parsed), candidate(lowId, parsed)), "raw", false);
+
+        assertEquals(30L, result.getWinner().getSource().getId());
+        assertTrue(result.isAmbiguous());
+        assertEquals(2, result.getTiedCandidates().size());
+    }
+
+    @Test
+    void allExplicitRulesFailingProducesNoCandidate() {
+        RuleSource first = rule(1L, "kind", explicitEq("kind", "a", 0));
+        RuleSource second = rule(2L, "kind", explicitEq("kind", "b", 0));
+
+        RuleSelectionEngine.Decision<String> result = engine.select(Arrays.asList(
+                candidate(first, fields("kind", "c")),
+                candidate(second, fields("kind", "c"))), "c", false);
+
+        assertFalse(result.isSelected());
+    }
+
+    @Test
+    void invalidMatchConfigurationIsRejectedDuringCompilation() {
+        RuleSource invalid = rule(1L, "kind", "\"match_params\":{"
+                + "\"enabled\":true,\"all\":[{\"field\":\"kind\","
+                + "\"op\":\"in\",\"value\":\"not-an-array\"}]},");
+
+        assertThrows(CompiledRuleCompiler.RuleCompilationException.class,
+                () -> cache.getMatchProfile(invalid));
+    }
+
+    @Test
+    void explicitProfileSupportsAllDocumentedOperatorsAndGroupSemantics() {
+        RuleSource source = rule(7L, "a", "\"match_params\":{\"enabled\":true,"
+                + "\"all\":["
+                + "{\"field\":\"a\",\"op\":\"exists\"},"
+                + "{\"field\":\"a\",\"op\":\"not_blank\"},"
+                + "{\"field\":\"a\",\"op\":\"eq\",\"value\":\"abc\"},"
+                + "{\"field\":\"a\",\"op\":\"ne\",\"value\":\"zzz\"},"
+                + "{\"field\":\"a\",\"op\":\"in\",\"value\":[\"abc\",\"def\"]},"
+                + "{\"field\":\"a\",\"op\":\"contains\",\"value\":\"b\"},"
+                + "{\"field\":\"a\",\"op\":\"regexp\",\"value\":\"^a.*\"},"
+                + "{\"field\":\"missing\",\"op\":\"not_exists\"},"
+                + "{\"source\":\"raw\",\"op\":\"not_blank\"},"
+                + "{\"source\":\"raw\",\"op\":\"contains\",\"value\":\"CEF:\"},"
+                + "{\"source\":\"raw\",\"op\":\"regexp\",\"value\":\"CEF:.*abc\"}],"
+                + "\"any\":[{\"field\":\"a\",\"op\":\"eq\",\"value\":\"no\"},"
+                + "{\"field\":\"a\",\"op\":\"eq\",\"value\":\"abc\"}],"
+                + "\"none\":[{\"field\":\"blocked\",\"op\":\"exists\"}]},");
+
+        RuleMatchProfile profile = cache.getMatchProfile(source);
+
+        assertTrue(profile.evaluate(fields("a", "abc"), "CEF: vendor abc").isMatched());
+        assertFalse(profile.evaluate(fields("a", "abc", "blocked", true),
+                "CEF: vendor abc").isMatched());
+    }
+
+    @Test
+    void rawSourceRejectsFieldOnlyOperations() {
+        RuleSource invalid = rule(8L, "a", "\"match_params\":{\"enabled\":true,"
+                + "\"all\":[{\"source\":\"raw\",\"op\":\"exists\"}]},");
+
+        assertThrows(CompiledRuleCompiler.RuleCompilationException.class,
+                () -> cache.getMatchProfile(invalid));
+    }
+
+    private static String explicitEq(String field, String value, int priority) {
+        return "\"match_params\":{\"enabled\":true,\"priority\":" + priority
+                + ",\"all\":[{\"field\":\"" + field + "\",\"op\":\"eq\","
+                + "\"value\":\"" + value + "\"}]},";
+    }
+
+    private static RuleSource rule(long id, String origins, String prefix) {
+        StringBuilder mappers = new StringBuilder();
+        if (!origins.isEmpty()) {
+            for (String field : origins.split(",")) {
+                if (mappers.length() > 0) mappers.append(',');
+                mappers.append("{\"origin_field\":\"").append(field)
+                        .append("\",\"dest_field\":\"").append(field)
+                        .append("\",\"action\":{\"type\":\"equal\",\"param\":{}}}");
+            }
+        }
+        String content = "{" + prefix + "\"mappers\":[" + mappers + "]}";
+        return new RuleSource(id, "rule-" + id, "rule-" + id,
+                "json", content, true, "0");
+    }
+
+    private static RuleSelectionEngine.Candidate<String> candidate(
+            RuleSource source, Map<String, Object> fields) {
+        return new RuleSelectionEngine.Candidate<>(source, fields, "payload-" + source.getId());
+    }
+
+    private static Map<String, Object> fields(Object... values) {
+        Map<String, Object> result = new LinkedHashMap<>();
+        for (int i = 0; i < values.length; i += 2) {
+            result.put(values[i].toString(), values[i + 1]);
+        }
+        return result;
+    }
+}

haobang-security-dm/syslog-consumer-rule/src/test/java/com/Modules/NormalData/metrics/NormalizationRuleMetricsTest.java

@@ -19,6 +19,9 @@ class NormalizationRuleMetricsTest {
         metrics.enrichment("device", "hit");
         metrics.invalidation("after_commit");
         metrics.messageParse("miss");
+        metrics.ruleSelection("selected");
+        metrics.ruleSelection("explicit_selected");
+        metrics.ruleSelectionObservation(3, 2.0d / 3.0d, 0.5d);
         metrics.ruleJsonParse();
         metrics.compiledCacheGauge(3, 1);
 
@@ -28,6 +31,9 @@ class NormalizationRuleMetricsTest {
         assertTrue(snapshot.containsKey("rule_source_cache_get_total"));
         assertTrue(snapshot.containsKey("enrichment_cache_get_total"));
         assertTrue(snapshot.containsKey("rule_invalidation_total"));
+        assertEquals(1L, ((Map<String, Long>) snapshot.get("rule_selection_total")).get("selected"));
+        assertEquals(3L, snapshot.get("rule_selection_candidate_count_total"));
+        assertEquals(2.0d / 3.0d, (Double) snapshot.get("rule_selection_match_rate_avg"), 0.000001d);
         assertEquals(1L, snapshot.get("rule_json_parse_total"));
     }
 }

haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/DmNormalizeRuleServiceUpdateTest.java

@@ -13,6 +13,7 @@ import org.springframework.test.util.ReflectionTestUtils;
 import java.util.Arrays;
 
 import static org.junit.jupiter.api.Assertions.assertFalse;
+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.Mockito.mock;
@@ -62,6 +63,21 @@ class DmNormalizeRuleServiceUpdateTest {
         verify(publisher, never()).publishEvent(any());
     }
 
+    @Test
+    void invalidMatchParamsAreRejectedBeforeRuleUpdate() {
+        when(mapper.selectById(11L)).thenReturn(validRule());
+        DmNormalizeRule patch = new DmNormalizeRule();
+        patch.setId(11L);
+        patch.setRuleContent("{\"match_params\":{\"enabled\":true,\"all\":[{"
+                + "\"field\":\"type\",\"op\":\"in\",\"value\":\"bad\"}]},"
+                + "\"mappers\":[]}");
+
+        assertThrows(CompiledRuleCompiler.RuleCompilationException.class,
+                () -> service.updateById(patch));
+        verify(mapper, never()).updateById(any());
+        verify(publisher, never()).publishEvent(any());
+    }
+
     private static DmNormalizeRule validRule() {
         DmNormalizeRule rule = new DmNormalizeRule();
         rule.setId(11L);

haobang-security-dm/syslog-consumer-rule/src/test/java/com/common/service/LateAlarmBucketWriterTest.java

@@ -0,0 +1,60 @@
+package com.common.service;
+
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.util.ReflectionTestUtils;
+
+class LateAlarmBucketWriterTest {
+
+  @Test
+  void groupsRowsByDeviceAndEventMinute() {
+    JdbcTemplate jdbcTemplate = mock(JdbcTemplate.class);
+    LateAlarmBucketWriter writer = new LateAlarmBucketWriter(jdbcTemplate);
+    ReflectionTestUtils.setField(writer, "enabled", true);
+
+    writer.record(
+        Arrays.asList(
+            row(126, LocalDateTime.of(2026, 9, 4, 10, 1, 2), 3),
+            row(126, LocalDateTime.of(2026, 9, 4, 10, 1, 59), 4),
+            row(126, LocalDateTime.of(2026, 9, 4, 10, 2, 1), 5)));
+
+    verify(jdbcTemplate, times(2))
+        .update(
+            anyString(),
+            any(),
+            any(),
+            any(),
+            any(),
+            any(),
+            any());
+  }
+
+  @Test
+  void rejectsAlarmRowsWithoutStableTimes() {
+    LateAlarmBucketWriter writer = new LateAlarmBucketWriter(mock(JdbcTemplate.class));
+    ReflectionTestUtils.setField(writer, "enabled", true);
+    Map<String, Object> invalid = new HashMap<>();
+    invalid.put("device_id", 126);
+
+    assertThrows(IllegalArgumentException.class, () -> writer.record(Arrays.asList(invalid)));
+  }
+
+  private Map<String, Object> row(int deviceId, LocalDateTime eventTime, int receiveSecond) {
+    Map<String, Object> row = new HashMap<>();
+    row.put("device_id", deviceId);
+    row.put("log_time", eventTime);
+    row.put("created_at", LocalDateTime.of(2026, 9, 5, 1, 0, receiveSecond));
+    return row;
+  }
+}

haobang-security-dm/syslog-consumer-rule/src/test/java/com/config/NormalizationCachePropertiesTest.java

@@ -22,6 +22,7 @@ class NormalizationCachePropertiesTest {
         assertFalse(properties.isJetcacheEnabled());
         assertFalse(properties.getRuleSourceCache().isEnabled());
         assertFalse(properties.getCompiledRule().isEnabled());
+        assertEquals(true, properties.getRuleSelection().isEnabled());
         assertEquals(1000, properties.getCompiledRule().getShadowTimeoutMs());
         assertEquals(30, properties.getRuleSourceCache().getLocalTtlSeconds());
         assertEquals(300, properties.getRuleSourceCache().getRemoteTtlSeconds());
@@ -58,6 +59,8 @@ class NormalizationCachePropertiesTest {
                     "normalization.batch-enrichment-cache.enabled"), resource);
             assertEquals("true", values.getProperty(
                     "normalization.message-parse-cache.enabled"), resource);
+            assertEquals("true", values.getProperty(
+                    "normalization.rule-selection.enabled"), resource);
 
             NormalizationOptimizationProperties bound = new Binder(
                     new MapConfigurationPropertySource(values))
@@ -72,6 +75,7 @@ class NormalizationCachePropertiesTest {
             assertEquals(true, bound.getCompiledRule().isRuleDetailLogEnabled(), resource);
             assertEquals(true, bound.getBatchEnrichmentCache().isEnabled(), resource);
             assertEquals(true, bound.getMessageParseCache().isEnabled(), resource);
+            assertEquals(true, bound.getRuleSelection().isEnabled(), resource);
         }
     }
 

haobang-security-dm/syslog-consumer-rule/src/test/java/com/influx/raw/InfluxRawV2ContractTest.java

@@ -29,9 +29,9 @@ class InfluxRawV2ContractTest {
     @Test void configurationRejectsChangingPersistentBucketCount() {
         RawStorageProperties properties=new RawStorageProperties();
         ReflectionTestUtils.setField(properties,"writeSchemaValue","V2");
-        ReflectionTestUtils.setField(properties,"readModeValue","V2_ONLY");
+        ReflectionTestUtils.setField(properties,"readModeValue","V1_V2");
         ReflectionTestUtils.setField(properties,"v1Bucket","yelangbucket");
-        ReflectionTestUtils.setField(properties,"v2Bucket","yelangbucket");
+        ReflectionTestUtils.setField(properties,"v2Bucket","yelangbucket_v2");
         ReflectionTestUtils.setField(properties,"uuidBucketCount",1024);
         ReflectionTestUtils.setField(properties,"queryMaxRows",10000);
         ReflectionTestUtils.setField(properties,"queryMaxResponseBytes",10485760L);
@@ -62,9 +62,9 @@ class InfluxRawV2ContractTest {
     }
 
     @Test void v2HasUuidFieldButNeverUuidTagAndUsesNanoseconds() {
-        RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket");
+        RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket_v2");
         InfluxRawPointFactory.RawPoint raw=new InfluxRawPointFactory(properties).create("20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5","message","20260814000000000","126","1","agent-01-syslog-topic",ReceiveTimeParser.parse("20260814000000000"),1_786_636_800_000_000_123L);
-        String line=raw.getPoint().toLineProtocol(); assertFalse(line.substring(0,line.indexOf(' ')).contains("uuid=")); assertTrue(line.contains("event_bucket=2085")); assertTrue(line.contains("uuid=\"20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5\"")); assertEquals(WritePrecision.NS,raw.getPoint().getPrecision()); assertEquals("yelangbucket",raw.getBucket());
+        String line=raw.getPoint().toLineProtocol(); assertFalse(line.substring(0,line.indexOf(' ')).contains("uuid=")); assertTrue(line.contains("event_bucket=2085")); assertTrue(line.contains("uuid=\"20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5\"")); assertEquals(WritePrecision.NS,raw.getPoint().getPrecision()); assertEquals("yelangbucket_v2",raw.getBucket());
     }
 
     @Test void onlyMissingPointTimeFallsBack() {
@@ -83,7 +83,7 @@ class InfluxRawV2ContractTest {
     }
 
     @Test void mixedLegacyAndNewMessagesAllProduceDistinctV2PointsAndRetryIsIdempotent() {
-        RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.isLegacyFallbackEnabled()).thenReturn(true); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket");
+        RawStorageProperties properties=mock(RawStorageProperties.class); when(properties.isLegacyFallbackEnabled()).thenReturn(true); when(properties.getWriteSchema()).thenReturn(RawStorageProperties.WriteSchema.V2); when(properties.getV2Bucket()).thenReturn("yelangbucket_v2");
         RawInfluxMetrics metrics=new RawInfluxMetrics(); PointTimeResolver resolver=new PointTimeResolver(properties,metrics); InfluxRawPointFactory factory=new InfluxRawPointFactory(properties); ReceiveTimeParser.ParsedReceiveTime parsed=ReceiveTimeParser.parse("20260814000000000");
         long first=resolver.resolve(false,null,parsed.getEpochMillis(),true); long second=resolver.resolve(false,null,parsed.getEpochMillis(),true); assertNotEquals(first,second);
         String id="20260814-a3d874a0-6f50-4f0c-9e7a-9c20303360f5"; String originalTopic="agent-01-syslog-topic";
@@ -100,7 +100,7 @@ class InfluxRawV2ContractTest {
 
     @Test void uuidCandidateFluxComputesBucketEscapesUuidAndSelectsEarliestGlobally() {
         String uuid="20260814-a\" |> die() //";
-        String query=InfluxDBClient.buildUuidV2CandidateQuery("yelangbucket",
+        String query=InfluxDBClient.buildUuidV2CandidateQuery("yelangbucket_v2",
                 "syslog_security_20260814",uuid,Instant.parse("2026-08-13T16:00:00Z"),
                 Instant.parse("2026-08-14T16:00:00Z"));
         assertTrue(query.contains("r.event_bucket == \""+InfluxEventBucket.bucketOf(uuid)+"\""));