refactor(mdp): 收敛采购交付 Authority 为单一中台写入

dwd_supplier_delivery 是 Rule01 的 Authority，此前除中台转换外还有两个业务 API 直写，
且都不写 mdp_transform_run_log——运行日志因此无法证明表里最终的数据出自谁。

ProcessOutsourceOrderService 把 delivery_status 硬编码为 'OPEN' 并放进 ODKU，
一次建单就能把中台刚判定的 COMPLETED / DELAYED 无条件翻回 OPEN；
它还另写一份 mdp_std_purchase_order，与中台的标准层构建互相覆盖。
DeliveryScheduleService 则写出 CANCELLED / CLOSED / PARTIAL 三个没有任何读方认识的值。
实测租户 797403760988229 的 2026-05-16「最新快照」共 2 行，两行都出自旁路写入方——
旁路足以凭空伪造一个当前快照。

移除这两处对 dwd_supplier_delivery 的直写，以及 ProcessOutsourceOrderService
对 mdp_std_purchase_order 的直写；两者各自的源单据事务与
mdp_std_delivery_schedule / mdp_std_process_outsource_order 写入保持不变。

运行期写入方由此只剩 S3MdpSyncTransformService，它写运行日志，
表的状态因此可以被运行日志完整解释——这正是 ProducerTrusted 的前提。

据此解冻 Rule01 的 ProducerTrusted（由 false 改为 true）。冻结时列出的解冻条件已全部达成：
单一写入方、membership 每轮淘汰、enrichment 租户谓词，
并补上冻结时尚未要求的第四项——空快照由发布证据指认，不再按时间兜底。

本地沙箱实测：4 个租户的当前快照共 473 行，全部由 S3 全量批次写入，非中台来源 0 行；
连续三轮跑批中 COMPLETED 计数稳定不变，无一行被翻回 OPEN。

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

@@ -18,6 +18,14 @@ public sealed class S8BusinessDatasetDefinitions : IS8DatasetDefinitionSource
     /// <summary>采购交付（订单行级）数据集编码。Provider 与规则的 dataset_code 必须逐字引用本常量。</summary>
     public const string PurchaseDeliveryCode = "PURCHASE_DELIVERY";
 
+    /// <summary>
+    /// 采购交付 Authority 的生产者作业码。<b>唯一来源</b>：
+    /// 下方 AuthoritySpec 与 <see cref="S8PurchaseDeliveryDataProvider"/> 的发布证据查询
+    /// 都引用本常量。两处各写一份字面量必然漂移，而漂移的表现是
+    /// 「健康判定看着 A 作业、取数却认 B 作业的发布」—— 两边各自都不报错。
+    /// </summary>
+    public const string PurchaseDeliveryProducerJobCode = "S3_MDP_SYNC_TRANSFORM";
+
     /// <summary>
     /// 采购交付数据集的能力声明。<b>唯一来源</b>：
     /// <see cref="S8PurchaseDeliveryDataProvider"/> 直接引用本字段，不另写一份字面量。
@@ -58,19 +66,43 @@ public sealed class S8BusinessDatasetDefinitions : IS8DatasetDefinitionSource
             Status = S8DatasetStatus.Defined,
             Capabilities = PurchaseDeliveryCapabilities,
 
