fix(mdp): 修复采购交付 Authority 当前态、租户隔离与空快照语义

S8 Rule01 读 dwd_supplier_delivery 的「当前快照」，此前该快照在三个方向上不可信。

一、enrichment JOIN 丢了租户谓词
表头与物料 enrichment 按 PurOrd / ItemNum 纯字符串匹配，跨租户命中。
MdpSqlScope.InjectTenantFactory 只改写 WHERE，结构上够不到 JOIN...ON，兜不了底。
症状不是行数膨胀——扇出被 uk_po_line 的 ODKU 吸收，最终行数不变，
但 item_name / supplier_code / order_date / buyer 由 MySQL 任取一行决定。
实测同一批数据：864/182 = 4.75 倍扇出、117 行跨租户；修复后 182/182 = 1.00 倍、跨租户 0。
改法：两个 JOIN 补 tenant 相等条件，并按 (租户, 键) 用 ROW_NUMBER 去重取最新一行。
不用 GROUP_CONCAT——raw_data 是 JSON 大字段而 group_concat_max_len 实测仅 1024。

二、标准层不是源的镜像，孤儿只增不减
INSERT 不带批次条件、每轮重读整张贴源表，贴源层又是纯 upsert 从不淘汰，
于是源侧已删除的行每轮被重新物化并盖上最新 sync_batch_id——连时间戳都认不出它是死的。
实测租户 797403760988229：源 182 行，标准层 518 行，其中 336 行源侧已不存在（64.9%）。
按 Rule01 判据评估当前快照会凭空建出 335 条永不恢复的异常。
改法：驱动行按 @BatchId 收窄，全量路径每轮淘汰非本批次行，DWD 追加同日淘汰。
淘汰闸门用 mdp_sync_log 的成功记录而不是「贴源层有没有行」——
后者会把源侧合法清空的租户永久锁在幽灵行上。
1.0.525 负责代码做不到的两件事：两个实体 sync_mode 切 FULL、一次性清存量孤儿。

三、合法的空会静默回落到旧快照
Provider 按「最新的那一天」挑当前快照。源侧被清空的那一轮产出 0 行 DWD，
当天根本不存在 stat_date 分区，同日淘汰也无行可删（它只能删「今天」），
于是回落到上一天，把一批早已作废的行当成当前态。
实测租户 838257212780613：源 10→0、STD 10→0、跑批 SUCCESS，Provider 仍返回前一日的 10 行。
改法：S3 在 DWD 写入与淘汰全部完成后，把 (table, batchId, statDate, currentRows)
写进 mdp_transform_run_log.summary_json；Provider 改为与该发布证据 JOIN 取数。
currentRows=0 是合法结果——成功发布的空快照；没有证据则判 NO_PUBLICATION → UNKNOWN，
由既有恢复闸门拦下。三种情形都由同一条 SQL 给出答案，回落路径在结构上不存在。
只有全量路径发布：增量批次只含变更行，指认成当前快照会让读者拿到残缺集合。

本地沙箱实测（9 租户全量跑批 SUCCESS）：
- 租户 797：source 182 / STD 182 / DWD current 182，only_source=0、only_authority=0，
  9 个关键字段 mismatch 全 0，phantom hits 335 → 0
- 源侧删 1 行 → 标准层与当前快照同步退出，贴源层残留 4 行不影响结果
- 源侧清空 → publish.currentRows=0、Provider 0 行，不再返回前一日的 10 行
- 无发布证据时 Provider 返回 0 行（旧口径会返回 220 行陈旧数据），
  运行日志实测 recovery_blocked reason=NO_PUBLICATION，发布落地后恢复正常命中

