feat(alarm): hybrid window aggregator and evaluator

Wires the windowed evaluation path that the previous P4 commits set up. With this commit AVG/MIN/MAX/SUM/COUNT/ALL/ANY rules actually run end- to-end: - ConditionEvaluator: utility extracted out of RuleEvaluatorImpl so both the LAST path and the windowed path apply identical operator semantics. evaluate(condition, value), recoveryConditionOf(...) and toBigDecimal(...) are the only public surfaces. - WindowDataSource: small interface with aggregate() (scalar fold for AVG/MIN/MAX/SUM/COUNT) and samples() (raw rows for ALL/ANY). AggregateOutcome carries (value, sampleCount) so callers can enforce minSamples. - LocalWindowDataSource: reads WindowSampleBuffer.snapshot(...) and folds in Java; AVG uses scale=6 / HALF_UP rounding to match PostgreSQL's AVG(numeric). Skips null numValue samples for numeric aggregates; COUNT counts every sample. - RepositoryWindowDataSource: pushes the aggregate to the time-series store via RepositoryService.aggregateInWindow / samplesInWindow introduced in the previous commit. Only POINT facts are supported. - HybridWindowDataSource (@Primary): routes by spec.duration — duration <= dc3.alarm.window.local-cutoff goes to local, larger goes to repository. - WindowedRuleEvaluator: dispatch by mode. AVG/MIN/MAX/SUM/COUNT pull the scalar and feed it to ConditionEvaluator; ALL/ANY pull samples and apply the rule's per-sample condition. minSamples gates both paths. Recovery uses the synthesized condition (Recovery.operator + Recovery.threshold) over the same window, mirroring the LAST path. - RuleEvaluatorImpl: rewritten to dispatch by parsed WindowSpec. LAST stays inline (no data-source plumbing needed); everything else delegates to the windowed evaluator. The previous warn-once gate for non-LAST modes is removed; invalid window specs (which should already be rejected by the save validator) are skipped with a single log line. Tests: - RuleEvaluatorImplTest is rewritten to pass a mock WindowedRuleEvaluator and verify dispatch. - WindowedRuleEvaluatorTest covers AVG/COUNT/ALL/ANY/recovery paths + minSamples gate + LAST defensive guard. - All 193 tests in dc3-common-data pass.
2026-09-01 15:33:10 +08:00 · 2026-05-21 21:30:07 +08:00
parent 3b01a16ad1
commit c7d927bec4
9 changed files with 940 additions and 134 deletions
@@ -0,0 +1,148 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.data.biz.alarm;
+
+import io.github.pnoker.common.constant.common.BaseConstant;
+import io.github.pnoker.common.entity.ext.RuleExt;
+import org.apache.commons.lang3.StringUtils;
+
+import java.math.BigDecimal;
+import java.util.Objects;
+
+/**
+ * Stateless rule-condition evaluator. Extracted so both
+ * {@link RuleEvaluatorImpl} (single-fact path) and {@link WindowedRuleEvaluator}
+ * (windowed path) apply the exact same operator semantics — once the windowed
+ * path has reduced its samples to a scalar (or is folding ALL/ANY per sample),
+ * it just calls back into here.
+ *
+ * @author pnoker
+ * @version 2026.5.21
+ * @since 2026.5.21
+ */
+public final class ConditionEvaluator {
+
+    private ConditionEvaluator() {
+        throw new IllegalStateException(BaseConstant.UTILITY_CLASS);
+    }
+
+    /**
+     * Evaluate {@code condition} against {@code value}. Returns false when the
+     * condition is ill-formed or the value cannot be coerced.
+     */
+    public static boolean evaluate(RuleExt.Condition condition, Object value) {
+        if (Objects.isNull(condition)) {
+            return false;
+        }
+        String operator = StringUtils.lowerCase(StringUtils.trim(condition.getOperator()));
+        if ("exists".equals(operator)) {
+            return Objects.nonNull(value);
+        }
+        if ("missing".equals(operator)) {
+            return Objects.isNull(value);
+        }
+        if (Objects.isNull(value)) {
+            return false;
+        }
+
+        BigDecimal actual = toBigDecimal(value);
+        if (isNumericOperator(operator, condition)) {
+            return evaluateNumeric(operator, actual, condition);
+        }
+        return evaluateText(operator, value, condition);
+    }
+
+    /**
+     * Synthesize the recovery-side condition: same field + unit, but the
+     * operator and threshold come from {@link RuleExt.Recovery}. Mirrors the
+     * legacy logic in {@code RuleEvaluatorImpl.recovers}.
+     */
+    public static RuleExt.Condition recoveryConditionOf(RuleExt.Condition condition, RuleExt.Recovery recovery) {
+        if (Objects.isNull(condition) || Objects.isNull(recovery)) {
+            return null;
+        }
+        return new RuleExt.Condition(
+                condition.getField(),
+                recovery.getOperator(),
+                null,
+                recovery.getThreshold(),
+                null,
+                null,
+                condition.getUnit());
+    }
+
+    /**
+     * Coerce a value into a {@link BigDecimal} for numeric comparisons.
+     * Returns null when the value isn't a number / numeric string.
+     */
+    public static BigDecimal toBigDecimal(Object value) {
+        try {
+            if (value instanceof BigDecimal decimal) {
+                return decimal;
+            }
+            if (value instanceof Number number) {
+                return new BigDecimal(number.toString());
+            }
+            if (value instanceof CharSequence text && StringUtils.isNotBlank(text)) {
+                return new BigDecimal(text.toString());
+            }
+            return null;
+        } catch (NumberFormatException e) {
+            return null;
+        }
+    }
+
+    private static boolean evaluateNumeric(String operator, BigDecimal actual, RuleExt.Condition condition) {
+        if (Objects.isNull(actual)) {
+            return false;
+        }
+        return switch (operator) {
+            case ">" -> condition.getThreshold() != null && actual.compareTo(condition.getThreshold()) > 0;
+            case ">=" -> condition.getThreshold() != null && actual.compareTo(condition.getThreshold()) >= 0;
+            case "<" -> condition.getThreshold() != null && actual.compareTo(condition.getThreshold()) < 0;
+            case "<=" -> condition.getThreshold() != null && actual.compareTo(condition.getThreshold()) <= 0;
+            case "==" -> condition.getThreshold() != null && actual.compareTo(condition.getThreshold()) == 0;
+            case "!=" -> condition.getThreshold() != null && actual.compareTo(condition.getThreshold()) != 0;
+            case "between" -> condition.getLow() != null && condition.getHigh() != null
+                    && actual.compareTo(condition.getLow()) >= 0 && actual.compareTo(condition.getHigh()) <= 0;
+            case "outside" -> condition.getLow() != null && condition.getHigh() != null
+                    && (actual.compareTo(condition.getLow()) < 0 || actual.compareTo(condition.getHigh()) > 0);
+            default -> false;
+        };
+    }
+
+    private static boolean evaluateText(String operator, Object value, RuleExt.Condition condition) {
+        String actual = Objects.toString(value, "");
+        String expected = Objects.toString(condition.getExpected(), "");
+        return switch (operator) {
+            case "==", "eq" -> StringUtils.equals(actual, expected);
+            case "!=", "ne" -> !StringUtils.equals(actual, expected);
+            case "contains" -> StringUtils.contains(actual, expected);
+            default -> false;
+        };
+    }
+
+    private static boolean isNumericOperator(String operator, RuleExt.Condition condition) {
+        return switch (operator) {
+            case ">", ">=", "<", "<=", "between", "outside" -> true;
+            case "==", "!=" -> condition.getThreshold() != null;
+            default -> false;
+        };
+    }
+
+}
@@ -0,0 +1,77 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.data.biz.alarm;
+
+import io.github.pnoker.common.data.entity.property.AlarmWindowProperties;
+import io.github.pnoker.common.enums.WindowMode;
+import lombok.RequiredArgsConstructor;
+import org.springframework.context.annotation.Primary;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Routes a window request to either the in-memory buffer (short windows) or
+ * the repository (long windows). The cutoff is configurable via
+ * {@code dc3.alarm.window.local-cutoff} (default 5 minutes); rules whose
+ * duration sits at or below the cutoff stay local, larger ones go to the
+ * time-series store.
+ *
+ * <p>Marked {@code @Primary} so injection of the bare {@link WindowDataSource}
+ * picks the hybrid by default. Tests inject the local / repository sources
+ * directly when they want to target one side.
+ *
+ * @author pnoker
+ * @version 2026.5.21
+ * @since 2026.5.21
+ */
+@Component
+@Primary
+@RequiredArgsConstructor
+public class HybridWindowDataSource implements WindowDataSource {
+
+    private final LocalWindowDataSource localWindowDataSource;
+
+    private final RepositoryWindowDataSource repositoryWindowDataSource;
+
+    private final AlarmWindowProperties properties;
+
+    @Override
+    public AggregateOutcome aggregate(WindowSpec spec, RuleFact fact, WindowMode mode) {
+        return select(spec).aggregate(spec, fact, mode);
+    }
+
+    @Override
+    public List<WindowSample> samples(WindowSpec spec, RuleFact fact) {
+        return select(spec).samples(spec, fact);
+    }
+
+    private WindowDataSource select(WindowSpec spec) {
+        if (Objects.isNull(spec) || Objects.isNull(spec.duration())) {
+            return localWindowDataSource;
+        }
+        Duration cutoff = properties.getLocalCutoff();
+        if (Objects.isNull(cutoff) || spec.duration().compareTo(cutoff) <= 0) {
+            return localWindowDataSource;
+        }
+        return repositoryWindowDataSource;
+    }
+
+}
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.data.biz.alarm;
+
+import io.github.pnoker.common.enums.WindowMode;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Reads window samples out of {@link WindowSampleBuffer} and folds them in
+ * Java. Fast enough for short windows (≤ a few thousand samples) — long
+ * windows route to {@link RepositoryWindowDataSource} instead.
+ *
+ * @author pnoker
+ * @version 2026.5.21
+ * @since 2026.5.21
+ */
+@Component
+@RequiredArgsConstructor
+public class LocalWindowDataSource implements WindowDataSource {
+
+    /**
+     * Scale used for AVG so callers don't have to deal with infinite-precision
+     * results. Mirrors the typical PostgreSQL {@code AVG(numeric)} behavior.
+     */
+    private static final int AVG_SCALE = 6;
+
+    private final WindowSampleBuffer windowSampleBuffer;
+
+    @Override
+    public AggregateOutcome aggregate(WindowSpec spec, RuleFact fact, WindowMode mode) {
+        List<WindowSample> samples = samples(spec, fact);
+        long count = samples.size();
+        if (mode == WindowMode.COUNT) {
+            return new AggregateOutcome(BigDecimal.valueOf(count), count);
+        }
+        long numericCount = samples.stream().filter(WindowSample::isNumeric).count();
+        if (numericCount == 0) {
+            return new AggregateOutcome(null, count);
+        }
+        BigDecimal value = switch (mode) {
+            case SUM -> samples.stream().filter(WindowSample::isNumeric)
+                    .map(s -> BigDecimal.valueOf(s.numValue()))
+                    .reduce(BigDecimal.ZERO, BigDecimal::add);
+            case MIN -> samples.stream().filter(WindowSample::isNumeric)
+                    .map(s -> BigDecimal.valueOf(s.numValue()))
+                    .min(BigDecimal::compareTo).orElse(null);
+            case MAX -> samples.stream().filter(WindowSample::isNumeric)
+                    .map(s -> BigDecimal.valueOf(s.numValue()))
+                    .max(BigDecimal::compareTo).orElse(null);
+            case AVG -> samples.stream().filter(WindowSample::isNumeric)
+                    .map(s -> BigDecimal.valueOf(s.numValue()))
+                    .reduce(BigDecimal.ZERO, BigDecimal::add)
+                    .divide(BigDecimal.valueOf(numericCount), AVG_SCALE, RoundingMode.HALF_UP);
+            default -> null;
+        };
+        return new AggregateOutcome(value, count);
+    }
+
+    @Override
+    public List<WindowSample> samples(WindowSpec spec, RuleFact fact) {
+        if (Objects.isNull(spec) || Objects.isNull(spec.duration())
+                || Objects.isNull(fact) || Objects.isNull(fact.getEntityId())
+                || Objects.isNull(fact.getAlarmTargetTypeFlag())) {
+            return List.of();
+        }
+        LocalDateTime to = Objects.requireNonNullElse(fact.getFactTime(), LocalDateTime.now());
+        LocalDateTime from = to.minus(spec.duration());
+        WindowSampleKey key = WindowSampleKey.of(fact.getTenantId(), fact.getAlarmTargetTypeFlag(), fact.getEntityId());
+        return windowSampleBuffer.snapshot(key, from, to);
+    }
+
+}
@@ -0,0 +1,135 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.data.biz.alarm;
+
+import io.github.pnoker.common.entity.bo.PointValueBO;
+import io.github.pnoker.common.entity.bo.WindowAggregateResult;
+import io.github.pnoker.common.entity.query.WindowAggregateRequest;
+import io.github.pnoker.common.enums.AlarmTargetTypeFlagEnum;
+import io.github.pnoker.common.enums.WindowMode;
+import io.github.pnoker.common.repository.RepositoryService;
+import io.github.pnoker.common.strategy.RepositoryStrategyFactory;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Long-window backend. Pushes the aggregate to the time-series repository
+ * (PostgreSQL today) so the alarm engine doesn't have to materialize an
+ * unbounded sample list in memory. ALL/ANY still pull raw rows here because
+ * the rule's condition is sample-by-sample.
+ *
+ * <p>Only POINT facts are supported — the time-series store is keyed on
+ * (tenantId, deviceId, pointId). Device/driver windowed alarms over long
+ * spans aren't a current use case; supporting them would require a different
+ * storage layout.
+ *
+ * @author pnoker
+ * @version 2026.5.21
+ * @since 2026.5.21
+ */
+@Slf4j
+@Component
+public class RepositoryWindowDataSource implements WindowDataSource {
+
+    @Override
+    public AggregateOutcome aggregate(WindowSpec spec, RuleFact fact, WindowMode mode) {
+        if (!isPointFact(fact) || Objects.isNull(spec) || Objects.isNull(spec.duration())) {
+            return AggregateOutcome.empty();
+        }
+        Long deviceId = longValue(fact.value("deviceId"));
+        if (Objects.isNull(deviceId) || deviceId <= 0) {
+            return AggregateOutcome.empty();
+        }
+        LocalDateTime to = Objects.requireNonNullElse(fact.getFactTime(), LocalDateTime.now());
+        LocalDateTime from = to.minus(spec.duration());
+
+        WindowAggregateRequest req = WindowAggregateRequest.builder()
+                .tenantId(fact.getTenantId())
+                .deviceId(deviceId)
+                .pointId(fact.getEntityId())
+                .function(mode.name())
+                .from(from)
+                .to(to)
+                .build();
+        try {
+            WindowAggregateResult result = currentRepository().aggregateInWindow(req);
+            return new AggregateOutcome(result.value(), result.sampleCount());
+        } catch (RuntimeException e) {
+            log.warn("Repository window aggregate failed, treating as empty; tenantId={}, pointId={}, mode={}",
+                    fact.getTenantId(), fact.getEntityId(), mode, e);
+            return AggregateOutcome.empty();
+        }
+    }
+
+    @Override
+    public List<WindowSample> samples(WindowSpec spec, RuleFact fact) {
+        if (!isPointFact(fact) || Objects.isNull(spec) || Objects.isNull(spec.duration())) {
+            return List.of();
+        }
+        Long deviceId = longValue(fact.value("deviceId"));
+        if (Objects.isNull(deviceId) || deviceId <= 0) {
+            return List.of();
+        }
+        LocalDateTime to = Objects.requireNonNullElse(fact.getFactTime(), LocalDateTime.now());
+        LocalDateTime from = to.minus(spec.duration());
+        try {
+            List<PointValueBO> rows = currentRepository().samplesInWindow(
+                    fact.getTenantId(), deviceId, fact.getEntityId(), from, to);
+            return rows.stream()
+                    .map(bo -> new WindowSample(bo.getNumValue(), bo.getCalValue(), bo.getCreateTime()))
+                    .toList();
+        } catch (RuntimeException e) {
+            log.warn("Repository window samples failed; tenantId={}, pointId={}",
+                    fact.getTenantId(), fact.getEntityId(), e);
+            return List.of();
+        }
+    }
+
+    private static boolean isPointFact(RuleFact fact) {
+        return Objects.nonNull(fact) && fact.getAlarmTargetTypeFlag() == AlarmTargetTypeFlagEnum.POINT
+                && Objects.nonNull(fact.getTenantId()) && Objects.nonNull(fact.getEntityId());
+    }
+
+    private static Long longValue(Object value) {
+        if (value instanceof Long longValue) {
+            return longValue;
+        }
+        if (value instanceof Number number) {
+            return number.longValue();
+        }
+        return null;
+    }
+
+    /**
+     * Look the strategy up dynamically — keeps this class compatible with the
+     * single-repository constraint enforced elsewhere without coupling to a
+     * specific implementation.
+     */
+    private RepositoryService currentRepository() {
+        List<RepositoryService> impls = RepositoryStrategyFactory.get();
+        if (impls.isEmpty()) {
+            throw new IllegalStateException("No repository service registered");
+        }
+        return impls.get(0);
+    }
+
+}
@@ -17,26 +17,26 @@

 package io.github.pnoker.common.data.biz.alarm;