-            // Authority 的写入方不止 MDP 转换：DeliveryScheduleService 与 ProcessOutsourceOrderService
-            // 两条 API 路径也直接写 dwd_supplier_delivery，且都不写 mdp_transform_run_log。
-            // 于是运行日志无法证明表里最终的数据出自谁 —— 判 UNKNOWN、拦截恢复判定。
+            // ProducerTrusted 由 false 解冻为 true。冻结时列出的三个解冻条件已全部达成：
+            //
+            // ① 单一写入方 —— DeliveryScheduleService 与 ProcessOutsourceOrderService 对
+            //    dwd_supplier_delivery 的直写已移除（前者曾写出 CANCELLED/CLOSED/PARTIAL 三个
+            //    没有任何读方认识的值，后者把 delivery_status 硬编码为 'OPEN' 并放进 ODKU，
+            //    一次建单就能把 MDP 刚判定的 COMPLETED / DELAYED 无条件翻回 OPEN）。
+            //    运行期写入方现在只剩 S3MdpSyncTransformService，它写 mdp_transform_run_log，
+            //    因此运行日志能完整解释这张表的状态。
+            //
+            // ② membership 淘汰 —— 标准层 INSERT 按 @BatchId 收窄驱动行，并在全量路径追加
+            //    每轮淘汰；DWD 追加同日淘汰。源侧退出的采购订单行会在下一轮跑批退出当前快照。
+            //    （此前实测租户 797403760988229 的标准层 518 行里有 336 行源侧已不存在。）
+            //
+            // ③ 租户 JOIN 谓词 —— 标准层的两个 enrichment JOIN 已补 tenant_id 相等条件，
+            //    并按 (租户, 键) 去重取最新一行。此前实测单租户扇出 3.95 倍。
+            //
+            // 同日淘汰还顺带满足了快照原子性：MAX(stat_date) 取到的必然是单一批次
+            // （此前实测同一天混入两个批次已发生 3 次）。
+            //
+            // ④ 空快照语义 —— 当前态改由**发布证据**指认（RequiresPublicationEvidence）。
+            //    此前 Provider 按「最新的那一天」挑当前快照，在合法的空面前会静默失效：
+            //    源侧清空的那一轮产出 0 行 DWD，当天没有 stat_date 分区，同日淘汰也无行可删，
+            //    于是回落到上一天的旧行。实测租户 838257212780613 源侧清零、STD 归 0、
+            //    跑批 SUCCESS，Provider 仍返回前一日的 10 行。
+            //    现在 currentRows=0 是「成功发布的空快照」，健康且允许恢复；
+            //    而没有发布证据判 NO_PUBLICATION → UNKNOWN，由 Wave2 闸门拦下恢复。
+            //
+            // 本数据集是 CurrentStateMirror，健康判定不读任何行数，
+            // 因此某租户源侧合法清空时仍判 HEALTHY、恢复正常放行。
             //
-            // 这只冻结「从缺席做负面推断」这一件事：建单与刷新照常，
-            // 因为 hit 是对某行存在的正面观测，写入方混杂不会凭空造出一行。
-            // 解冻条件：Authority 收敛为单一写入方、补上 membership 淘汰、补齐租户 JOIN 谓词。
+            // ⚠️ 以上结论已在本地沙箱跑通（membership 182/182/182、扇出 1.00×、
+            //    跨租户 0、phantom 0、空快照 0 行不回落）。共享库运行态验证待统一部署后完成。
             AuthoritySpec = new Health.S8AuthoritySpec
             {
-                ProducerJobCode = "S3_MDP_SYNC_TRANSFORM",
+                ProducerJobCode = PurchaseDeliveryProducerJobCode,
                 AuthorityKind = Health.S8AuthorityKind.CurrentStateMirror,
-                ProducerTrusted = false,
-                UntrustedReasonCode = Health.S8AuthorityHealthReason.Rule01MultiWriterUntrusted
+                RequiresPublicationEvidence = true
             },
             // S8-RULE01-PURCHASE-DELIVERY-CLOSURE-1：due_at 的业务语义已完成血缘取证，
             // 从「UAT_EXECUTION_MAPPING / 待生产确认」升级为「与中台自身逾期判据同源」。

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

@@ -847,79 +847,20 @@ public class DeliveryScheduleService : IDynamicApiController, ITransient
                 new SugarParameter("@BatchId", batchId),
                 new SugarParameter("@Now", now));
 