一并修掉一个会让整轮跑批 FAILED 的作用域注入陷阱：派生表外层是 FROM (，
别名正则匹配不到括号会回退成空前缀，生成裸 factory_id / 歧义的 tenant_id。
本批三条语句改为自带 @TenantId，注入对其为恒等变换，并由契约测试用真注入器守住。

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.524</AssemblyVersion>
-    <FileVersion>1.0.524</FileVersion>
-    <Version>1.0.524</Version>
+    <AssemblyVersion>1.0.525</AssemblyVersion>
+    <FileVersion>1.0.525</FileVersion>
+    <Version>1.0.525</Version>
   </PropertyGroup>
 
   <ItemGroup>
@@ -787,6 +787,12 @@
     <None Update="UpdateScripts\1.0.521.verify.sql">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
+    <None Update="UpdateScripts\1.0.525.sql">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+    <None Update="UpdateScripts\1.0.525.verify.sql">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
     <None Update="UpdateScripts\UAT-PLACEHOLDER-MENU-HIDE.ops.sql">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>

server/Admin.NET.Web.Entry/UpdateScripts/1.0.525.sql

@@ -0,0 +1,96 @@
+-- =====================================================================================
+-- 1.0.525 · S3 · 让 mdp_std_purchase_order 恢复「源的镜像」语义（Rule01 Authority 整改）
+--
+-- ── 缺陷 ────────────────────────────────────────────────────────────────────────────
+-- 标准层的 INSERT 不带任何批次条件，每轮重读整张贴源表；贴源层又是纯 upsert、从不淘汰。
+-- 于是源侧已删除的采购订单行会：
+--   ① 永远留在 mdp_stg_purchase_order；
+--   ② 每轮被重新物化进 mdp_std_purchase_order，并被盖上最新的 sync_batch_id
+--      —— 所以「看时间戳」根本认不出它们是陈旧数据。
+--
+-- 实测（aidopdev，2026-09-08）租户 797403760988229：
+--   源侧 PurOrdDetail 182 行；贴源最新批次 182 行（与源逐行相等）；
+--   标准层 518 行，其中 336 行源侧已不存在（64.9%），且 518 行共享同一个批次号。
+--   DWD 是标准层的 1:1 复制（518/518，两侧差集均为 0）——
+--   即孤儿首现于标准层，不是 DWD 也不是 Provider。
+--
+-- 另有身份断裂放大了它：source_biz_key 的格式与 source_system 都换过代
+--   （旧 `Domain:PurOrd:Line` + AIDOP  →  新的裸 RecID + AIDOPDEV_MYSQL），
+-- 而贴源唯一键是 (source_system, source_table, source_biz_key)，
+-- 两代永远不会互相 ODKU 覆盖，因此并存至今（实测 568 行旧格式 / 258 行新格式）。
+--
+-- ── 影响面 ──────────────────────────────────────────────────────────────────────────
+-- S8 Rule01（采购交付延期预警）读 dwd_supplier_delivery 的最新快照。
+-- 若按其判据评估租户 797 当前快照：483 行命中，其中 335 行（69%）对应的采购订单行
+-- 在源侧已不存在 —— 启用即凭空建出 335 条永不恢复的异常。
+--
+-- ── 修法 ────────────────────────────────────────────────────────────────────────────
+-- 代码侧（本迁移的前提，随同一版本发布）：
+--   · 标准层 INSERT 按 @BatchId 收窄驱动行；
+--   · 每轮追加淘汰步骤，删掉本租户下批次号不等于 @BatchId 的行；
+--   · DWD 追加同日淘汰，使 MAX(stat_date) 取到的必然是单一批次。
+-- 本脚本负责两件代码做不到的事：
+--   ① 把两个实体的 sync_mode 由 INCR 改为 FULL（下方 ① 的理由）；
+--   ② 一次性清掉已经沉淀的孤儿（代码只能阻止新增，删不掉存量）。
+-- =====================================================================================
+
+-- ── ① 让声明与实际行为一致：这两个实体每轮都必须全量重读 ─────────────────────────────
+--
+-- 「不在本批次里」只有在**每轮都是全量重读**时才等于「源侧已删除」。
+-- S3 的全量作业确实以 fullRefresh:true 调用，但 RunInboundAsync 与 MdpHotWatchService 不是，
+-- 而贴源层里已经存在这样一个部分批次（`hot-20260813003331`，3 行）——
+-- 若它恰好是最新批次，按批次淘汰会把该租户的标准层几乎删空。
+-- 只有 sync_mode='FULL' 能让 MdpSyncWindowResolver 在**所有**调用路径上强制 FullRefresh=true。
+--
+-- 只改这两个正式实体；*_API 与 *_SQLSERVER 变体当前未被调度，保持原样不动。
+UPDATE `mdp_entity`
+   SET `sync_mode` = 'FULL',
+       `incr_column` = NULL,
+       `update_time` = NOW()
+ WHERE `entity_code` IN ('S3_PURCHASE_ORDER_MASTER', 'S3_PURCHASE_ORDER_DETAIL')
+   AND `sync_mode` <> 'FULL';
+
+-- ── ② 一次性清除标准层已沉淀的孤儿 ──────────────────────────────────────────────────
+--
+-- 判据：该 (tenant_id, po_no, po_line) 在源表 PurOrdDetail 里已不存在。
+-- 直接对源表求证而不是对贴源层 —— 贴源层本身也在累积，拿它当基准会漏掉一部分。
+--
+-- ⚠️ 安全闸门 —— 「该租户在源表里确实有数据」：
+--    若某租户在 PurOrdDetail 里一行都没有（从未接入 / 源侧被清空），
+--    没有这道闸门就会把该租户的标准层整个删空。这里宁可不删，也不冒清空的风险。
+--    （源侧合法清空的场景由代码侧每轮淘汰处理，它用 mdp_sync_log 的成功记录做闸门，
+--      能区分「拉取成功且确实 0 行」与「拉取没跑成」。）
+--
+-- 规模：实测全表 787 行、孤儿 339 行，DELETE 走 uk_po_line 主键回表，毫秒级。
+DELETE s FROM `mdp_std_purchase_order` s
+ WHERE EXISTS (SELECT 1 FROM `PurOrdDetail` g WHERE g.`tenant_id` = s.`tenant_id`)
+   AND NOT EXISTS (
+        SELECT 1 FROM `PurOrdDetail` d
+         WHERE d.`tenant_id` = s.`tenant_id`
+           AND d.`PurOrd` = s.`po_no`
+           AND CAST(d.`Line` AS CHAR) = s.`po_line`);
+
+-- ── ③ 清除 DWD 当前快照里已失去标准层依据的行 ───────────────────────────────────────
+--
+-- Rule01 只读每租户 stat_date 最大的那一天。②之后标准层已经干净，
+-- 但当天的 DWD 快照仍留着孤儿 —— 代码侧的同日淘汰要等下一轮跑批才会生效，
+-- 而部署后规则可能先于跑批被启用。这里把当前快照一次性对齐。
+--
+-- 只处理每租户的最新 stat_date，不动历史快照（历史是留档，Provider 不读）。
+-- 用派生表而不是相关子查询：dwd_supplier_delivery 实测 41,422 行，
+-- 相关子查询在该表上已多次超时（历史上有 600s 迁移超时的先例）。
+DELETE w FROM `dwd_supplier_delivery` w
+  JOIN (SELECT `tenant_id`, MAX(`stat_date`) AS `latest_stat_date`
+          FROM `dwd_supplier_delivery` GROUP BY `tenant_id`) t
+    ON t.`tenant_id` = w.`tenant_id` AND t.`latest_stat_date` = w.`stat_date`
+ WHERE NOT EXISTS (
+        SELECT 1 FROM `mdp_std_purchase_order` s
+         WHERE s.`tenant_id` = w.`tenant_id`
+           AND s.`po_no` = w.`po_no`
+           AND s.`po_line` = w.`po_line`);
+
+-- ── 回滚（注释，非可执行）──
+-- UPDATE mdp_entity SET sync_mode='INCR', incr_column='UpdateTime'
+--  WHERE entity_code IN ('S3_PURCHASE_ORDER_MASTER','S3_PURCHASE_ORDER_DETAIL');
+-- 被删除的孤儿不做回滚：它们对应的采购订单行在源侧已不存在，
+-- 重建方式是重跑 S3 全量跑批（标准层与 DWD 均为纯派生）。

server/Admin.NET.Web.Entry/UpdateScripts/1.0.525.verify.sql

@@ -0,0 +1,54 @@
+-- 1.0.525.verify.sql — 每条 SELECT 都必须返回真值（0 或 NULL 即判失败并中断启动）
+--
+-- 与 1.0.521.verify 同法：正向断言「该清的清干净了」，反向断言「不该清的一行没少」。
+-- 反向断言不是可选的 —— 一个把整张表删空的脚本同样能让正向断言全过。
+
+-- ① 两个正式实体已切到 FULL，且不再带增量水位。
+--    incr_column 必须为空：只要它还在，MdpDbPullExecutor 就会发射游标谓词，
+--    产出的批次就是「只含变更行」的部分批次，按批次淘汰会误删存量。
+SELECT COUNT(*) = 2
+  FROM mdp_entity
+ WHERE entity_code IN ('S3_PURCHASE_ORDER_MASTER', 'S3_PURCHASE_ORDER_DETAIL')
+   AND sync_mode = 'FULL'
+   AND incr_column IS NULL;
+
+-- ② 标准层里不再有源侧已不存在的采购订单行（限定源表里确实有该租户数据）。
+SELECT COUNT(*) = 0
+  FROM mdp_std_purchase_order s
+ WHERE EXISTS (SELECT 1 FROM PurOrdDetail g WHERE g.tenant_id = s.tenant_id)
+   AND NOT EXISTS (
+        SELECT 1 FROM PurOrdDetail d
+         WHERE d.tenant_id = s.tenant_id
+           AND d.PurOrd = s.po_no
+           AND CAST(d.Line AS CHAR) = s.po_line);
+
+-- ③ 反向断言：源侧存在的采购订单行**一行都没被删掉**。
+--    只有正向断言的话，「把整张表删空」也能通过 ②。
+SELECT COUNT(*) = 0
+  FROM PurOrdDetail d
+  JOIN PurOrdMaster m ON m.RecID = d.PurOrdRecID AND m.tenant_id = d.tenant_id
+ WHERE EXISTS (SELECT 1 FROM mdp_std_purchase_order x WHERE x.tenant_id = d.tenant_id)
+   AND NOT EXISTS (
+        SELECT 1 FROM mdp_std_purchase_order s
+         WHERE s.tenant_id = d.tenant_id
+           AND s.po_no = d.PurOrd
+           AND s.po_line = CAST(d.Line AS CHAR));
+
+-- ④ 每租户的当前 DWD 快照里，不再有失去标准层依据的行。
+SELECT COUNT(*) = 0
+  FROM dwd_supplier_delivery w
+  JOIN (SELECT tenant_id, MAX(stat_date) AS latest_stat_date
+          FROM dwd_supplier_delivery GROUP BY tenant_id) t
+    ON t.tenant_id = w.tenant_id AND t.latest_stat_date = w.stat_date
+ WHERE NOT EXISTS (
+        SELECT 1 FROM mdp_std_purchase_order s
+         WHERE s.tenant_id = w.tenant_id AND s.po_no = w.po_no AND s.po_line = w.po_line);
+
+-- ⑤ 反向断言：当前快照没有被删空 —— 至少还有一个租户留着行。
+--    若源侧确实全空，本条会失败；届时应人工确认属实后再放行，
+--    而不是把它删掉。宁可多一次人工确认，也不要让「删空」静默通过。
+SELECT COUNT(*) > 0
+  FROM dwd_supplier_delivery w
+  JOIN (SELECT tenant_id, MAX(stat_date) AS latest_stat_date
+          FROM dwd_supplier_delivery GROUP BY tenant_id) t
+    ON t.tenant_id = w.tenant_id AND t.latest_stat_date = w.stat_date;

server/Plugins/Admin.NET.Plugin.AiDOP.Tests/DataPlatform/S3PurchaseAuthorityContractTests.cs

@@ -0,0 +1,326 @@
+using Admin.NET.Plugin.AiDOP.DataPlatform.MdpRebuild;
+using Xunit;
+
+namespace Admin.NET.Plugin.AiDOP.Tests.DataPlatform;
+
+/// <summary>
+/// 采购交付 Authority（<c>mdp_std_purchase_order</c> → <c>dwd_supplier_delivery</c>）的生产契约。
+///
+/// <para>这条链是 S8 Rule01 的唯一数据来源。它曾同时坏在三个地方，且三处根因都落在同一段
+/// STD 构建 SQL 上：丢了租户谓词（跨租户串值）、丢了批次作用域（孤儿只增不减）、
+/// 以及有两个业务 API 绕过它直写下游 DWD。本文件守住修好之后的形状。</para>
+/// </summary>
+public class S3PurchaseAuthorityContractTests
+{
+    private static string Source(params string[] parts)
+    {
+        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(new[] { dir!.FullName, "Admin.NET.Plugin.AiDOP" }.Concat(parts).ToArray());
+        Assert.True(File.Exists(full), $"源码文件不存在，路径需同步更新：{full}");
+        return File.ReadAllText(full);
+    }
+
+    private static string S3Service() => Source("Supply", "S3MdpSyncTransformService.cs");
+
+    private static int At(string code, string anchor)
+    {
+        var i = code.IndexOf(anchor, StringComparison.Ordinal);
+        Assert.True(i >= 0, $"未找到锚点：{anchor}");
+        return i;
+    }
+
+    /// <summary>取 <c>mdp_std_purchase_order</c> 那一条 INSERT 的 SQL 区间。</summary>
+    private static string StdInsertBlock()
+    {
+        var code = S3Service();
+        var start = At(code, "INSERT INTO mdp_std_purchase_order");
+        var end = code.IndexOf("\"\"\"", start, StringComparison.Ordinal);
+        Assert.True(end > start, "mdp_std_purchase_order 的 INSERT 未正常收尾");
+        return code[start..end];
+    }
+
+    /// <summary>
+    /// STD 构建的两个 enrichment JOIN 必须自带租户谓词。
+    ///
+    /// <para><b>为什么不能靠作用域注入兜底</b>：<c>MdpSqlScope.InjectTenantFactory</c> 的实现是
+    /// <c>Regex.Replace(sql, @"\bWHERE\b", ...)</c>，它只改写 <c>WHERE</c>，
+    /// 对 <c>JOIN ... ON</c> 零覆盖。这不是配置问题，是机制上够不到。</para>
+    ///
+    /// <para><b>后果不是行数膨胀</b>：扇出会被 <c>uk_po_line(tenant_id, po_no, po_line)</c> 的
+    /// ON DUPLICATE KEY UPDATE 吸收，最终行数不变，但 item_name / supplier_code / order_date / buyer
+    /// 这些 enrichment 值由 MySQL 任取一行决定 —— 症状是取值随机化，不是报错。
+    /// 实测单租户扇出 870/220 = 3.95 倍，item_code 跨租户重名 5529 组。</para>
+    /// </summary>
+    [Fact]
+    public void StdBuild_EnrichmentJoins_CarryTenantPredicate()
+    {
+        var block = StdInsertBlock();
+
+        // 表头 enrichment：按 PurOrd 字符串匹配，必须限定同租户。
+        Assert.Contains("m ON m.tenant_id = d.tenant_id", block);
+        Assert.Contains("m.pur_ord = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd'))", block);
+
+        // 物料 enrichment：按 ItemNum 字符串匹配，同样必须限定同租户。
+        Assert.Contains("i ON i.tenant_id = d.tenant_id", block);
+        Assert.Contains("i.item_num = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum'))", block);
+    }
+
+    /// <summary>
+    /// DWD 构建的三个 JOIN 本来就是租户安全的，本批不得把它们改坏。
+    /// <para>它们已带 <c>po.tenant_id=X.tenant_id AND po.factory_id=X.factory_id</c>，
+    /// 泄漏点在 STD 而不在这里 —— 这条断言是防止后续改动误伤。</para>
+    /// </summary>
+    [Fact]
+    public void DwdBuild_JoinsRemainTenantAndFactoryScoped()
+    {
+        var code = S3Service();
+        var start = At(code, "INSERT INTO dwd_supplier_delivery");
+        var end = code.IndexOf("\"\"\"", start, StringComparison.Ordinal);
+        var block = code[start..end];
+
+        Assert.Contains("ds ON po.tenant_id=ds.tenant_id AND po.factory_id=ds.factory_id", block);
+        Assert.Contains("dr ON po.tenant_id=dr.tenant_id AND po.factory_id=dr.factory_id", block);
+        Assert.Contains("s ON po.tenant_id=s.tenant_id AND po.factory_id=s.factory_id", block);
+    }
+
+    /// <summary>
+    /// STD 构建必须按本轮批次收窄驱动行。
+    /// <para>少了它，INSERT 会把整张贴源历史重新物化，并给死行盖上最新批次戳 ——
+    /// 实测租户 797 因此在 STD 沉淀了 336 行源侧已不存在的采购订单行（518 行中的 64.9%），
+    /// 且全部 518 行共享同一个批次号，连时间戳都认不出哪些是死的。</para>
+    /// </summary>
+    [Fact]
+    public void StdBuild_DrivingRows_ScopedToCurrentBatch()
+    {
+        var block = StdInsertBlock();
+        Assert.Contains("AND d.source_table='PurOrdDetail' AND d.sync_batch_id=@BatchId", block);
+    }
+
+    /// <summary>
+    /// enrichment 必须按 (租户, 键) 去重取最新一行，且**不得**用 GROUP_CONCAT 实现。
+    /// <para>贴源层同时存在两代 <c>source_biz_key</c>（旧 <c>Domain:PurOrd:Line</c> 与新的裸 RecID），
+    /// 两代永远不会互相 ODKU 覆盖，因此同一个 PurOrd/ItemNum 在同租户内有多行 —— 不去重就仍会扇出。</para>
+    /// <para>禁用 GROUP_CONCAT 的理由是硬的：<c>raw_data</c> 是 JSON 大字段，
+    /// 而 <c>group_concat_max_len</c> 实测仅 1024，拼接必然截断、首元素本身就可能超限。</para>
+    /// </summary>
+    [Fact]
+    public void StdBuild_EnrichmentDeduplicated_WithRowNumberNotGroupConcat()
+    {
+        var block = StdInsertBlock();
+
+        Assert.Contains("ROW_NUMBER() OVER (", block);
+        Assert.Contains("PARTITION BY x0.tenant_id, JSON_UNQUOTE(JSON_EXTRACT(x0.raw_data,'$.PurOrd'))", block);
+        Assert.Contains("PARTITION BY y0.tenant_id, JSON_UNQUOTE(JSON_EXTRACT(y0.raw_data,'$.ItemNum'))", block);
+
+        // 可执行 SQL 里不得出现 GROUP_CONCAT（注释里提到它是允许的）。
+        var sqlOnly = string.Join('\n', block.Split('\n').Where(l => !l.TrimStart().StartsWith("--", StringComparison.Ordinal)));
+        Assert.DoesNotContain("GROUP_CONCAT", sqlOnly);
+    }
+
+    /// <summary>
+    /// 每轮淘汰必须存在，且只在全量路径执行。
+    /// <para>只加批次过滤只能让死行「可识别」，不能「退役」—— 运行期链路里没有任何 DELETE。
+    /// 而增量批次只含变更行，按它淘汰会把未变更的存量行全删掉，所以必须按路径分流。</para>
+    /// </summary>
+    [Fact]
+    public void StdBuild_RetiresStaleRows_OnlyOnFullPath()
+    {
+        var code = S3Service();
+
+        Assert.Contains("DELETE s FROM mdp_std_purchase_order s", code);
+        Assert.Contains("IFNULL(s.sync_batch_id,'') <> @BatchId", code);
+
+        // 安全闸门必须用 mdp_sync_log 的成功记录，而不是「贴源层有没有行」——
+        // 后者会把源侧合法清空的租户永久锁在幽灵行上。
+        Assert.Contains("FROM mdp_sync_log l", code);
+        Assert.Contains("l.status='SUCCESS'", code);
+
+        // 路径分流
+        Assert.Contains("BuildStandardCommands(string batchId, DateTime now, bool retireStale)", code);
+        Assert.Contains("TransformStandardAsync(batchId, now, retireStale: true, cancellationToken)", code);
+        Assert.Contains("TransformStandardAsync(batchId, now, retireStale: false, cancellationToken)", code);
+    }
+
+    /// <summary>取一条语句的 SQL 区间（从锚点到原始字符串字面量的收尾 <c>"""</c>）。</summary>
+    private static string StatementBlock(string startAnchor)
+    {
+        var code = S3Service();
+        var start = At(code, startAnchor);
+        var end = code.IndexOf("\"\"\"", start, StringComparison.Ordinal);
+        Assert.True(end > start, $"语句未正常收尾：{startAnchor}");
+        return code[start..end];
+    }
+
+    /// <summary>
+    /// 本文件新增/改写的每条 SQL，经 <see cref="MdpSqlScope.InjectTenantFactory"/> 处理后
+    /// 都不得出现「注入到不存在的列上」。
+    ///
+    /// <para><b>为什么必须用真注入器跑一遍，而不是比对源码文本</b>：这个坑有两个入口，
+    /// 光看源码任何一处都不显眼，只有跑一遍注入才暴露。二者本地沙箱都实测炸过：</para>
+    ///
+    /// <list type="number">
+    /// <item><b>空前缀</b> —— <c>ResolveSourcePrefix</c> 反向找最近的同层 <c>FROM 表 别名</c>。
+    /// 派生表外层是 <c>FROM (</c>，别名正则匹配不到括号，于是回退成空前缀，生成裸
+    /// <c>factory_id</c>；派生表 SELECT 列表里没有这一列 → 整轮转换失败。</item>
+    /// <item><b>别名指向没有该列的表</b> —— 淘汰语句的 EXISTS 子查询若写成
+    /// <c>l.tenant_id = s.tenant_id</c>（关联外层、不含 <c>@TenantId</c>），会被注入
+    /// <c>COALESCE(NULLIF(l.factory_id,0),1)=@FactoryId</c>，而 <c>mdp_sync_log</c> 没有
+    /// <c>factory_id</c> 列 → 同样整轮失败。</item>
+    /// </list>
+    ///
+    /// <para>失败形态是<b>整轮跑批 FAILED</b>，不是少几行数据 —— 而 Authority 一轮不产出，
+    /// S8 的恢复闸门就判 DEGRADED、整个租户的恢复停摆。所以这条守卫值得用真注入器跑。</para>
+    /// </summary>
+    [Theory]
+    [InlineData("INSERT INTO mdp_std_purchase_order")]
+    [InlineData("DELETE s FROM mdp_std_purchase_order s")]
+    [InlineData("DELETE w FROM dwd_supplier_delivery w")]
+    public void ScopeInjection_NeverTargetsAColumnThatDoesNotExist(string anchor)
+    {
+        var injected = MdpSqlScope.InjectTenantFactory(StatementBlock(anchor));
+
+        // ① 空前缀：注入器没能定位别名。在本文件里所有表都带别名，因此这必定是缺陷。
+        Assert.DoesNotContain("WHERE tenant_id=@TenantId", injected);
+
+        // ② 前缀指向的表没有 factory_id：派生表别名（x/y/m/i）与 mdp_sync_log（l）都没有这一列。
+        foreach (var alias in new[] { "x", "y", "m", "i", "l" })
+            Assert.DoesNotContain($"NULLIF({alias}.factory_id", injected);
+    }
+
+    /// <summary>
+    /// 驱动行必须**显式**带上租户与工厂谓词，不能依赖注入器补。
+    ///
+    /// <para>历史上它确实是靠注入的：<c>ResolveSourcePrefix</c> 反向找到 <c>FROM ... d</c>
+    /// 就能补出 <c>d.</c> 前缀。但两个 enrichment 派生表插进 <c>FROM ... d</c> 与本
+    /// <c>WHERE</c> 之间以后，反向扫描会先撞上派生表的 <c>FROM (</c>，别名正则匹配不到括号
+    /// → 回退成空前缀，生成裸 <c>tenant_id</c>。而派生表 m / i 也各自暴露 <c>tenant_id</c>，
+    /// 该列立刻变成歧义列 —— 整轮转换 FAILED，不是少几行。</para>
+    ///
+    /// <para>所以这不是风格偏好：只要这条 INSERT 还带派生表，作用域就必须写死。</para>
+    /// </summary>
+    [Fact]
+    public void DrivingRows_CarryTenantAndFactoryExplicitly()
+    {
+        Assert.Contains(
+            "WHERE d.tenant_id=@TenantId AND COALESCE(NULLIF(d.factory_id,0),1)=@FactoryId",
+            StatementBlock("INSERT INTO mdp_std_purchase_order"));
+    }
+
+    /// <summary>
+    /// Rule01 Authority 链上的三条语句必须**完全自带**作用域 —— 注入对它们应当是恒等变换。
+    ///
+    /// <para>这是本文件与注入器之间的边界声明：这三条 SQL 的作用域由字面 SQL 唯一决定，
+    /// 不交给一个够不到 <c>JOIN...ON</c>、认不出派生表别名、且对 SQL 注释同样生效的正则去猜。
+    /// 恒等变换是这里唯一可接受的结果；任何非恒等都意味着有一个 <c>WHERE</c> 漏了
+    /// <c>@TenantId</c>，而它下一步会以整轮跑批失败的形式暴露。</para>
+    /// </summary>
+    [Theory]
+    [InlineData("INSERT INTO mdp_std_purchase_order")]
+    [InlineData("DELETE s FROM mdp_std_purchase_order s")]
+    [InlineData("DELETE w FROM dwd_supplier_delivery w")]
+    public void AuthorityStatements_AreFullySelfScoped(string anchor)
+    {
+        var raw = StatementBlock(anchor);
+        Assert.Equal(raw, MdpSqlScope.InjectTenantFactory(raw));
+    }
+
+    /// <summary>
+    /// DWD 必须做同日淘汰，且**放在 INSERT 之后**。
+    /// <para>放之前会让快照短暂为空；此处无事务，并发的 Rule01 取数会读到空快照，
+    /// 把全部在办异常判成已恢复 —— 那正是 Authority 健康门禁要防的形态。</para>
+    /// <para>该淘汰同时保证 <c>MAX(stat_date)</c> 取到的是单一批次：实测同一天混入两个批次
+    /// 已发生 3 次（如 2026-06-07 的 178 行 + 7 行）。</para>
+    /// </summary>
+    [Fact]
+    public void DwdBuild_RetiresSameDayStaleRows_AfterInsert()
+    {
+        var code = S3Service();
+        var insertAt = At(code, "INSERT INTO dwd_supplier_delivery");
+        var deleteAt = code.IndexOf("DELETE w FROM dwd_supplier_delivery w", insertAt, StringComparison.Ordinal);
+
+        Assert.True(deleteAt > insertAt, "同日淘汰必须在 INSERT 之后，否则快照会短暂为空");
+        Assert.Contains("w.stat_date=@StatDate", code);
+        Assert.Contains("IFNULL(w.sync_batch_id,'') <> @BatchId", code);
+        Assert.Contains("BuildDwdAsync(batchId, now, retireStale: true, result, cancellationToken)", code);
+        Assert.Contains("BuildDwdAsync(batchId, now, retireStale: false, result, cancellationToken)", code);
+    }
+
+    /// <summary>
+    /// <c>dwd_supplier_delivery</c> 与 <c>mdp_std_purchase_order</c> 在运行期<b>只能有一个写入方</b>。
+    ///
+    /// <para>它们是 Rule01 的 Authority。此前另有两个业务 API 直写：
+    /// <c>ProcessOutsourceOrderService</c> 把 <c>delivery_status</c> 硬编码为 <c>'OPEN'</c> 并放进
+    /// ON DUPLICATE KEY UPDATE，一次建单就能把 MDP 刚判定的 COMPLETED / DELAYED 无条件翻回 OPEN；
+    /// <c>DeliveryScheduleService</c> 则写出 CANCELLED / CLOSED / PARTIAL 三个没有任何读方认识的值。</para>
+    ///
+    /// <para>更严重的是它们能凭空造出一个「最新快照」：Rule01 取 <c>MAX(stat_date)</c> 那一天的全部行，
+    /// 实测租户 797403760988229 的 2026-05-16 快照只有 2 行、且全部出自旁路写入 ——
+    /// 中台停摆时，一次业务 API 调用就让 Rule01 在 2 行残缺数据上评估，而快照看上去还很新鲜。</para>
+    ///
+    /// <para>用「非注释行里出现写入语句」判定，避免被解释性注释里的表名误判。</para>
+    /// </summary>
+    [Fact]
+    public void AuthorityTables_HaveExactlyOneRuntimeWriter()
+    {
+        // (文件, 表, 期望写入处数)
+        var expectations = new[]
+        {
+            ("S3MdpSyncTransformService.cs",   "dwd_supplier_delivery",   1),
+            ("DeliveryScheduleService.cs",     "dwd_supplier_delivery",   0),
+            ("ProcessOutsourceOrderService.cs","dwd_supplier_delivery",   0),
+            ("S3MdpSyncTransformService.cs",   "mdp_std_purchase_order",  1),
+            ("DeliveryScheduleService.cs",     "mdp_std_purchase_order",  0),
+            ("ProcessOutsourceOrderService.cs","mdp_std_purchase_order",  0)
+        };
+
+        foreach (var (file, table, expected) in expectations)
+        {
+            var writes = Source("Supply", file).Split('\n')
+                .Where(l => !l.TrimStart().StartsWith("//", StringComparison.Ordinal))
+                .Count(l => l.Contains($"INSERT INTO {table}", StringComparison.Ordinal)
+                         || l.Contains($"UPDATE {table}", StringComparison.Ordinal));
+
+            Assert.True(writes == expected,
+                $"{file} 对 {table} 的运行期写入应为 {expected} 处，实测 {writes} 处");
+        }
+    }
+
+    /// <summary>
+    /// 旁路写入方不得把 <c>delivery_status</c> 硬编码后写进 Authority。
+    /// <para>状态只能由正式 MDP 转换从源侧派生（COMPLETED / DELAYED / OPEN 三值）。</para>
+    /// </summary>
+    [Fact]
+    public void BypassServices_NoLongerHardcodeDeliveryStatus()
+    {
+        foreach (var f in new[] { "DeliveryScheduleService.cs", "ProcessOutsourceOrderService.cs" })
+        {
+            var code = Source("Supply", f);
+            var sqlOnly = string.Join('\n', code.Split('\n')
+                .Where(l => !l.TrimStart().StartsWith("//", StringComparison.Ordinal)));
+
+            Assert.DoesNotContain("delivery_status=VALUES(delivery_status)", sqlOnly);
+            Assert.DoesNotContain("'CANCELLED'", sqlOnly);
+            Assert.DoesNotContain("'PARTIAL'", sqlOnly);
+        }
+    }
+
+    /// <summary>
+    /// 租户是唯一安全边界，不得被工厂替代。
+    /// <para>该链的源侧根本没有可用的数字工厂号（源表只有 varchar 的 Site/POSite，
+    /// 三个写入方都写字面量 1），所以 enrichment JOIN 只加租户、不加工厂 ——
+    /// 加了反而会在 factory_id 为 NULL 与 1 混存时把行丢掉。</para>
+    /// </summary>
+    [Fact]
+    public void StdBuild_EnrichmentJoins_DoNotSubstituteFactoryForTenant()
+    {
+        var block = StdInsertBlock();
+
+        // 不得出现"只按工厂不按租户"的写法。
+        Assert.DoesNotContain("m.factory_id = d.factory_id", block);
+        Assert.DoesNotContain("i.factory_id = d.factory_id", block);
+    }
+}

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

@@ -0,0 +1,249 @@
+using System.Reflection;
+using Admin.NET.Plugin.AiDOP.Service.S8.Rules.DataAccess.Providers;
+using Admin.NET.Plugin.AiDOP.Service.S8.Rules.Health;
+using Xunit;
+
+namespace Admin.NET.Plugin.AiDOP.Tests.S8;
+
+/// <summary>
+/// Rule01「当前快照」的发布契约。
+///
+/// <para><b>这一条守的是一个只在「合法的空」上才现形的缺陷。</b>
+/// Provider 原来按「最新的那一天」挑当前快照。源侧被清空的那一轮产出 0 行 DWD，
+/// 于是当天根本不存在 <c>stat_date</c> 分区，中台的同日淘汰也无行可删（它只能删「今天」），
+/// 「最新的那一天」就回落到<b>上一天</b>，把一批早已作废的行当成当前态返回。</para>
+///
+/// <para>本地沙箱实测：租户 838257212780613 源侧 10 → 0、STD 10 → 0、S3 跑批 SUCCESS，
+/// 而 DWD 当前快照仍返回前一日的 10 行。行数、状态、批次号全都「看起来正常」，
+/// 没有任何一处报错 —— 这类缺陷只能靠契约守住。</para>
+/// </summary>
+public class S8Rule01PublicationContractTests
+{
+    private static string Source(params string[] parts)
+    {
+        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(new[] { dir!.FullName, "Admin.NET.Plugin.AiDOP" }.Concat(parts).ToArray());
+        Assert.True(File.Exists(full), $"源码文件不存在，路径需同步更新：{full}");
+        return File.ReadAllText(full);
+    }
+
+    private static string S3Service() => Source("Supply", "S3MdpSyncTransformService.cs");
+
+    private static string Resolver() =>
+        Source("Service", "S8", "Rules", "Health", "S8MdpAuthorityHealthResolver.cs");
+
+    private static S8AuthoritySpec Rule01Spec()
+    {
+        var spec = new S8BusinessDatasetDefinitions()
+            .GetDefinitions()
+            .Single(d => d.DatasetCode == S8BusinessDatasetDefinitions.PurchaseDeliveryCode)
+            .AuthoritySpec;
+        Assert.NotNull(spec);
+        return spec!;
+    }
+
+    // ══════════════════ 生产者侧：发布证据 ══════════════════
+
+    /// <summary>
+    /// 发布证据只在<b>全量</b>路径产生。
+    ///
+    /// <para>增量路径（<c>RunInboundAsync</c>）同样以 SUCCESS 收尾，但它只处理变更行。
+    /// 把这样一个批次指认成「当前快照」，读者拿到的就是一个残缺集合 ——
+    /// 而且它会以「有发布证据、状态成功」的样子出现，比没有证据更危险。</para>
+    /// </summary>
+    [Fact]
+    public void Publication_IsProducedOnlyOnFullPath()
+    {
+        var code = S3Service();
+        var publishAt = code.IndexOf("result.Publication = await PublishCurrentSnapshotAsync", StringComparison.Ordinal);
+        Assert.True(publishAt > 0, "全量路径必须写发布证据");
+
+        // 发布语句必须落在 if (retireStale) 之内 —— 取其之前最近的一段代码验证。
+        var guardAt = code.LastIndexOf("if (retireStale)", publishAt, StringComparison.Ordinal);
+        Assert.True(guardAt > 0 && publishAt - guardAt < 400,
+            "发布必须被 retireStale 守卫包住，否则增量批次会被指认成当前快照");
+    }
+
+    /// <summary>
+    /// 发布证据必须在 DWD 的全部写入命令执行完之后才产生。
+    ///
+    /// <para>顺序颠倒的后果不是报错，是读者拿到一个还没灌完（或还混着上一批残留）的快照，
+    /// 而运行日志会把它记成一次正常发布。</para>
+    ///
+    /// <para><b>比对的是执行顺序，不是源码顺序</b>：同日淘汰是
+    /// <c>BuildDwdCommands</c> 这个迭代器 yield 出来的最后一条命令，
+    /// 而该方法在文件里排在 <c>BuildDwdAsync</c> 之后 —— 按源码位置比会得出相反的结论。
+    /// 真正的保证是：发布语句排在<b>命令循环之后</b>。淘汰在命令序列内部的位置，
+    /// 由 <c>S3PurchaseAuthorityContractTests.DwdBuild_RetiresSameDayStaleRows_AfterInsert</c> 守。</para>
+    /// </summary>
+    [Fact]
+    public void Publication_HappensAfterAllDwdCommandsExecuted()
+    {
+        var code = S3Service();
+        var bodyAt = code.IndexOf("private async Task<int> BuildDwdAsync(", StringComparison.Ordinal);
+        Assert.True(bodyAt > 0, "BuildDwdAsync 不存在");
+
+        var loopAt = code.IndexOf("foreach (var command in BuildDwdCommands(", bodyAt, StringComparison.Ordinal);
+        var shortageAt = code.IndexOf("InsertMaterialShortageAsync(batchId, now)", bodyAt, StringComparison.Ordinal);
+        var publishAt = code.IndexOf("result.Publication = await PublishCurrentSnapshotAsync", bodyAt, StringComparison.Ordinal);
+
+        Assert.True(loopAt > bodyAt, "未找到 DWD 命令执行循环");
+        Assert.True(publishAt > loopAt, "发布必须在 DWD 命令循环之后");
+        Assert.True(publishAt > shortageAt, "发布必须在 DWD 阶段的全部写入之后");
+    }
+
+    /// <summary>
+    /// 发布行数必须用<b>复核查询</b>得到，不能用 INSERT 的影响行数。
+    /// <para>ODKU 的影响行数（插入 1 / 更新 2 / 无变化 0）从来不等于「当前有几行」。</para>
+    /// </summary>
+    [Fact]
+    public void Publication_CountsRowsByVerificationQuery()
+    {
+        var code = S3Service();
+        Assert.Contains("SELECT COUNT(*) FROM dwd_supplier_delivery", code);
+        Assert.Contains("AND sync_batch_id=@BatchId", code);
+        Assert.Contains("AND stat_date=@StatDate", code);
+    }
+
+    /// <summary>
+    /// 运行摘要必须带上定位当前快照所需的全部字段，且用序列化器生成而非字符串拼接。
+    /// <para><c>currentRows = 0</c> 是合法值 —— 它正是「成功发布了一个空快照」的表达，
+    /// 下游据此把「正常的空」与「没发布成」区分开。</para>
+    /// </summary>
+    [Fact]
+    public void RunSummaryJson_CarriesPublicationEvidence()
+    {
+        var code = S3Service();
+
+        Assert.Contains("JsonSerializer.Serialize", code);
+        foreach (var field in new[] { "table =", "batchId =", "statDate =", "factoryId =", "currentRows =" })
+            Assert.Contains(field, code);
+
+        // 没发布时必须是 null，不能退化成一个 currentRows=0 的假证据 ——
+        // 那会把「没发布」伪装成「发布了空快照」，正好抹掉两者的区别。
+        Assert.Contains("publish = result.Publication is null", code);
+    }
+
+    // ══════════════════ 读者侧：Provider ══════════════════
+
+    /// <summary>
+    /// Provider 必须按发布证据取数，且<b>结构上</b>不存在回落到其它 stat_date 的路径。
+    /// </summary>
+    [Fact]
+    public void Provider_ReadsPublishedSnapshot_WithNoFallbackPath()
+    {
+        var sql = S8PurchaseDeliveryDataProvider.Sql;
+
+        Assert.Contains("WITH pub AS", sql);
+        Assert.Contains("'$.publish.batchId'", sql);
+        Assert.Contains("'$.publish.statDate'", sql);
+        Assert.Contains("d.sync_batch_id = p.batch_id", sql);
+
+        // 只认终态成功的生产运行：该表有大量永不回收的 RUNNING 孤儿行。
+        Assert.Contains("r.status = 'SUCCESS'", sql);
+        Assert.Contains("r.end_time IS NOT NULL", sql);
+
+        // 旧的「取最新那一天」写法必须彻底消失 —— 留着它就等于留着回落路径。
+        Assert.DoesNotContain("ORDER BY x.stat_date DESC", sql);
+        Assert.DoesNotContain("MAX(stat_date)", sql);
+        Assert.DoesNotContain("MAX(d.stat_date)", sql);
+    }
+
+    /// <summary>
+    /// 发布证据的租户过滤必须是<b>严格相等</b>。
+    /// <para><c>mdp_transform_run_log</c> 确实存在 <c>tenant_id = 0</c> 的平台行；
+    /// 写成 <c>OR tenant_id = 0</c> 会让一条平台行同时充当所有租户的发布证据。</para>
+    /// </summary>
+    [Fact]
+    public void Provider_PublicationLookup_IsStrictlyTenantScoped()
+    {
+        var sql = S8PurchaseDeliveryDataProvider.Sql;
+
+        Assert.Contains("r.tenant_id = @TenantId", sql);
+        Assert.DoesNotContain("r.tenant_id = 0", sql);
+        Assert.DoesNotContain("OR tenant_id = 0", sql);
+    }
+
+    /// <summary>
+    /// 生产者作业码只有一个来源：目录与取数各写一份必然漂移，
+    /// 而漂移的表现是「健康判定看着 A 作业、取数却认 B 作业的发布」，两边各自都不报错。
+    /// </summary>
+    [Fact]
+    public void ProducerJobCode_HasSingleSource()
+    {
+        Assert.Equal(
+            S8BusinessDatasetDefinitions.PurchaseDeliveryProducerJobCode,
+            Rule01Spec().ProducerJobCode);
+
+        var provider = Source("Service", "S8", "Rules", "DataAccess", "Providers",
+            "S8PurchaseDeliveryDataProvider.cs");
+        Assert.Contains("S8BusinessDatasetDefinitions.PurchaseDeliveryProducerJobCode", provider);
+        Assert.DoesNotContain("\"S3_MDP_SYNC_TRANSFORM\"", provider);
+    }
+
+    // ══════════════════ 健康判定侧 ══════════════════
+
+    /// <summary>Rule01 的当前态必须由发布证据指认，不得退回按时间兜底。</summary>
+    [Fact]
+    public void Rule01Spec_RequiresPublicationEvidence()
+    {
+        var spec = Rule01Spec();
+        Assert.True(spec.RequiresPublicationEvidence);
+        Assert.True(spec.ProducerTrusted);
+        Assert.Equal(S8AuthorityKind.CurrentStateMirror, spec.AuthorityKind);
+    }
+
+    /// <summary>
+    /// 需要发布证据时，「最近一次成功」只认<b>带证据</b>的那一次。
+    /// <para>否则一次增量成功就会把锚点从上一轮完整发布挪走，新鲜度与失败锚点全部算错。</para>
+    /// </summary>
+    [Fact]
+    public void Resolver_LastSuccess_FiltersOnPublicationWhenRequired()
+    {
+        var code = Resolver();
+
+        Assert.Contains("AND JSON_EXTRACT(r.summary_json, '$.publish.batchId') IS NOT NULL", code);
+        Assert.Contains("spec.RequiresPublicationEvidence", code);
+        Assert.Contains("S8AuthorityHealthReason.NoPublication", code);
+    }
+
+    /// <summary>
+    /// 「没有发布证据」与「压根没成功过」是两件事，必须有各自的原因码。
+    /// <para>结论同为 UNKNOWN、同样拦恢复；分开只是为了排查时一眼看出是哪一种。</para>
+    /// </summary>
+    [Fact]
+    public void NoPublicationReason_IsDistinct_AndBlocksRecovery()
+    {
+        Assert.NotEqual(S8AuthorityHealthReason.NoSuccessfulRun, S8AuthorityHealthReason.NoPublication);
+        Assert.NotEqual(S8AuthorityHealthReason.Healthy, S8AuthorityHealthReason.NoPublication);
+
+        var result = new S8AuthorityHealthResult
+        {
+            State = S8AuthorityHealthState.Unknown,
+            ReasonCode = S8AuthorityHealthReason.NoPublication
+        };
+        Assert.False(result.AllowsRecovery);
+    }
+
+    /// <summary>
+    /// 空快照<b>不是</b>一种新状态。
+    /// <para>成功发布的空快照就是 HEALTHY，恢复照常放行；缺的是证据时才判 UNKNOWN。
+    /// 词表里绝不能冒出 HEALTHY_EMPTY 这种第四态 —— 那会让每个读方各自决定它算不算健康。</para>
+    /// </summary>
+    [Fact]
+    public void StateVocabulary_HasNoHealthyEmpty()
+    {
+        var states = typeof(S8AuthorityHealthState)
+            .GetFields(BindingFlags.Public | BindingFlags.Static | BindingFlags.FlattenHierarchy)
+            .Where(f => f.IsLiteral && !f.IsInitOnly && f.FieldType == typeof(string))
+            .Select(f => (string)f.GetRawConstantValue()!)
+            .ToArray();
+
+        Assert.Equal(3, states.Length);
+        Assert.DoesNotContain("HEALTHY_EMPTY", states);
+    }
+}

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

@@ -60,9 +60,26 @@ public sealed class S8PurchaseDeliveryDataProvider : IS8MonitoringDataProvider
     /// 强行相等匹配会让每条规则恒返 0 行。客户运营模式为 1 Tenant = 1 Factory，
     /// 因此租户过滤已构成完整的数据隔离边界。
     ///
-    /// <b>快照收敛用 ORDER BY … DESC LIMIT 1 而非 MAX()</b>：<c>idx_stat_date</c> 为
-    /// <c>(tenant_id, stat_date)</c>，聚合写法在大租户上实测 143ms（扫 34,112 条索引项），
-    /// 改为反向索引扫描后 2.7ms（约 53×）。语义等价：无数据时同样返回 NULL → 外层 0 行。
+    /// <b>当前快照由发布证据指认，不再按时间挑</b>。这里曾经写的是
+    /// <c>stat_date = (SELECT stat_date … ORDER BY stat_date DESC LIMIT 1)</c>，
+    /// 即「取最新的那一天」。它在**合法的空**面前会静默失效：
+    /// 源侧被清空的那一轮产出 0 行 DWD，当天根本不存在 <c>stat_date</c> 分区，
+    /// 中台的同日淘汰也无行可删（它只能删「今天」），于是「最新的那一天」回落到<b>上一天</b>，
+    /// 把一批早已作废的行当成当前态返回。实测：租户 838257212780613 源侧清零、
+    /// STD 归 0、S3 跑批 SUCCESS，而本 Provider 仍返回前一日的 10 行。
+    ///
+    /// 改为与 <c>mdp_transform_run_log</c> 的发布证据 JOIN，读者与生产者对「哪一批是当前」
+    /// 达成同一个事实。三种情形都由这一条 SQL 直接给出正确答案，<b>没有兜底分支</b>：
+    /// <list type="bullet">
+    ///   <item><c>currentRows &gt; 0</c> → JOIN 命中该批次，返回该批次的行；</item>
+    ///   <item><c>currentRows = 0</c>（合法空）→ 没有行带这个批次号，自然 0 行；</item>
+    ///   <item>没有发布证据 → CTE 为空，JOIN 无从命中，同样 0 行（fail-safe）。</item>
+    /// </list>
+    /// 关键在于：<b>任何一种情形都不会去看别的 stat_date</b> —— 回落这条路径在结构上不存在。
+    ///
+    /// 同时按 <c>stat_date</c> 与 <c>sync_batch_id</c> 两个条件收窄：前者让查询仍能走
+    /// <c>idx_stat_date(tenant_id, stat_date)</c>，后者才是身份判据。只有 batch 相等时，
+    /// 单靠 stat_date 无法排除同日残留的其它批次。
     ///
     /// <c>po_no</c> / <c>po_line</c> 均为 NOT NULL，故只判空串；两者共同构成对象身份，
     /// 任一为空会产生退化的 "#" 身份，必须排除。
@@ -87,6 +104,18 @@ public sealed class S8PurchaseDeliveryDataProvider : IS8MonitoringDataProvider
     /// </remarks>
     internal const string Sql =
         """
+        WITH pub AS (
+            SELECT JSON_UNQUOTE(JSON_EXTRACT(r.summary_json, '$.publish.batchId'))  AS batch_id,
+                   JSON_UNQUOTE(JSON_EXTRACT(r.summary_json, '$.publish.statDate')) AS stat_date
+            FROM mdp_transform_run_log r
+            WHERE r.job_code = @ProducerJobCode
+              AND r.tenant_id = @TenantId
+              AND r.status = 'SUCCESS'
+              AND r.end_time IS NOT NULL
+              AND JSON_EXTRACT(r.summary_json, '$.publish.batchId') IS NOT NULL
+            ORDER BY r.start_time DESC, r.id DESC
+            LIMIT 1
+        )
         SELECT
             CONCAT(d.po_no, '#', d.po_line)                        AS SourceObjectId,
             d.po_no                                                AS RelatedObjectCode,
@@ -108,12 +137,10 @@ public sealed class S8PurchaseDeliveryDataProvider : IS8MonitoringDataProvider
             d.risk_level     AS RiskLevel,
             d.stat_date      AS StatDate
         FROM dwd_supplier_delivery d
+        JOIN pub p
+          ON d.stat_date = p.stat_date
+         AND d.sync_batch_id = p.batch_id
         WHERE d.tenant_id = @TenantId
-          AND d.stat_date = (SELECT x.stat_date
-                             FROM dwd_supplier_delivery x
-                             WHERE x.tenant_id = @TenantId
-                             ORDER BY x.stat_date DESC
-                             LIMIT 1)
           AND d.po_no <> ''
           AND d.po_line <> ''
         """;
@@ -144,7 +171,12 @@ public sealed class S8PurchaseDeliveryDataProvider : IS8MonitoringDataProvider
 
         var rows = await _db.Ado.SqlQueryAsync<PurchaseDeliveryRow>(
             Sql,
-            new SugarParameter[] { new("@TenantId", request.TenantId) });
+            new SugarParameter[]
+            {
+                new("@TenantId", request.TenantId),
+                // 生产者作业码取自数据集定义，不在这里另写一份字面量 —— 目录与取数各写一份必然漂移。
+                new("@ProducerJobCode", S8BusinessDatasetDefinitions.PurchaseDeliveryProducerJobCode)
+            });
 
         var budget = ResolveSourceObjectIdBudget(request);
         var mapped = new List<S8MonitoringRow>(rows.Count);

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

