feat(s8): 新增 Authority 健康门禁，阻断中台异常导致的假恢复

三条 S8 规则共用同一条恢复判定：某个 dedup_key 不在本轮 hits 里，
即累计 miss，达到阈值后写 recovered_at。这是全链路唯一一处
「从缺席做负面推断」，而它默认假设了取数是可信的。

一旦中台断供、快照没发布成功、或跑批只写了一半，本该命中的行会凭空消失，
于是真实异常被静默标成「已恢复」——没有报错、没有通知，异常就没了。
Rule03 现在就处在这个状态：贴源停更、当前快照未发布，Provider 恒取 0 行。

但不能简单地「取数为空就拦」：业务对象全部合法退场时本来就该是 0 行，
挡掉它会让真实恢复永不发生，且同样静默。所以要区分的不是空与非空，
而是 Authority 可不可信。

── 契约 ──
状态只有三个：HEALTHY / DEGRADED / UNKNOWN，仅 HEALTHY 放行恢复。
刻意不设 HEALTHY_EMPTY 状态——「可信且本轮 0 行」是 HEALTHY 与行数的组合观测，
设成状态就必须在状态机里读行数，而与行数解耦正是本门禁的全部意义。
判定核心 S8AuthorityHealthEvaluator 是纯函数，入参里不存在任何行数字段，
因此「空集就拦」在结构上写不出来。

判据（按租户 + 数据集，禁止全局最后成功时间）：
生产者最近一次终态 SUCCESS + 未超新鲜度窗口 + 无更新的终态失败
+ 无更新且未终态的在途运行；快照型 Authority 另加发布不变量。

── 门禁只挡恢复 ──
建单与刷新照常：hit 是「这一行确实存在且越界」的正面证据，
数据不可信不会凭空造出一行，却完全可能让一行消失。风险不对称，处置也不对称。
拦截时不动 ConsecutiveMissCount（已累计的 miss 来自可信轮次，仍然有效），
不把 run 标失败（否则连续三次会自动暂停规则一小时，中台跑一次批就够了），
不抛异常打断本轮。拦截写 RECOVERY_BLOCKED detection_log 并带原因码与结构化 payload
——这道门本身是静默的，事后必须能从库里回答「为什么这轮没做恢复」。

── 三条规则的判定 ──
Rule01 恒 UNKNOWN：dwd_supplier_delivery 除 MDP 转换外还有两个 API 路径直写
且都不写运行日志，运行日志证明不了数据出处。解冻条件是 Authority 收敛为单一写入方。
Rule02 只看生产运行：mdp_std_so 是当前态镜像，没有发布步骤、读者也不按批次取数，
不存在可被破坏的表级状态；唯一候选「每租户单一 sync_batch_id」既非必要也非充分。
Rule03 加发布不变量：恰好一个当前批次、等于最近成功运行的批次、当前快照内
material_role 非空（用绝对计数，空快照上真空成立）。刻意不校验 bom_level
——转换侧对非数字层级有意留 NULL，列为必填等于按设计制造假 DEGRADED。
当前批次数为 0 时判 UNKNOWN 而非「健康的空」：生产侧发布语句在本轮零行时命中 0 行、
不留任何批次标记，合法空快照与发布失败字面等同。猜成健康的代价是清空真实预警，
故保守拦截。消除它需生产侧补发布标记，属下一批。

── 三处采集侧的坑（都有实测依据）──
租户谓词严格相等，不复用 MdpMonitorService.BuildMdpRunLogTenantWhere 的
(tenant_id = @TenantId OR tenant_id = 0)：该表确有平台行（S1 有 111 条，
S5_PURCHASE_RECEIPT_MDP_SYNC 有 325 条平台级 SUCCESS），一条就能让所有租户同时判健康。
必须显式过滤终态：表内有 298 条永不回收的 RUNNING 行（S1 占 42 条、最老 570 小时），
裸 ORDER BY start_time DESC 会取到孤儿行，其 end_time 为 NULL。
回溯窗口不是可选优化：无界回溯在长期停摆的租户上实测单次 146ms，加 7 天上界后 7.3ms。
新鲜度窗口显式声明而非按作业周期推导：声明 60 分钟而实测相邻成功间隔 28-39 分钟，
近 14 天最大间隔 8.04 小时、超 3 小时的有 46 次，按 3×周期会产生 46 次假 STALE。

── 验证 ──
纯函数矩阵 28 项（CASE A-H + 原因码穷尽性 + 状态词表封闭），
接线结构守卫 9 项（CASE I/J/K + 观测性 + 花括号深度证明 hit 循环未被裹进门禁）。
反向验证 5 条变异全部被捕获：放行判定恒 true(22 红) / 租户谓词加 OR 0 /
拦截分支碰 miss 计数 / 门禁前加 hits.Count==0 短路 / 整段移除门禁(5 红)。

本地 sandbox 三条规则实跑：
  Rule01 → RECOVERY_BLOCKED x3，原因码 RULE01_MULTI_WRITER_UNTRUSTED，
           同期仍写出 56 条 REFRESHED，30 条异常 0 条被恢复
  Rule02 / Rule03 → 判 HEALTHY，无拦截，Rule03 还正常恢复了 1 条异常
  三条规则 last_status 均 SUCCESS、consecutive_failure_count 0、paused_until 空

server 1.0.522 → 1.0.523

server/Admin.NET.Web.Entry/Admin.NET.Web.Entry.csproj

@@ -11,9 +11,9 @@
     <GenerateSatelliteAssembliesForCore>true</GenerateSatelliteAssembliesForCore>
     <Copyright>Admin.NET</Copyright>
     <Description>Admin.NET 通用权限开发平台</Description>
-    <AssemblyVersion>1.0.522</AssemblyVersion>
-    <FileVersion>1.0.522</FileVersion>
-    <Version>1.0.522</Version>
+    <AssemblyVersion>1.0.523</AssemblyVersion>
+    <FileVersion>1.0.523</FileVersion>
+    <Version>1.0.523</Version>
   </PropertyGroup>
 
   <ItemGroup>

server/Plugins/Admin.NET.Plugin.AiDOP.Tests/S8/S8AuthorityHealthGateTests.cs

