AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/S8WatchSchedulerService.cs

using Admin.NET.Plugin.AiDOP.Entity.S8;
using Admin.NET.Plugin.AiDOP.Infrastructure.S8;
using Admin.NET.Plugin.AiDOP.Service.S8.Rules;
using Admin.NET.Plugin.AiDOP.Service.S8.Rules.DataAccess;
using Admin.NET.Plugin.AiDOP.Service.S8.Rules.Definitions;
using Microsoft.Extensions.Logging;
using SqlSugar;
using System.Globalization;
using System.Text.Json;

namespace Admin.NET.Plugin.AiDOP.Service.S8;

/// <summary>
/// 监视规则轮询调度服务（首轮存根）。
/// 后续接入 Admin.NET 定时任务机制后，由调度器周期调用 <see cref="RunOnceAsync"/>，
/// 按各规则的 PollIntervalSeconds 逐条评估并生成异常记录。
/// </summary>
public class S8WatchSchedulerService : ITransient
{
    private readonly SqlSugarRepository<AdoS8WatchRule> _ruleRep;
    private readonly SqlSugarRepository<AdoS8Exception> _exceptionRep;
    private readonly SqlSugarRepository<AdoS8ExceptionType> _exceptionTypeRep;
    private readonly S8NotificationService _notificationService;
    private readonly S8NotificationLayerResolver _notificationLayerResolver;
    private readonly S8ImpactMetricsService _impactMetricsService;
    private readonly S8ManualReportService _manualReportService;
    // S8-STEP6E：Watch 主链的业务建单边界（唯一最窄 seam）。仅 :1482 的自动建单走它；
    // legacy debug 主链的 CreateFromWatchAsync 保持直连 _manualReportService，行为不变。
    private readonly IS8ExceptionReportDispatcher _reportDispatcher;
    private readonly S8TimeoutRuleEvaluator _timeoutEvaluator;
    private readonly S8ShortageRuleEvaluator _shortageEvaluator;
    private readonly S8OutOfRangeRuleEvaluator _outOfRangeEvaluator;
    private readonly ILogger<S8WatchSchedulerService> _logger;
    private readonly SqlSugarRepository<AdoS8DetectionLog> _detectionLogRep;
    private readonly SqlSugarRepository<AdoS8RuleDetectionState> _detectionStateRep;
    private readonly IS8RuleCatalog _ruleCatalog;

    private const string DetectionTriggerSource = "WATCH_SCHEDULER";
    private const string DetectResultCreated = "CREATED";
    private const string DetectResultRefreshed = "REFRESHED";
    private const string DetectResultRecovered = "RECOVERED";
    private const string DetectResultNoHit = "NO_HIT";
    private const string DetectResultEvaluateFailed = "EVALUATE_FAILED";

    // S8-DETECTION-LOG-WRITE-REDUCE-P1-1：REFRESHED 明细写入限频窗口。
    // 同 (tenant, factory, rule_code, dedup_key) 在 10 分钟内最多写一条 REFRESHED detection_log；
    // 业务侧 RefreshDetectionAsync / BackfillLegacyExceptionAsync 不受限频影响。
    private const int RefreshedRateLimitMinutes = 10;

    private const string DefaultTriggerType = "VALUE_DEVIATION";
    private const string SqlDataSourceType = "SQL";

    // G01-05 未闭环状态集合：复用自 S8ExceptionService 当前 pendingStatuses 事实口径
    // （见 S8ExceptionService.GetPagedAsync 中 pendingStatuses 的定义，两处必须保持一致）。
    // 这不是“自定义未闭环集合”；若现有口径调整，两处需同步修改。
    private static readonly string[] UnclosedExceptionStatuses =
        { "NEW", "ASSIGNED", "IN_PROGRESS", "PENDING_VERIFICATION" };

    public S8WatchSchedulerService(
        SqlSugarRepository<AdoS8WatchRule> ruleRep,
        SqlSugarRepository<AdoS8Exception> exceptionRep,
        SqlSugarRepository<AdoS8ExceptionType> exceptionTypeRep,
        S8NotificationService notificationService,
        S8NotificationLayerResolver notificationLayerResolver,
        S8ImpactMetricsService impactMetricsService,
        S8ManualReportService manualReportService,
        IS8ExceptionReportDispatcher reportDispatcher,
        S8TimeoutRuleEvaluator timeoutEvaluator,
        S8ShortageRuleEvaluator shortageEvaluator,
        S8OutOfRangeRuleEvaluator outOfRangeEvaluator,
        ILogger<S8WatchSchedulerService> logger,
        SqlSugarRepository<AdoS8DetectionLog> detectionLogRep,
        SqlSugarRepository<AdoS8RuleDetectionState> detectionStateRep,
        IS8RuleCatalog ruleCatalog)
    {
        _ruleRep = ruleRep;
        _exceptionRep = exceptionRep;
        _exceptionTypeRep = exceptionTypeRep;
        _notificationService = notificationService;
        _notificationLayerResolver = notificationLayerResolver;
        _impactMetricsService = impactMetricsService;
        _manualReportService = manualReportService;
        _reportDispatcher = reportDispatcher;
        _timeoutEvaluator = timeoutEvaluator;
        _shortageEvaluator = shortageEvaluator;
        _outOfRangeEvaluator = outOfRangeEvaluator;
        _logger = logger;
        _detectionLogRep = detectionLogRep;
        _detectionStateRep = detectionStateRep;
        _ruleCatalog = ruleCatalog;
    }

    /// <summary>
    /// S8-TENANT-ONLY-BATCH5：调度遍历改为 <b>Tenant → Rule</b>。
    ///
    /// <para>原实现返回 DISTINCT (tenant_id, factory_id)，于是同一租户在 N 个工厂下会被 tick N 次；
    /// 而规则的运行策略按 Batch 2 的口径每租户只有一条，重复 tick 只会互相抢同一把租约。</para>
    ///
    /// <para>同时删掉 <c>AND r.factory_id &gt; 0</c>：Batch 2 供给出来的运行策略行
    /// <c>factory_id</c> 恒为 0，那条谓词会让它们**永远不被调度、且不报任何错**
    /// —— 页面显示已启用，实际从不执行。这是本批必须与供给同时闭合的缺口。</para>
    /// </summary>
    public async Task<List<long>> ListEnabledTenantsAsync()
    {
        return await _ruleRep.Context.Ado.SqlQueryAsync<long>(
            """
            SELECT DISTINCT r.tenant_id
            FROM ado_s8_watch_rule r
            INNER JOIN SysTenant t ON t.Id = r.tenant_id AND t.Status = 1
            WHERE r.enabled = 1
              AND r.tenant_id > 0
            ORDER BY r.tenant_id
            """);
    }






    // 取任意一条匹配的未闭环异常 Id 作为“是否存在重复单”的拦截依据。
    // 首版只需要“存在性”，不关心“最早 / 最新”；不在 G01-05 处理排序语义。

    /// <summary>
    /// 自动建单入口（debug run-once 使用；生产调度走 <see cref="RunDispatchTickAsync"/>）。
    ///
    /// S8-LEGACY-SQL-RESIDUAL-CLEANUP-3：原先此处还有一段旧 AlertRule 兼容主链
    /// （EvaluateDedupAsync → EvaluateHitsAsync → QueryDeviceRowsAsync → 直连 SQL），
    /// 该链自行开 SqlSugarScope 执行 rule.expression，
    /// 且只装载 rule_type 为空的未分类历史规则 + 需同场景恰好一条可运行 AlertRule。
    /// 经证实其唯一入口是 404 门禁的 debug controller，且 LoadExecutionRulesAsync 的过滤条件
    /// 在当前数据下恒返回空，已整体删除。本方法现在只保留三类正式 evaluator 路径。
    /// </summary>
    public async Task<List<S8WatchCreationResult>> CreateExceptionsAsync(long tenantId)
    {
        var results = new List<S8WatchCreationResult>();

        // R6 RunId：本次 CreateExceptionsAsync 调用对应的统一关联 id，落入 detection_log。
        var runId = Guid.NewGuid().ToString("N").Substring(0, 16);
        results.AddRange(await ProcessRulesByTypeAsync(tenantId, _timeoutEvaluator, S8TimeoutRuleEvaluator.RuleTypeCode, runId));
        results.AddRange(await ProcessRulesByTypeAsync(tenantId, _shortageEvaluator, S8ShortageRuleEvaluator.RuleTypeCode, runId));
        results.AddRange(await ProcessRulesByTypeAsync(tenantId, _outOfRangeEvaluator, S8OutOfRangeRuleEvaluator.RuleTypeCode, runId));

        return results;
    }