@@ -31,6 +31,16 @@ public static class S8AuthorityHealthReason
     /// <summary>回溯窗口内没有任何一次成功的生产运行。含「从未同步」与「停摆超过回溯窗口」两种。</summary>
     public const string NoSuccessfulRun = "NO_SUCCESSFUL_RUN";
 
+    /// <summary>
+    /// 回溯窗口内有成功运行，但没有一次带着<b>发布证据</b>（<c>summary_json.publish</c>）。
+    ///
+    /// <para>只对 <c>RequiresPublicationEvidence</c> 的数据集适用。它与
+    /// <see cref="NoSuccessfulRun"/> 是两件事：跑批成功了，但运行日志证明不了
+    /// 「当前快照」是哪一批 —— 这时任何按时间兜底挑出来的快照都可能是上一轮的。
+    /// 保守判 UNKNOWN，让 Wave2 闸门拦下恢复。</para>
+    /// </summary>
+    public const string NoPublication = "NO_PUBLICATION";
+
     /// <summary>最近一次成功距今超过 StaleWindow。</summary>
     public const string Stale = "STALE";
 

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

@@ -62,6 +62,22 @@ public sealed class S8AuthoritySpec
     /// <summary>ProducerTrusted 为 false 时的原因码，见 <see cref="S8AuthorityHealthReason"/>。</summary>
     public string UntrustedReasonCode { get; init; } = S8AuthorityHealthReason.Rule01MultiWriterUntrusted;
 