-            await _db.Ado.ExecuteCommandAsync(
-                """
-                INSERT INTO dwd_supplier_delivery
-                (tenant_id, factory_id, stat_date, po_no, po_line, po_type, supplier_code, supplier_name, item_code, item_name, order_qty, schedule_qty, delivery_qty, receipt_qty, return_qty, remaining_qty, due_date, need_date, last_delivery_date, delivery_status, source_system, sync_batch_id, calc_time)
-                SELECT
-                    @TenantId,
-                    1,
-                    @StatDate,
-                    IFNULL(m.PurOrd, @PoNumber),
-                    CAST(IFNULL(d.Line, @PoLineInt) AS CHAR),
-                    IFNULL(m.Potype, ''),
-                    IFNULL(m.Supp, @SupplierCode),
-                    @Supplier,
-                    IFNULL(d.ItemNum, @ItemNum),
-                    IFNULL(i.Descr, ''),
-                    IFNULL(d.QtyOrded, @SchedQty),
-                    @SchedQty,
-                    @SentQty,
-                    IFNULL(d.QtyReceived, IFNULL(d.RctQty, 0)),
-                    @ReturnQty,
-                    @RestQty,
-                    d.DueDate,
-                    @NeedDate,
-                    @LastSentDate,
-                    CASE
-                        WHEN IFNULL(@Status, '') = 'C' THEN 'CANCELLED'
-                        WHEN IFNULL(@RestQty, 0) <= 0 THEN 'CLOSED'
-                        WHEN IFNULL(@SentQty, 0) > 0 THEN 'PARTIAL'
-                        ELSE 'OPEN'
-                    END,
-                    'AIDOP',
-                    @BatchId,
-                    @Now
-                FROM PurOrdMaster m
-                JOIN PurOrdDetail d ON m.RecID = d.PurOrdRecID
-                LEFT JOIN ItemMaster i ON d.ItemNum = i.ItemNum AND (m.tenant_id = i.tenant_id OR m.tenant_id IS NULL)
-                WHERE m.PurOrd = @PoNumber AND d.Line = @PoLineInt
-                LIMIT 1
-                ON DUPLICATE KEY UPDATE
-                    supplier_code=VALUES(supplier_code),
-                    supplier_name=VALUES(supplier_name),
-                    item_code=VALUES(item_code),
-                    item_name=VALUES(item_name),
-                    order_qty=VALUES(order_qty),
-                    schedule_qty=VALUES(schedule_qty),
-                    delivery_qty=VALUES(delivery_qty),
-                    receipt_qty=VALUES(receipt_qty),
-                    return_qty=VALUES(return_qty),
-                    remaining_qty=VALUES(remaining_qty),
-                    due_date=VALUES(due_date),
-                    need_date=VALUES(need_date),
-                    last_delivery_date=VALUES(last_delivery_date),
-                    delivery_status=VALUES(delivery_status),
-                    sync_batch_id=VALUES(sync_batch_id),
-                    calc_time=VALUES(calc_time),
-                    update_time=CURRENT_TIMESTAMP
-                """,
-                new SugarParameter("@TenantId", ResolveScheduleTenant(schedule)),
-                new SugarParameter("@StatDate", now.Date),
-                new SugarParameter("@PoNumber", schedule.PoNumber),
-                new SugarParameter("@PoLineInt", schedule.PoLine),
-                new SugarParameter("@SupplierCode", schedule.SupplierCode),
-                new SugarParameter("@Supplier", schedule.Supplier),
-                new SugarParameter("@ItemNum", schedule.ItemNum),
-                new SugarParameter("@SchedQty", schedule.SchedQty),
-                new SugarParameter("@SentQty", schedule.SentQty),
-                new SugarParameter("@ReturnQty", schedule.ReturnQty),
-                new SugarParameter("@RestQty", schedule.RestQty),
-                new SugarParameter("@NeedDate", schedule.NeedDate),
-                new SugarParameter("@LastSentDate", schedule.LastSentDate),
-                new SugarParameter("@Status", schedule.Status),
-                new SugarParameter("@BatchId", batchId),
-                new SugarParameter("@Now", now));
+        // S8-RULE01-AUTHORITY-SINGLE-WRITER：已移除对 dwd_supplier_delivery 的直写。
+        //
+        // 该表是 Rule01 的 Authority，必须只有一个运行期写入方（S3 的 MDP 转换）。
+        // 这里原本直读 PurOrdMaster/PurOrdDetail/ItemMaster/SuppMaster 后直接 upsert DWD，
+        // 且把 delivery_status 硬编码为 'OPEN' 并放进 ON DUPLICATE KEY UPDATE ——
+        // 于是一次建单调用就能把 MDP 刚判定的 COMPLETED / DELAYED 无条件翻回 OPEN。
+        //
+        // 更严重的是它能凭空造出一个「最新快照」：Rule01 取 MAX(stat_date) 那一天的全部行，
+        // 实测租户 797403760988229 的 2026-05-16 快照只有 2 行、且全部出自本旁路 ——
+        // MDP 停摆时，一次业务 API 调用就让 Rule01 在 2 行残缺数据上评估，而快照看起来还很新鲜。
+        //
+        // 业务侧只负责写源单据；DWD 由 S3 全量转换派生，最长 60 分钟可见延迟。
+        // 若将来需要更快可见，用 ModuleRebuildService.EnqueueAsync("S3", ...) 在**事务提交之后**
+        // 触发一次正式重建，而不是重新引入第二个写入方。
         }
     }
 

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