-import io.github.pnoker.common.constant.service.AlarmConstant;
 import io.github.pnoker.common.data.entity.bo.RuleBO;
 import io.github.pnoker.common.entity.ext.RuleExt;
+import io.github.pnoker.common.enums.WindowMode;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang3.StringUtils;
 import org.springframework.stereotype.Service;

-import java.math.BigDecimal;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;

 /**
- * Deterministic evaluator for structured alarm rules.
+ * Deterministic evaluator for structured alarm rules. Dispatches by window
+ * mode: LAST (or no window) evaluates the rule's condition against the fact's
+ * named field directly; everything else delegates to
+ * {@link WindowedRuleEvaluator}.
 *
- * <p>Window-aware rules are not yet implemented in this release. The only
- * supported {@code RuleExt.Window.mode} value is {@code LAST} (or null/blank,
- * which is treated as LAST). Any other mode (AVG/MIN/MAX/SUM/COUNT/ALL/ANY) is
- * skipped at evaluation time with a one-time WARN per rule id, so a rule
- * configured against an unimplemented mode will not silently behave as LAST.
+ * <p>Invalid window specs (unknown mode, malformed duration) should already
+ * be rejected upstream by the save validator. If one slips through, the
+ * evaluator skips it with a one-time warn per rule id.
 *
 * @author pnoker
 * @version 2025.9.0
@@ -44,20 +44,30 @@ import java.util.concurrent.ConcurrentHashMap;
 */