+    /// <summary>
+    /// 该数据集的「当前态」是否必须由<b>发布证据</b>（<c>summary_json.publish</c>）指认。
+    ///
+    /// <para><b>为什么需要这个开关</b>：<c>CURRENT_STATE_MIRROR</c> 类的 Authority 没有
+    /// <c>is_current_flag</c> 这样的物理翻牌列，读者只能靠某种「最新」约定去挑当前快照。
+    /// 而「最新」在<b>合法的空</b>面前会失效 —— 源侧被清空的那一轮不产出任何行，
+    /// 于是当天根本不存在分区，按 <c>MAX(stat_date)</c> 之类的约定会**静默回落到上一天**，
+    /// 把一批早已作废的行当成当前态。实测：租户 838257212780613 源侧清零、STD 归 0、
+    /// 跑批 SUCCESS，DWD 当前快照仍返回前一日的 10 行。</para>
+    ///
+    /// <para>置 true 后：当前态由「最近一次带 publish 证据的成功运行」唯一指认，
+    /// <c>publish.currentRows = 0</c> 是<b>合法且健康</b>的空；没有任何发布证据则判
+    /// <see cref="S8AuthorityHealthReason.NoPublication"/> → UNKNOWN，绝不按时间兜底。</para>
+    /// </summary>
+    public bool RequiresPublicationEvidence { get; init; }
+
     // ── 以下仅 AuthorityKind = PUBLISHED_SNAPSHOT 时使用 ──────────────────────────────
     // 声明在这里而不是写死在 Resolver 里，是为了让 Resolver 保持通用：
     // 它只按声明去查，内部不出现 switch (datasetCode) 或 switch (tableName)。

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