    /// <summary>
    /// R2 TIMEOUT 类规则主链：薄包装，复用 <see cref="ProcessRulesByTypeAsync"/>。RunId 由内部生成。
    /// </summary>
    public Task<List<S8WatchCreationResult>> ProcessTimeoutRulesAsync(long tenantId) =>
        ProcessRulesByTypeAsync(tenantId, _timeoutEvaluator, S8TimeoutRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));

    /// <summary>
    /// R3 SHORTAGE 类规则主链：薄包装，复用 <see cref="ProcessRulesByTypeAsync"/>。RunId 由内部生成。
    /// </summary>
    public Task<List<S8WatchCreationResult>> ProcessShortageRulesAsync(long tenantId) =>
        ProcessRulesByTypeAsync(tenantId, _shortageEvaluator, S8ShortageRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));

    /// <summary>
    /// R3-OUT_OF_RANGE-REWRITE-1：OUT_OF_RANGE 类规则主链。
    /// 复用 <see cref="ProcessRulesByTypeAsync"/>，并对历史 dedup_key=NULL 的旧记录做 compat fallback：
    /// (source_rule_id=rule.Id AND related_object_code=hit AND status!=CLOSED AND dedup_key IS NULL AND is_deleted=0)
    /// 命中则 backfill 6 列，避免重复建单。RunId 由内部生成。
    /// </summary>
    public Task<List<S8WatchCreationResult>> ProcessOutOfRangeRulesAsync(long tenantId) =>
        ProcessRulesByTypeAsync(tenantId, _outOfRangeEvaluator, S8OutOfRangeRuleEvaluator.RuleTypeCode, Guid.NewGuid().ToString("N").Substring(0, 16));

    /// <summary>
    /// R2/R3 通用规则主链。S8-SCHED-CLEANUP-LEGACY-PATH-1：本方法已收敛为 thin wrapper，
    /// 单规则处理（evaluator → reconcile → hit 循环 → CREATED/REFRESHED/NO_HIT/EVALUATE_FAILED 日志）
    /// 全部下沉至 <see cref="ProcessSingleRuleAsync"/>，与 Job tick 路径
    /// (<see cref="RunDispatchTickAsync"/> → <see cref="RunSingleRuleAsync"/> → ProcessSingleRuleAsync)
    /// 共享同一份逻辑，避免双维护。
    ///
    /// 调用方仅有 <see cref="CreateExceptionsAsync"/>（debug run-once / Process*RulesAsync 公共薄包装）。
    /// debug run-once 不持 lease（保留"手动立即跑"语义），但本方法在每条规则结束时通过
    /// <see cref="ApplyRunOnceCompletionAsync"/> 同步更新 watch_rule.last_run_at / last_status /
    /// last_error / last_duration_ms / last_run_id / consecutive_failure_count，缩小 run-once 与
    /// Job tick 之间的 last_* 状态分裂；不动 lock_token / running_started_at / paused_until，避免
    /// 与正在持锁运行的 Job 撕扯。
    /// </summary>
    private async Task<List<S8WatchCreationResult>> ProcessRulesByTypeAsync(
        long tenantId, IS8RuleEvaluator evaluator, string ruleType, string runId)
    {
        var aggregate = new List<S8WatchCreationResult>();

        var rules = await _ruleRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                        && x.Enabled
                        && x.RuleType == ruleType)
            .ToListAsync();
        if (rules.Count == 0) return aggregate;

        foreach (var rule in rules.OrderBy(x => x.Id))
        {
            // S8-RULE-GOVERNANCE-BATCH1：无代码定义的规则不进入执行链（与 Job tick 路径同一口径）。
            if (!_ruleCatalog.IsDefined(rule.RuleCode))
            {
                _logger.LogWarning(
                    "run_once_skip_no_definition ruleId={RuleId} ruleCode={RuleCode}", rule.Id, rule.RuleCode);
                aggregate.Add(BuildSkipResult(rule, S8RuleCatalog.ReasonNotFound, null));
                continue;
            }

            // S8-RUN-ONCE-LEASE-AWARENESS-1：debug run-once 不持 lease，但若该 rule 已被 Scheduler Job
            // 通过 PickReadyRulesAsync 抢锁（lock_token 非空且 lock_until > now），run-once 跳过该 rule，
            // 不写 last_*、不写 detection_log、不动锁，以避免与 Job tick 并发评估同一 rule 而互相覆盖运行态。
            // 锁状态用 ToListAsync 的初始快照判定；run-once 与 Job 的微秒级竞速无法在不抢 lease 的前提下
            // 完全消除（已登记为 follow-up 风险）。
            if (!string.IsNullOrEmpty(rule.LockToken)
                && rule.LockUntil.HasValue
                && rule.LockUntil.Value > DateTime.Now)
            {
                _logger.LogInformation(
                    "run_once_skip_locked ruleId={RuleId} ruleCode={RuleCode} ruleType={RuleType} lockToken={Token} lockedBy={By} lockUntil={Until}",
                    rule.Id, rule.RuleCode, ruleType, rule.LockToken, rule.LockedBy, rule.LockUntil);
                aggregate.Add(BuildSkipResult(rule, "rule_locked_by_scheduler", null));
                continue;
            }

            var sw = System.Diagnostics.Stopwatch.StartNew();
            S8RuleRunResult completion;
            try
            {
                var ruleResults = await ProcessSingleRuleAsync(tenantId, rule, ruleType, evaluator, runId);
                aggregate.AddRange(ruleResults);
                completion = new S8RuleRunResult
                {
                    Success = true,
                    Stats = new S8RuleRunStats
                    {
                        Hits = ruleResults.Count,
                        Created = ruleResults.Count(r => r.Created),
                        Refreshed = ruleResults.Count(r => r.Reason == "duplicate_pending"),
                        Pending = ruleResults.Count(r => r.Reason == "antiflap_pending_hit"),
                        Failed = ruleResults.Count(r => r.Reason == "create_failed" || r.Reason == "refresh_failed" || r.Reason == "antiflap_failed" || r.Reason == "evaluate_failed")
                    }
                };
            }
            catch (Exception ex)
            {
                // ProcessSingleRuleAsync 在 evaluator 抛 S8RuleEvaluatorException 时已写 EVALUATE_FAILED
                // detection_log 后再 throw（供 RunSingleRuleAsync 标 Success=false）。本路径无 lease，
                // 吞异常以保留"逐规则失败不影响其他规则"的旧 ProcessRulesByTypeAsync 语义。
                _logger.LogWarning(ex, "process_rule_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
                aggregate.Add(BuildSkipResult(rule, "evaluate_failed", ex.Message));
                completion = new S8RuleRunResult { Success = false, ErrorMessage = ex.Message, Stats = new() };
            }
            sw.Stop();

            try
            {
                await ApplyRunOnceCompletionAsync(rule, completion, (int)sw.ElapsedMilliseconds, runId);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "run_once_completion_write_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
            }
        }

        return aggregate;
    }

    /// <summary>
    /// S8-SCHED-CLEANUP-LEGACY-PATH-1：debug run-once 的 last_* 状态回写（无 lease 版）。
    /// 与 <see cref="OnRuleCompletedAsync"/> 的语义平行，差异在于：
    /// - 不要求 lock_token 匹配（run-once 不持 lease）
    /// - 不写 lock_token / locked_by / lock_until / running_started_at（不与 Job lease 撕扯）
    /// - 不做 consecutive_failure_count 阈值的 paused_until 自动暂停（debug 路径不应自动暂停 demo rule）
    /// 写入：last_run_at / next_run_at / last_status / last_error / last_duration_ms / last_run_id /
    ///       consecutive_failure_count / updated_at。
    /// </summary>
    private async Task ApplyRunOnceCompletionAsync(AdoS8WatchRule rule, S8RuleRunResult result, int durationMs, string runId)
    {
        var now = DateTime.Now;
        var nextRunAt = now.AddSeconds(NormalizePollInterval(rule.PollIntervalSeconds));
        if (result.Success)
        {
            await _ruleRep.Context.Updateable<AdoS8WatchRule>()
                .SetColumns(x => new AdoS8WatchRule
                {
                    LastRunAt = now,
                    NextRunAt = nextRunAt,
                    LastStatus = "SUCCESS",
                    LastError = null,
                    LastDurationMs = durationMs,
                    LastRunId = runId,
                    ConsecutiveFailureCount = 0,
                    UpdatedAt = now
                })
                .Where(x => x.Id == rule.Id)
                .ExecuteCommandAsync();
        }
        else
        {
            var errorTrunc = Truncate(result.ErrorMessage, 500);
            await _ruleRep.Context.Updateable<AdoS8WatchRule>()
                .SetColumns(x => new AdoS8WatchRule
                {
                    LastRunAt = now,
                    NextRunAt = nextRunAt,
                    LastStatus = "FAILED",
                    LastError = errorTrunc,
                    LastDurationMs = durationMs,
                    LastRunId = runId,
                    ConsecutiveFailureCount = x.ConsecutiveFailureCount + 1,
                    UpdatedAt = now
                })
                .Where(x => x.Id == rule.Id)
                .ExecuteCommandAsync();
        }
    }

    // S8-SCHED-EXEC-1：trigger / recover 抗抖计数兜底，null / <1 / >10 一律按 1，避免非法配置导致永远不建单 / 永远不恢复。
    private static int NormalizeAntiflapCount(int raw)
    {
        if (raw < 1 || raw > 10) return 1;
        return raw;
    }

    /// <summary>
    /// 命中时累加 detection_state.consecutive_hit_count；未存在则插入 hitCount=1。
    /// 返回当前 state 行（含 Id）以及命中后的 hitCount。
    /// 注意：本函数不消费 trigger_count_required；上游决定是否进入 CreateFromHitAsync。
    /// </summary>
    private async Task<(AdoS8RuleDetectionState? state, int hitCount)> UpsertDetectionStateOnHitAsync(
        long tenantId, AdoS8WatchRule rule, S8RuleHit hit)
    {
        var now = DateTime.Now;
        var existing = await _detectionStateRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                        && x.RuleCode == rule.RuleCode
                        && x.DedupKey == hit.DedupKey)
            .FirstAsync();

        if (existing == null)
        {
            var fresh = new AdoS8RuleDetectionState
            {
                TenantId = tenantId,
                FactoryId = rule.FactoryId,
                RuleCode = rule.RuleCode,
                DedupKey = hit.DedupKey,
                SourceObjectType = string.IsNullOrEmpty(hit.SourceObjectType) ? null : hit.SourceObjectType,
                SourceObjectId = string.IsNullOrEmpty(hit.SourceObjectId) ? null : hit.SourceObjectId,
                ConsecutiveHitCount = 1,
                ConsecutiveMissCount = 0,
                LastSeenAt = now,
                LastHitAt = now,
                CreatedAt = now,
                UpdatedAt = now
            };
            // BUG-S8-DETECTION-STATE-ACTIVE-EXC-ID-TRIGGER1-001：必须回填 fresh.Id，否则
            // trigger=1 首 tick 建单后 UPDATE state SET active_exception_id WHERE id=state.Id
            // 命中 0 行（state.Id 为默认 0）。沿用 S8ManualReportService 既定模式。
            fresh = await _detectionStateRep.AsInsertable(fresh).ExecuteReturnEntityAsync();
            return (fresh, 1);
        }

        var newHitCount = existing.ConsecutiveHitCount + 1;
        await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
            .SetColumns(x => new AdoS8RuleDetectionState
            {
                ConsecutiveHitCount = x.ConsecutiveHitCount + 1,
                ConsecutiveMissCount = 0,
                LastSeenAt = now,
                LastHitAt = now,
                SourceObjectType = string.IsNullOrEmpty(hit.SourceObjectType) ? existing.SourceObjectType : hit.SourceObjectType,
                SourceObjectId = string.IsNullOrEmpty(hit.SourceObjectId) ? existing.SourceObjectId : hit.SourceObjectId,
                UpdatedAt = now
            })
            .Where(x => x.Id == existing.Id)
            .ExecuteCommandAsync();
        existing.ConsecutiveHitCount = newHitCount;
        return (existing, newHitCount);
    }

    private async Task<long> FindOpenExceptionByDedupKeyAsync(long tenantId, string dedupKey)
    {
        var ids = await _exceptionRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                        && !x.IsDeleted
                        && x.Status != "CLOSED"
                        && x.DedupKey == dedupKey)
            .Select(x => x.Id)
            .Take(1)
            .ToListAsync();
        return ids.Count > 0 ? ids[0] : 0L;
    }

    /// <summary>
    /// R3 OUT_OF_RANGE compat fallback 查找：用 (source_rule_id, related_object_code, status!=CLOSED,
    /// dedup_key IS NULL, is_deleted=0) 严格条件定位旧 AlertRule 主链留下的历史记录。
    /// </summary>
    private async Task<long> FindLegacyOutOfRangeExceptionAsync(long tenantId, long sourceRuleId, string relatedObjectCode)
    {
        if (string.IsNullOrWhiteSpace(relatedObjectCode)) return 0L;
        var ids = await _exceptionRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                        && !x.IsDeleted
                        && x.Status != "CLOSED"
                        && x.DedupKey == null
                        && x.SourceRuleId == sourceRuleId
                        && x.RelatedObjectCode == relatedObjectCode)
            .Select(x => x.Id)
            .Take(1)
            .ToListAsync();
        return ids.Count > 0 ? ids[0] : 0L;
    }

    /// <summary>
    /// R3 OUT_OF_RANGE compat fallback backfill：把历史记录的 R1 新 6 列（dedup_key/source_rule_code/
    /// source_object_type/source_object_id/source_payload/last_detected_at）写入，并刷新 updated_at。
    /// </summary>
    private async Task BackfillLegacyExceptionAsync(long exceptionId, S8RuleHit hit)
    {
        await _exceptionRep.Context.Updateable<AdoS8Exception>()
            .SetColumns(x => new AdoS8Exception
            {
                DedupKey = hit.DedupKey,
                SourceRuleCode = hit.SourceRuleCode,
                SourceObjectType = hit.SourceObjectType,
                SourceObjectId = hit.SourceObjectId,
                SourcePayload = hit.SourcePayload,
                LastDetectedAt = hit.DetectedAt,
                UpdatedAt = DateTime.Now
            })
            .Where(x => x.Id == exceptionId)
            .ExecuteCommandAsync();
    }

    /// <summary>
    /// R5 恢复时间最小闭环：对当前 rule 下未关闭、有 dedup_key、recovered_at 仍为 NULL 的异常，
    /// 凡不在本轮 hits.dedup_key 集合内的，写入 recovered_at = now、updated_at = now。
    /// 仅写这 2 列；不动 status / assignee / verifier / source_payload / last_detected_at；
    /// recovered_at 一旦写入，本轮不做复发清空。
    /// R6 返回 recoveredIds 供上游决定是否写 NO_HIT 日志，并对每个 recovered exception 写一条 RECOVERED 日志。
    /// </summary>
    private async Task<List<long>> ReconcileRecoveriesForRuleAsync(
        long tenantId, AdoS8WatchRule rule, string ruleType, List<S8RuleHit> hits, string runId)
    {
        var hitDedupKeys = hits
            .Where(h => !string.IsNullOrWhiteSpace(h.DedupKey))
            .Select(h => h.DedupKey)
            .ToHashSet(StringComparer.Ordinal);

        var candidates = await _exceptionRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                        && !x.IsDeleted
                        && x.Status != "CLOSED"
                        && x.SourceRuleCode == rule.RuleCode
                        && x.DedupKey != null
                        && x.RecoveredAt == null)
            .Select(x => new { x.Id, x.DedupKey, x.SourceObjectType, x.SourceObjectId, x.RelatedObjectCode, x.ConsecutiveMissCount })
            .ToListAsync();
        if (candidates.Count == 0) return new List<long>();

        var now = DateTime.Now;
        var recoverRequired = NormalizeAntiflapCount(rule.RecoverCountRequired);
        var recoveredIds = new List<long>();
        foreach (var c in candidates)
        {
            if (hitDedupKeys.Contains(c.DedupKey!)) continue;

            // S8-SCHED-EXEC-1：恢复抗抖累计。
            // 1) 每次未命中：异常 ConsecutiveMissCount += 1，ConsecutiveHitCount 清零；
            // 2) miss < recover_count_required：仅累计，不写 recovered_at、不写 RECOVERED；
            // 3) miss >= recover_count_required：写 recovered_at、写 RECOVERED 日志。
            var newMissCount = c.ConsecutiveMissCount + 1;
            await _exceptionRep.Context.Updateable<AdoS8Exception>()
                .SetColumns(x => new AdoS8Exception
                {
                    ConsecutiveMissCount = x.ConsecutiveMissCount + 1,
                    ConsecutiveHitCount = 0,
                    UpdatedAt = now
                })
                .Where(x => x.Id == c.Id)
                .ExecuteCommandAsync();

            // detection_state 同步累计 miss（建单后 state.active_exception_id 仍指向 c.Id）。
            await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
                .SetColumns(x => new AdoS8RuleDetectionState
                {
                    ConsecutiveMissCount = x.ConsecutiveMissCount + 1,
                    ConsecutiveHitCount = 0,
                    LastSeenAt = now,
                    LastMissAt = now,
                    UpdatedAt = now
                })
                .Where(x => x.TenantId == tenantId
                                && x.RuleCode == rule.RuleCode
                            && x.DedupKey == c.DedupKey)
                .ExecuteCommandAsync();

            if (newMissCount < recoverRequired)
            {
                _logger.LogInformation(
                    "antiflap_pending_recovery ruleCode={RuleCode} dedupKey={DedupKey} missCount={Miss} recoverRequired={Required}",
                    rule.RuleCode, c.DedupKey, newMissCount, recoverRequired);
                continue;
            }

            await _exceptionRep.Context.Updateable<AdoS8Exception>()
                .SetColumns(x => new AdoS8Exception
                {
                    RecoveredAt = now,
                    UpdatedAt = now
                })
                .Where(x => x.Id == c.Id)
                .ExecuteCommandAsync();
            recoveredIds.Add(c.Id);

            await WriteDetectionLogAsync(new AdoS8DetectionLog
            {
                TenantId = tenantId, FactoryId = rule.FactoryId,
                RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
                SourceObjectType = c.SourceObjectType, SourceObjectId = c.SourceObjectId,
                RelatedObjectCode = c.RelatedObjectCode, DedupKey = c.DedupKey,
                DetectResult = DetectResultRecovered,
                ExceptionId = c.Id,
                DetectedAt = now,
                PayloadSnapshot = JsonSerializer.Serialize(new { ruleId = rule.Id, ruleCode = rule.RuleCode, reason = "no_longer_hit", missCount = newMissCount, recoverRequired }),
                RunId = runId, TriggerSource = DetectionTriggerSource,
                Remark = "Rule no longer hit; recovered_at marked"
            });

            // S8-NOTIFY-WIRE-RECOVERED-1：detection_log 已写入、recovered_at 已落库后挂入恢复通知。
            // 通知失败仅 LogWarning，绝不影响恢复状态/检测日志。
            await TryDispatchRecoveredLayerNotificationAsync(c.Id);
        }

        if (recoveredIds.Count > 0)
        {
            _logger.LogInformation(
                "rule_recovered ruleCode={RuleCode} ruleType={RuleType} recoveredCount={Count} recoveredIds={Ids}",
                rule.RuleCode, ruleType, recoveredIds.Count, string.Join(",", recoveredIds));
        }
        return recoveredIds;
    }

    /// <summary>R6 通用 hit 日志构造（CREATED / REFRESHED 共用）。</summary>
    private static AdoS8DetectionLog BuildHitLog(
        long tenantId, AdoS8WatchRule rule, string ruleType, S8RuleHit hit,
        string detectResult, long exceptionId, string runId) => new()
    {
        TenantId = tenantId, FactoryId = rule.FactoryId,
        RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
        SourceObjectType = hit.SourceObjectType, SourceObjectId = hit.SourceObjectId,
        RelatedObjectCode = hit.RelatedObjectCode, DedupKey = hit.DedupKey,
        DetectResult = detectResult,
        ExceptionId = exceptionId,
        DetectedAt = hit.DetectedAt,
        PayloadSnapshot = hit.SourcePayload,
        RunId = runId,
        TriggerSource = DetectionTriggerSource
    };

    /// <summary>R6 日志写入：失败仅 LogWarning，不阻断主链；不抛异常。</summary>
    private async Task WriteDetectionLogAsync(AdoS8DetectionLog log)
    {
        try
        {
            await _detectionLogRep.InsertAsync(log);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "detection_log_write_failed runId={RunId} detectResult={Result} ruleCode={RuleCode} exceptionId={ExceptionId}",
                log.RunId, log.DetectResult, log.RuleCode, log.ExceptionId);
        }
    }

    /// <summary>
    /// S8-DETECTION-LOG-WRITE-REDUCE-P1-1：REFRESHED 明细限频判定。
    /// 查询 cutoff = detectedAt - RefreshedRateLimitMinutes 之后，
    /// 是否已有同 (tenant_id, factory_id, rule_code, dedup_key, DetectResult=REFRESHED) 的 detection_log；
    /// 命中既有索引 idx_s8_detection_log_dedup_time。查询失败仅 LogWarning 后返回 false，
    /// 宁可多写一条 REFRESHED 明细，也不阻断调度主链路；ruleCode / dedupKey 空白时同样返回 false（保留写入）。
    /// 业务侧 RefreshDetectionAsync / BackfillLegacyExceptionAsync 不在限频范围。
    /// </summary>
    private async Task<bool> HasRecentRefreshedDetectionLogAsync(
        long tenantId, string ruleCode, string dedupKey, DateTime detectedAt)
    {
        if (string.IsNullOrWhiteSpace(ruleCode) || string.IsNullOrWhiteSpace(dedupKey))
            return false;

        var cutoff = detectedAt.AddMinutes(-RefreshedRateLimitMinutes);
        try
        {
            return await _detectionLogRep.AsQueryable()
                .Where(x => x.TenantId == tenantId
                                && x.DedupKey == dedupKey
                            && x.RuleCode == ruleCode
                            && x.DetectResult == DetectResultRefreshed
                            && x.DetectedAt >= cutoff)
                .AnyAsync();
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex,
                "detection_log_refreshed_ratelimit_check_failed ruleCode={RuleCode} dedupKey={DedupKey}",
                ruleCode, dedupKey);
            return false;
        }
    }

    private static string Truncate(string? s, int max) =>
        string.IsNullOrEmpty(s) ? string.Empty : (s.Length <= max ? s : s.Substring(0, max));

    private async Task RefreshDetectionAsync(long exceptionId, S8RuleHit hit)
    {
        // S8-SCHED-EXEC-1：刷新阶段抗抖累计 + 复发清空 recovered_at。
        // ConsecutiveHitCount += 1（用 SetColumns 内表达式完成原子自增）；ConsecutiveMissCount 归零。
        // RecoveredAt 不为 NULL 时（复发）一并清空，保持业务对"再次命中即视为活跃"的预期。
        await _exceptionRep.Context.Updateable<AdoS8Exception>()
            .SetColumns(x => new AdoS8Exception
            {
                LastDetectedAt = hit.DetectedAt,
                SourcePayload = hit.SourcePayload,
                ConsecutiveHitCount = x.ConsecutiveHitCount + 1,
                ConsecutiveMissCount = 0,
                RecoveredAt = null,
                UpdatedAt = DateTime.Now
            })
            .Where(x => x.Id == exceptionId)
            .ExecuteCommandAsync();
    }

    private static S8WatchCreationResult BuildCreatedResult(AdoS8WatchRule rule, S8RuleHit hit, long exceptionId) =>
        new()
        {
            DedupResult = new S8WatchDedupResult
            {
                Hit = ToWatchHit(rule, hit),
                CanCreate = true,
                MatchedExceptionId = null,
                Reason = "no_pending"
            },
            Created = true,
            Skipped = false,
            CreatedExceptionId = exceptionId,
            Reason = "auto_created",
            ErrorMessage = null
        };

    private static S8WatchCreationResult BuildSkippedDuplicate(AdoS8WatchRule rule, S8RuleHit hit, long matchedId) =>
        new()
        {
            DedupResult = new S8WatchDedupResult
            {
                Hit = ToWatchHit(rule, hit),
                CanCreate = false,
                MatchedExceptionId = matchedId,
                Reason = "duplicate_pending"
            },
            Created = false,
            Skipped = true,
            CreatedExceptionId = null,
            Reason = "duplicate_pending",
            ErrorMessage = null
        };

    private static S8WatchCreationResult BuildSkipResult(AdoS8WatchRule rule, string reason, string? error, S8RuleHit? hit = null) =>
        new()
        {
            DedupResult = new S8WatchDedupResult
            {
                Hit = hit != null ? ToWatchHit(rule, hit) : new S8WatchHitResult { SourceRuleId = rule.Id, SourceRuleCode = rule.RuleCode },
                CanCreate = false,
                MatchedExceptionId = null,
                Reason = reason
            },
            Created = false,
            Skipped = true,
            CreatedExceptionId = null,
            Reason = reason,
            ErrorMessage = error
        };

    private static S8WatchHitResult ToWatchHit(AdoS8WatchRule rule, S8RuleHit hit) => new()
    {
        SourceRuleId = hit.SourceRuleId == 0 ? rule.Id : hit.SourceRuleId,
        SourceRuleCode = string.IsNullOrEmpty(hit.SourceRuleCode) ? rule.RuleCode : hit.SourceRuleCode,
        RelatedObjectCode = hit.RelatedObjectCode,
        Severity = hit.Severity,
        OccurrenceDeptId = hit.OccurrenceDeptId,
        ResponsibleDeptId = hit.ResponsibleDeptId,
        SourcePayload = hit.SourcePayload
    };


    // G01-04 首版最小比较符集合：>, >=, <, <=。
    // 允许首尾空格；非此集合的一律视为“比较符非法”，由调用方跳过。










    private static string NormalizeColumnName(string columnName) =>
        columnName.Replace("_", string.Empty, StringComparison.Ordinal).Trim().ToUpperInvariant();





    // ============================================================
    // S8-SCHED-EXEC-1：DB 驱动调度执行层
    // ============================================================

    private const int LeaseDurationMinutes = 5;
    private const int AutoPauseFailureThreshold = 3;
    private const int AutoPauseDurationHours = 1;
    private const int DefaultPollIntervalSeconds = 300;
    private const int MinPollIntervalSeconds = 60;
    private const int MaxPollIntervalSeconds = 86400;

    /// <summary>
    /// S8-SCHED-EXEC-1：释放过期 lease（lock_until &lt; NOW），不修改 last_status / last_error，仅清空 lock 三件套 + running_started_at。
    /// 返回释放的行数。
    /// </summary>
    public async Task<int> ResetExpiredLeasesAsync(long tenantId)
    {
        var now = DateTime.Now;
        var affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
            .SetColumns(x => new AdoS8WatchRule
            {
                LockToken = null,
                LockedBy = null,
                LockUntil = null,
                RunningStartedAt = null,
                UpdatedAt = now
            })
            .Where(x => x.TenantId == tenantId
                        && x.LockUntil != null
                        && x.LockUntil < now)
            .ExecuteCommandAsync();
        if (affected > 0)
        {
            _logger.LogWarning(
                "lease_reset tenantId={Tenant} releasedCount={Count}",
                tenantId, affected);
        }
        return affected;
    }

    /// <summary>
    /// S8-SCHED-EXEC-1：到期规则候选 + 乐观 UPDATE 抢锁，返回成功抢到的 lease 列表。
    /// 抢锁条件：enabled=1 AND (paused_until IS NULL OR paused_until &lt;= NOW)
    ///         AND (next_run_at IS NULL OR next_run_at &lt;= NOW)
    ///         AND (lock_until IS NULL OR lock_until &lt;= NOW)。
    /// 抢锁回写：lock_token / locked_by / lock_until = NOW + 5min / running_started_at = NOW / last_run_id = runId。
    /// affectedRows == 1 才算抢到；后续 OnRuleCompletedAsync 必须按 lockToken 回写，避免旧进程覆盖新 lease。
    /// </summary>
    public async Task<List<S8RuleLease>> PickReadyRulesAsync(long tenantId, int batchSize, string lockedBy, string runId)
    {
        if (batchSize <= 0) batchSize = 16;
        var now = DateTime.Now;

        var candidates = await _ruleRep.AsQueryable()
            .Where(x => x.TenantId == tenantId
                        && x.Enabled
                        // S8-STANDARD-DATASET-HARD-CUTOVER-1：取数唯一依赖 dataset_code。
                        // 保留这条谓词而非依赖 Enable Gate：Gate 只在"切换启用"那一刻生效，
                        // 而 dataset_code 可能在此之后被清空（历史遗留行、直连改库）。
                        // 一条没有数据集的规则跑起来只会每 tick 抛一次 dataset_code_missing。
                        //
                        // 刻意做成 **SQL 谓词而非内存过滤**：Take(batchSize) 在过滤之前生效，
                        // 内存过滤会让无效候选占满名额，导致同租户的有效规则被饿死。
                        && x.DatasetCode != null && x.DatasetCode != ""
                        && (x.PausedUntil == null || x.PausedUntil <= now)
                        && (x.NextRunAt == null || x.NextRunAt <= now)
                        && (x.LockUntil == null || x.LockUntil <= now))
            .OrderBy(x => x.NextRunAt, OrderByType.Asc)
            .OrderBy(x => x.Id, OrderByType.Asc)
            .Take(batchSize)
            .Select(x => new { x.Id, x.RuleCode, x.RuleType })
            .ToListAsync();
        if (candidates.Count == 0) return new();

        var leases = new List<S8RuleLease>();
        foreach (var c in candidates)
        {
            var token = Guid.NewGuid().ToString("N");
            var lockUntil = DateTime.Now.AddMinutes(LeaseDurationMinutes);
            var runningAt = DateTime.Now;
            // 乐观 UPDATE：再校验一次条件，affectedRows=1 才算抢到。
            var affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
                .SetColumns(x => new AdoS8WatchRule
                {
                    LockToken = token,
                    LockedBy = lockedBy,
                    LockUntil = lockUntil,
                    RunningStartedAt = runningAt,
                    LastRunId = runId,
                    UpdatedAt = runningAt
                })
                .Where(x => x.Id == c.Id
                            && x.Enabled
                            // 抢锁的乐观 UPDATE 复查同一条件：候选查询与本次 UPDATE 之间存在时间窗，
                            // 期间该行的 dataset_code 可能被清空。
                            && x.DatasetCode != null && x.DatasetCode != ""
                            && (x.PausedUntil == null || x.PausedUntil <= runningAt)
                            && (x.NextRunAt == null || x.NextRunAt <= runningAt)
                            && (x.LockUntil == null || x.LockUntil <= runningAt))
                .ExecuteCommandAsync();
            if (affected == 1)
            {
                leases.Add(new S8RuleLease
                {
                    RuleId = c.Id,
                    RuleCode = c.RuleCode,
                    RuleType = c.RuleType,
                    LockToken = token,
                    LockedBy = lockedBy,
                    LockUntil = lockUntil,
                    RunId = runId,
                    AcquiredAt = runningAt
                });
            }
        }
        return leases;
    }

    /// <summary>
    /// S8-SCHED-EXEC-1：执行单条已抢锁规则的 evaluator → 抗抖去重 → 建单/刷新 → 恢复 reconcile。
    /// 不释放 lease（OnRuleCompletedAsync 负责）；evaluator 抛异常时 Result.Success=false 并保留 ErrorMessage。
    /// </summary>
    public async Task<S8RuleRunResult> RunSingleRuleAsync(long tenantId, S8RuleLease lease)
    {
        // S8-P0-1-SCHEDULER-TRUSTED-SCOPE-1：执行入口必须按本次 tick 的可信作用域绑行。
        // 原实现只按 lease.RuleId 装载，不校验 rule 归属；一旦 lease 来源被改写 / 未来新增调用方，
        // 就会在 A 租户的 scope 上下文里执行 B 租户的规则，而下游全部以传入的 tenantId/factoryId 落库。
        // 这里同时做「谓词绑行」与「归属复核」两层，任一不符即 fail-fast，绝不静默降级。
        var rule = await _ruleRep.AsQueryable()
            .Where(x => x.Id == lease.RuleId && x.TenantId == tenantId)
            .FirstAsync();
        if (rule == null)
        {
            return new S8RuleRunResult { Success = false, ErrorMessage = "rule_not_found", Stats = new() };
        }
        if (rule.TenantId != tenantId)
        {
            _logger.LogError(
                "rule_scope_mismatch ruleId={RuleId} ruleTenant={RuleTenant} scopeTenant={ScopeTenant}",
                rule.Id, rule.TenantId, tenantId);
            return new S8RuleRunResult { Success = false, ErrorMessage = "rule_scope_mismatch", Stats = new() };
        }

        // S8-RULE-GOVERNANCE-BATCH1：分派的真源是**代码定义**，不是 DB 上的 rule_type 列。
        // 那一列现在只是 provisioning 维护的投影；若有人手工改它，规则的执行方式不应随之改变。
        // 同时这里是「没有代码定义的规则不得进入执行链」的把关点 —— 在 Create API 仍存在的
        // 过渡期（Batch 3 才退役），它保证业务即使造出任意 rule_code 也调不动调度器。
        var definition = _ruleCatalog.TryGet(rule.RuleCode);
        if (definition == null)
        {
            _logger.LogWarning(
                "rule_definition_not_found ruleId={RuleId} ruleCode={RuleCode} tenant={Tenant}",
                rule.Id, rule.RuleCode, tenantId);
            return new S8RuleRunResult { Success = false, ErrorMessage = S8RuleCatalog.ReasonNotFound, Stats = new() };
        }

        var ruleType = definition.RuleType;

        IS8RuleEvaluator? evaluator = ruleType switch
        {
            S8TimeoutRuleEvaluator.RuleTypeCode => _timeoutEvaluator,
            S8ShortageRuleEvaluator.RuleTypeCode => _shortageEvaluator,
            S8OutOfRangeRuleEvaluator.RuleTypeCode => _outOfRangeEvaluator,
            _ => null
        };
        if (evaluator == null)
        {
            return new S8RuleRunResult { Success = false, ErrorMessage = $"unsupported_rule_type:{ruleType}", Stats = new() };
        }

        try
        {
            var results = await ProcessSingleRuleAsync(tenantId, rule, ruleType, evaluator, lease.RunId);
            var stats = new S8RuleRunStats
            {
                Hits = results.Count,
                Created = results.Count(r => r.Created),
                Refreshed = results.Count(r => r.Reason == "duplicate_pending"),
                Pending = results.Count(r => r.Reason == "antiflap_pending_hit"),
                Failed = results.Count(r => r.Reason == "create_failed" || r.Reason == "refresh_failed" || r.Reason == "antiflap_failed" || r.Reason == "evaluate_failed")
            };
            return new S8RuleRunResult { Success = true, Stats = stats };
        }
        catch (Exception ex)
        {
            return new S8RuleRunResult
            {
                Success = false,
                ErrorMessage = ex.Message,
                Stats = new()
            };
        }
    }

    /// <summary>
    /// S8-SCHED-EXEC-1：单规则处理（evaluator → reconcile → hit 循环）。
    /// 与 ProcessRulesByTypeAsync 内单规则循环体语义一致；此处抽出便于新调度路径直接调用单条 rule。
    /// </summary>
    private async Task<List<S8WatchCreationResult>> ProcessSingleRuleAsync(
        long tenantId, AdoS8WatchRule rule, string ruleType,
        IS8RuleEvaluator evaluator, string runId)
    {
        var results = new List<S8WatchCreationResult>();
        List<S8RuleHit> hits;
        try
        {
            hits = await evaluator.EvaluateAsync(tenantId, rule);
        }
        catch (Exception ex)
        {
            var failureReason = ex is S8RuleEvaluatorException sre ? sre.Reason : ex.GetType().Name;
            await WriteDetectionLogAsync(new AdoS8DetectionLog
            {
                TenantId = tenantId, FactoryId = rule.FactoryId,
                RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
                SourceObjectType = rule.SourceObjectType,
                DetectResult = DetectResultEvaluateFailed,
                DetectedAt = DateTime.Now,
                FailureReason = failureReason,
                FailureMessage = Truncate(ex.Message, 1000),
                RunId = runId, TriggerSource = DetectionTriggerSource
            });
            results.Add(BuildSkipResult(rule, "evaluate_failed", ex.Message));
            throw;
        }

        List<long> recoveredIds;
        try
        {
            recoveredIds = await ReconcileRecoveriesForRuleAsync(tenantId, rule, ruleType, hits, runId);
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "recovery_reconcile_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
            recoveredIds = new();
        }

        // S8-DETECTION-LOG-WRITE-REDUCE-P1-1：NO_HIT 不再写 detection_log 明细。
        // 规则级"被定期评估"信号由 ado_s8_watch_rule.LastRunAt / LastStatus / LastRunId / LastDurationMs
        // 在 OnRuleCompletedAsync / ApplyRunOnceCompletionAsync 中承接，无需 detection_log 冗余记录。
        // DetectResultNoHit 常量保留（历史数据反查 / 未来如需恢复明细写入）。

        foreach (var hit in hits)
        {
            if (string.IsNullOrWhiteSpace(hit.DedupKey))
            {
                results.Add(BuildSkipResult(rule, "missing_dedup_key", null, hit));
                continue;
            }

            long matchedId;
            try
            {
                matchedId = await FindOpenExceptionByDedupKeyAsync(tenantId, hit.DedupKey);
            }
            catch (Exception ex)
            {
                results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
                continue;
            }

            if (matchedId > 0)
            {
                try
                {
                    await RefreshDetectionAsync(matchedId, hit);
                    if (!await HasRecentRefreshedDetectionLogAsync(tenantId, rule.RuleCode, hit.DedupKey, hit.DetectedAt))
                    {
                        await WriteDetectionLogAsync(BuildHitLog(tenantId, rule, ruleType, hit, DetectResultRefreshed, matchedId, runId));
                    }
                    results.Add(BuildSkippedDuplicate(rule, hit, matchedId));
                }
                catch (Exception ex)
                {
                    results.Add(BuildSkipResult(rule, "refresh_failed", ex.Message, hit));
                }
                continue;
            }

            if (string.Equals(ruleType, S8OutOfRangeRuleEvaluator.RuleTypeCode, StringComparison.OrdinalIgnoreCase))
            {
                long compatId;
                try
                {
                    compatId = await FindLegacyOutOfRangeExceptionAsync(tenantId, rule.Id, hit.RelatedObjectCode);
                }
                catch (Exception ex)
                {
                    results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
                    continue;
                }
                if (compatId > 0)
                {
                    try
                    {
                        await BackfillLegacyExceptionAsync(compatId, hit);
                        if (!await HasRecentRefreshedDetectionLogAsync(tenantId, rule.RuleCode, hit.DedupKey, hit.DetectedAt))
                        {
                            await WriteDetectionLogAsync(BuildHitLog(tenantId, rule, ruleType, hit, DetectResultRefreshed, compatId, runId));
                        }
                        results.Add(BuildSkippedDuplicate(rule, hit, compatId));
                    }
                    catch (Exception ex)
                    {
                        results.Add(BuildSkipResult(rule, "refresh_failed", ex.Message, hit));
                    }
                    continue;
                }
            }

            bool typeExists;
            try
            {
                typeExists = await _exceptionTypeRep.AsQueryable()
                    // S8-TENANT-ONLY-BATCH5：去掉工厂条件。这是一次**存在性**检查，
                    // 少一个条件只会更宽松：factory_id=0 的供给规则也能看到本租户的工厂级异常类型行，
                    // 否则它只能匹配到平台默认行、找不到就整条规则 exception_type_missing。
                    // 注意 t.TenantId == 0 是平台默认哨兵（S8ConfigScope），不是工厂语义，保留。
                    .Where(t => t.TypeCode == hit.ExceptionTypeCode
                                && (t.TenantId == 0 || t.TenantId == tenantId)
                                && t.Enabled)
                    .AnyAsync();
            }
            catch (Exception ex)
            {
                results.Add(BuildSkipResult(rule, "query_failed", ex.Message, hit));
                continue;
            }

            if (!typeExists)
            {
                results.Add(BuildSkipResult(rule, "exception_type_missing", null, hit));
                continue;
            }

            int hitCount;
            AdoS8RuleDetectionState? state;
            try
            {
                (state, hitCount) = await UpsertDetectionStateOnHitAsync(tenantId, rule, hit);
            }
            catch (Exception ex)
            {
                results.Add(BuildSkipResult(rule, "antiflap_failed", ex.Message, hit));
                continue;
            }

            var triggerRequired = NormalizeAntiflapCount(rule.TriggerCountRequired);
            if (hitCount < triggerRequired)
            {
                _logger.LogInformation(
                    "antiflap_pending_hit ruleCode={RuleCode} dedupKey={DedupKey} hitCount={HitCount} trigger={Trigger}",
                    rule.RuleCode, hit.DedupKey, hitCount, triggerRequired);
                results.Add(BuildSkipResult(rule, "antiflap_pending_hit", null, hit));
                continue;
            }

            try
            {
                var entity = await _reportDispatcher.CreateFromHitAsync(tenantId, rule.FactoryId, hit);
                await _exceptionRep.Context.Updateable<AdoS8Exception>()
                    .SetColumns(x => new AdoS8Exception
                    {
                        ConsecutiveHitCount = hitCount,
                        ConsecutiveMissCount = 0,
                        UpdatedAt = DateTime.Now
                    })
                    .Where(x => x.Id == entity.Id)
                    .ExecuteCommandAsync();
                if (state != null)
                {
                    await _detectionStateRep.Context.Updateable<AdoS8RuleDetectionState>()
                        .SetColumns(x => new AdoS8RuleDetectionState
                        {
                            ActiveExceptionId = entity.Id,
                            UpdatedAt = DateTime.Now
                        })
                        .Where(x => x.Id == state.Id)
                        .ExecuteCommandAsync();
                }
                await WriteDetectionLogAsync(BuildHitLog(tenantId, rule, ruleType, hit, DetectResultCreated, entity.Id, runId));
                await TryDispatchLayerNotificationAsync(entity);
                results.Add(BuildCreatedResult(rule, hit, entity.Id));
            }
            catch (Exception ex)
            {
                results.Add(BuildSkipResult(rule, "create_failed", ex.Message, hit));
            }
        }

        return results;
    }

    /// <summary>
    /// S8-SCHED-EXEC-1：lease 执行完成回写。
    /// 必须 WHERE id = lease.RuleId AND lock_token = lease.LockToken；affectedRows = 0 视为 lease 丢失，记录 Warning，不覆盖状态。
    /// 失败 ≥ 阈值（默认 3）写 paused_until = NOW + 1h。
    /// </summary>
    public async Task OnRuleCompletedAsync(long tenantId, S8RuleLease lease, S8RuleRunResult result, int durationMs)
    {
        var rule = await _ruleRep.AsQueryable().Where(x => x.Id == lease.RuleId).FirstAsync();
        if (rule == null)
        {
            _logger.LogWarning("lease_complete_rule_missing ruleId={RuleId}", lease.RuleId);
            return;
        }

        var now = DateTime.Now;
        var effectiveInterval = NormalizePollInterval(rule.PollIntervalSeconds);
        var nextRunAt = now.AddSeconds(effectiveInterval);
        int affected;
        if (result.Success)
        {
            affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
                .SetColumns(x => new AdoS8WatchRule
                {
                    LastRunAt = now,
                    NextRunAt = nextRunAt,
                    LastStatus = "SUCCESS",
                    LastError = null,
                    LastDurationMs = durationMs,
                    ConsecutiveFailureCount = 0,
                    LockToken = null,
                    LockedBy = null,
                    LockUntil = null,
                    RunningStartedAt = null,
                    UpdatedAt = now
                })
                .Where(x => x.Id == lease.RuleId && x.LockToken == lease.LockToken)
                .ExecuteCommandAsync();
        }
        else
        {
            var errorTrunc = Truncate(result.ErrorMessage, 500);
            var newFailures = rule.ConsecutiveFailureCount + 1;
            DateTime? pausedUntil = rule.PausedUntil;
            string? pauseReason = rule.PauseReason;
            if (newFailures >= AutoPauseFailureThreshold)
            {
                pausedUntil = now.AddHours(AutoPauseDurationHours);
                pauseReason = Truncate($"AUTO_PAUSED_AFTER_{AutoPauseFailureThreshold}_FAILURES: {errorTrunc}", 64);
            }

            affected = await _ruleRep.Context.Updateable<AdoS8WatchRule>()
                .SetColumns(x => new AdoS8WatchRule
                {
                    LastRunAt = now,
                    NextRunAt = nextRunAt,
                    LastStatus = "FAILED",
                    LastError = errorTrunc,
                    LastDurationMs = durationMs,
                    ConsecutiveFailureCount = x.ConsecutiveFailureCount + 1,
                    PausedUntil = pausedUntil,
                    PauseReason = pauseReason,
                    LockToken = null,
                    LockedBy = null,
                    LockUntil = null,
                    RunningStartedAt = null,
                    UpdatedAt = now
                })
                .Where(x => x.Id == lease.RuleId && x.LockToken == lease.LockToken)
                .ExecuteCommandAsync();
        }

        if (affected == 0)
        {
            _logger.LogWarning(
                "lease_lost_on_complete ruleId={RuleId} ruleCode={RuleCode} lockToken={LockToken}",
                lease.RuleId, lease.RuleCode, lease.LockToken);
        }
    }

    /// <summary>
    /// S8-SCHED-EXEC-1：单 tick 完整流程。Job / debug 调度入口。
    /// 1) ResetExpiredLeasesAsync
    /// 2) PickReadyRulesAsync(batchSize)
    /// 3) 每条 rule 独立 try/catch 调用 RunSingleRuleAsync + OnRuleCompletedAsync
    /// 单条规则失败不影响其他规则；整 tick 不抛异常。
    /// </summary>
    public async Task<S8DispatchTickResult> RunDispatchTickAsync(long tenantId, int batchSize, string lockedBy)
    {
        var tickId = Guid.NewGuid().ToString("N").Substring(0, 8);
        var runId = Guid.NewGuid().ToString("N").Substring(0, 16);
        var summary = new S8DispatchTickResult { TickId = tickId, RunId = runId };

        try
        {
            summary.LeaseReleased = await ResetExpiredLeasesAsync(tenantId);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "tick_reset_lease_failed tickId={TickId}", tickId);
        }

        List<S8RuleLease> leases;
        try
        {
            leases = await PickReadyRulesAsync(tenantId, batchSize, lockedBy, runId);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "tick_pick_failed tickId={TickId}", tickId);
            return summary;
        }
        summary.Picked = leases.Count;

        foreach (var lease in leases)
        {
            var sw = System.Diagnostics.Stopwatch.StartNew();
            S8RuleRunResult runResult;
            try
            {
                runResult = await RunSingleRuleAsync(tenantId, lease);
            }
            catch (Exception ex)
            {
                runResult = new S8RuleRunResult { Success = false, ErrorMessage = ex.Message, Stats = new() };
            }
            sw.Stop();
            try
            {
                await OnRuleCompletedAsync(tenantId, lease, runResult, (int)sw.ElapsedMilliseconds);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "tick_complete_failed tickId={TickId} ruleId={RuleId} ruleCode={RuleCode}", tickId, lease.RuleId, lease.RuleCode);
            }

            _logger.LogInformation(
                "tick_rule_done tickId={TickId} runId={RunId} ruleId={RuleId} ruleCode={RuleCode} status={Status} durationMs={Dur} hits={Hits} created={Created} refreshed={Refreshed} pending={Pending} failed={Failed} error={Error}",
                tickId, runId, lease.RuleId, lease.RuleCode,
                runResult.Success ? "SUCCESS" : "FAILED",
                sw.ElapsedMilliseconds,
                runResult.Stats.Hits, runResult.Stats.Created, runResult.Stats.Refreshed, runResult.Stats.Pending, runResult.Stats.Failed,
                runResult.ErrorMessage);

            if (runResult.Success)
            {
                summary.Success++;
                summary.Created += runResult.Stats.Created;
                summary.Refreshed += runResult.Stats.Refreshed;
                summary.Pending += runResult.Stats.Pending;
                summary.PerRuleFailed += runResult.Stats.Failed;
            }
            else
            {
                summary.Failed++;
            }
        }
        return summary;
    }

    private static int NormalizePollInterval(int raw)
    {
        if (raw < MinPollIntervalSeconds || raw > MaxPollIntervalSeconds) return DefaultPollIntervalSeconds;
        return raw;
    }

    /// <summary>
    /// S8-NOTIFY-WIRE-WATCH-1：异常自动建单成功后，非破坏性挂入通知分层路由。
    /// 全程异常隔离：任何异常仅 LogWarning，不抛回主流程，不影响 detection_log / 事务 /
    /// 异常状态机；该方法独立于 CreateFromWatchAsync / CreateFromHitAsync 的事务边界
    /// （两者已 commit 后才返回 entity，故在此调用安全）。
    /// sceneCode：先从 entity.SceneCode；空则 fallback 到 "S8_DEMO_DEFAULT"（demo 路径）。
    /// severity：直接取 entity.Severity（CreateFromWatchAsync/CreateFromHitAsync 已保证非空）。
    /// </summary>
    private async Task TryDispatchLayerNotificationAsync(AdoS8Exception entity)
    {
        if (entity == null || entity.Id <= 0) return;
        try
        {
            var sceneCode = string.IsNullOrWhiteSpace(entity.SceneCode) ? "S8_DEMO_DEFAULT" : entity.SceneCode;
            // S8-SEVERITY-FOLLOW-SERIOUS-STANDARDIZE-EXEC-1：写入前 Normalize，落 FOLLOW/SERIOUS。
            var severity = S8SeverityCode.Normalize(entity.Severity);

            // S8-DEMO-IMPACT-SORT-NOTICE-1：附带 30 天影响统计（重复次数 / 累计损失 / 建议关注级别）。
            // 计算失败不阻断通知主链路；snap==null 时 BuildNotification 仅写基础 Context。
            ExceptionImpactSnapshot? snap = null;
            try
            {
                snap = await _impactMetricsService.ComputeOneAsync(entity);
            }
            catch (Exception ex)
            {
                _logger.LogWarning(ex, "notify_impact_compute_throw exceptionId={ExceptionId}", entity.Id);
            }

            var baseContent =
                $"异常 {entity.ExceptionCode}：{entity.Title}（场景 {sceneCode}，严重度 {severity}" +
                (string.IsNullOrWhiteSpace(entity.SourceRuleCode) ? "" : $"，规则 {entity.SourceRuleCode}") + "）";
            var content = snap == null
                ? baseContent
                : baseContent
                    + Environment.NewLine
                    + $"过去30天同类异常发生 {snap.RepeatCount30d} 次，已关闭累计损失 {snap.CumulativeLossHours30d.ToString("0.#")} 小时，建议关注级别：{snap.SuggestedAttentionLabel}";

            await _notificationLayerResolver.DispatchByLayerAsync(new S8NotificationLayerResolver.DispatchByLayerInput
            {
                TenantId = entity.TenantId,
                FactoryId = entity.FactoryId,
                ExceptionId = entity.Id,
                ExceptionNo = entity.ExceptionCode,
                SceneCode = sceneCode,
                Severity = severity,
                Title = entity.Title ?? string.Empty,
                Content = content,
                Status = entity.Status,
                SourceRuleCode = entity.SourceRuleCode,
                JumpUrl = $"/aidop/s8/exceptions/{entity.Id}",
                RepeatCount30d = snap?.RepeatCount30d,
                CumulativeLossHours30d = snap?.CumulativeLossHours30d,
                SuggestedAttentionLevel = snap?.SuggestedAttentionLevel,
                SuggestedAttentionLabel = snap?.SuggestedAttentionLabel,
                ImpactReason = snap?.ImpactReason,
            });
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "notify_dispatch_throw exceptionId={ExceptionId}", entity.Id);
        }
    }

    /// <summary>
    /// S8-NOTIFY-WIRE-RECOVERED-1：异常恢复（recovered_at 已写入、RECOVERED detection_log 已落库）后，
    /// 非破坏性挂入分层通知。全程异常隔离：任何异常仅 LogWarning，不抛回主流程，不影响 detection_log /
    /// 状态机；call site 已在事务边界外（recovered 路径无事务）。
    /// 重新读取 entity 拿场景/严重度/编号/状态/规则代码（恢复事件相对低频，1 次额外读可接受）。
    /// S8-DEMO-IMPACT-SORT-NOTICE-1：恢复事件**不**追加影响统计字段，保持原有恢复语义不变。
    /// </summary>
    private async Task TryDispatchRecoveredLayerNotificationAsync(long exceptionId)
    {
        if (exceptionId <= 0) return;
        try
        {
            var entity = await _exceptionRep.GetByIdAsync(exceptionId);
            if (entity == null)
            {
                _logger.LogWarning("notify_recovered_dispatch_entity_missing exceptionId={ExceptionId}", exceptionId);
                return;
            }
            var sceneCode = string.IsNullOrWhiteSpace(entity.SceneCode) ? "S8_DEMO_DEFAULT" : entity.SceneCode;
            // S8-SEVERITY-FOLLOW-SERIOUS-STANDARDIZE-EXEC-1：写入前 Normalize，落 FOLLOW/SERIOUS。
            var severity = S8SeverityCode.Normalize(entity.Severity);
            var title = $"【已恢复】{entity.ExceptionCode}";
            var content =
                $"异常 {entity.ExceptionCode} 已恢复，场景 {sceneCode}，严重度 {severity}" +
                (string.IsNullOrWhiteSpace(entity.SourceRuleCode) ? "" : $"，规则 {entity.SourceRuleCode}");

            await _notificationLayerResolver.DispatchByLayerAsync(new S8NotificationLayerResolver.DispatchByLayerInput
            {
                TenantId = entity.TenantId,
                FactoryId = entity.FactoryId,
                ExceptionId = entity.Id,
                ExceptionNo = entity.ExceptionCode,
                SceneCode = sceneCode,
                Severity = severity,
                Title = title,
                Content = content,
                Status = entity.Status,
                SourceRuleCode = entity.SourceRuleCode,
                JumpUrl = $"/aidop/s8/exceptions/{entity.Id}",
                Recovered = true,
            });
        }
        catch (Exception ex)
        {
            _logger.LogWarning(ex, "notify_recovered_dispatch_throw exceptionId={ExceptionId}", exceptionId);
        }
    }
}