@Slf4j
@Service
+@RequiredArgsConstructor
 public class RuleEvaluatorImpl implements RuleEvaluator {

-    private final Set<Long> warnedUnsupportedRules = ConcurrentHashMap.newKeySet();
+    private final WindowedRuleEvaluator windowedRuleEvaluator;
+
+    private final Set<Long> warnedInvalidRules = ConcurrentHashMap.newKeySet();

    @Override
    public boolean matches(RuleBO rule, RuleFact fact) {
-        if (!isWindowModeSupported(rule)) {
+        if (Objects.isNull(rule) || Objects.isNull(fact)) {
            return false;
        }
+        WindowSpec spec = parseSpec(rule);
+        if (!spec.valid()) {
+            return false;
+        }
+        if (spec.mode() != WindowMode.LAST) {
+            return windowedRuleEvaluator.matches(rule, fact, spec);
+        }
        RuleExt.Condition condition = condition(rule);
-        if (Objects.isNull(condition) || Objects.isNull(fact)) {
+        if (Objects.isNull(condition)) {
            return false;
        }
-        return evaluate(condition, fact.value(condition.getField()));
+        return ConditionEvaluator.evaluate(condition, fact.value(condition.getField()));
    }

    @Override
@@ -66,127 +76,43 @@ public class RuleEvaluatorImpl implements RuleEvaluator {
                || Objects.isNull(fact)) {
            return false;
        }
-        if (!isWindowModeSupported(rule)) {
+        WindowSpec spec = parseSpec(rule);
+        if (!spec.valid()) {
            return false;
        }
+        if (spec.mode() != WindowMode.LAST) {
+            return windowedRuleEvaluator.recovers(rule, fact, spec);
+        }
        RuleExt.Recovery recovery = rule.getRuleExt().getContent().getRecovery();
        RuleExt.Condition condition = rule.getRuleExt().getContent().getCondition();
        if (Objects.isNull(recovery) || !Boolean.TRUE.equals(recovery.getEnabled()) || Objects.isNull(condition)) {
            return false;
        }
-        RuleExt.Condition recoveryCondition = new RuleExt.Condition(
-                condition.getField(),
-                recovery.getOperator(),
-                null,
-                recovery.getThreshold(),
-                null,
-                null,
-                condition.getUnit());
-        return evaluate(recoveryCondition, fact.value(condition.getField()));
+        RuleExt.Condition recoveryCondition = ConditionEvaluator.recoveryConditionOf(condition, recovery);
+        return ConditionEvaluator.evaluate(recoveryCondition, fact.value(condition.getField()));
    }

-    private boolean isWindowModeSupported(RuleBO rule) {
-        if (Objects.isNull(rule) || Objects.isNull(rule.getRuleExt())
-                || Objects.isNull(rule.getRuleExt().getContent())) {
-            return true;
+    private WindowSpec parseSpec(RuleBO rule) {
+        RuleExt.Window window = window(rule);
+        WindowSpec spec = WindowSpecParser.parse(window);
+        if (!spec.valid() && Objects.nonNull(rule.getId()) && warnedInvalidRules.add(rule.getId())) {
+            log.warn("Skipping rule[{}] because window spec is invalid: {}", rule.getId(), spec.reason());
        }
-        RuleExt.Window window = rule.getRuleExt().getContent().getWindow();
-        if (Objects.isNull(window) || StringUtils.isBlank(window.getMode())) {
-            return true;
-        }
-        if (StringUtils.equalsIgnoreCase(window.getMode(), AlarmConstant.WINDOW_MODE_LAST)) {
-            return true;
-        }
-        if (Objects.nonNull(rule.getId()) && warnedUnsupportedRules.add(rule.getId())) {
-            log.warn("Skipping rule[{}] because window mode '{}' is not yet supported; only LAST is implemented",
-                    rule.getId(), window.getMode());
-        }
-        return false;
+        return spec;
    }

-    private RuleExt.Condition condition(RuleBO rule) {
-        if (Objects.isNull(rule) || Objects.isNull(rule.getRuleExt()) || Objects.isNull(rule.getRuleExt().getContent())) {
+    private static RuleExt.Window window(RuleBO rule) {
+        if (Objects.isNull(rule.getRuleExt()) || Objects.isNull(rule.getRuleExt().getContent())) {
+            return null;
+        }
+        return rule.getRuleExt().getContent().getWindow();
+    }
+
+    private static RuleExt.Condition condition(RuleBO rule) {
+        if (Objects.isNull(rule.getRuleExt()) || Objects.isNull(rule.getRuleExt().getContent())) {
            return null;
        }
        return rule.getRuleExt().getContent().getCondition();
    }

-    private boolean evaluate(RuleExt.Condition condition, Object value) {
-        String operator = StringUtils.lowerCase(StringUtils.trim(condition.getOperator()));
-        if ("exists".equals(operator)) {
-            return Objects.nonNull(value);
-        }
-        if ("missing".equals(operator)) {
-            return Objects.isNull(value);
-        }
-        if (Objects.isNull(value)) {
-            return false;
-        }
-
-        BigDecimal actual = toBigDecimal(value);
-        if (isNumericOperator(operator, condition)) {
-            return evaluateNumeric(operator, actual, condition);
-        }
-        return evaluateText(operator, value, condition);
-    }
-
-    private boolean evaluateNumeric(String operator, BigDecimal actual, RuleExt.Condition condition) {
-        if (Objects.isNull(actual)) {
-            return false;
-        }
-        return switch (operator) {
-            case ">" -> condition.getThreshold() != null && compare(actual, condition.getThreshold()) > 0;
-            case ">=" -> condition.getThreshold() != null && compare(actual, condition.getThreshold()) >= 0;
-            case "<" -> condition.getThreshold() != null && compare(actual, condition.getThreshold()) < 0;
-            case "<=" -> condition.getThreshold() != null && compare(actual, condition.getThreshold()) <= 0;
-            case "==" -> condition.getThreshold() != null && compare(actual, condition.getThreshold()) == 0;
-            case "!=" -> condition.getThreshold() != null && compare(actual, condition.getThreshold()) != 0;
-            case "between" -> condition.getLow() != null && condition.getHigh() != null
-                    && compare(actual, condition.getLow()) >= 0 && compare(actual, condition.getHigh()) <= 0;
-            case "outside" -> condition.getLow() != null && condition.getHigh() != null
-                    && (compare(actual, condition.getLow()) < 0 || compare(actual, condition.getHigh()) > 0);
-            default -> false;
-        };
-    }
-
-    private boolean evaluateText(String operator, Object value, RuleExt.Condition condition) {
-        String actual = Objects.toString(value, "");
-        String expected = Objects.toString(condition.getExpected(), "");
-        return switch (operator) {
-            case "==", "eq" -> StringUtils.equals(actual, expected);
-            case "!=", "ne" -> !StringUtils.equals(actual, expected);
-            case "contains" -> StringUtils.contains(actual, expected);
-            default -> false;
-        };
-    }
-
-    private boolean isNumericOperator(String operator, RuleExt.Condition condition) {
-        return switch (operator) {
-            case ">", ">=", "<", "<=", "between", "outside" -> true;
-            case "==", "!=" -> condition.getThreshold() != null;
-            default -> false;
-        };
-    }
-
-    private int compare(BigDecimal actual, BigDecimal expected) {
-        return actual.compareTo(expected);
-    }
-
-    private BigDecimal toBigDecimal(Object value) {
-        try {
-            if (value instanceof BigDecimal decimal) {
-                return decimal;
-            }
-            if (value instanceof Number number) {
-                return new BigDecimal(number.toString());
-            }
-            if (value instanceof CharSequence text && StringUtils.isNotBlank(text)) {
-                return new BigDecimal(text.toString());
-            }
-            return null;
-        } catch (NumberFormatException e) {
-            return null;
-        }
-    }
-
 }
@@ -0,0 +1,70 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.data.biz.alarm;
+
+import io.github.pnoker.common.enums.WindowMode;
+
+import java.math.BigDecimal;
+import java.util.List;
+
+/**
+ * Pluggable backend for windowed alarm evaluation. Two production
+ * implementations exist:
+ *
+ * <ul>
+ *   <li>{@link LocalWindowDataSource} reads from {@link WindowSampleBuffer}
+ *       — fast, but bounded by retention.</li>
+ *   <li>{@link RepositoryWindowDataSource} pushes the aggregate to the
+ *       time-series store — durable, but with per-evaluation latency.</li>
+ * </ul>
+ *
+ * <p>{@link HybridWindowDataSource} routes between them by window duration.
+ *
+ * @author pnoker
+ * @version 2026.5.21
+ * @since 2026.5.21
+ */
+public interface WindowDataSource {
+
+    /**
+     * Return the scalar aggregate (AVG/MIN/MAX/SUM/COUNT) for the rule's
+     * window. The result also carries the sample count so callers can enforce
+     * the {@code minSamples} guard. Modes that do not reduce to a scalar
+     * (LAST/ALL/ANY) should not be passed in.
+     */
+    AggregateOutcome aggregate(WindowSpec spec, RuleFact fact, WindowMode mode);
+
+    /**
+     * Pull the raw samples in the rule's window, ordered oldest → newest.
+     * Used by ALL/ANY where the rule condition runs sample-by-sample.
+     */
+    List<WindowSample> samples(WindowSpec spec, RuleFact fact);
+
+    /**
+     * Aggregate result + sample count. The value is null when the window had
+     * no usable samples (numeric aggregate over an empty / all-null window).
+     */
+    record AggregateOutcome(BigDecimal value, long sampleCount) {
+
+        public static AggregateOutcome empty() {
+            return new AggregateOutcome(null, 0L);
+        }
+
+    }
+
+}
@@ -0,0 +1,153 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.data.biz.alarm;
+
+import io.github.pnoker.common.data.entity.bo.RuleBO;
+import io.github.pnoker.common.entity.ext.RuleExt;
+import io.github.pnoker.common.enums.WindowMode;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * Window-aware rule evaluation. Handles every {@link WindowMode} other than
+ * {@link WindowMode#LAST} (which {@link RuleEvaluatorImpl} continues to
+ * service inline because it does not need the data-source plumbing).
+ *
+ * <p>Behavior:
+ *
+ * <ul>
+ *   <li>AVG/MIN/MAX/SUM/COUNT — pull a scalar aggregate from
+ *       {@link WindowDataSource#aggregate(WindowSpec, RuleFact, WindowMode)}
+ *       and feed it into {@link ConditionEvaluator}; minSamples gate enforces
+ *       the rule's "needs at least N samples" hint.</li>
+ *   <li>ALL — every sample in the window must satisfy the rule's per-sample
+ *       condition.</li>
+ *   <li>ANY — at least one sample must satisfy.</li>
+ * </ul>
+ *
+ * <p>Recovery uses the synthesized recovery condition (operator + threshold
+ * from {@code RuleExt.Recovery}) but the same window + mode. That is the
+ * legacy behavior — recovery does not get a separate window definition.
+ *
+ * @author pnoker
+ * @version 2026.5.21
+ * @since 2026.5.21
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class WindowedRuleEvaluator {
+
+    private final WindowDataSource windowDataSource;
+
+    /**
+     * Evaluate the rule's firing branch against the windowed samples.
+     */
+    public boolean matches(RuleBO rule, RuleFact fact, WindowSpec spec) {
+        RuleExt.Condition condition = condition(rule);
+        if (Objects.isNull(condition) || Objects.isNull(spec) || !spec.valid() || spec.mode() == WindowMode.LAST) {
+            return false;
+        }
+        return evaluate(spec, fact, condition);
+    }
+
+    /**
+     * Evaluate the rule's recovery branch against the windowed samples. Mirrors
+     * {@link RuleEvaluatorImpl#recovers} for the LAST path: only fires when
+     * recovery is enabled, the operator is non-null, and the synthesized
+     * condition holds.
+     */
+    public boolean recovers(RuleBO rule, RuleFact fact, WindowSpec spec) {
+        if (Objects.isNull(rule) || Objects.isNull(rule.getRuleExt())
+                || Objects.isNull(rule.getRuleExt().getContent())) {
+            return false;
+        }
+        RuleExt.Recovery recovery = rule.getRuleExt().getContent().getRecovery();
+        RuleExt.Condition condition = rule.getRuleExt().getContent().getCondition();
+        if (Objects.isNull(recovery) || !Boolean.TRUE.equals(recovery.getEnabled())
+                || Objects.isNull(condition) || Objects.isNull(spec) || !spec.valid()
+                || spec.mode() == WindowMode.LAST) {
+            return false;
+        }
+        return evaluate(spec, fact, ConditionEvaluator.recoveryConditionOf(condition, recovery));
+    }
+
+    private boolean evaluate(WindowSpec spec, RuleFact fact, RuleExt.Condition condition) {
+        if (Objects.isNull(condition)) {
+            return false;
+        }
+        WindowMode mode = spec.mode();
+        return switch (mode) {
+            case AVG, MIN, MAX, SUM, COUNT -> evaluateAggregate(spec, fact, mode, condition);
+            case ALL, ANY -> evaluateFold(spec, fact, mode, condition);
+            default -> false;
+        };
+    }
+
+    private boolean evaluateAggregate(WindowSpec spec, RuleFact fact, WindowMode mode, RuleExt.Condition condition) {
+        WindowDataSource.AggregateOutcome outcome = windowDataSource.aggregate(spec, fact, mode);
+        if (outcome.sampleCount() < spec.minSamples()) {
+            return false;
+        }
+        return ConditionEvaluator.evaluate(condition, outcome.value());
+    }
+
+    private boolean evaluateFold(WindowSpec spec, RuleFact fact, WindowMode mode, RuleExt.Condition condition) {
+        List<WindowSample> samples = windowDataSource.samples(spec, fact);
+        if (samples.size() < spec.minSamples()) {
+            return false;
+        }
+        if (samples.isEmpty()) {
+            return false;
+        }
+        return switch (mode) {
+            case ALL -> samples.stream().allMatch(s -> ConditionEvaluator.evaluate(condition, sampleValue(condition, s)));
+            case ANY -> samples.stream().anyMatch(s -> ConditionEvaluator.evaluate(condition, sampleValue(condition, s)));
+            default -> false;
+        };
+    }
+
+    /**
+     * Pick the sample's numeric or text projection depending on the field the
+     * rule's condition selects. Supports {@code numValue}, {@code calValue} /
+     * {@code value} / {@code rawValue}; everything else falls back to the
+     * numeric projection — the condition's operator coerces from there.
+     */
+    private static Object sampleValue(RuleExt.Condition condition, WindowSample sample) {
+        if (Objects.isNull(condition) || Objects.isNull(condition.getField())) {
+            return sample.numValue();
+        }
+        return switch (condition.getField()) {
+            case "numValue" -> sample.numValue();
+            case "calValue", "value", "rawValue" -> sample.calValue();
+            default -> sample.numValue();
+        };
+    }
+
+    private static RuleExt.Condition condition(RuleBO rule) {
+        if (Objects.isNull(rule) || Objects.isNull(rule.getRuleExt()) || Objects.isNull(rule.getRuleExt().getContent())) {
+            return null;
+        }
+        return rule.getRuleExt().getContent().getCondition();
+    }
+
+}
@@ -28,10 +28,15 @@ import java.util.List;
 import java.util.Map;

 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;

 class RuleEvaluatorImplTest {

-    private final RuleEvaluatorImpl evaluator = new RuleEvaluatorImpl();
+    private final WindowedRuleEvaluator windowedRuleEvaluator = mock(WindowedRuleEvaluator.class);
+    private final RuleEvaluatorImpl evaluator = new RuleEvaluatorImpl(windowedRuleEvaluator);

    @Test
    void matchesNumericThresholdRule() {
@@ -86,39 +91,37 @@ class RuleEvaluatorImplTest {
    }

    @Test
-    void rejectsNonLastWindowModeOnMatches() {
+    void delegatesNonLastModesToWindowedEvaluator() {
        RuleBO rule = rule(">", BigDecimal.valueOf(80), null);
        rule.getRuleExt().getContent().setWindow(new RuleExt.Window("AVG", "PT3M", 3));
        RuleFact fact = fact(Map.of("numValue", BigDecimal.valueOf(86)));
+        when(windowedRuleEvaluator.matches(any(), any(), any())).thenReturn(true);

-        // The current fact would otherwise satisfy the threshold, but AVG mode is
-        // not yet implemented; refuse to fall back to LAST semantics silently.
-        assertThat(evaluator.matches(rule, fact)).isFalse();
+        assertThat(evaluator.matches(rule, fact)).isTrue();
+        verify(windowedRuleEvaluator).matches(any(), any(), any());
    }

    @Test
-    void rejectsNonLastWindowModeOnRecovers() {
+    void delegatesNonLastModesToWindowedEvaluatorOnRecovery() {
        RuleBO rule = rule(">", BigDecimal.valueOf(80), null);
        rule.getRuleExt().getContent().setRecovery(new RuleExt.Recovery(true, "<=", BigDecimal.valueOf(75), "PT2M"));
        rule.getRuleExt().getContent().setWindow(new RuleExt.Window("COUNT", "PT3M", 3));
        RuleFact fact = fact(Map.of("numValue", BigDecimal.valueOf(72)));
+        when(windowedRuleEvaluator.recovers(any(), any(), any())).thenReturn(true);

-        assertThat(evaluator.recovers(rule, fact)).isFalse();
+        assertThat(evaluator.recovers(rule, fact)).isTrue();
+        verify(windowedRuleEvaluator).recovers(any(), any(), any());
    }

    @Test
-    void warnsOnceForRepeatedRejectionOfSameRule() {
-        // Two evaluations of the same rule should not double-log; the rejection
-        // warning is rate-limited per rule id via a ConcurrentHashMap.
+    void invalidWindowSpecIsSkipped() {
+        // Malformed duration → spec is invalid → matches returns false without
+        // touching the windowed evaluator.
        RuleBO rule = rule(">", BigDecimal.valueOf(80), null);
-        rule.getRuleExt().getContent().setWindow(new RuleExt.Window("MAX", "PT3M", 3));
+        rule.getRuleExt().getContent().setWindow(new RuleExt.Window("AVG", "5 minutes", 3));
        RuleFact fact = fact(Map.of("numValue", BigDecimal.valueOf(86)));

        assertThat(evaluator.matches(rule, fact)).isFalse();
-        assertThat(evaluator.matches(rule, fact)).isFalse();
-        // No assertions on log output here — the dedup guarantee is internal —
-        // but the test exists so a regression that rips the dedup will at least
-        // be visible in the test trace.
    }

    private RuleBO rule(String operator, BigDecimal threshold, String expected) {
@@ -0,0 +1,200 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package io.github.pnoker.common.data.biz.alarm;
+
+import io.github.pnoker.common.data.entity.bo.RuleBO;
+import io.github.pnoker.common.entity.ext.RuleExt;
+import io.github.pnoker.common.enums.AlarmTargetTypeFlagEnum;
+import io.github.pnoker.common.enums.WindowMode;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.math.BigDecimal;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class WindowedRuleEvaluatorTest {
+
+    @Mock
+    private WindowDataSource windowDataSource;
+
+    @InjectMocks
+    private WindowedRuleEvaluator evaluator;
+
+    private static RuleBO rule(String operator, BigDecimal threshold, RuleExt.Recovery recovery) {
+        RuleExt.Content content = new RuleExt.Content(
+                new RuleExt.Condition("numValue", operator, null, threshold, null, null, "C"),
+                null,
+                recovery,
+                "P1",
+                "ALARM",
+                List.of("temperature"));
+        RuleExt ext = new RuleExt(content);
+        ext.setType("POINT_VALUE_RULE");
+        ext.setVersion(1);
+        RuleBO rule = new RuleBO();
+        rule.setId(1L);
+        rule.setRuleCode("temp-high");
+        rule.setRuleExt(ext);
+        return rule;
+    }
+
+    private static RuleFact fact() {
+        return new RuleFact(7L, AlarmTargetTypeFlagEnum.POINT, 11L, null, LocalDateTime.now(), Map.of());
+    }
+
+    private static WindowSpec spec(WindowMode mode, int minSamples) {
+        return WindowSpec.ok(mode, Duration.ofMinutes(3), minSamples);
+    }
+
+    @Test
+    void avgFiresWhenAggregateAboveThreshold() {
+        when(windowDataSource.aggregate(any(), any(), eq(WindowMode.AVG)))
+                .thenReturn(new WindowDataSource.AggregateOutcome(BigDecimal.valueOf(85), 5));
+
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(), spec(WindowMode.AVG, 3));
+        assertThat(fired).isTrue();
+    }
+
+    @Test
+    void avgDoesNotFireBelowThreshold() {
+        when(windowDataSource.aggregate(any(), any(), eq(WindowMode.AVG)))
+                .thenReturn(new WindowDataSource.AggregateOutcome(BigDecimal.valueOf(72), 5));
+
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(), spec(WindowMode.AVG, 3));
+        assertThat(fired).isFalse();
+    }
+
+    @Test
+    void respectsMinSamplesGate() {
+        when(windowDataSource.aggregate(any(), any(), eq(WindowMode.AVG)))
+                .thenReturn(new WindowDataSource.AggregateOutcome(BigDecimal.valueOf(85), 2));
+
+        // 2 samples in the window but minSamples=5 → don't fire even though
+        // the aggregate would otherwise satisfy the threshold.
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(), spec(WindowMode.AVG, 5));
+        assertThat(fired).isFalse();
+    }
+
+    @Test
+    void countModeUsesSampleCountAsValue() {
+        when(windowDataSource.aggregate(any(), any(), eq(WindowMode.COUNT)))
+                .thenReturn(new WindowDataSource.AggregateOutcome(BigDecimal.valueOf(7), 7));
+
+        boolean fired = evaluator.matches(rule(">=", BigDecimal.valueOf(5), null), fact(), spec(WindowMode.COUNT, 1));
+        assertThat(fired).isTrue();
+    }
+
+    @Test
+    void allModeFiresWhenEverySampleSatisfies() {
+        List<WindowSample> samples = List.of(
+                new WindowSample(85.0, "85", LocalDateTime.now()),
+                new WindowSample(82.0, "82", LocalDateTime.now()),
+                new WindowSample(90.0, "90", LocalDateTime.now()));
+        when(windowDataSource.samples(any(), any())).thenReturn(samples);
+
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(), spec(WindowMode.ALL, 3));
+        assertThat(fired).isTrue();
+    }
+
+    @Test
+    void allModeFailsWhenOneSampleDoesNotSatisfy() {
+        List<WindowSample> samples = List.of(
+                new WindowSample(85.0, "85", LocalDateTime.now()),
+                new WindowSample(70.0, "70", LocalDateTime.now()),
+                new WindowSample(90.0, "90", LocalDateTime.now()));
+        when(windowDataSource.samples(any(), any())).thenReturn(samples);
+
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(), spec(WindowMode.ALL, 3));
+        assertThat(fired).isFalse();
+    }
+
+    @Test
+    void anyModeFiresWhenAtLeastOneSampleSatisfies() {
+        List<WindowSample> samples = List.of(
+                new WindowSample(70.0, "70", LocalDateTime.now()),
+                new WindowSample(72.0, "72", LocalDateTime.now()),
+                new WindowSample(85.0, "85", LocalDateTime.now()));
+        when(windowDataSource.samples(any(), any())).thenReturn(samples);
+
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(), spec(WindowMode.ANY, 1));
+        assertThat(fired).isTrue();
+    }
+
+    @Test
+    void anyModeFailsWhenNoSampleSatisfies() {
+        List<WindowSample> samples = List.of(
+                new WindowSample(70.0, "70", LocalDateTime.now()),
+                new WindowSample(75.0, "75", LocalDateTime.now()));
+        when(windowDataSource.samples(any(), any())).thenReturn(samples);
+
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(), spec(WindowMode.ANY, 1));
+        assertThat(fired).isFalse();
+    }
+
+    @Test
+    void allModeRespectsMinSamples() {
+        // Only 1 sample present but minSamples=3 → don't fire even if it
+        // satisfies the condition.
+        when(windowDataSource.samples(any(), any())).thenReturn(List.of(
+                new WindowSample(85.0, "85", LocalDateTime.now())));
+
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(), spec(WindowMode.ALL, 3));
+        assertThat(fired).isFalse();
+    }
+
+    @Test
+    void recoveryUsesSynthesizedConditionAndWindow() {
+        when(windowDataSource.aggregate(any(), any(), eq(WindowMode.AVG)))
+                .thenReturn(new WindowDataSource.AggregateOutcome(BigDecimal.valueOf(70), 5));
+        RuleExt.Recovery recovery = new RuleExt.Recovery(true, "<=", BigDecimal.valueOf(75), "PT2M");
+
+        boolean recovered = evaluator.recovers(
+                rule(">", BigDecimal.valueOf(80), recovery), fact(), spec(WindowMode.AVG, 3));
+        assertThat(recovered).isTrue();
+    }
+
+    @Test
+    void recoveryNoOpsWhenRecoveryDisabled() {
+        RuleExt.Recovery disabled = new RuleExt.Recovery(false, "<=", BigDecimal.valueOf(75), "PT2M");
+        boolean recovered = evaluator.recovers(
+                rule(">", BigDecimal.valueOf(80), disabled), fact(), spec(WindowMode.AVG, 3));
+        assertThat(recovered).isFalse();
+    }
+
+    @Test
+    void rejectsLastModeAtThisEntrypoint() {
+        // LAST is handled by RuleEvaluatorImpl directly; the windowed
+        // evaluator must not be reachable for LAST. Defensive guard.
+        boolean fired = evaluator.matches(rule(">", BigDecimal.valueOf(80), null), fact(),
+                WindowSpec.last());
+        assertThat(fired).isFalse();
+    }
+
+}