@@ -61,7 +61,28 @@ public class S8MdpAuthorityHealthResolver : IS8AuthorityHealthResolver, ITransie
             }
 
             var lookbackFrom = observedAt - spec.LookbackWindow;
-            var lastSuccess = await QueryLastSuccessAsync(tenantId, spec.ProducerJobCode, lookbackFrom);
+            var lastSuccess = await QueryLastSuccessAsync(
+                tenantId, spec.ProducerJobCode, lookbackFrom, spec.RequiresPublicationEvidence);
+
+            // 需要发布证据却一条都没有：跑批可能成功过，但没人能指认「当前快照」是哪一批。
+            // 这与「压根没成功过」是两件事，故给独立原因码；结论同为 UNKNOWN、拦恢复。
+            // 判定放在这里而不是纯函数里，是因为「有没有 publish」是一个查询事实，
+            // 而 S8AuthorityHealthEvaluator 的契约是不读行、不读库。
+            if (spec.RequiresPublicationEvidence && lastSuccess is null)
+            {
+                var anySuccess = await QueryLastSuccessAsync(
+                    tenantId, spec.ProducerJobCode, lookbackFrom, requirePublication: false);
+                if (anySuccess is not null)
+                {
+                    return new S8AuthorityHealthResult
+                    {
+                        DatasetCode = datasetCode, TenantId = tenantId, ObservedAt = observedAt,
+                        State = S8AuthorityHealthState.Unknown,
+                        ReasonCode = S8AuthorityHealthReason.NoPublication,
+                        Reason = "回溯窗口内有成功运行，但没有一次带发布证据，无法指认当前快照"
+                    };
+                }
+            }
 
             // 锚点：有成功则以其结束时刻为界，否则以回溯窗口起点为界。
             var anchor = lastSuccess?.EndTime ?? lookbackFrom;