/// <summary>
/// G01-05 去重结果对象。仅服务 G01-06 建单前拦截，由 CanCreate 单决策位决定是否建单。
/// 只服务首版唯一场景 S2（迁移后由 S2S6_PRODUCTION 切到单模块 S2）+ 唯一 trigger_type VALUE_DEVIATION + 设备对象。
/// 不预留多 trigger_type / 平台化去重扩展结构。
/// Reason 值域：no_pending / duplicate_pending / missing_dedup_key / query_failed。
/// </summary>
public sealed class S8WatchDedupResult
{
    public S8WatchHitResult Hit { get; set; } = new();
    public bool CanCreate { get; set; }
    public long? MatchedExceptionId { get; set; }
    public string Reason { get; set; } = string.Empty;
}

/// <summary>
/// G01-06 建单结果对象。仅服务 G-01 首版主线验收，由 Created / Skipped 两位决定结局。
/// 只服务首版唯一场景 S2（迁移后由 S2S6_PRODUCTION 切到单模块 S2）+ 唯一 trigger_type VALUE_DEVIATION + 设备对象。
/// 不预留多 trigger_type / 平台化工单扩展结构。
/// Reason 值域：auto_created / create_failed / 透传自 DedupResult.Reason。
/// </summary>
public sealed class S8WatchCreationResult
{
    public S8WatchDedupResult DedupResult { get; set; } = new();
    public bool Created { get; set; }
    public bool Skipped { get; set; }
    public long? CreatedExceptionId { get; set; }
    public string Reason { get; set; } = string.Empty;
    public string? ErrorMessage { get; set; }
}