@@ -0,0 +1,358 @@
+using System.Reflection;
+using Admin.NET.Plugin.AiDOP.Service.S8.Rules.Health;
+using Xunit;
+
+namespace Admin.NET.Plugin.AiDOP.Tests.S8;
+
+/// <summary>
+/// S8-AUTHORITY-HEALTH-GATE：恢复放行门禁的判定逻辑。
+///
+/// <para><b>本文件同时守两种相反的失效，两边都会静默</b>：</para>
+/// <list type="number">
+///   <item><b>门禁太松</b> —— 中台断供时取数变空，恢复判定会把全部在办异常判成「已恢复」。
+///     没有报错、没有通知，异常就消失了。</item>
+///   <item><b>门禁太紧</b> —— 若有人把门禁写成「取数为空就拦」，真实恢复将永不发生，
+///     异常在列表里堆到天荒地老，而调度器每轮都报成功。
+///     <see cref="B_HealthyEmpty_StillAllowsRecovery"/> 就是为这一侧准备的反向守卫。</item>
+/// </list>
+///
+/// <para>判定是纯函数（<see cref="S8AuthorityHealthEvaluator.Evaluate"/>），
+/// 不接数据库、不接 DI，因此 A–H 全部可在普通单测里跑。
+/// 「怎么把运行日志查出来」是 <c>S8MdpAuthorityHealthResolver</c> 的事，不在此处断言。</para>
+/// </summary>
+public class S8AuthorityHealthGateTests
+{
+    private const long TenantA = 838257186181189L;
+    private const string Dataset = "ORDER_DELIVERY_RISK";
+
+    /// <summary>固定「现在」，让所有时间边界可逐分钟锁定。</summary>
+    private static readonly DateTime Now = new(2026, 9, 8, 12, 0, 0);
+
+    private static readonly TimeSpan Stale = TimeSpan.FromHours(1);
+
+    private static S8AuthorityObservation Obs(
+        DateTime? lastSuccessEndAt = null,
+        bool specFound = true,
+        bool producerTrusted = true,
+        bool newerFailure = false,
+        string? newerFailureStatus = null,
+        DateTime? inFlightStartAt = null,
+        string snapshot = S8SnapshotInvariant.NotApplicable,
+        string? snapshotDetail = null) => new()
+    {
+        DatasetCode = Dataset,
+        TenantId = TenantA,
+        ObservedAt = Now,
+        SpecFound = specFound,
+        ProducerTrusted = producerTrusted,
+        LastSuccessEndAt = lastSuccessEndAt,
+        HasNewerTerminalFailure = newerFailure,
+        NewerTerminalFailureStatus = newerFailureStatus,
+        OldestNewerInFlightStartAt = inFlightStartAt,
+        SnapshotInvariant = snapshot,
+        SnapshotDetail = snapshotDetail,
+        StaleWindow = Stale
+    };
+
+    /// <summary>一次完全健康的观测：30 分钟前成功，早于 1 小时的新鲜度窗口。</summary>
+    private static S8AuthorityObservation Healthy() => Obs(lastSuccessEndAt: Now.AddMinutes(-30));
+
+    // ══════════════════ CASE A / B：放行侧 ══════════════════
+
+    /// <summary>CASE A：Authority 可信 → 恢复正常放行。</summary>
+    [Fact]
+    public void A_Healthy_AllowsRecovery()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(Healthy());
+
+        Assert.Equal(S8AuthorityHealthState.Healthy, r.State);
+        Assert.Equal(S8AuthorityHealthReason.Healthy, r.ReasonCode);
+        Assert.Equal(Now.AddMinutes(-30), r.LastSuccessAt);
+        Assert.Equal(Now.AddMinutes(-30) + Stale, r.StaleAfter);
+        Assert.True(r.AllowsRecovery);
+    }
+
+    /// <summary>
+    /// CASE B（<b>本文件最重要的一条</b>）：Authority 可信但本轮取数为空 → <b>恢复仍然放行</b>。
+    ///
+    /// <para>「空集」是恢复的<b>正常输入</b>而非异常信号 —— 全部业务对象都合法退场时，
+    /// 本来就该看到 0 行。把门禁写成「空集就拦」，恢复将永不发生且没有任何报错。</para>
+    ///
+    /// <para>断言分两层：行为上放行；<b>结构上根本拿不到行数</b> ——
+    /// 判定的入参类型里不存在任何与行数/命中数相关的字段，
+    /// 于是「顺手加个空集判断」在编译期就做不到。这一层比行为断言更耐改。</para>
+    /// </summary>
+    [Fact]
+    public void B_HealthyEmpty_StillAllowsRecovery()
+    {
+        // 行为层：与 CASE A 走的是同一组判据 —— 判定里根本没有「空/非空」这个分支。
+        Assert.True(S8AuthorityHealthEvaluator.Evaluate(Healthy()).AllowsRecovery);
+
+        // 结构层一：判定入参里不得出现任何行数概念。
+        var banned = new[] { "hit", "row", "count", "empty", "any", "size", "length" };
+        foreach (var p in typeof(S8AuthorityObservation).GetProperties())
+        {
+            foreach (var b in banned)
+            {
+                Assert.False(p.Name.Contains(b, StringComparison.OrdinalIgnoreCase),
+                    $"S8AuthorityObservation.{p.Name} 疑似把行数引入了健康判定");
+            }
+        }
+
+        // 结构层二：放行判定是结果对象上的无参属性，没有任何可供传入行数的入口。
+        var allows = typeof(S8AuthorityHealthResult).GetProperty(nameof(S8AuthorityHealthResult.AllowsRecovery));
+        Assert.NotNull(allows);
+        Assert.Empty(allows!.GetMethod!.GetParameters());
+
+        // 结构层三：Evaluate 只有一个入参，就是观测值本身。
+        var evaluate = typeof(S8AuthorityHealthEvaluator).GetMethod(
+            nameof(S8AuthorityHealthEvaluator.Evaluate), BindingFlags.Public | BindingFlags.Static);
+        Assert.NotNull(evaluate);
+        Assert.Single(evaluate!.GetParameters());
+        Assert.Equal(typeof(S8AuthorityObservation), evaluate.GetParameters()[0].ParameterType);
+    }
+
+    // ══════════════════ CASE C–H：拦截侧 ══════════════════
+
+    /// <summary>CASE C：回溯窗口内一次成功都没有 → UNKNOWN。「查不到」不等于「没问题」。</summary>
+    [Fact]
+    public void C_NoSuccessfulRun_IsUnknownAndBlocks()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(Obs(lastSuccessEndAt: null));
+
+        Assert.Equal(S8AuthorityHealthState.Unknown, r.State);
+        Assert.Equal(S8AuthorityHealthReason.NoSuccessfulRun, r.ReasonCode);
+        Assert.Null(r.LastSuccessAt);
+        Assert.Null(r.StaleAfter);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>CASE C'：数据集压根没声明 AuthoritySpec → UNKNOWN，绝不默认健康。</summary>
+    [Fact]
+    public void C2_NoAuthoritySpec_IsUnknownAndBlocks()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(Obs(specFound: false, lastSuccessEndAt: Now.AddMinutes(-1)));
+
+        Assert.Equal(S8AuthorityHealthState.Unknown, r.State);
+        Assert.Equal(S8AuthorityHealthReason.NoAuthoritySpec, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>CASE D：最近一次成功已超过新鲜度窗口 → DEGRADED / STALE。</summary>
+    [Fact]
+    public void D_SuccessTooOld_IsStaleAndBlocks()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(Obs(lastSuccessEndAt: Now.AddMinutes(-90)));
+
+        Assert.Equal(S8AuthorityHealthState.Degraded, r.State);
+        Assert.Equal(S8AuthorityHealthReason.Stale, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>CASE D'：恰好落在窗口边界上仍算健康 —— 阈值是「超过」才降级。</summary>
+    [Fact]
+    public void D2_ExactlyAtStaleBoundary_IsStillHealthy()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(Obs(lastSuccessEndAt: Now - Stale));
+
+        Assert.Equal(S8AuthorityHealthState.Healthy, r.State);
+        Assert.True(r.AllowsRecovery);
+    }
+
+    /// <summary>
+    /// CASE E：成功之后又开了一轮、还没跑完 → IN_FLIGHT。
+    /// 此刻 Authority 可能只写了一半，恢复判定不能在这个瞬间做。
+    /// </summary>
+    [Fact]
+    public void E_NewerRunInFlight_Blocks()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(
+            Obs(lastSuccessEndAt: Now.AddMinutes(-40), inFlightStartAt: Now.AddMinutes(-5)));
+
+        Assert.Equal(S8AuthorityHealthState.Degraded, r.State);
+        Assert.Equal(S8AuthorityHealthReason.InFlight, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>
+    /// CASE E'：在途运行的年龄已超过新鲜度窗口 → STUCK_RUN，与 IN_FLIGHT 分开报。
+    ///
+    /// <para>运行日志里的 RUNNING 行没有任何应用侧回收机制（失败路径会吞掉自己的写入异常，
+    /// 注释里写的是「遗留 RUNNING 行可由运维手动清理」，而并不存在这样的清理作业）。
+    /// 实测该表有 298 条永不终态的 RUNNING 行，S1 占 42 条、最老已 570 小时。
+    /// 若与「正在正常跑」混为一谈，运维分不清该等还是该查 —— 两者都拦，但排障动作不同。</para>
+    /// </summary>
+    [Fact]
+    public void E2_StuckRun_IsReportedSeparately()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(
+            Obs(lastSuccessEndAt: Now.AddHours(-5), inFlightStartAt: Now.AddHours(-4)));
+
+        Assert.Equal(S8AuthorityHealthState.Degraded, r.State);
+        Assert.Equal(S8AuthorityHealthReason.StuckRun, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>CASE F：成功之后的那一轮失败了 → LATEST_FAILED。</summary>
+    [Fact]
+    public void F_NewerRunFailed_Blocks()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(
+            Obs(lastSuccessEndAt: Now.AddMinutes(-40), newerFailure: true, newerFailureStatus: "FAILED"));
+
+        Assert.Equal(S8AuthorityHealthState.Degraded, r.State);
+        Assert.Equal(S8AuthorityHealthReason.LatestFailed, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>
+    /// CASE F'：失败发生在最近一次成功<b>之前</b> → 不影响判定。
+    /// 历史失败是常态；若一律降级，门禁会永久拦死。
+    /// （观测层已按「锚点之后」过滤，此处等价于 newerFailure = false。）
+    /// </summary>
+    [Fact]
+    public void F2_OlderFailure_DoesNotDegrade()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(
+            Obs(lastSuccessEndAt: Now.AddMinutes(-20), newerFailure: false));
+
+        Assert.Equal(S8AuthorityHealthState.Healthy, r.State);
+        Assert.True(r.AllowsRecovery);
+    }
+
+    /// <summary>CASE G：成功之后出现终态 PARTIAL → LATEST_PARTIAL。部分成功的快照不足以支撑恢复判定。</summary>
+    [Fact]
+    public void G_PartialRun_Blocks()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(
+            Obs(lastSuccessEndAt: Now.AddMinutes(-40), newerFailure: true, newerFailureStatus: "PARTIAL"));
+
+        Assert.Equal(S8AuthorityHealthState.Degraded, r.State);
+        Assert.Equal(S8AuthorityHealthReason.LatestPartial, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>
+    /// CASE G'：未知的终态状态一律按失败处置，不得被当成健康放行。
+    /// <para>该表的 status 值域实测还包含 <c>SUCCESS_WITH_WARNING</c> 与 <c>CANCELED</c>
+    /// （由 S0 维度刷新写出，当前不出现在 S1 上）。采集层用的是
+    /// <c>status &lt;&gt; 'SUCCESS'</c> 而非白名单，正是为了让将来新增的终态状态
+    /// 默认落到「失败」而不是「健康」这一侧。</para>
+    /// </summary>
+    [Fact]
+    public void G2_UnknownTerminalStatus_IsTreatedAsFailure()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(
+            Obs(lastSuccessEndAt: Now.AddMinutes(-40), newerFailure: true, newerFailureStatus: "SUCCESS_WITH_WARNING"));
+
+        Assert.Equal(S8AuthorityHealthState.Degraded, r.State);
+        Assert.Equal(S8AuthorityHealthReason.LatestFailed, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>CASE G''：快照不变量被破坏 → 即使生产运行成功也降级。</summary>
+    [Fact]
+    public void G3_SnapshotInvariantViolated_Blocks()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(Obs(
+            lastSuccessEndAt: Now.AddMinutes(-10),
+            snapshot: S8SnapshotInvariant.Violated,
+            snapshotDetail: "同时存在 2 个当前批次"));
+
+        Assert.Equal(S8AuthorityHealthState.Degraded, r.State);
+        Assert.Equal(S8AuthorityHealthReason.SnapshotInvariantFailed, r.ReasonCode);
+        Assert.Contains("2 个当前批次", r.Reason);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>
+    /// CASE G'''：快照发布无法确认 → UNKNOWN，<b>不猜</b>。
+    ///
+    /// <para>这是 Rule03 今天的真实处境：生产侧的发布语句在本轮零行时命中 0 行、
+    /// 在库里不留任何批次标记，于是「源侧本轮确实没有数据」与「发布压根没跑成」字面等同。
+    /// 猜成健康的代价是把真实缺料预警整体判成已恢复，因此保守拦截。
+    /// 消除它需要生产侧补一个发布标记 —— 那是下一批的工作，不在本轮范围。</para>
+    /// </summary>
+    [Fact]
+    public void G4_SnapshotPublicationUnconfirmed_IsUnknownAndBlocks()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(Obs(
+            lastSuccessEndAt: Now.AddMinutes(-10),
+            snapshot: S8SnapshotInvariant.Unconfirmed,
+            snapshotDetail: "当前批次数为 0"));
+
+        Assert.Equal(S8AuthorityHealthState.Unknown, r.State);
+        Assert.Equal(S8AuthorityHealthReason.SnapshotPublicationUnconfirmed, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    /// <summary>
+    /// CASE H：Authority 有多个写入方且部分不留运行日志 → UNKNOWN，
+    /// 且必须<b>压过一切健康信号</b>。
+    ///
+    /// <para>本用例刻意喂进一次「一分钟前刚成功、快照不变量也成立」的完美观测，
+    /// 它不得把结论洗白 —— 因为问题不在于生产跑没跑成，
+    /// 而在于运行日志根本证明不了表里最终的数据出自谁。</para>
+    /// </summary>
+    [Fact]
+    public void H_MultiWriterAuthority_IsUnknownRegardlessOfRunHealth()
+    {
+        var r = S8AuthorityHealthEvaluator.Evaluate(Obs(
+            producerTrusted: false,
+            lastSuccessEndAt: Now.AddMinutes(-1),
+            snapshot: S8SnapshotInvariant.Satisfied));
+
+        Assert.Equal(S8AuthorityHealthState.Unknown, r.State);
+        Assert.Equal(S8AuthorityHealthReason.Rule01MultiWriterUntrusted, r.ReasonCode);
+        Assert.False(r.AllowsRecovery);
+    }
+
+    // ══════════════════ 穷尽性与词表封闭 ══════════════════
+
+    public static IEnumerable<object[]> AllNonHealthyReasonCodes() =>
+        typeof(S8AuthorityHealthReason)
+            .GetFields(BindingFlags.Public | BindingFlags.Static | BindingFlags.FlattenHierarchy)
+            .Where(f => f.IsLiteral && !f.IsInitOnly && f.FieldType == typeof(string))
+            .Select(f => (string)f.GetRawConstantValue()!)
+            .Where(v => v != S8AuthorityHealthReason.Healthy)
+            .Select(v => new object[] { v });
+
+    /// <summary>
+    /// 放行只有一个出口：<c>State == HEALTHY</c>。
+    /// <para>用反射枚举<b>全部</b>原因码 —— 将来有人新增一个却忘了想清楚放不放行时，
+    /// 这条会自动变红。默认必须是拦。</para>
+    /// </summary>
+    [Theory]
+    [MemberData(nameof(AllNonHealthyReasonCodes))]
+    public void EveryNonHealthyReason_Blocks(string reasonCode)
+    {
+        foreach (var state in new[] { S8AuthorityHealthState.Degraded, S8AuthorityHealthState.Unknown })
+        {
+            var r = new S8AuthorityHealthResult { State = state, ReasonCode = reasonCode };
+            Assert.False(r.AllowsRecovery, $"state={state} reason={reasonCode} 不应放行");
+        }
+    }
+
+    /// <summary>状态是封闭三元集合，不得出现第四种（尤其不得出现 HEALTHY_EMPTY）。</summary>
+    [Fact]
+    public void StateVocabularyIsClosed_AndHasNoHealthyEmpty()
+    {
+        var states = typeof(S8AuthorityHealthState)
+            .GetFields(BindingFlags.Public | BindingFlags.Static | BindingFlags.FlattenHierarchy)
+            .Where(f => f.IsLiteral && f.FieldType == typeof(string))
+            .Select(f => (string)f.GetRawConstantValue()!)
+            .OrderBy(v => v, StringComparer.Ordinal)
+            .ToArray();
+
+        Assert.Equal(new[] { "DEGRADED", "HEALTHY", "UNKNOWN" }, states);
+    }
+
+    /// <summary>端口是接口且实现可被自动注册，调度器才能换实现而不改自身。</summary>
+    [Fact]
+    public void Resolver_IsASeamInterface()
+    {
+        Assert.True(typeof(IS8AuthorityHealthResolver).IsInterface);
+        Assert.True(typeof(IS8AuthorityHealthResolver)
+            .IsAssignableFrom(typeof(S8MdpAuthorityHealthResolver)));
+    }
+}

server/Plugins/Admin.NET.Plugin.AiDOP.Tests/S8/S8RecoveryGateWiringTests.cs

@@ -0,0 +1,304 @@
+using System.Reflection;
+using Admin.NET.Plugin.AiDOP.Service.S8;
+using Admin.NET.Plugin.AiDOP.Service.S8.Rules.Health;
+using Xunit;
+
+namespace Admin.NET.Plugin.AiDOP.Tests.S8;
+
+/// <summary>
+/// S8-AUTHORITY-HEALTH-GATE：门禁在调度器里的<b>接线位置与爆炸半径</b>。
+///
+/// <para><b>为什么是源码扫描而不是行为测试</b>：<see cref="S8WatchSchedulerService"/>
+/// 的构造参数里有 8 个 <c>SqlSugarRepository&lt;T&gt;</c>，其无参构造读取
+/// <c>SqlSugarSetup.ITenant</c> 静态字段，没有 Furion 启动就是 NRE ——
+/// 本仓从未、也无法在测试里实例化调度器（<c>grep "new S8WatchSchedulerService"</c> 全仓 0 命中）。
+/// 既有的 <c>S8SchedulerScopeGuardTests</c> / <c>S8SystemRecoveredTimelineTests</c>
+/// 走的都是同一条路。</para>
+///
+/// <para><b>判定逻辑不在这里</b>，在 <see cref="S8AuthorityHealthGateTests"/>（纯函数、可断言）。
+/// 本文件只回答一个问题：门禁挡住的到底是<b>哪一段</b>。
+/// 挡多了，正面观测停摆；挡少了，中台停更时照样假恢复。</para>
+/// </summary>
+public class S8RecoveryGateWiringTests
+{
+    private const string SchedulerSrc = "Service/S8/S8WatchSchedulerService.cs";
+
+    // ───────────────── helpers ─────────────────
+
+    /// <summary>读取源码并剥掉注释行 —— 断言不应被注释里的字面量满足或破坏。</summary>
+    private static string CodeOnly(string relative)
+    {
+        var dir = new DirectoryInfo(AppContext.BaseDirectory);
+        while (dir != null && !Directory.Exists(Path.Combine(dir.FullName, "Admin.NET.Plugin.AiDOP")))
+            dir = dir.Parent;
+        Assert.NotNull(dir);
+
+        var full = Path.Combine(dir!.FullName, "Admin.NET.Plugin.AiDOP",
+            relative.Replace('/', Path.DirectorySeparatorChar));
+        Assert.True(File.Exists(full), $"源码文件不存在，路径需同步更新：{full}");
+
+        return string.Join('\n', File.ReadAllLines(full)
+            .Where(l =>
+            {
+                var t = l.TrimStart();
+                return !t.StartsWith("///", StringComparison.Ordinal)
+                       && !t.StartsWith("//", StringComparison.Ordinal);
+            }));
+    }
+
+    private static string Scheduler() => CodeOnly(SchedulerSrc);
+
+    private static int At(string code, string anchor)
+    {
+        var i = code.IndexOf(anchor, StringComparison.Ordinal);
+        Assert.True(i >= 0, $"未找到锚点：{anchor}");
+        return i;
+    }
+
+    private static string Slice(string code, string from, string to)
+    {
+        var a = At(code, from);
+        var b = code.IndexOf(to, a, StringComparison.Ordinal);
+        return b > a ? code[a..b] : code[a..];
+    }
+
+    private static string ProcessBody() => Slice(Scheduler(),
+        "private async Task<List<S8WatchCreationResult>> ProcessSingleRuleAsync",
+        "public async Task OnRuleCompletedAsync");
+
+    private static string ReconcileBody() => Slice(Scheduler(),
+        "private async Task<List<long>> ReconcileRecoveriesForRuleAsync",
+        "private static AdoS8DetectionLog BuildHitLog");
+
+    private static string RunSingleBody() => Slice(Scheduler(),
+        "public async Task<S8RuleRunResult> RunSingleRuleAsync",
+        "private async Task<List<S8WatchCreationResult>> ProcessSingleRuleAsync");
+
+    /// <summary>
+    /// 计算 <paramref name="index"/> 处的花括号嵌套深度。
+    /// <para>只在<b>单个方法体切片</b>内使用：日志模板里的 <c>{RuleCode}</c> 之类占位符成对出现、
+    /// 净增为 0，不影响相对深度比较。若将来该方法引入 raw string SQL，本助手需先剥离字符串字面量。</para>
+    /// </summary>
+    private static int DepthAt(string code, int index)
+    {
+        var d = 0;
+        for (var i = 0; i < index; i++)
+        {
+            if (code[i] == '{') d++;
+            else if (code[i] == '}') d--;
+        }
+        return d;
+    }
+
+    private static int Count(string code, string needle)
+    {
+        int n = 0, i = 0;
+        while ((i = code.IndexOf(needle, i, StringComparison.Ordinal)) >= 0) { n++; i += needle.Length; }
+        return n;
+    }
+
+    // ───────────────── 接线基本形态 ─────────────────
+
+    /// <summary>调度器必须依赖端口接口，且不得自己知道运行日志表名。</summary>
+    [Fact]
+    public void Scheduler_DependsOnTheResolverSeamInterface()
+    {
+        var ctor = typeof(S8WatchSchedulerService).GetConstructors().Single();
+        Assert.Contains(ctor.GetParameters(), p => p.ParameterType == typeof(IS8AuthorityHealthResolver));
+
+        var code = Scheduler();
+        Assert.DoesNotContain("new S8MdpAuthorityHealthResolver", code);
+
+        // 物理表名属适配层的知识；调度器只传 DatasetCode。
+        Assert.DoesNotContain("mdp_transform_run_log", code);
+        Assert.DoesNotContain("dwd_requirement_examine_detail", code);
+        Assert.DoesNotContain("mdp_std_so", code);
+    }
+
+    /// <summary>门禁必须在 reconcile <b>之前</b>，且位于 ProcessSingleRuleAsync 内。</summary>
+    [Fact]
+    public void Gate_RunsBeforeRecoveryReconcile()
+    {
+        var body = ProcessBody();
+        Assert.True(At(body, "_authorityHealthResolver.ResolveAsync") < At(body, "health.AllowsRecovery"));
+        Assert.True(At(body, "health.AllowsRecovery") < At(body, "ReconcileRecoveriesForRuleAsync("));
+    }
+
+    // ───────────────── CASE B（结构侧）：空集不得成为拦截理由 ─────────────────
+
+    /// <summary>
+    /// CASE B：门禁与「本轮有没有命中」<b>完全无关</b>。
+    /// <para>取数与门禁之间不得出现任何对结果规模的判断 ——
+    /// 一旦有人补上 <c>if (hits.Count == 0)</c>，真实恢复会永远不再发生，
+    /// 而日志上每一轮都是成功。</para>
+    /// </summary>
+    [Fact]
+    public void B_GateIsNotConditionedOnHitCount()
+    {
+        var body = ProcessBody();
+        var window = body[At(body, "await evaluator.EvaluateAsync")..At(body, "ReconcileRecoveriesForRuleAsync(")];
+
+        Assert.DoesNotContain("hits.Count ==", window);
+        Assert.DoesNotContain("hits.Count >", window);
+        Assert.DoesNotContain("hits.Any(", window);
+        Assert.DoesNotContain("!hits.Any", window);
+
+        // reconcile 自身也不得新增空集早退 —— 空 hits 正是「全部恢复」的正常输入。
+        var reconcile = ReconcileBody();
+        Assert.DoesNotContain("hits.Count == 0", reconcile);
+        Assert.DoesNotContain("!hits.Any()", reconcile);
+    }
+
+    // ───────────────── CASE I：抗抖计数不得被拦截路径触碰 ─────────────────
+
+    /// <summary>
+    /// CASE I：门禁拦截时 <c>ConsecutiveMissCount</c> <b>既不加也不减</b>。
+    ///
+    /// <para>实现方式必须是「整段跳过 reconcile」，而<b>不是</b>在 reconcile 内部某处 return ——
+    /// 后者会让候选查询与 miss 自增的相对顺序变成定时炸弹：今天写在自增之前，
+    /// 明天挪到之后，就变成「拦截期间照样累计 miss，中台一恢复立刻集体假恢复」。</para>
+    /// </summary>
+    [Fact]
+    public void I_BlockedPath_LeavesConsecutiveMissCountUntouched()
+    {
+        var code = Scheduler();
+        var reconcile = ReconcileBody();
+
+        const string inc = "ConsecutiveMissCount = x.ConsecutiveMissCount + 1";
+        Assert.Equal(2, Count(code, inc));
+        Assert.Equal(2, Count(reconcile, inc));   // 两处自增必须全部落在 reconcile 内
+
+        // 门禁绝不能被搬进 reconcile —— 那样自增与拦截的先后无法再被静态保证。
+        Assert.DoesNotContain("_authorityHealthResolver", reconcile);
+        Assert.DoesNotContain("AllowsRecovery", reconcile);
+        Assert.DoesNotContain("S8AuthorityHealth", reconcile);
+
+        // 拦截分支自身不碰任何抗抖计数与恢复标记。
+        var body = ProcessBody();
+        var blocked = body[At(body, "if (!health.AllowsRecovery)")..At(body, "ReconcileRecoveriesForRuleAsync(")];
+        Assert.DoesNotContain("ConsecutiveMissCount", blocked);
+        Assert.DoesNotContain("ConsecutiveHitCount", blocked);
+        Assert.DoesNotContain("RecoveredAt", blocked);
+    }
+
+    // ───────────────── CASE J：正面观测不受影响 ─────────────────
+
+    /// <summary>
+    /// CASE J：门禁拦住的<b>只有恢复</b>，建单与刷新照常。
+    ///
+    /// <para>用花括号深度证明 <c>foreach (var hit in hits)</c> 与方法首行同级 ——
+    /// 它没有被裹进门禁的 if/else 里。这比 <c>Assert.Contains</c> 强：
+    /// 把 foreach 缩进到 else 分支内，字符串照样能匹配到，深度不会。</para>
+    ///
+    /// <para>正面观测停摆是比假恢复更早被发现、但同样严重的失效：
+    /// 中台停更期间新异常一条也建不出来，而页面上一片安静。</para>
+    /// </summary>
+    [Fact]
+    public void J_BlockedPath_StillRunsCreateAndRefresh()
+    {
+        var body = ProcessBody();
+
+        var baseline = At(body, "var results = new List<S8WatchCreationResult>();");
+        var gate = At(body, "if (!health.AllowsRecovery)");
+        var loop = At(body, "foreach (var hit in hits)");
+
+        Assert.True(loop > gate, "hit 循环必须在门禁之后");
+        Assert.Equal(DepthAt(body, baseline), DepthAt(body, loop));
+    }
+
+    // ───────────────── CASE K：不判失败、不触发自动暂停 ─────────────────
+
+    /// <summary>
+    /// CASE K：门禁拦截<b>不是一次失败的 run</b>。
+    ///
+    /// <para>若拦截走 <c>Success = false</c>，规则完成回调会累计 <c>ConsecutiveFailureCount</c>，
+    /// 达到阈值即写 <c>paused_until</c> 把规则自动暂停一小时。
+    /// 于是中台跑一次批，监控就睡一小时 —— 而页面上只显示「最近结果：失败」，
+    /// 没人知道是数据源的问题。</para>
+    /// </summary>
+    [Fact]
+    public void K_BlockedPath_DoesNotFailTheRunNorAutoPause()
+    {
+        var body = ProcessBody();
+        var blocked = body[At(body, "if (!health.AllowsRecovery)")..At(body, "foreach (var hit in hits)")];
+
+        Assert.DoesNotContain("throw", blocked);
+        Assert.DoesNotContain("Success = false", blocked);
+        Assert.DoesNotContain("S8RuleRunResult", blocked);
+        Assert.DoesNotContain("Enabled = false", blocked);
+        Assert.DoesNotContain("PausedUntil", blocked);
+
+        // 门禁不得上移到 RunSingleRuleAsync —— 那一层的失败语义就是整条 run 失败。
+        Assert.DoesNotContain("_authorityHealthResolver", RunSingleBody());
+    }
+
+    /// <summary>
+    /// CASE K'：探测自身失败必须<b>降级为 UNKNOWN</b>，不得让异常穿到上层变成整条 run 失败 ——
+    /// 那等于绕过本条守卫达成同一个后果。该兜底落在适配实现里，此处断言它确实存在。
+    /// </summary>
+    [Fact]
+    public void K2_ResolverFailure_DegradesToUnknown_NotRunFailure()
+    {
+        var resolver = CodeOnly("Service/S8/Rules/Health/S8MdpAuthorityHealthResolver.cs");
+        var probe = resolver[At(resolver, "catch (Exception ex)")..];
+
+        Assert.Contains("S8AuthorityHealthState.Unknown", probe);
+        Assert.Contains("S8AuthorityHealthReason.ResolverFailed", probe);
+        Assert.DoesNotContain("throw;", probe);
+    }
+
+    // ───────────────── 观测性：拦截必须留痕 ─────────────────
+
+    /// <summary>
+    /// 拦截期间异常只增不减，这件事<b>必须能被查到</b>。
+    /// <para>只写日志不够 —— 日志会滚走，而 detection_log 是可查的。
+    /// 原因码要落 <c>failure_reason</c>，否则只知道「被拦了」、不知道为什么。</para>
+    /// </summary>
+    [Fact]
+    public void BlockedPath_WritesRecoveryBlockedDetectionLog()
+    {
+        var body = ProcessBody();
+        var blocked = body[At(body, "if (!health.AllowsRecovery)")..At(body, "foreach (var hit in hits)")];
+
+        Assert.Contains("WriteDetectionLogAsync", blocked);
+        Assert.Contains("DetectResult = DetectResultRecoveryBlocked", blocked);
+        Assert.Contains("FailureReason = health.ReasonCode", blocked);
+        Assert.Contains("PayloadSnapshot", blocked);
+
+        // detect_result 列是 varchar(32)，常量不得超长。
+        var f = typeof(S8WatchSchedulerService).GetField("DetectResultRecoveryBlocked",
+            BindingFlags.NonPublic | BindingFlags.Static | BindingFlags.Public);
+        Assert.NotNull(f);
+        var v = (string)f!.GetRawConstantValue()!;
+        Assert.Equal("RECOVERY_BLOCKED", v);
+        Assert.True(v.Length <= 32);
+    }
+
+    // ───────────────── 适配层的租户隔离 ─────────────────
+
+    /// <summary>
+    /// 适配层的租户谓词必须是严格相等。
+    ///
+    /// <para>仓内 <c>MdpMonitorService.BuildMdpRunLogTenantWhere</c> 用的是
+    /// <c>(tenant_id = @TenantId OR tenant_id = 0)</c> —— <b>本类不得复用</b>：
+    /// 该表确实存在 <c>tenant_id = 0</c> 的平台行（实测 S1 有 111 条，
+    /// 且 <c>S5_PURCHASE_RECEIPT_MDP_SYNC</c> 有 325 条平台级 SUCCESS），
+    /// 一条这样的行会让所有租户同时被判成健康。</para>
+    /// </summary>
+    [Fact]
+    public void Resolver_UsesStrictTenantEquality()
+    {
+        var resolver = CodeOnly("Service/S8/Rules/Health/S8MdpAuthorityHealthResolver.cs");
+
+        Assert.DoesNotContain("OR tenant_id = 0", resolver);
+        Assert.DoesNotContain("BuildMdpRunLogTenantWhere", resolver);
+        Assert.Equal(3, Count(resolver, "r.tenant_id = @TenantId"));   // 三条运行日志查询各一次
+
+        // 必须显式过滤终态，否则会取到永不回收的孤儿 RUNNING 行。
+        Assert.Contains("r.status = 'SUCCESS'", resolver);
+        Assert.Contains("r.end_time IS NOT NULL", resolver);
+
+        // 回溯窗口不是可选优化：无界回溯在长期停摆的租户上实测单次 146ms。
+        Assert.Contains("r.start_time >= @LookbackFrom", resolver);
+    }
+}

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/DataAccess/Providers/S8BusinessDatasetDefinitions.cs

@@ -57,6 +57,21 @@ public sealed class S8BusinessDatasetDefinitions : IS8DatasetDefinitionSource
             Kind = S8DatasetKind.Object,
             Status = S8DatasetStatus.Defined,
             Capabilities = PurchaseDeliveryCapabilities,
+
+            // Authority 的写入方不止 MDP 转换：DeliveryScheduleService 与 ProcessOutsourceOrderService
+            // 两条 API 路径也直接写 dwd_supplier_delivery，且都不写 mdp_transform_run_log。
+            // 于是运行日志无法证明表里最终的数据出自谁 —— 判 UNKNOWN、拦截恢复判定。
+            //
+            // 这只冻结「从缺席做负面推断」这一件事：建单与刷新照常，
+            // 因为 hit 是对某行存在的正面观测，写入方混杂不会凭空造出一行。
+            // 解冻条件：Authority 收敛为单一写入方、补上 membership 淘汰、补齐租户 JOIN 谓词。
+            AuthoritySpec = new Health.S8AuthoritySpec
+            {
+                ProducerJobCode = "S3_MDP_SYNC_TRANSFORM",
+                AuthorityKind = Health.S8AuthorityKind.CurrentStateMirror,
+                ProducerTrusted = false,
+                UntrustedReasonCode = Health.S8AuthorityHealthReason.Rule01MultiWriterUntrusted
+            },
             // S8-RULE01-PURCHASE-DELIVERY-CLOSURE-1：due_at 的业务语义已完成血缘取证，
             // 从「UAT_EXECUTION_MAPPING / 待生产确认」升级为「与中台自身逾期判据同源」。
             //

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/DataAccess/Providers/S8MaterialShortageDatasetDefinitions.cs

@@ -122,6 +122,25 @@ public sealed class S8MaterialShortageDatasetDefinitions : IS8DatasetDefinitionS
             Kind = S8DatasetKind.Object,
             Status = S8DatasetStatus.Defined,
             Capabilities = MaterialShortageCapabilities,
+
+            // dwd_requirement_examine_detail 是「按批次累积 + 末尾原子翻牌发布」的当前快照表，
+            // 因此除生产运行成功外，还必须校验发布不变量：
+            // 生产运行成功 ≠ 快照发布成功 —— 发布语句在两条 DWD INSERT 之后才执行，
+            // 而 run 直到最末尾才标记 SUCCESS，中间任何一步没落地都会留下一个「成功但没有当前快照」的库态。
+            //
+            // 必填列只列 material_role：它由带 ELSE 兜底的 CASE 产出，结构上不可能为空。
+            // 刻意**不列** bom_level —— 转换侧对非数字层级有意留 NULL（"脏值留 NULL，不猜层级"），
+            // 把它当必填等于按设计制造假 DEGRADED。
+            AuthoritySpec = new Health.S8AuthoritySpec
+            {
+                ProducerJobCode = "S1_MDP_SYNC_TRANSFORM",
+                AuthorityKind = Health.S8AuthorityKind.PublishedSnapshot,
+                SnapshotTable = "dwd_requirement_examine_detail",
+                SnapshotBatchColumn = "calc_batch_id",
+                SnapshotCurrentFlagColumn = "is_current_flag",
+                SnapshotRequiredColumns = new[] { "material_role" }
+            },
+
             Description =
                 "数据 Authority = 数据中台 dwd_requirement_examine_detail（单表，无 JOIN，不直读任何业务源表）；" +
                 "过滤 is_current_flag = 1（每租户+工厂恰好一个当前批次；DWD 按 calc_batch_id 累积历史批次，" +

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/DataAccess/Providers/S8OrderDeliveryDatasetDefinitions.cs

@@ -57,6 +57,20 @@ public sealed class S8OrderDeliveryDatasetDefinitions : IS8DatasetDefinitionSour
             Kind = S8DatasetKind.Object,
             Status = S8DatasetStatus.Defined,
             Capabilities = OrderDeliveryRiskCapabilities,
+
+            // mdp_std_so 是当前态镜像：upsert 覆盖、没有「发布/翻牌」这一步，
+            // 读者（本数据集的 Provider）也从不按批次取数 ——
+            // 表里因此不存在一个「与本轮是否完成相关、且可能被破坏」的表级状态，
+            // 健康度完全由生产运行本身决定，不加表级不变量。
+            // （唯一的候选「每租户单一 sync_batch_id」既非必要也非充分：
+            //  它今天恒为真只是孤儿行被反复盖上当前批次戳的副产物，
+            //  而最新批次过滤生效后它会合法地变成 false。）
+            AuthoritySpec = new Health.S8AuthoritySpec
+            {
+                ProducerJobCode = "S1_MDP_SYNC_TRANSFORM",
+                AuthorityKind = Health.S8AuthorityKind.CurrentStateMirror
+            },
+
             Description =
                 "来源 crm_seorderentry（订单行）左连 crm_seorder（主单）；粒度为订单行" +
                 "（bill_no + entry_seq，实测 390/390 唯一）；" +

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/DataAccess/S8DatasetDefinition.cs

@@ -27,6 +27,15 @@ public sealed class S8DatasetDefinition
     /// <summary>说明（可选）。</summary>
     public string? Description { get; init; }
 
+    /// <summary>
+    /// Authority 声明：健康判定需要知道的一切（生产者作业、新鲜度窗口、快照不变量）。
+    ///
+    /// <para>为 null 表示该数据集尚未声明 Authority ——
+    /// <see cref="Health.IS8AuthorityHealthResolver"/> 会判 UNKNOWN 并拦截恢复判定，
+    /// 但<b>不影响</b>取数、命中评估与建单/刷新。保守方向：宁可不恢复，不可误恢复。</para>
+    /// </summary>
+    public Health.S8AuthoritySpec? AuthoritySpec { get; init; }
+
     /// <summary>定义本身是否处于可用状态（不含 Provider 判断）。</summary>
     public bool IsActive => string.Equals(Status, S8DatasetStatus.Defined, StringComparison.OrdinalIgnoreCase);
 }

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/Health/IS8AuthorityHealthResolver.cs

@@ -0,0 +1,22 @@
+namespace Admin.NET.Plugin.AiDOP.Service.S8.Rules.Health;
+
+/// <summary>
+/// Authority 健康判定端口。
+///
+/// <para><b>为什么是独立端口而不是塞进 Provider</b>：
+/// <c>IS8MonitoringDataProvider</c> 的接口注释已明确写明新鲜度「属数据平台侧，
+/// 待其提供 freshness 接口后由本层消费，不在 S8 内造替代实现」。
+/// 做成端口 + 临时的 MDP 适配实现，等中台真正提供数据集级 freshness 时只换实现类，
+/// 调度器、取数网关与三个 Provider 一行都不用动。</para>
+///
+/// <para><b>为什么不放在调度器里内联</b>：内联会让调度器重新知道物理表名与作业编码 ——
+/// 那正是取数网关硬切换时刚清理掉的耦合，且必然长成 per-rule 的分支。</para>
+/// </summary>
+public interface IS8AuthorityHealthResolver
+{
+    /// <summary>
+    /// 判定某租户下某数据集的 Authority 当前是否可信。
+    /// <para>实现必须 fail-safe：自身异常一律返回 UNKNOWN（拦截恢复），绝不返回 HEALTHY。</para>
+    /// </summary>
+    Task<S8AuthorityHealthResult> ResolveAsync(long tenantId, string datasetCode);
+}

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/Health/S8AuthorityHealth.cs

@@ -0,0 +1,96 @@
+namespace Admin.NET.Plugin.AiDOP.Service.S8.Rules.Health;
+
+/// <summary>
+/// Authority 健康状态。
+///
+/// <para><b>只有三个状态，且 HEALTHY_EMPTY 不在其中</b>：
+/// 「Authority 可信但本轮确实一行都没有」是 <c>HEALTHY</c> 与 <c>rows.Count == 0</c> 的
+/// 组合观测，不是一个独立状态。把它设成状态，就必须在状态机里读行数 ——
+/// 而本门禁存在的全部意义，正是让恢复判定<b>与行数解耦</b>。</para>
+/// </summary>
+public static class S8AuthorityHealthState
+{
+    /// <summary>本轮取数可信：生产者最近一次成功、够新、无更新的失败/在途运行、快照不变量成立。与返回了几行无关。</summary>
+    public const string Healthy = "HEALTHY";
+
+    /// <summary>已知不可信：生产失败 / 在途 / 超期 / 快照不变量被破坏。运维动作 = 去修管线。</summary>
+    public const string Degraded = "DEGRADED";
+
+    /// <summary>无法建立可信性：从未成功、无 AuthoritySpec、发布无法确认、探测自身失败。运维动作 = 去查这条链是否跑起来过。</summary>
+    public const string Unknown = "UNKNOWN";
+}
+
+/// <summary>健康判定的原因码。回执与 detection_log 用它解释「为什么本轮没做恢复」。</summary>
+public static class S8AuthorityHealthReason
+{
+    public const string Healthy = "HEALTHY";
+
+    /// <summary>该数据集没有声明 AuthoritySpec —— 无从判断，保守拦截。</summary>
+    public const string NoAuthoritySpec = "NO_AUTHORITY_SPEC";
+
+    /// <summary>回溯窗口内没有任何一次成功的生产运行。含「从未同步」与「停摆超过回溯窗口」两种。</summary>
+    public const string NoSuccessfulRun = "NO_SUCCESSFUL_RUN";
+
+    /// <summary>最近一次成功距今超过 StaleWindow。</summary>
+    public const string Stale = "STALE";
+
+    /// <summary>存在比最近一次成功更新、且尚未终态的生产运行，年龄仍在 StaleWindow 内（正常跑批中）。</summary>
+    public const string InFlight = "IN_FLIGHT";
+
+    /// <summary>同上，但年龄已超过 StaleWindow —— 大概率是永不回收的孤儿 RUNNING 行。</summary>
+    public const string StuckRun = "STUCK_RUN";
+
+    /// <summary>最近一次成功之后出现了终态失败。</summary>
+    public const string LatestFailed = "LATEST_FAILED";
+
+    /// <summary>最近一次成功之后出现了终态 PARTIAL。</summary>
+    public const string LatestPartial = "LATEST_PARTIAL";
+
+    /// <summary>Authority 的快照不变量被破坏（如同时存在多个当前批次、当前快照里有必填列为空）。</summary>
+    public const string SnapshotInvariantFailed = "SNAPSHOT_INVARIANT_FAILED";
+
+    /// <summary>
+    /// 快照型 Authority 无法确认「本轮发布是否成功」。
+    /// <para>这是 Rule03 的现状：合法空快照与发布失败在库里字面等同（都表现为 0 个当前批次），
+    /// 生产侧不留任何 batch marker。保守判 UNKNOWN，不猜。</para>
+    /// </summary>
+    public const string SnapshotPublicationUnconfirmed = "SNAPSHOT_PUBLICATION_UNCONFIRMED";
+
+    /// <summary>Authority 存在多个写入方且部分不留运行日志，run-log 无法证明最终 provenance（Rule01）。</summary>
+    public const string Rule01MultiWriterUntrusted = "RULE01_MULTI_WRITER_UNTRUSTED";
+
+    /// <summary>健康探测自身抛错。fail-safe：判 UNKNOWN、拦截恢复，绝不当成 HEALTHY。</summary>
+    public const string ResolverFailed = "RESOLVER_FAILED";
+}
+
+/// <summary>健康判定结果。<b>刻意不是 bool</b> —— 回执必须能回答「为什么」。</summary>
+public sealed class S8AuthorityHealthResult
+{
+    public string DatasetCode { get; init; } = string.Empty;
+
+    public long TenantId { get; init; }
+
+    /// <summary>见 <see cref="S8AuthorityHealthState"/>。</summary>
+    public string State { get; init; } = S8AuthorityHealthState.Unknown;
+
+    /// <summary>见 <see cref="S8AuthorityHealthReason"/>。</summary>
+    public string ReasonCode { get; init; } = S8AuthorityHealthReason.NoAuthoritySpec;
+
+    /// <summary>人类可读说明，带上判定用到的实际数值。</summary>
+    public string Reason { get; init; } = string.Empty;
+
+    public DateTime ObservedAt { get; init; }
+
+    /// <summary>最近一次成功生产运行的结束时刻；无则为 null。</summary>
+    public DateTime? LastSuccessAt { get; init; }
+
+    /// <summary>超过该时刻即判 STALE（= LastSuccessAt + StaleWindow）；LastSuccessAt 为 null 时为 null。</summary>
+    public DateTime? StaleAfter { get; init; }
+
+    /// <summary>
+    /// 是否允许执行恢复判定。<b>只有 HEALTHY 放行</b>。
+    /// <para>注意这里没有、也不允许有任何 rows.Count 分支：
+    /// 「Authority 可信 + 本轮 0 行」必须放行恢复，那是合法的业务全退场。</para>
+    /// </summary>
+    public bool AllowsRecovery => string.Equals(State, S8AuthorityHealthState.Healthy, StringComparison.Ordinal);
+}

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/Health/S8AuthorityHealthEvaluator.cs

@@ -0,0 +1,148 @@
+namespace Admin.NET.Plugin.AiDOP.Service.S8.Rules.Health;
+
+/// <summary>快照发布不变量的观测结果。</summary>
+public static class S8SnapshotInvariant
+{
+    /// <summary>该 Authority 是当前态镜像，无发布步骤 ⇒ 不适用。</summary>
+    public const string NotApplicable = "NOT_APPLICABLE";
+
+    /// <summary>恰好一个当前批次，且等于最近一次成功运行的批次；当前快照内无必填列为空。</summary>
+    public const string Satisfied = "SATISFIED";
+
+    /// <summary>零个当前批次 —— 合法空快照与发布失败在库里字面等同，无法区分。</summary>
+    public const string Unconfirmed = "UNCONFIRMED";
+
+    /// <summary>多个当前批次，或当前批次与最近成功运行不符，或当前快照内存在必填列为空。</summary>
+    public const string Violated = "VIOLATED";
+}
+
+/// <summary>健康判定所需的全部观测值。由适配层采集，判定本身不碰数据库。</summary>
+public sealed class S8AuthorityObservation
+{
+    public string DatasetCode { get; init; } = string.Empty;
+
+    public long TenantId { get; init; }
+
+    public DateTime ObservedAt { get; init; }
+
+    /// <summary>数据集是否声明了 AuthoritySpec。</summary>
+    public bool SpecFound { get; init; }
+
+    /// <summary>见 <see cref="S8AuthoritySpec.ProducerTrusted"/>。</summary>
+    public bool ProducerTrusted { get; init; } = true;
+
+    public string UntrustedReasonCode { get; init; } = S8AuthorityHealthReason.Rule01MultiWriterUntrusted;
+
+    /// <summary>回溯窗口内最近一次成功生产运行的<b>结束</b>时刻。null = 窗口内没有成功。</summary>
+    public DateTime? LastSuccessEndAt { get; init; }
+
+    /// <summary>最近一次成功之后是否出现了终态非成功的运行。</summary>
+    public bool HasNewerTerminalFailure { get; init; }
+
+    /// <summary>上一条的具体状态字面量（用于区分 LATEST_FAILED / LATEST_PARTIAL）。</summary>
+    public string? NewerTerminalFailureStatus { get; init; }
+
+    /// <summary>比最近一次成功更新、且尚未终态的运行中，最早那条的开始时刻。null = 没有在途运行。</summary>
+    public DateTime? OldestNewerInFlightStartAt { get; init; }
+
+    /// <summary>见 <see cref="S8SnapshotInvariant"/>。</summary>
+    public string SnapshotInvariant { get; init; } = S8SnapshotInvariant.NotApplicable;
+
+    /// <summary>快照不变量被破坏时的细节，写进 Reason 便于排障。</summary>
+    public string? SnapshotDetail { get; init; }
+
+    public TimeSpan StaleWindow { get; init; } = TimeSpan.FromHours(12);
+}
+
+/// <summary>
+/// Authority 健康判定的<b>纯函数核心</b>。
+///
+/// <para>刻意不依赖数据库、时钟与任何服务：所有输入都在
+/// <see cref="S8AuthorityObservation"/> 里显式给出，因此 A–K 全部场景都能用普通单测覆盖，
+/// 不需要连库、不需要 <c>AIDOP_IT</c> 门控（该门控在未设变量时是静默 return、显示为绿，
+/// 核心判定绝不能只靠它守）。</para>
+///
+/// <para><b>不变量</b>：本文件<b>不存在</b>任何行数入参，因此
+/// 「取数为空 ⇒ 拦截恢复」这条错误逻辑在结构上写不出来。</para>
+/// </summary>
+public static class S8AuthorityHealthEvaluator
+{
+    public static S8AuthorityHealthResult Evaluate(S8AuthorityObservation o)
+    {
+        ArgumentNullException.ThrowIfNull(o);
+
+        var staleAfter = o.LastSuccessEndAt.HasValue
+            ? o.LastSuccessEndAt.Value + o.StaleWindow
+            : (DateTime?)null;
+
+        S8AuthorityHealthResult Result(string state, string reasonCode, string reason) => new()
+        {
+            DatasetCode = o.DatasetCode,
+            TenantId = o.TenantId,
+            State = state,
+            ReasonCode = reasonCode,
+            Reason = reason,
+            ObservedAt = o.ObservedAt,
+            LastSuccessAt = o.LastSuccessEndAt,
+            StaleAfter = staleAfter
+        };
+
+        // ① 没有声明 AuthoritySpec —— 无从判断，保守拦截。
+        if (!o.SpecFound)
+            return Result(S8AuthorityHealthState.Unknown, S8AuthorityHealthReason.NoAuthoritySpec,
+                $"数据集 {o.DatasetCode} 未声明 AuthoritySpec，无法判定 Authority 健康度");
+
+        // ② 生产者不可信（多写入方且部分不留运行日志）—— run-log 证明不了 provenance。
+        if (!o.ProducerTrusted)
+            return Result(S8AuthorityHealthState.Unknown, o.UntrustedReasonCode,
+                $"数据集 {o.DatasetCode} 的 Authority 存在多个写入方且部分不写运行日志，运行日志无法证明数据出处");
+
+        // ③ 回溯窗口内没有任何成功运行。含「从未同步」与「停摆超过回溯窗口」。
+        if (!o.LastSuccessEndAt.HasValue)
+            return Result(S8AuthorityHealthState.Unknown, S8AuthorityHealthReason.NoSuccessfulRun,
+                $"回溯窗口内没有 {o.DatasetCode} 的成功生产运行");
+
+        // ④ 最近一次成功之后出现终态失败 —— 之后的数据状态不可信。
+        //    先于 STALE 判定，因为「失败」比「过期」更具体、更有排障价值。
+        if (o.HasNewerTerminalFailure)
+        {
+            var isPartial = string.Equals(o.NewerTerminalFailureStatus, "PARTIAL", StringComparison.OrdinalIgnoreCase);
+            return Result(S8AuthorityHealthState.Degraded,
+                isPartial ? S8AuthorityHealthReason.LatestPartial : S8AuthorityHealthReason.LatestFailed,
+                $"最近一次成功（{o.LastSuccessEndAt:yyyy-MM-dd HH:mm:ss}）之后出现终态 " +
+                $"{o.NewerTerminalFailureStatus ?? "非成功"} 运行");
+        }
+
+        // ⑤ 存在比最近一次成功更新、且尚未终态的运行 —— 本轮 Authority 可能只写了一半。
+        //    IN_FLIGHT 与 STUCK_RUN 都拦截，区别只在排障动作（等它跑完 / 去清孤儿行）。
+        if (o.OldestNewerInFlightStartAt.HasValue)
+        {
+            var age = o.ObservedAt - o.OldestNewerInFlightStartAt.Value;
+            var stuck = age > o.StaleWindow;
+            return Result(S8AuthorityHealthState.Degraded,
+                stuck ? S8AuthorityHealthReason.StuckRun : S8AuthorityHealthReason.InFlight,
+                $"存在比最近一次成功更新且未终态的生产运行（开始于 " +
+                $"{o.OldestNewerInFlightStartAt:yyyy-MM-dd HH:mm:ss}，已历时 {age.TotalMinutes:F0} 分钟）");
+        }
+
+        // ⑥ 最近一次成功已过期。
+        if (o.ObservedAt > staleAfter!.Value)
+            return Result(S8AuthorityHealthState.Degraded, S8AuthorityHealthReason.Stale,
+                $"最近一次成功于 {o.LastSuccessEndAt:yyyy-MM-dd HH:mm:ss}，" +
+                $"已超过 {o.StaleWindow.TotalHours:F1} 小时的新鲜度窗口");
+
+        // ⑦ 快照不变量被破坏。
+        if (string.Equals(o.SnapshotInvariant, S8SnapshotInvariant.Violated, StringComparison.Ordinal))
+            return Result(S8AuthorityHealthState.Degraded, S8AuthorityHealthReason.SnapshotInvariantFailed,
+                $"当前快照不变量被破坏：{o.SnapshotDetail ?? "(无细节)"}");
+
+        // ⑧ 快照发布无法确认 —— 合法空快照与发布失败字面等同，不猜，判 UNKNOWN。
+        if (string.Equals(o.SnapshotInvariant, S8SnapshotInvariant.Unconfirmed, StringComparison.Ordinal))
+            return Result(S8AuthorityHealthState.Unknown, S8AuthorityHealthReason.SnapshotPublicationUnconfirmed,
+                $"无法确认本轮快照是否发布成功：{o.SnapshotDetail ?? "当前批次数为 0"}");
+
+        // ⑨ 通过。注意：走到这里从未读过任何行数 —— Authority 可信但一行都没有，同样是 HEALTHY。
+        return Result(S8AuthorityHealthState.Healthy, S8AuthorityHealthReason.Healthy,
+            $"最近一次成功于 {o.LastSuccessEndAt:yyyy-MM-dd HH:mm:ss}，在新鲜度窗口内，无更新的失败或在途运行");
+    }
+}

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/Health/S8AuthoritySpec.cs

@@ -0,0 +1,88 @@
+namespace Admin.NET.Plugin.AiDOP.Service.S8.Rules.Health;
+
+/// <summary>Authority 的快照语义种类。决定除「生产运行是否成功」之外还要不要校验表级不变量。</summary>
+public static class S8AuthorityKind
+{
+    /// <summary>当前态镜像表（如 <c>mdp_std_so</c>）：upsert 覆盖、无发布步骤、读者不按批次取数 ⇒ 无表级不变量可校验。</summary>
+    public const string CurrentStateMirror = "CURRENT_STATE_MIRROR";
+
+    /// <summary>当前快照表（如 <c>dwd_requirement_examine_detail</c>）：按批次累积 + 末尾原子翻牌发布 ⇒ 必须校验发布不变量。</summary>
+    public const string PublishedSnapshot = "PUBLISHED_SNAPSHOT";
+}
+
+/// <summary>
+/// 数据集的 Authority 声明：健康判定需要知道的一切，全部声明在数据集定义处。
+///
+/// <para><b>为什么放在数据集定义而不是规则上</b>：不变量是 per-dataset 的。
+/// Rule02 与 Rule03 共用同一个生产者作业（实测 <c>job_code</c> 同为 <c>S1_MDP_SYNC_TRANSFORM</c>、
+/// 同一次 run、同一个 batch_id），仅靠运行日志无法区分二者；把不变量挂在数据集上，
+/// Resolver 内部就不需要 <c>switch (ruleCode)</c>。</para>
+///
+/// <para><b>调度器不感知物理表名</b>：它只传 <c>DatasetCode</c>，
+/// 由 MDP 适配实现去查具体的表与运行日志。</para>
+/// </summary>
+public sealed class S8AuthoritySpec
+{
+    /// <summary>生产该 Authority 的转换作业编码，对应 <c>mdp_transform_run_log.job_code</c>。</summary>
+    public string ProducerJobCode { get; init; } = string.Empty;
+
+    /// <summary>见 <see cref="S8AuthorityKind"/>。</summary>
+    public string AuthorityKind { get; init; } = S8AuthorityKind.CurrentStateMirror;
+
+    /// <summary>
+    /// 最近一次成功距今超过该时长即判 STALE。
+    ///
+    /// <para><b>必须显式声明，不从作业周期推导</b>。实测四条理由：
+    /// ① 声明周期 60 分钟，而同租户实际相邻成功间隔为 28–39 分钟，声明值高估；
+    /// ② 近 14 天 3530 个间隔中，最大 8.04 小时、超过 3 小时的有 46 次 ——
+    ///    若按「3 × 周期 = 3 小时」判定，14 天内会产生 46 次假 STALE；
+    /// ③ 触发器调度的是「入队」而非「执行」，每租户实际起跑时间还叠加了不可控的队列延迟；
+    /// ④ 触发器 Args 的单位与形态不统一（PeriodTrigger 与 CronTrigger 混存，
+    ///    <c>[Period(ms)]</c> 与 <c>[PeriodSeconds(s)]</c> 都序列化成裸数字）。
+    /// 且该表可被作业管理界面在运行期改写，不应让门禁阈值随之漂移。</para>
+    /// </summary>
+    public TimeSpan StaleWindow { get; init; } = TimeSpan.FromHours(12);
+
+    /// <summary>
+    /// 查找「最近一次成功」时的回溯窗口。<b>不是可选优化</b>：
+    /// 实测无界回溯在长期停摆的租户上单次耗时 146ms（反向扫描 7122 条索引项），
+    /// 加 7 天上界后降到 7.3ms。且语义上等价 —— 「回溯窗口内没有成功」本身就是 STALE 结论。
+    /// </summary>
+    public TimeSpan LookbackWindow { get; init; } = TimeSpan.FromDays(7);
+
+    /// <summary>
+    /// 生产者是否可信。
+    /// <para>false 表示该 Authority 存在多个写入方、且部分写入方不写运行日志，
+    /// 于是 run-log 无法证明表里最终的数据出自谁 —— 此时一律判 UNKNOWN、拦截恢复。
+    /// 这是 Rule01 的现状（<c>dwd_supplier_delivery</c> 除 MDP 转换外，
+    /// 还有两个 API 路径直写且不留 run-log）。</para>
+    /// </summary>
+    public bool ProducerTrusted { get; init; } = true;
+
+    /// <summary>ProducerTrusted 为 false 时的原因码，见 <see cref="S8AuthorityHealthReason"/>。</summary>
+    public string UntrustedReasonCode { get; init; } = S8AuthorityHealthReason.Rule01MultiWriterUntrusted;
+
+    // ── 以下仅 AuthorityKind = PUBLISHED_SNAPSHOT 时使用 ──────────────────────────────
+    // 声明在这里而不是写死在 Resolver 里，是为了让 Resolver 保持通用：
+    // 它只按声明去查，内部不出现 switch (datasetCode) 或 switch (tableName)。
+    // 调度器仍然只传 DatasetCode，不接触任何物理表名。
+
+    /// <summary>快照表名。</summary>
+    public string? SnapshotTable { get; init; }
+
+    /// <summary>快照表里标识批次的列。</summary>
+    public string SnapshotBatchColumn { get; init; } = "calc_batch_id";
+
+    /// <summary>快照表里标识「属于当前批次」的列（值为 1 即当前）。</summary>
+    public string SnapshotCurrentFlagColumn { get; init; } = "is_current_flag";
+
+    /// <summary>
+    /// 当前快照内不允许为空的列。
+    /// <para>用<b>绝对计数</b>校验（<c>COUNT(*) WHERE 当前 AND 列 IS NULL</c> 必须为 0），
+    /// 而不是空值率 —— 计数在空快照上天然真空成立，与 HEALTHY_EMPTY 零冲突；
+    /// 写成比率则会在空集上 0 除、语义未定义。</para>
+    /// <para>只填「结构上不可能为空」的列。例如 <c>bom_level</c> 就<b>不能</b>放进来 ——
+    /// 转换侧对非数字层级刻意留 NULL（"脏值留 NULL，不猜层级"），把它列为必填等于按设计制造假 DEGRADED。</para>
+    /// </summary>
+    public string[] SnapshotRequiredColumns { get; init; } = Array.Empty<string>();
+}

server/Plugins/Admin.NET.Plugin.AiDOP/Service/S8/Rules/Health/S8MdpAuthorityHealthResolver.cs

@@ -0,0 +1,295 @@
+using Admin.NET.Plugin.AiDOP.Service.S8.Rules.DataAccess;
+using Microsoft.Extensions.Logging;
+using SqlSugar;
+using System.Text.RegularExpressions;
+
+namespace Admin.NET.Plugin.AiDOP.Service.S8.Rules.Health;
+
+/// <summary>
+/// <see cref="IS8AuthorityHealthResolver"/> 的数据中台适配实现：
+/// 从 <c>mdp_transform_run_log</c> 采集生产运行事实，必要时再校验快照表的发布不变量，
+/// 然后交给纯函数 <see cref="S8AuthorityHealthEvaluator"/> 判定。
+///
+/// <para>本类是「临时适配」：中台一旦提供数据集级 freshness 接口，
+/// 换掉本类即可，端口与调用方不动。</para>
+/// </summary>
+public class S8MdpAuthorityHealthResolver : IS8AuthorityHealthResolver, ITransient
+{
+    /// <summary>合法标识符（表名 / 列名）。声明值虽由开发者书写，仍不直接拼进 SQL 而先做白名单校验。</summary>
+    private static readonly Regex IdentifierPattern = new("^[A-Za-z_][A-Za-z0-9_]*$", RegexOptions.Compiled);
+
+    private readonly ISqlSugarClient _db;
+    private readonly IS8DatasetCatalog _datasetCatalog;
+    private readonly ILogger<S8MdpAuthorityHealthResolver> _logger;
+
+    public S8MdpAuthorityHealthResolver(
+        ISqlSugarClient db,
+        IS8DatasetCatalog datasetCatalog,
+        ILogger<S8MdpAuthorityHealthResolver> logger)
+    {
+        _db = db;
+        _datasetCatalog = datasetCatalog;
+        _logger = logger;
+    }
+
+    public async Task<S8AuthorityHealthResult> ResolveAsync(long tenantId, string datasetCode)
+    {
+        var observedAt = DateTime.Now;
+        try
+        {
+            var spec = _datasetCatalog.Find(datasetCode)?.AuthoritySpec;
+
+            // 未声明 AuthoritySpec / 声明不完整 → 交给纯函数判 UNKNOWN，不在这里编造默认值。
+            if (spec is null || string.IsNullOrWhiteSpace(spec.ProducerJobCode))
+            {
+                return S8AuthorityHealthEvaluator.Evaluate(new S8AuthorityObservation
+                {
+                    DatasetCode = datasetCode, TenantId = tenantId, ObservedAt = observedAt, SpecFound = false
+                });
+            }
+
+            // 生产者不可信时不必查库 —— 结论与运行日志无关。
+            if (!spec.ProducerTrusted)
+            {
+                return S8AuthorityHealthEvaluator.Evaluate(new S8AuthorityObservation
+                {
+                    DatasetCode = datasetCode, TenantId = tenantId, ObservedAt = observedAt,
+                    SpecFound = true, ProducerTrusted = false,
+                    UntrustedReasonCode = spec.UntrustedReasonCode,
+                    StaleWindow = spec.StaleWindow
+                });
+            }
+
+            var lookbackFrom = observedAt - spec.LookbackWindow;
+            var lastSuccess = await QueryLastSuccessAsync(tenantId, spec.ProducerJobCode, lookbackFrom);
+
+            // 锚点：有成功则以其结束时刻为界，否则以回溯窗口起点为界。
+            var anchor = lastSuccess?.EndTime ?? lookbackFrom;
+
+            var newerFailure = await QueryNewerTerminalFailureAsync(tenantId, spec.ProducerJobCode, anchor);
+            var newerInFlight = await QueryNewerInFlightAsync(tenantId, spec.ProducerJobCode, anchor);
+
+            var (invariant, detail) = await ResolveSnapshotInvariantAsync(tenantId, spec, lastSuccess?.BatchId);
+
+            return S8AuthorityHealthEvaluator.Evaluate(new S8AuthorityObservation
+            {
+                DatasetCode = datasetCode,
+                TenantId = tenantId,
+                ObservedAt = observedAt,
+                SpecFound = true,
+                ProducerTrusted = true,
+                LastSuccessEndAt = lastSuccess?.EndTime,
+                HasNewerTerminalFailure = newerFailure is not null,
+                NewerTerminalFailureStatus = newerFailure?.Status,
+                OldestNewerInFlightStartAt = newerInFlight,
+                SnapshotInvariant = invariant,
+                SnapshotDetail = detail,
+                StaleWindow = spec.StaleWindow
+            });
+        }
+        catch (Exception ex)
+        {
+            // fail-safe：探测自身失败一律判 UNKNOWN、拦截恢复。绝不因为查不动库就放行。
+            _logger.LogWarning(ex,
+                "authority_health_probe_failed tenant={Tenant} dataset={Dataset}", tenantId, datasetCode);
+
+            return new S8AuthorityHealthResult
+            {
+                DatasetCode = datasetCode, TenantId = tenantId, ObservedAt = observedAt,
+                State = S8AuthorityHealthState.Unknown,
+                ReasonCode = S8AuthorityHealthReason.ResolverFailed,
+                Reason = $"Authority 健康探测失败：{ex.GetType().Name}"
+            };
+        }
+    }
+
+    /// <summary>
+    /// 最近一次<b>终态成功</b>的生产运行。
+    ///
+    /// <para><b>三处刻意为之，改动前先读完</b>：</para>
+    /// <list type="number">
+    ///   <item>租户条件是严格相等，<b>绝不能</b>写成 <c>(tenant_id = @TenantId OR tenant_id = 0)</c>。
+    ///     仓内 <c>MdpMonitorService.BuildMdpRunLogTenantWhere</c> 就是后者，本类不得复用：
+    ///     该表确实存在 <c>tenant_id = 0</c> 的平台行（实测 S1 有 111 条、
+    ///     且 <c>S5_PURCHASE_RECEIPT_MDP_SYNC</c> 有 325 条平台级 SUCCESS），
+    ///     一条这样的行会让所有租户同时判成健康。</item>
+    ///   <item>必须显式过滤终态。该表有 298 条永不回收的 RUNNING 行（S1 占 42 条，最老 570 小时），
+    ///     裸 <c>ORDER BY start_time DESC LIMIT 1</c> 会取到孤儿行，其 <c>end_time</c> 为 NULL，
+    ///     后续新鲜度运算全部失效。</item>
+    ///   <item>按 <c>start_time</c> 排序（走 <c>idx_job_start</c> 的反向索引扫描，无 filesort），
+    ///     但新鲜度用取到那行的 <c>end_time</c> 计算 —— 那才是 Authority 真正变成当前态的时刻。</item>
+    /// </list>
+    /// </summary>
+    private async Task<RunRow?> QueryLastSuccessAsync(long tenantId, string jobCode, DateTime lookbackFrom)
+    {
+        const string sql =
+            """
+            SELECT r.batch_id AS BatchId, r.status AS Status, r.start_time AS StartTime, r.end_time AS EndTime
+            FROM mdp_transform_run_log r
+            WHERE r.job_code = @JobCode
+              AND r.tenant_id = @TenantId
+              AND r.status = 'SUCCESS'
+              AND r.end_time IS NOT NULL
+              AND r.start_time >= @LookbackFrom
+            ORDER BY r.start_time DESC, r.id DESC
+            LIMIT 1
+            """;
+
+        var rows = await _db.Ado.SqlQueryAsync<RunRow>(sql, new SugarParameter[]
+        {
+            new("@JobCode", jobCode), new("@TenantId", tenantId), new("@LookbackFrom", lookbackFrom)
+        });
+        return rows.FirstOrDefault();
+    }
+
+    /// <summary>
+    /// 锚点之后是否出现终态非成功的运行。
+    /// <para>用 <c>status &lt;&gt; 'SUCCESS'</c> 而非 <c>IN ('FAILED', ...)</c>：
+    /// 将来若新增未知的终态状态，它会被归为「失败」而不是被静默当成健康 —— 失败方向必须保守。</para>
+    /// </summary>
+    private async Task<RunRow?> QueryNewerTerminalFailureAsync(long tenantId, string jobCode, DateTime anchor)
+    {
+        const string sql =
+            """
+            SELECT r.batch_id AS BatchId, r.status AS Status, r.start_time AS StartTime, r.end_time AS EndTime
+            FROM mdp_transform_run_log r
+            WHERE r.job_code = @JobCode
+              AND r.tenant_id = @TenantId
+              AND r.end_time IS NOT NULL
+              AND r.status <> 'SUCCESS'
+              AND r.start_time > @Anchor
+            ORDER BY r.start_time DESC, r.id DESC
+            LIMIT 1
+            """;
+
+        var rows = await _db.Ado.SqlQueryAsync<RunRow>(sql, new SugarParameter[]
+        {
+            new("@JobCode", jobCode), new("@TenantId", tenantId), new("@Anchor", anchor)
+        });
+        return rows.FirstOrDefault();
+    }
+
+    /// <summary>
+    /// 锚点之后是否存在尚未终态的运行，取最早那条的开始时刻。
+    /// <para><c>end_time IS NULL</c> 与 <c>status = 'RUNNING'</c> 在该表上实测完全等价（298/298 双向成立），
+    /// 用前者是因为它对将来新增的非终态状态同样成立。</para>
+    /// </summary>
+    private async Task<DateTime?> QueryNewerInFlightAsync(long tenantId, string jobCode, DateTime anchor)
+    {
+        const string sql =
+            """
+            SELECT MIN(r.start_time) AS StartTime
+            FROM mdp_transform_run_log r
+            WHERE r.job_code = @JobCode
+              AND r.tenant_id = @TenantId
+              AND r.end_time IS NULL
+              AND r.start_time > @Anchor
+            """;
+
+        var rows = await _db.Ado.SqlQueryAsync<RunRow>(sql, new SugarParameter[]
+        {
+            new("@JobCode", jobCode), new("@TenantId", tenantId), new("@Anchor", anchor)
+        });
+        return rows.FirstOrDefault()?.StartTime;
+    }
+
+    /// <summary>
+    /// 快照型 Authority 的发布不变量。
+    ///
+    /// <para>当前批次数 = 0 时返回 <see cref="S8SnapshotInvariant.Unconfirmed"/> 而非「健康的空」：
+    /// 生产侧的发布语句在本轮零行时命中 0 行、在库里不留任何 batch marker，
+    /// 于是「源侧本轮确实没有数据」与「发布压根没跑成」字面等同。
+    /// 这一点必须保守 —— 猜成健康会让真实预警被整体判成已恢复。
+    /// 消除它需要生产侧补一个发布标记，属下一批的工作。</para>
+    ///
+    /// <para>作用域只按租户。快照的真实作用域是 (租户, 工厂)，但运行日志表没有 factory 列，
+    /// 无法把两侧对齐；若某租户将来出现多工厂，本查询会看到多个当前批次并判 VIOLATED ——
+    /// 是拦截而非放行，失败方向安全。</para>
+    /// </summary>
+    private async Task<(string Invariant, string? Detail)> ResolveSnapshotInvariantAsync(
+        long tenantId, S8AuthoritySpec spec, string? lastSuccessBatchId)
+    {
+        if (!string.Equals(spec.AuthorityKind, S8AuthorityKind.PublishedSnapshot, StringComparison.Ordinal))
+            return (S8SnapshotInvariant.NotApplicable, null);
+
+        if (string.IsNullOrWhiteSpace(spec.SnapshotTable))
+            return (S8SnapshotInvariant.Violated, "AuthorityKind 声明为 PUBLISHED_SNAPSHOT 但未声明 SnapshotTable");
+
+        var table = RequireIdentifier(spec.SnapshotTable, nameof(spec.SnapshotTable));
+        var batchCol = RequireIdentifier(spec.SnapshotBatchColumn, nameof(spec.SnapshotBatchColumn));
+        var flagCol = RequireIdentifier(spec.SnapshotCurrentFlagColumn, nameof(spec.SnapshotCurrentFlagColumn));
+
+        var batchSql =
+            $"""
+             SELECT COUNT(DISTINCT s.`{batchCol}`) AS CurrentBatches, MIN(s.`{batchCol}`) AS BatchId
+             FROM `{table}` s
+             WHERE s.tenant_id = @TenantId AND s.`{flagCol}` = 1
+             """;
+
+        var batchRows = await _db.Ado.SqlQueryAsync<SnapshotRow>(
+            batchSql, new SugarParameter[] { new("@TenantId", tenantId) });
+        var batch = batchRows.FirstOrDefault();
+        var currentBatches = batch?.CurrentBatches ?? 0;
+
+        if (currentBatches == 0)
+            return (S8SnapshotInvariant.Unconfirmed,
+                $"{table} 在租户 {tenantId} 下当前批次数为 0（合法空快照与发布失败在库里无法区分）");
+
+        if (currentBatches > 1)
+            return (S8SnapshotInvariant.Violated,
+                $"{table} 在租户 {tenantId} 下同时存在 {currentBatches} 个当前批次");
+
+        if (!string.IsNullOrWhiteSpace(lastSuccessBatchId)
+            && !string.Equals(batch!.BatchId, lastSuccessBatchId, StringComparison.Ordinal))
+        {
+            return (S8SnapshotInvariant.Violated,
+                $"{table} 的当前批次 {batch.BatchId} 与最近一次成功运行的批次 {lastSuccessBatchId} 不一致");
+        }
+
+        // 必填列校验用绝对计数，空快照上天然真空成立。
+        foreach (var column in spec.SnapshotRequiredColumns)
+        {
+            var col = RequireIdentifier(column, nameof(spec.SnapshotRequiredColumns));
+            var nullSql =
+                $"""
+                 SELECT COUNT(*) AS NullCount
+                 FROM `{table}` s
+                 WHERE s.tenant_id = @TenantId AND s.`{flagCol}` = 1 AND s.`{col}` IS NULL
+                 """;
+
+            var nullRows = await _db.Ado.SqlQueryAsync<SnapshotRow>(
+                nullSql, new SugarParameter[] { new("@TenantId", tenantId) });
+            var nullCount = nullRows.FirstOrDefault()?.NullCount ?? 0;
+
+            if (nullCount > 0)
+                return (S8SnapshotInvariant.Violated,
+                    $"{table} 当前快照中 {col} 为空的行数 = {nullCount}");
+        }
+
+        return (S8SnapshotInvariant.Satisfied, null);
+    }
+
+    private static string RequireIdentifier(string value, string field)
+    {
+        if (!IdentifierPattern.IsMatch(value))
+            throw new InvalidOperationException($"AuthoritySpec.{field} 不是合法标识符：{value}");
+        return value;
+    }
+
+    /// <summary>运行日志投影行。属性名与 SELECT 别名一一对应。</summary>
+    private sealed class RunRow
+    {
+        public string? BatchId { get; set; }
+        public string? Status { get; set; }
+        public DateTime? StartTime { get; set; }
+        public DateTime? EndTime { get; set; }
+    }
+
+    /// <summary>快照校验投影行。</summary>
+    private sealed class SnapshotRow
+    {
+        public int CurrentBatches { get; set; }
+        public string? BatchId { get; set; }
+        public int NullCount { get; set; }
+    }
+}

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

@@ -37,6 +37,7 @@ public class S8WatchSchedulerService : ITransient
     private readonly IS8RuleCatalog _ruleCatalog;
     // S8-RULE-READINESS-1：与 Enable / RunNow 共用的就绪门禁。
     private readonly S8RuleReadinessGate _readinessGate;
+    private readonly Rules.Health.IS8AuthorityHealthResolver _authorityHealthResolver;
 
     private const string DetectionTriggerSource = "WATCH_SCHEDULER";
     private const string DetectResultCreated = "CREATED";
@@ -48,6 +49,13 @@ public class S8WatchSchedulerService : ITransient
     private const string DetectResultNoHit = "NO_HIT";
     private const string DetectResultEvaluateFailed = "EVALUATE_FAILED";
 
+    /// <summary>
+    /// 本轮因 Authority 不可信而跳过了恢复判定。
+    /// <para>必须落库而不是只写 text log —— 这道门本身是静默的：跳过恢复不产生任何用户可见变化，
+    /// 事后必须能从库里回答「为什么这一轮没做恢复」。</para>
+    /// </summary>
+    private const string DetectResultRecoveryBlocked = "RECOVERY_BLOCKED";
+
     // S8-DETECTION-LOG-WRITE-REDUCE-P1-1：REFRESHED 明细写入限频窗口。
     // 同 (tenant, factory, rule_code, dedup_key) 在 10 分钟内最多写一条 REFRESHED detection_log；
     // 业务侧 RefreshDetectionAsync / BackfillLegacyExceptionAsync 不受限频影响。
@@ -78,7 +86,8 @@ public class S8WatchSchedulerService : ITransient
         SqlSugarRepository<AdoS8DetectionLog> detectionLogRep,
         SqlSugarRepository<AdoS8RuleDetectionState> detectionStateRep,
         IS8RuleCatalog ruleCatalog,
-        S8RuleReadinessGate readinessGate)
+        S8RuleReadinessGate readinessGate,
+        Rules.Health.IS8AuthorityHealthResolver authorityHealthResolver)
     {
         _ruleRep = ruleRep;
         _exceptionRep = exceptionRep;
@@ -96,6 +105,7 @@ public class S8WatchSchedulerService : ITransient
         _detectionStateRep = detectionStateRep;
         _ruleCatalog = ruleCatalog;
         _readinessGate = readinessGate;
+        _authorityHealthResolver = authorityHealthResolver;
     }
 
     /// <summary>
@@ -1034,15 +1044,70 @@ public class S8WatchSchedulerService : ITransient
             throw;
         }
 
+        // ── Authority Health Gate ───────────────────────────────────────────────────────
+        // 恢复判定是唯一一处「从缺席做负面推断」的地方：某个 dedup_key 不在本轮 hits 里，
+        // 就认定它已经恢复。这条推断只有在 Authority 可信时才成立 ——
+        // 中台断供、快照没发布成功、跑批只写了一半，都会让本该命中的行凭空消失，
+        // 于是真实异常被静默标成「已恢复」，且没有任何人会收到通知。
+        //
+        // 门禁**只挡恢复这一件事**：上面的建单与刷新照常执行。
+        // 因为 hit 是「这一行确实存在且越界」的正面证据 —— 数据不可信不会凭空造出一行，
+        // 却完全可能让一行消失。风险不对称，处置也不对称。
+        //
+        // 注意这里没有、也不允许有 hits.Count 的分支：Authority 可信而本轮 0 行，
+        // 是「业务对象全部合法退场」，必须放行恢复。挡空集会把这种情况一起冻死。
+        var health = await _authorityHealthResolver.ResolveAsync(tenantId, rule.DatasetCode ?? string.Empty);
+
         List<long> recoveredIds;
-        try
+        if (!health.AllowsRecovery)
         {
-            recoveredIds = await ReconcileRecoveriesForRuleAsync(tenantId, rule, ruleType, hits, runId);
+            recoveredIds = new();
+
+            await WriteDetectionLogAsync(new AdoS8DetectionLog
+            {
+                TenantId = tenantId, FactoryId = rule.FactoryId,
+                RuleId = rule.Id, RuleCode = rule.RuleCode, RuleType = ruleType, SceneCode = rule.SceneCode,
+                SourceObjectType = rule.SourceObjectType,
+                DetectResult = DetectResultRecoveryBlocked,
+                DetectedAt = DateTime.Now,
+                FailureReason = health.ReasonCode,
+                FailureMessage = Truncate(health.Reason, 1000),
+                PayloadSnapshot = JsonSerializer.Serialize(new
+                {
+                    datasetCode = health.DatasetCode,
+                    state = health.State,
+                    reasonCode = health.ReasonCode,
+                    observedAt = health.ObservedAt,
+                    lastSuccessAt = health.LastSuccessAt,
+                    staleAfter = health.StaleAfter,
+                    hitCount = hits.Count
+                }),
+                RunId = runId, TriggerSource = DetectionTriggerSource
+            });
+
+            _logger.LogInformation(
+                "recovery_blocked ruleCode={RuleCode} tenant={Tenant} dataset={Dataset} " +
+                "state={State} reason={Reason} lastSuccessAt={LastSuccessAt}",
+                rule.RuleCode, tenantId, health.DatasetCode, health.State, health.ReasonCode, health.LastSuccessAt);
+
+            // 刻意不做的三件事：
+            //   · 不动 ConsecutiveMissCount —— 已累计的 miss 都来自可信轮次，仍然有效；
+            //     本轮既不加也不减，Authority 恢复后接着原节奏推进即可。
+            //   · 不把本次 rule run 标成失败 —— 那会累加 ConsecutiveFailureCount，
+            //     达到阈值即自动暂停规则一小时；中台跑一次批就能把规则停掉。
+            //   · 不抛异常打断本轮 —— 正面观测已经处理完，没有理由回滚它们。
         }
-        catch (Exception ex)
+        else
         {
-            _logger.LogWarning(ex, "recovery_reconcile_failed ruleCode={RuleCode} ruleType={RuleType}", rule.RuleCode, ruleType);
-            recoveredIds = new();
+            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 明细。