@@ -120,10 +141,17 @@ public class S8MdpAuthorityHealthResolver : IS8AuthorityHealthResolver, ITransie
     ///     但新鲜度用取到那行的 <c>end_time</c> 计算 —— 那才是 Authority 真正变成当前态的时刻。</item>
     /// </list>
     /// </summary>
-    private async Task<RunRow?> QueryLastSuccessAsync(long tenantId, string jobCode, DateTime lookbackFrom)
+    /// <param name="requirePublication">
+    /// true 时只认<b>带发布证据</b>的成功运行。用于 <c>RequiresPublicationEvidence</c> 的数据集：
+    /// 增量跑批（<c>RunInboundAsync</c>）同样以 SUCCESS 收尾，但它只处理变更行、不发布当前态，
+    /// 因此不带 publish。若不加这层过滤，一次增量成功就会把「最近一次成功」的锚点从上一轮
+    /// 完整发布挪走，新鲜度与失败锚点都会算错。
+    /// </param>
+    private async Task<RunRow?> QueryLastSuccessAsync(
+        long tenantId, string jobCode, DateTime lookbackFrom, bool requirePublication)
     {
-        const string sql =
-            """
+        var sql =
+            $"""
             SELECT r.batch_id AS BatchId, r.status AS Status, r.start_time AS StartTime, r.end_time AS EndTime,
                    JSON_UNQUOTE(JSON_EXTRACT(r.summary_json, '$.publish.batchId'))     AS PublishedBatchId,
                    JSON_EXTRACT(r.summary_json, '$.publish.currentRows')               AS PublishedCurrentRows
@@ -133,6 +161,9 @@ public class S8MdpAuthorityHealthResolver : IS8AuthorityHealthResolver, ITransie
               AND r.status = 'SUCCESS'
               AND r.end_time IS NOT NULL
               AND r.start_time >= @LookbackFrom
+              {(requirePublication
+                  ? "AND JSON_EXTRACT(r.summary_json, '$.publish.batchId') IS NOT NULL"
+                  : string.Empty)}
             ORDER BY r.start_time DESC, r.id DESC
             LIMIT 1
             """;

server/Plugins/Admin.NET.Plugin.AiDOP/Supply/S3MdpSyncTransformService.cs

@@ -58,11 +58,11 @@ public class S3MdpSyncTransformService : ITransient
                 await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Staging, 1, 35, "拉取源数据完成", result.StageRows, ModuleRebuildStages.Staging));
 
                 await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Standard, 2, 35, "正在标准化数据"));
-                result.StandardRows = await TransformStandardAsync(batchId, now, cancellationToken);
+                result.StandardRows = await TransformStandardAsync(batchId, now, retireStale: true, cancellationToken);
                 await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Standard, 2, 50, "标准化数据完成", result.StandardRows, ModuleRebuildStages.Standard));
 
                 await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Dwd, 3, 50, "正在生成 DWD 明细"));
-                result.DwdRows = await BuildDwdAsync(batchId, now, cancellationToken);
+                result.DwdRows = await BuildDwdAsync(batchId, now, retireStale: true, result, cancellationToken);
                 await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Dwd, 3, 65, "生成 DWD 明细完成", result.DwdRows, ModuleRebuildStages.Dwd));
 
                 await ReportAsync(reportProgress, new ModuleProgressUpdate(ModuleRebuildStages.Kpi, 4, 65, "正在重算 KPI"));
@@ -157,8 +157,8 @@ public class S3MdpSyncTransformService : ITransient
             var codes = entityCodes?.ToList() ?? S3MdpEntityConfig.All.Select(x => x.EntityCode).ToList();
             result.StageRows = await _stagingPuller.PullEntitiesAsync(
                 codes, batchId, tenantId, fullRefresh, "S3_MDP_INBOUND", cancellationToken);
-            result.StandardRows = await TransformStandardAsync(batchId, now, cancellationToken);
-            result.DwdRows = await BuildDwdAsync(batchId, now, cancellationToken);
+            result.StandardRows = await TransformStandardAsync(batchId, now, retireStale: false, cancellationToken);
+            result.DwdRows = await BuildDwdAsync(batchId, now, retireStale: false, result, cancellationToken);
             result.KpiRows = await BuildS3KpiValuesAsync(now, cancellationToken);
             result.AtomicRows = await _atomicBuild.BuildSupplyPurchaseDomainForAllDatesAsync(
                 _runScope.TenantId, _runScope.FactoryId, batchId, cancellationToken);
@@ -244,10 +244,11 @@ public class S3MdpSyncTransformService : ITransient
         }
     }
 
-    private async Task<int> TransformStandardAsync(string batchId, DateTime now, CancellationToken cancellationToken)
+    private async Task<int> TransformStandardAsync(
+        string batchId, DateTime now, bool retireStale, CancellationToken cancellationToken)
     {
         var total = 0;
-        foreach (var command in BuildStandardCommands(batchId, now))
+        foreach (var command in BuildStandardCommands(batchId, now, retireStale))
         {
             cancellationToken.ThrowIfCancellationRequested();
             total += await _db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
@@ -255,10 +256,12 @@ public class S3MdpSyncTransformService : ITransient
         return total;
     }
 
-    private async Task<int> BuildDwdAsync(string batchId, DateTime now, CancellationToken cancellationToken)
+    private async Task<int> BuildDwdAsync(
+        string batchId, DateTime now, bool retireStale, S3MdpSyncTransformResult result,
+        CancellationToken cancellationToken)
     {
         var total = 0;
-        foreach (var command in BuildDwdCommands(batchId, now))
+        foreach (var command in BuildDwdCommands(batchId, now, retireStale))
         {
             cancellationToken.ThrowIfCancellationRequested();
             total += await _db.Ado.ExecuteCommandAsync(command.Sql, command.Parameters);
@@ -266,9 +269,60 @@ public class S3MdpSyncTransformService : ITransient
         total += await UpsertMaterialReadinessDwdAsync(batchId, now);
         await DeleteCurrentShortageAsync(now.Date);
         total += await InsertMaterialShortageAsync(batchId, now);
+
+        // 发布证据：必须在 DWD 写入**与同日淘汰**都完成之后。
+        // 只有全量路径发布 —— 增量批次只含变更行，把它指认成「当前快照」会让读者看到残缺集合。
+        if (retireStale)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            result.Publication = await PublishCurrentSnapshotAsync(batchId, now);
+        }
+
         return total;
     }
 
+    /// <summary>
+    /// 复核并记录本轮发布的当前快照，供 S8 Rule01 精确指认「当前态是哪一批」。
+    ///
+    /// <para><b>为什么不能靠 MAX(stat_date) 之类的时间约定</b>：源侧被合法清空的那一轮
+    /// 产出 0 行 DWD，于是当天根本不存在 <c>stat_date</c> 分区，同日淘汰也无行可删
+    /// （它只能删「今天」）。读者按最新日期兜底就会**静默回落到上一天**，把一批早已作废的行
+    /// 当成当前态。实测：租户 838257212780613 源侧清零、STD 归 0、跑批 SUCCESS，
+    /// 而 DWD 当前快照仍返回前一日的 10 行。</para>
+    ///
+    /// <para>返回的 <c>CurrentRows = 0</c> 是<b>合法结果</b> —— 成功发布了一个空快照。
+    /// 把它记进运行日志，下游才能把「正常的空」与「没发布成」区分开：前者健康且允许恢复，
+    /// 后者无证据、判 UNKNOWN 并拦下恢复。</para>
+    ///
+    /// <para>行数用<b>复核查询</b>得到，不用 INSERT 的影响行数：ODKU 的影响行数
+    /// （插入计 1、更新计 2、无变化计 0）从来不等于「当前有几行」。</para>
+    /// </summary>
+    private async Task<S3MdpSnapshotPublication> PublishCurrentSnapshotAsync(string batchId, DateTime now)
+    {
+        var scope = RequireScope();
+        var currentRows = await _db.Ado.GetIntAsync(
+            """
+            SELECT COUNT(*) FROM dwd_supplier_delivery
+             WHERE tenant_id=@TenantId
+               AND COALESCE(NULLIF(factory_id,0),1)=@FactoryId
+               AND stat_date=@StatDate
+               AND sync_batch_id=@BatchId
+            """,
+            new SugarParameter("@TenantId", scope.TenantId),
+            new SugarParameter("@FactoryId", scope.FactoryId),
+            new SugarParameter("@StatDate", now.Date),
+            new SugarParameter("@BatchId", batchId));
+
+        return new S3MdpSnapshotPublication
+        {
+            Table = "dwd_supplier_delivery",
+            BatchId = batchId,
+            StatDate = now.Date,
+            FactoryId = scope.FactoryId,
+            CurrentRows = currentRows
+        };
+    }
+
     private async Task<int> BuildS3KpiValuesAsync(DateTime now, CancellationToken cancellationToken)
     {
         var affected = 0;
@@ -564,7 +618,7 @@ public class S3MdpSyncTransformService : ITransient
             new SugarParameter("@TargetResolvedAt", snap.ResolvedAt));
     }
 
-    private IEnumerable<S3MdpSqlCommand> BuildStandardCommands(string batchId, DateTime now)
+    private IEnumerable<S3MdpSqlCommand> BuildStandardCommands(string batchId, DateTime now, bool retireStale)
     {
         yield return Cmd(
             """
@@ -712,11 +766,75 @@ public class S3MdpSyncTransformService : ITransient
                    JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.WorkOrd')),
                    d.source_biz_key, @BatchId, @Now
             FROM mdp_stg_purchase_order d
-            LEFT JOIN mdp_stg_purchase_order m ON m.source_table='PurOrdMaster'
-                AND JSON_UNQUOTE(JSON_EXTRACT(m.raw_data,'$.PurOrd')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd'))
-            LEFT JOIN mdp_stg_item i ON i.source_table='ItemMaster'
-                AND JSON_UNQUOTE(JSON_EXTRACT(i.raw_data,'$.ItemNum')) = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum'))
-            WHERE d.source_table='PurOrdDetail' AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd')), '') <> ''
+            -- ── enrichment JOIN 的两个约束，缺一不可 ────────────────────────────────────
+            --
+            -- ① 必须自带租户谓词。MdpSqlScope.InjectTenantFactory 只把 WHERE 改写成
+            --    `WHERE d.tenant_id=@TenantId AND ...`（实现是 Regex.Replace(sql, @"\bWHERE\b", ...)），
+            --    它**结构上够不到 JOIN...ON**。少了它，采购单号与物料号就成了跨租户的字符串匹配：
+            --    实测 mdp_stg_item 里 ItemMaster 的 item_code 有 5529 组跨租户重名（最多横跨 3 个租户）。
+            --
+            -- ② 必须按 (租户, 键) 去重取最新一行。贴源层同时存在两代 source_biz_key
+            --    （旧 `Domain:PurOrd:Line` 与新的裸 RecID，因 uk_source_key 含 source_system 与
+            --    biz_key，两代永远不会互相 ODKU 覆盖），同一个 PurOrd/ItemNum 因此有多行。
+            --    不去重则同租户内仍会扇出 —— 单租户实测 870/220 = 3.95 倍。
+            --
+            -- 扇出被 uk_po_line(tenant_id,po_no,po_line) 的 ODKU 吸收，所以症状不是行数膨胀，
+            -- 而是 item_name / supplier_code / order_date / buyer 这些值由 MySQL 任取一行决定。
+            -- 这里不按 @BatchId 收窄 enrichment：增量路径下未变更的主数据不在本批次里，
+            -- 收窄会把它们打成 NULL。取最新一行才是「当前已知主数据」的正确语义。
+            -- 去重用 ROW_NUMBER 而不是 GROUP_CONCAT+SUBSTRING_INDEX：raw_data 是 JSON 大字段，
+            -- 而 group_concat_max_len 实测仅 1024，拼接必然截断且首元素本身就可能超限。
+            --
+            -- ③ 派生表里的每个 WHERE 都必须**自带** @TenantId。这不是冗余，是在关掉注入器：
+            --    Inject 只在 WHERE 后 240 字符内出现 @TenantId 时才跳过，否则一律改写。
+            --    而 ResolveSourcePrefix 沿 SQL 反向找最近的同层 `FROM x 别名` 来定别名前缀，
+            --    派生表外层是 `FROM (` —— 别名正则匹配不到括号，于是回退成**空前缀**，
+            --    生成裸 `factory_id`；派生表的 SELECT 列表里没有这一列，
+            --    整轮转换直接报 Unknown column 'factory_id' 失败（本地沙箱实测复现）。
+            --    （此处刻意不写出该报错的完整英文原文：注入器对 SQL 注释同样生效，
+            --      原文里的 where 一词会被当成关键字改写，徒增排查噪音。）
+            --    自带 @TenantId 后注入被跳过，作用域完全由这里的字面 SQL 决定。
+            --    只按租户不按工厂：驱动行 d 已被注入器限定到 (@TenantId,@FactoryId)，
+            --    而主数据（PurOrdMaster / ItemMaster）的 factory_id 未必与明细一致，
+            --    多加一层工厂过滤会把 supplier_code / item_name 静默打成 NULL。
+            LEFT JOIN (
+                SELECT x.tenant_id, x.pur_ord, x.raw_data FROM (
+                    SELECT x0.tenant_id, x0.raw_data,
+                           JSON_UNQUOTE(JSON_EXTRACT(x0.raw_data,'$.PurOrd')) AS pur_ord,
+                           ROW_NUMBER() OVER (
+                               PARTITION BY x0.tenant_id, JSON_UNQUOTE(JSON_EXTRACT(x0.raw_data,'$.PurOrd'))
+                               ORDER BY x0.sync_time DESC, x0.id DESC) AS rn
+                      FROM mdp_stg_purchase_order x0
+                     WHERE x0.tenant_id=@TenantId AND x0.source_table='PurOrdMaster') x
+                 WHERE x.tenant_id=@TenantId AND x.rn=1
+            ) m ON m.tenant_id = d.tenant_id
+                AND m.pur_ord = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd'))
+            LEFT JOIN (
+                SELECT y.tenant_id, y.item_num, y.raw_data FROM (
+                    SELECT y0.tenant_id, y0.raw_data,
+                           JSON_UNQUOTE(JSON_EXTRACT(y0.raw_data,'$.ItemNum')) AS item_num,
+                           ROW_NUMBER() OVER (
+                               PARTITION BY y0.tenant_id, JSON_UNQUOTE(JSON_EXTRACT(y0.raw_data,'$.ItemNum'))
+                               ORDER BY y0.sync_time DESC, y0.id DESC) AS rn
+                      FROM mdp_stg_item y0
+                     WHERE y0.tenant_id=@TenantId AND y0.source_table='ItemMaster') y
+                 WHERE y.tenant_id=@TenantId AND y.rn=1
+            ) i ON i.tenant_id = d.tenant_id
+                AND i.item_num = JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.ItemNum'))
+            -- 驱动行按本轮批次收窄：staging 与本转换共用同一个 @BatchId
+            -- （RunFullAsync 生成一次，依次传给 SyncStagingAsync / TransformStandardAsync / BuildDwdAsync），
+            -- 因此「本批次里的行」就是「本轮源侧的行」，比"最新批次"子查询更确定。
+            -- 少了它，本 INSERT 会把整张贴源历史重新物化，并把死行盖上最新批次戳
+            -- —— 实测租户 797 因此在 STD 沉淀了 336 行源侧已不存在的采购订单行（518 行中的 64.9%）。
+            -- 驱动行的租户/工厂作用域在这里写死，不再交给注入器。
+            -- 上面两个派生表插进 `FROM ... d` 与本筛选子句之间以后，ResolveSourcePrefix 反向找
+            -- 最近的同层 `FROM 表 别名` 时会先撞上派生表的 `FROM (`，别名正则匹配不到括号 →
+            -- 回退成空前缀，生成裸 `tenant_id` —— 而派生表 m/i 也各自暴露 tenant_id，
+            -- 该列随即变成歧义列。本 SQL 因此必须自带前缀，且注入对它应当是恒等变换。
+            -- （注释里刻意避开 WHERE 一词：注入器对注释同样生效，只有紧跟 @TenantId 时才跳过。）
+            WHERE d.tenant_id=@TenantId AND COALESCE(NULLIF(d.factory_id,0),1)=@FactoryId
+              AND d.source_table='PurOrdDetail' AND d.sync_batch_id=@BatchId
+              AND IFNULL(JSON_UNQUOTE(JSON_EXTRACT(d.raw_data,'$.PurOrd')), '') <> ''
             ON DUPLICATE KEY UPDATE po_type=VALUES(po_type), supplier_code=VALUES(supplier_code), item_code=VALUES(item_code),
                 item_name=VALUES(item_name), order_qty=VALUES(order_qty), received_qty=VALUES(received_qty), returned_qty=VALUES(returned_qty),
                 due_date=VALUES(due_date), need_date=VALUES(need_date), order_date=VALUES(order_date), status=VALUES(status),
@@ -724,6 +842,35 @@ public class S3MdpSyncTransformService : ITransient
                 update_time=CURRENT_TIMESTAMP
             """, batchId, now);
 
+        // ── 每轮淘汰：让标准层重新成为源的镜像 ────────────────────────────────────────
+        //
+        // 只加批次过滤是不够的。过滤只让死行「不再被刷新」，而运行期链路里没有任何 DELETE，
+        // 死行会带着旧批次号永远留下 —— 那只是把问题从「认不出」变成「认得出但仍在」。
+        //
+        // 判据：本轮 INSERT 已给所有存活行盖上 @BatchId，
+        // 因此该租户下 sync_batch_id 不等于 @BatchId 的行，就是本轮源侧已不存在的行。
+        //
+        // 安全闸门用 mdp_sync_log 而不是「贴源层有没有行」：后者对「源侧真的清空了」的合法租户
+        // 会永久拒绝淘汰、把幽灵行锁死；前者精确表达「本轮拉取对该租户成功了」——
+        // 实测 0 行租户同样留下 SUCCESS 记录（如 1300000000777：90 次运行 / 90 次 0 行 / 90 次 SUCCESS），
+        // 因此源侧合法清空的租户能被正确清干净。
+        //
+        // 仅在全量路径执行：增量批次只含变更行，按它淘汰会把未变更的存量行全删掉。
+        if (retireStale)
+        {
+            yield return Cmd(
+                """
+                DELETE s FROM mdp_std_purchase_order s
+                 WHERE s.tenant_id=@TenantId
+                   AND COALESCE(NULLIF(s.factory_id,0),1)=@FactoryId
+                   AND IFNULL(s.sync_batch_id,'') <> @BatchId
+                   AND EXISTS (SELECT 1 FROM mdp_sync_log l
+                                WHERE l.tenant_id=@TenantId
+                                  AND l.sync_batch_id=@BatchId
+                                  AND l.status='SUCCESS')
+                """, batchId, now);
+        }
+
         yield return Cmd(
             """
             INSERT INTO mdp_std_delivery_schedule
@@ -832,7 +979,7 @@ public class S3MdpSyncTransformService : ITransient
             """, batchId, now);
     }
 
-    private IEnumerable<S3MdpSqlCommand> BuildDwdCommands(string batchId, DateTime now)
+    private IEnumerable<S3MdpSqlCommand> BuildDwdCommands(string batchId, DateTime now, bool retireStale)
     {
         yield return Cmd(
             """
@@ -895,6 +1042,31 @@ public class S3MdpSyncTransformService : ITransient
                 risk_level=VALUES(risk_level), sync_batch_id=VALUES(sync_batch_id), calc_time=VALUES(calc_time), update_time=CURRENT_TIMESTAMP
             """, batchId, now);
 
+        // ── 同日快照淘汰：让 MAX(stat_date) 成为「单一批次」而不是「某一天」 ──────────────
+        //
+        // Rule01 的 Provider 取 stat_date 最大的那一天，然后读该天全部行。
+        // 但 DWD 是按 (tenant, factory, po_no, po_line, stat_date) 唯一键的 upsert，
+        // 同一天被跑两轮时，前一轮产出而后一轮没产出的行会**保留旧批次号留在同一天**——
+        // 于是 MAX(stat_date) 取到的是「半个旧批 + 半个新批」。实测已发生 3 次
+        // （如 2026-06-07 的 178 行 + 7 行分属两个批次）。
+        //
+        // 本轮 INSERT 已给所有应存在的行盖上 @BatchId，故同日里批次号不等于 @BatchId 的行
+        // 要么是本轮源侧已不存在的，要么是旁路写入方留下的（它们用 S3_DELIVERY_ / S3_OUTSOURCE_ 前缀）。
+        //
+        // 放在 INSERT **之后**而不是之前：先删后插会让快照短暂为空，
+        // 而这里没有事务，并发的 Rule01 取数会读到一个空快照并把全部在办异常判成已恢复。
+        if (retireStale)
+        {
+            yield return Cmd(
+                """
+                DELETE w FROM dwd_supplier_delivery w
+                 WHERE w.tenant_id=@TenantId
+                   AND COALESCE(NULLIF(w.factory_id,0),1)=@FactoryId
+                   AND w.stat_date=@StatDate
+                   AND IFNULL(w.sync_batch_id,'') <> @BatchId
+                """, batchId, now);
+        }
+
         yield return Cmd("DELETE FROM dwd_supplier_risk WHERE stat_date=@StatDate", batchId, now);
         yield return Cmd(
             """
@@ -1176,9 +1348,30 @@ public class S3MdpSyncTransformService : ITransient
         return string.IsNullOrWhiteSpace(triggerType) ? "AUTO" : triggerType.Trim().ToUpperInvariant();
     }
 
+    /// <summary>
+    /// 运行摘要。改用 JsonSerializer 而非字符串插值：batchId 今天不可能含引号，
+    /// 但拼 JSON 是会被后来者继承的脆弱写法，且这里要加的 publish 是嵌套对象。
+    /// </summary>
     private static string BuildRunSummaryJson(S3MdpSyncTransformResult result)
     {
-        return $$"""{"batchId":"{{result.BatchId}}","stageRows":{{result.StageRows}},"standardRows":{{result.StandardRows}},"dwdRows":{{result.DwdRows}},"kpiRows":{{result.KpiRows}}}""";
+        return System.Text.Json.JsonSerializer.Serialize(new
+        {
+            batchId = result.BatchId,
+            stageRows = result.StageRows,
+            standardRows = result.StandardRows,
+            dwdRows = result.DwdRows,
+            kpiRows = result.KpiRows,
+            publish = result.Publication is null
+                ? null
+                : new
+                {
+                    table = result.Publication.Table,
+                    batchId = result.Publication.BatchId,
+                    statDate = result.Publication.StatDate.ToString("yyyy-MM-dd"),
+                    factoryId = result.Publication.FactoryId,
+                    currentRows = result.Publication.CurrentRows
+                }
+        });
     }
 
     private static string ResolveKpiValueTable(int metricLevel)
@@ -1270,6 +1463,35 @@ public sealed class S3MdpSyncTransformResult
     public int DwdRows { get; set; }
     public int KpiRows { get; set; }
     public int AtomicRows { get; set; }
+
+    /// <summary>
+    /// 本轮发布的当前快照。<b>null 表示本轮没有发布</b>（增量路径，或全量路径中途抛错）。
+    /// <para>失败方向是保守的：抛错时调用方写 FAILED，本字段保持 null，
+    /// 于是运行日志里不会出现「成功了但发布证据是半批」的记录。</para>
+    /// </summary>
+    public S3MdpSnapshotPublication? Publication { get; set; }
+}
+
+/// <summary>
+/// 一次当前快照发布的结果。与 <c>status='SUCCESS'</c> 写在同一条 UPDATE 里
+/// （见 <c>MarkTransformRunSuccessAsync</c>），因此不存在「已成功但没有发布证据」的中间态。
+/// </summary>
+public sealed class S3MdpSnapshotPublication
+{
+    /// <summary>被发布为当前快照的表。</summary>
+    public string Table { get; set; } = string.Empty;
+
+    /// <summary>本轮批次号。读者据此确认「当前快照」出自哪一次运行。</summary>
+    public string BatchId { get; set; } = string.Empty;
+
+    /// <summary>本轮快照所在的 stat_date。与 BatchId 一起构成读者的精确定位键。</summary>
+    public DateTime StatDate { get; set; }
+
+    /// <summary>发布作用域的工厂号。运行日志表本身没有工厂列，只能由这里承载。</summary>
+    public long FactoryId { get; set; }
+
+    /// <summary>发布后该批次的当前行数。<b>0 是合法值</b>，表示成功发布了一个空快照。</summary>
+    public int CurrentRows { get; set; }
 }
 
 public sealed class S3MaterialRefreshResult