/// <summary>
/// G01-04 命中结果对象。承载 G01-05 去重与 G01-06 建单所需最小追溯字段，
/// 仅服务首版唯一场景 S2（迁移后由 S2S6_PRODUCTION 切到单模块 S2）+ 唯一 trigger_type VALUE_DEVIATION + 设备对象。
/// 不预留多 trigger_type / 多场景 / 平台化扩展结构。
/// </summary>
public sealed class S8WatchHitResult
{
    public long SourceRuleId { get; set; }
    public string SourceRuleCode { get; set; } = string.Empty;
    public string RelatedObjectCode { get; set; } = string.Empty;
    public decimal CurrentValue { get; set; }
    public decimal ThresholdValue { get; set; }
    public string TriggerCondition { get; set; } = string.Empty;
    public string Severity { get; set; } = string.Empty;
    public long? OccurrenceDeptId { get; set; }
    public long? ResponsibleDeptId { get; set; }
    public string SourcePayload { get; set; } = string.Empty;
}

/// <summary>S8-SCHED-EXEC-1：lease 抢占成功后传递的最小标识对象。</summary>
public sealed class S8RuleLease
{
    public long RuleId { get; set; }
    public string RuleCode { get; set; } = string.Empty;
    public string? RuleType { get; set; }
    public string LockToken { get; set; } = string.Empty;
    public string LockedBy { get; set; } = string.Empty;
    public DateTime LockUntil { get; set; }
    public string RunId { get; set; } = string.Empty;
    public DateTime AcquiredAt { get; set; }
}

/// <summary>S8-SCHED-EXEC-1：单条规则执行结果，OnRuleCompletedAsync 据此更新状态。</summary>
public sealed class S8RuleRunResult
{
    public bool Success { get; set; }
    public string? ErrorMessage { get; set; }
    public S8RuleRunStats Stats { get; set; } = new();
}

public sealed class S8RuleRunStats
{
    public int Hits { get; set; }
    public int Created { get; set; }
    public int Refreshed { get; set; }
    public int Pending { get; set; }
    public int Failed { get; set; }
}

/// <summary>S8-SCHED-EXEC-1：单 tick 调度结果聚合。</summary>
public sealed class S8DispatchTickResult
{
    public string TickId { get; set; } = string.Empty;
    public string RunId { get; set; } = string.Empty;
    public int LeaseReleased { get; set; }
    public int Picked { get; set; }
    public int Success { get; set; }
    public int Failed { get; set; }
    public int Created { get; set; }
    public int Refreshed { get; set; }
    public int Pending { get; set; }
    public int PerRuleFailed { get; set; }
}