@@ -352,59 +352,13 @@ public class ProcessOutsourceOrderService : IDynamicApiController, ITransient
     private async Task UpsertProcessOutsourceMdpAsync(string purOrd, DateTime now)
     {
         var batchId = $"S3_OUTSOURCE_{now:yyyyMMddHHmmss}";
-        await _db.Ado.ExecuteCommandAsync(
-            """
-            INSERT INTO mdp_std_purchase_order
-            (tenant_id, source_system, po_no, po_line, po_type, supplier_code, item_code, item_name, order_qty, received_qty, returned_qty, due_date, need_date, order_date, status, buyer, work_order, source_biz_key, sync_batch_id, sync_time)
-            SELECT
-                NULLIF(m.tenant_id, 0),
-                'AIDOP',
-                IFNULL(m.PurOrd, ''),
-                CAST(IFNULL(d.Line, 0) AS CHAR),
-                IFNULL(m.Potype, ''),
-                IFNULL(m.Supp, ''),
-                IFNULL(d.ItemNum, ''),
-                IFNULL(i.Descr, ''),
-                IFNULL(d.QtyOrded, 0),
-                IFNULL(d.QtyReceived, IFNULL(d.RctQty, 0)),
-                IFNULL(d.QtyReturned, 0),
-                d.DueDate,
-                d.NeedDate,
-                m.OrdDate,
-                CASE WHEN IFNULL(LENGTH(m.Status), 0) = 0 THEN 'R' ELSE m.Status END,
-                m.Buyer,
-                m.WorkOrd,
-                CONCAT(IFNULL(m.PurOrd,''), '|', IFNULL(d.Line, 0)),
-                @BatchId,
-                @Now
-            FROM PurOrdMaster m
-            JOIN PurOrdDetail d ON m.RecID = d.PurOrdRecID
-            LEFT JOIN ItemMaster i ON d.ItemNum = i.ItemNum
-            WHERE m.PurOrd = @PurOrd
-              AND IFNULL(d.ItemNum, '') <> ''
-              AND NULLIF(m.tenant_id, 0) IS NOT NULL
-            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),
-                buyer=VALUES(buyer),
-                work_order=VALUES(work_order),
-                sync_batch_id=VALUES(sync_batch_id),
-                sync_time=VALUES(sync_time),
-                update_time=CURRENT_TIMESTAMP
-            """,
-            new SugarParameter("@PurOrd", purOrd),
-            new SugarParameter("@BatchId", batchId),
-            new SugarParameter("@Now", now));
-
+        // S8-RULE01-AUTHORITY-SINGLE-WRITER：已移除对 mdp_std_purchase_order 的直写。
+        //
+        // 标准层现在是「本轮批次的镜像」：S3 全量转换按 @BatchId 收窄并淘汰不属于本批次的行。
+        // 本旁路写入的行带的是 S3_OUTSOURCE_* 批次号，下一轮全量必然把它淘汰掉，
+        // 留着它只会让标准层短暂出现一批镜像不认识的行，且与「单一 Authority 写入方」相矛盾。
+        //
+        // 采购订单行的标准层由 S3 从源表统一派生；本服务只负责写源单据。
         await _db.Ado.ExecuteCommandAsync(
             """
             INSERT INTO mdp_std_process_outsource_order
@@ -448,61 +402,20 @@ public class ProcessOutsourceOrderService : IDynamicApiController, ITransient
             new SugarParameter("@BatchId", batchId),
             new SugarParameter("@Now", now));
 
-        await _db.Ado.ExecuteCommandAsync(
-            """
-            INSERT INTO dwd_supplier_delivery
-            (tenant_id, factory_id, stat_date, po_no, po_line, po_type, supplier_code, supplier_name, item_code, item_name, order_qty, schedule_qty, delivery_qty, receipt_qty, return_qty, remaining_qty, due_date, need_date, delivery_status, source_system, sync_batch_id, calc_time)
-            SELECT
-                NULLIF(m.tenant_id, 0),
-                1,
-                @StatDate,
-                IFNULL(m.PurOrd, ''),
-                CAST(IFNULL(d.Line, 0) AS CHAR),
-                IFNULL(m.Potype, ''),
-                IFNULL(m.Supp, ''),
-                IFNULL(s.SortName, ''),
-                IFNULL(d.ItemNum, ''),
-                IFNULL(i.Descr, ''),
-                IFNULL(d.QtyOrded, 0),
-                0,
-                0,
-                IFNULL(d.QtyReceived, IFNULL(d.RctQty, 0)),
-                IFNULL(d.QtyReturned, 0),
-                GREATEST(IFNULL(d.QtyOrded, 0) - IFNULL(d.QtyReceived, IFNULL(d.RctQty, 0)) - IFNULL(d.QtyReturned, 0), 0),
-                d.DueDate,
-                d.NeedDate,
-                'OPEN',
-                'AIDOP',
-                @BatchId,
-                @Now
-            FROM PurOrdMaster m
-            JOIN PurOrdDetail d ON m.RecID = d.PurOrdRecID
-            LEFT JOIN ItemMaster i ON d.ItemNum = i.ItemNum
-            LEFT JOIN SuppMaster s ON m.Domain = s.Domain AND m.Supp = s.Supp
-            WHERE m.PurOrd = @PurOrd
-              AND IFNULL(d.ItemNum, '') <> ''
-              AND NULLIF(m.tenant_id, 0) IS NOT NULL
-            ON DUPLICATE KEY UPDATE
-                po_type=VALUES(po_type),
-                supplier_code=VALUES(supplier_code),
-                supplier_name=VALUES(supplier_name),
-                item_code=VALUES(item_code),
-                item_name=VALUES(item_name),
-                order_qty=VALUES(order_qty),
-                receipt_qty=VALUES(receipt_qty),
-                return_qty=VALUES(return_qty),
-                remaining_qty=VALUES(remaining_qty),
-                due_date=VALUES(due_date),
-                need_date=VALUES(need_date),
-                delivery_status=VALUES(delivery_status),
-                sync_batch_id=VALUES(sync_batch_id),
-                calc_time=VALUES(calc_time),
-                update_time=CURRENT_TIMESTAMP
-            """,
-            new SugarParameter("@PurOrd", purOrd),
-            new SugarParameter("@StatDate", now.Date),
-            new SugarParameter("@BatchId", batchId),
-            new SugarParameter("@Now", now));
+        // S8-RULE01-AUTHORITY-SINGLE-WRITER：已移除对 dwd_supplier_delivery 的直写。
+        //
+        // 该表是 Rule01 的 Authority，必须只有一个运行期写入方（S3 的 MDP 转换）。
+        // 这里原本直读 PurOrdMaster/PurOrdDetail/ItemMaster/SuppMaster 后直接 upsert DWD，
+        // 且把 delivery_status 硬编码为 'OPEN' 并放进 ON DUPLICATE KEY UPDATE ——
+        // 于是一次建单调用就能把 MDP 刚判定的 COMPLETED / DELAYED 无条件翻回 OPEN。
+        //
+        // 更严重的是它能凭空造出一个「最新快照」：Rule01 取 MAX(stat_date) 那一天的全部行，
+        // 实测租户 797403760988229 的 2026-05-16 快照只有 2 行、且全部出自本旁路 ——
+        // MDP 停摆时，一次业务 API 调用就让 Rule01 在 2 行残缺数据上评估，而快照看起来还很新鲜。
+        //
+        // 业务侧只负责写源单据；DWD 由 S3 全量转换派生，最长 60 分钟可见延迟。
+        // 若将来需要更快可见，用 ModuleRebuildService.EnqueueAsync("S3", ...) 在**事务提交之后**
+        // 触发一次正式重建，而不是重新引入第二个写入方。
     }
 
     [DisplayName("保存工序外协订单")]