fix(aidop): rebuild polluted outsource delivery DWD (MIGRATION NOT EXECUTED)

新增 P0-A 数据纠偏 migration 与配套 verify、契约测试。**脚本尚未在任何库执行。**

未执行原因：Stop-Write Gate FAIL。实测后端进程存活、
job_smart_ops_kpi_mdp_bootstrap 于 20:07:13 刚触发、
job_s3_mdp_sync_transform NextRunTime=21:07:14，写入必然恢复。
按规约不带着活跃写入方做 DELETE / TRUNCATE / ALTER。

文件名用占位 WIP-P0A-CLEANUP.sql（CLAUDE.md §二·7）：本仓与 origin/master 分叉，
且 1.0.536~538 已撞号（同号不同内容），此刻无法取正式版本号。
占位文件未注册 csproj Copy 条目，不会被 AutoVersionUpdate 拾取执行。
待分叉解决后 rename 为 1.0.<next>.sql 并补注册。

migration 步骤（顺序不可颠倒）：
  Step 0 门禁 G1-G5，任一不成立即 SIGNAL 中止
         G1 集合完备：population_a + survivors == std_total（断言集合关系而非硬编码行数）
         G2 血缘双向差集为 0（EXISTS 回溯 stg 的 RoutingOpDetail 分区）
         G3 幸存者全部是真实 PW  G4 overlap=0  G5 待保全子集非空
  Step 1 保全 DWD 真实 PO 子集 + 元数据留证（复用 1.0.350 的 cleanup_log 范式）
  Step 2 精准清 std 种群 A（范式 A：CREATE TABLE LIKE + INSERT IGNORE + DELETE）
  Step 3 TRUNCATE DWD（非分批 DELETE：后者不归还磁盘，却多付约 5-8GB binlog 与数十 GB undo）
  Step 4 空表期 schema 收口：po_no/po_line NOT NULL DEFAULT ''、
         唯一键换为 (tenant_id,factory_id,stat_date,po_no,po_line)、
         租户守卫 CHECK 按 CHECK_CLAUSE 动态定位后收敛到稳定名
  Step 5 回插保全历史，显式排除 tenant_id IN (0,1,1300000000001)

两处实测校正（否则会静默出错）：
  - 保全谓词必须是 IFNULL(po_no,'')<>'' 而非 po_no IS NOT NULL。
    代码侧 IFNULL 补丁上线后种群 A 的 po_no 变成空串而非 NULL，
    用 IS NOT NULL 当时会多保 56 行垃圾。
  - 血缘回溯不得附加 source_system 等值条件。stg 侧两代分属
    AIDOP(colon 形态) 与 AIDOPDEV_MYSQL(digits 形态)，std 侧被硬编码为 'AIDOP'，
    加上该条件会假阴 75939 行。

只读 dry-run（aidopdev，仅 SELECT）：
  G1 117154 + 4 = 117158 ✓   G2 双向差集 0/0 ✓
  G3 0 ✓  G4 0 ✓  G5 保全 31 行 / 1 grain / 2026-08-16~09-15 ✓
  列类型核对：po_no varchar(100) / po_line varchar(50)，与 MODIFY 语句一致 ✓

测试：新增 11 条结构契约测试，mutation 验证有效
（保全谓词退回 IS NOT NULL → T6 红；schema 收口提前到清数据之前 → T7 红）。

server/Admin.NET.Web.Entry/UpdateScripts/WIP-P0A-CLEANUP.sql

@@ -0,0 +1,298 @@
+-- WIP-P0A-CLEANUP.sql
+--   P0-A 数据纠偏：精准清 std 种群 A + 保全 DWD 不可再生历史 + 清空重建 DWD + schema 收口。
+--
+--   ⚠️ 占位文件名（CLAUDE.md §二·7）：本仓与 origin/master 当前处于分叉状态且 1.0.536~538 已撞号，
+--      无法在此刻取正式版本号。提交前 rename 为 1.0.<next>.sql 并注册 csproj Copy 条目。
+--
+-- ============================================================================
+-- 背景（全部为只读实测结论，证据见本批审计回执）
+--   dwd_process_outsource_delivery 曾混装两个业务对象：
+--     种群 A = RoutingOpDetail（S0 标准工艺路线明细，主数据），无 work_order / po_no / po_line，
+--              其 std 物化把 po_no/po_line 硬编码为 NULL；而 uk_work_op_stat 含这两列，
+--              MySQL 唯一索引 NULL-distinct 语义使 ODKU 永不命中 → 每轮全量追加。
+--     种群 B = PurOrdMaster(Potype='PW') JOIN PurOrdDetail，真实工序委外采购订单行。
+--   代码侧止血已完成（S3_ROUTING_OUTSOURCE 入站注册与 std INSERT 已移除，DWD 投影已补
+--   IFNULL(po_no,'') 使 po_no 不再为 NULL）。本脚本只处理存量。
+--
+--   Option A 目标粒度：一行 = 一条真实 PW 委外订单行在某 stat_date 的交付状态。
+--   目标唯一键：(tenant_id, factory_id, stat_date, po_no, po_line)。
+--
+--   历史 stat_date 不可复现（@StatDate = 运行时墙钟日期，无回填入参），且 31 行对应的源单
+--   已从 PurOrdMaster 物理删除 → 保全那 31 行就是历史恢复方案，不做"重新计算"。
+-- ============================================================================
+SET NAMES utf8mb4;
+
+-- ===========================================================================
+-- Step 0 · 前置门禁（fail-fast；任一不成立即中止，绝不带着错误前提往下做）
+--   断言集合关系而非硬编码行数：行数会随跑批漂移，集合关系不会。
+-- ===========================================================================
+
+-- 种群 A 主谓词（与审计一致：批次来源 + biz_key 形态 + 结构特征三重冗余）
+SET @std_total := (SELECT COUNT(*) FROM `mdp_std_process_outsource_order`);
+SET @pop_a := (
+  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
+  WHERE `sync_batch_id`  LIKE 'S3\_MDP\_%'
+    AND `source_biz_key` NOT LIKE '%|%'
+    AND `work_order` = ''
+    AND `po_no`   IS NULL
+    AND `po_line` IS NULL
+);
+SET @survivors := (
+  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
+  WHERE NOT (`sync_batch_id` LIKE 'S3\_MDP\_%'
+         AND `source_biz_key` NOT LIKE '%|%'
+         AND `work_order` = ''
+         AND `po_no`   IS NULL
+         AND `po_line` IS NULL)
+);
+SELECT @std_total AS std_total, @pop_a AS population_a, @survivors AS survivors;
+
+-- G1 · 集合完备：种群 A + 幸存者 必须恰好等于总量（不允许存在第三类）
+SET @sql := IF(@pop_a + @survivors <> @std_total,
+  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G1: population_a + survivors <> std_total — predicate incomplete, abort'",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- G2 · 血缘交叉验证：主谓词命中集必须与 stg 血缘回溯集完全一致（双向差集为 0）
+--      注意：不得在 EXISTS 里附加 source_system 等值条件 —— stg 侧两代分属
+--      AIDOP(colon 形态) 与 AIDOPDEV_MYSQL(digits 形态)，而 std 侧被统一硬编码为 'AIDOP'，
+--      加上该条件会假阴约 7.6 万行。
+SET @main_not_prov := (
+  SELECT COUNT(*) FROM `mdp_std_process_outsource_order` o
+  WHERE o.`sync_batch_id` LIKE 'S3\_MDP\_%' AND o.`source_biz_key` NOT LIKE '%|%'
+    AND o.`work_order` = '' AND o.`po_no` IS NULL AND o.`po_line` IS NULL
+    AND NOT EXISTS (SELECT 1 FROM `mdp_stg_work_order_material` s
+                    WHERE s.`source_table` = 'RoutingOpDetail'
+                      AND s.`source_biz_key` = o.`source_biz_key`)
+);
+SET @prov_not_main := (
+  SELECT COUNT(*) FROM `mdp_std_process_outsource_order` o
+  WHERE EXISTS (SELECT 1 FROM `mdp_stg_work_order_material` s
+                WHERE s.`source_table` = 'RoutingOpDetail'
+                  AND s.`source_biz_key` = o.`source_biz_key`)
+    AND NOT (o.`sync_batch_id` LIKE 'S3\_MDP\_%' AND o.`source_biz_key` NOT LIKE '%|%'
+         AND o.`work_order` = '' AND o.`po_no` IS NULL AND o.`po_line` IS NULL)
+);
+SELECT @main_not_prov AS main_not_provenance, @prov_not_main AS provenance_not_main;
+SET @sql := IF(@main_not_prov > 0 OR @prov_not_main > 0,
+  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G2: cleanup predicate disagrees with stg provenance — abort'",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- G3 · 幸存者必须全部是真实 PW 行（biz_key 形如 PurOrd|Line 且 po_no 非空）
+SET @bad_survivor := (
+  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
+  WHERE NOT (`sync_batch_id` LIKE 'S3\_MDP\_%' AND `source_biz_key` NOT LIKE '%|%'
+         AND `work_order` = '' AND `po_no` IS NULL AND `po_line` IS NULL)
+    AND (`source_biz_key` NOT LIKE '%|%' OR IFNULL(`po_no`,'') = '')
+);
+SET @sql := IF(@bad_survivor > 0,
+  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G3: survivor set contains non-PW rows — abort'",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- G4 · overlap：种群 A 与真实 PW 集不得有交集
+SET @overlap := (
+  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
+  WHERE `sync_batch_id` LIKE 'S3\_MDP\_%' AND `source_biz_key` NOT LIKE '%|%'
+    AND `work_order` = '' AND `po_no` IS NULL AND `po_line` IS NULL
+    AND `source_biz_key` LIKE '%|%'
+);
+SET @sql := IF(@overlap > 0,
+  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G4: population_a overlaps real PW rows — abort'",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- G5 · 待保全的 DWD 历史必须非空且 po_no/po_line 干净
+--      ⚠️ 谓词必须是 IFNULL(po_no,'') <> '' 而不是 po_no IS NOT NULL：
+--      代码侧 IFNULL 补丁上线后，种群 A 的行 po_no 变成**空串**而非 NULL，
+--      用 IS NOT NULL 会把这些垃圾一并当成"不可再生历史"保下来。
+SET @preserve_cnt := (SELECT COUNT(*) FROM `dwd_process_outsource_delivery` WHERE IFNULL(`po_no`,'') <> '');
+SELECT @preserve_cnt AS dwd_rows_to_preserve;
+SET @sql := IF(@preserve_cnt = 0,
+  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G5: no real-PO rows found in DWD — refuse to truncate blindly'",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- ===========================================================================
+-- Step 1 · 保全 DWD 不可再生历史 + 元数据留证
+--   只备份真实 PO 子集（约 31 行），不复制 3269 万行错误派生数据：
+--   零消费者 / 零 FK / 可重建（今后）/ 历史上已 TRUNCATE 过 / 权威源仍在。
+-- ===========================================================================
+CREATE TABLE IF NOT EXISTS `dwd_process_outsource_delivery_keep_p0a` LIKE `dwd_process_outsource_delivery`;
+INSERT IGNORE INTO `dwd_process_outsource_delivery_keep_p0a`
+SELECT * FROM `dwd_process_outsource_delivery` WHERE IFNULL(`po_no`,'') <> '';
+
+-- 保全完整性断言：备份行数必须等于源侧真实 PO 行数
+SET @kept := (SELECT COUNT(*) FROM `dwd_process_outsource_delivery_keep_p0a`);
+SET @sql := IF(@kept <> @preserve_cnt,
+  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-S1: preserved row count mismatch — abort before truncate'",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- 元数据留证（复用 1.0.350 既有范式与表）
+CREATE TABLE IF NOT EXISTS `aidop_migration_derived_cleanup_log` (
+  `id` BIGINT NOT NULL AUTO_INCREMENT,
+  `migration_version` VARCHAR(32) NOT NULL,
+  `table_name` VARCHAR(128) NOT NULL,
+  `estimated_rows` BIGINT NOT NULL DEFAULT 0,
+  `data_length` BIGINT NOT NULL DEFAULT 0,
+  `reason` VARCHAR(500) NOT NULL,
+  `captured_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_migration_cleanup` (`migration_version`,`table_name`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+
+INSERT INTO `aidop_migration_derived_cleanup_log`
+(`migration_version`,`table_name`,`estimated_rows`,`data_length`,`reason`)
+SELECT 'WIP-P0A','dwd_process_outsource_delivery',IFNULL(TABLE_ROWS,0),IFNULL(DATA_LENGTH,0),
+       CONCAT('P0-A：RoutingOpDetail 主数据被误建模为委外交付事实，叠加 uk 含可空 po_no 致 ODKU 永不命中，膨胀至 3269 万行；',
+              '真实 PO 子集 ', @preserve_cnt, ' 行已存入 dwd_process_outsource_delivery_keep_p0a 并于 schema 收口后回插；',
+              '其余为错误派生数据，零消费者、零 FK、可由 std 重新生成，不做全量备份')
+FROM information_schema.TABLES
+WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
+ON DUPLICATE KEY UPDATE `reason`=VALUES(`reason`), `estimated_rows`=VALUES(`estimated_rows`), `data_length`=VALUES(`data_length`);
+
+-- ===========================================================================
+-- Step 2 · 精准清 std 种群 A（保留真实 PW；不得 TRUNCATE 整表）
+--   备份沿用 1.0.351/352/353 的范式 A：CREATE TABLE LIKE + INSERT IGNORE + DELETE。
+-- ===========================================================================
+CREATE TABLE IF NOT EXISTS `mdp_std_process_outsource_order_bak_p0a` LIKE `mdp_std_process_outsource_order`;
+INSERT IGNORE INTO `mdp_std_process_outsource_order_bak_p0a`
+SELECT * FROM `mdp_std_process_outsource_order`
+WHERE `sync_batch_id`  LIKE 'S3\_MDP\_%'
+  AND `source_biz_key` NOT LIKE '%|%'
+  AND `work_order` = ''
+  AND `po_no`   IS NULL
+  AND `po_line` IS NULL;
+
+DELETE FROM `mdp_std_process_outsource_order`
+WHERE `sync_batch_id`  LIKE 'S3\_MDP\_%'
+  AND `source_biz_key` NOT LIKE '%|%'
+  AND `work_order` = ''
+  AND `po_no`   IS NULL
+  AND `po_line` IS NULL;
+
+-- 清理后立即断言：种群 A 归零，且幸存者数量与门禁阶段一致
+SET @pop_a_after := (
+  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
+  WHERE `sync_batch_id` LIKE 'S3\_MDP\_%' AND `source_biz_key` NOT LIKE '%|%'
+    AND `work_order` = '' AND `po_no` IS NULL AND `po_line` IS NULL
+);
+SET @std_after := (SELECT COUNT(*) FROM `mdp_std_process_outsource_order`);
+SELECT @pop_a_after AS population_a_after, @std_after AS std_total_after;
+SET @sql := IF(@pop_a_after <> 0 OR @std_after <> @survivors,
+  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-S2: std cleanup result mismatch — abort before DWD truncate'",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+-- 注：tenant 0 / 1300000000001 的真实 PW 行仍保留在 std。
+--     "进不了 DWD"（被 CHECK 拦）不等于"应该从 std 删除"。
+
+-- ===========================================================================
+-- Step 3 · 清空 DWD
+--   选 TRUNCATE 而非分批 DELETE：错误数据无业务价值、零 FK、零消费者，
+--   且 DELETE 会产生约 5-8GB binlog（ROW+FULL image）与数十 GB undo，
+--   却**不把空间归还 OS**；TRUNCATE 立即归还约 7.9GB。
+-- ===========================================================================
+TRUNCATE TABLE `dwd_process_outsource_delivery`;
+
+-- ===========================================================================
+-- Step 4 · 空表期 schema 收口（此时行数为 0，DDL 代价最低且不会因存量 NULL 失败）
+-- ===========================================================================
+
+-- 4.1 po_no / po_line → NOT NULL DEFAULT ''（幂等：仅当当前仍 nullable 时执行）
+SET @po_no_nullable := (SELECT IS_NULLABLE FROM information_schema.COLUMNS
+  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery' AND COLUMN_NAME='po_no');
+SET @sql := IF(@po_no_nullable = 'YES',
+  "ALTER TABLE `dwd_process_outsource_delivery` MODIFY COLUMN `po_no` VARCHAR(100) NOT NULL DEFAULT ''",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+SET @po_line_nullable := (SELECT IS_NULLABLE FROM information_schema.COLUMNS
+  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery' AND COLUMN_NAME='po_line');
+SET @sql := IF(@po_line_nullable = 'YES',
+  "ALTER TABLE `dwd_process_outsource_delivery` MODIFY COLUMN `po_line` VARCHAR(50) NOT NULL DEFAULT ''",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- 4.2 唯一键改为 Option A 粒度：(tenant_id, factory_id, stat_date, po_no, po_line)
+--     旧键 uk_work_op_stat 含 work_order/op_code，对 RoutingOpDetail 来源无区分度。
+SET @has_old_uk := (SELECT COUNT(*) FROM information_schema.STATISTICS
+  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery' AND INDEX_NAME='uk_work_op_stat');
+SET @sql := IF(@has_old_uk > 0,
+  'ALTER TABLE `dwd_process_outsource_delivery` DROP INDEX `uk_work_op_stat`',
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+SET @has_new_uk := (SELECT COUNT(*) FROM information_schema.STATISTICS
+  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery' AND INDEX_NAME='uk_po_line_stat');
+SET @sql := IF(@has_new_uk = 0,
+  'ALTER TABLE `dwd_process_outsource_delivery` ADD UNIQUE INDEX `uk_po_line_stat` (`tenant_id`,`factory_id`,`stat_date`,`po_no`,`po_line`)',
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- 4.3 租户守卫 CHECK：按**实际名**处理，不得沿用 1.0.353 声明的名字
+--     实测漂移：1.0.353 声明 ck_dwd_process_outsource_valid_tenant，
+--     但该表在 1.0.353 之后被 CodeFirst 重建过，约束被匿名重建为
+--     dwd_process_outsource_delivery_chk_1（全库唯一的自动命名 CHECK）。
+--     此处按语义（CHECK_CLAUSE 含 tenant_id 且含 1300000000001）定位，统一收敛到稳定名。
+SET @tenant_chk_cnt := (
+  SELECT COUNT(*) FROM information_schema.CHECK_CONSTRAINTS cc
+  JOIN information_schema.TABLE_CONSTRAINTS tc
+    ON tc.CONSTRAINT_SCHEMA=cc.CONSTRAINT_SCHEMA AND tc.CONSTRAINT_NAME=cc.CONSTRAINT_NAME
+  WHERE tc.TABLE_SCHEMA=DATABASE() AND tc.TABLE_NAME='dwd_process_outsource_delivery'
+    AND tc.CONSTRAINT_TYPE='CHECK'
+    AND cc.CHECK_CLAUSE LIKE '%tenant_id%' AND cc.CHECK_CLAUSE LIKE '%1300000000001%'
+);
+SET @sql := IF(@tenant_chk_cnt > 1,
+  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-S4: duplicate tenant-guard CHECK constraints found — manual review required'",
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+SET @tenant_chk_name := (
+  SELECT tc.CONSTRAINT_NAME FROM information_schema.CHECK_CONSTRAINTS cc
+  JOIN information_schema.TABLE_CONSTRAINTS tc
+    ON tc.CONSTRAINT_SCHEMA=cc.CONSTRAINT_SCHEMA AND tc.CONSTRAINT_NAME=cc.CONSTRAINT_NAME
+  WHERE tc.TABLE_SCHEMA=DATABASE() AND tc.TABLE_NAME='dwd_process_outsource_delivery'
+    AND tc.CONSTRAINT_TYPE='CHECK'
+    AND cc.CHECK_CLAUSE LIKE '%tenant_id%' AND cc.CHECK_CLAUSE LIKE '%1300000000001%'
+  LIMIT 1
+);
+-- 若现存名不是稳定名，则丢弃后按稳定名重建（空表期，代价可忽略）
+SET @sql := IF(@tenant_chk_name IS NOT NULL AND @tenant_chk_name <> 'ck_dwd_process_outsource_valid_tenant',
+  CONCAT('ALTER TABLE `dwd_process_outsource_delivery` DROP CHECK `', @tenant_chk_name, '`'),
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+SET @has_stable_chk := (SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS
+  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
+    AND CONSTRAINT_TYPE='CHECK' AND CONSTRAINT_NAME='ck_dwd_process_outsource_valid_tenant');
+SET @sql := IF(@has_stable_chk = 0,
+  'ALTER TABLE `dwd_process_outsource_delivery` ADD CONSTRAINT `ck_dwd_process_outsource_valid_tenant` CHECK (`tenant_id` NOT IN (0,1,1300000000001))',
+  'DO 0');
+PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
+
+-- ===========================================================================
+-- Step 5 · 回插保全的历史
+--   显式排除非法租户（不依赖隐式 filter），并显式要求 po_no/po_line 非空。
+-- ===========================================================================
+INSERT INTO `dwd_process_outsource_delivery`
+SELECT * FROM `dwd_process_outsource_delivery_keep_p0a`
+WHERE `tenant_id` NOT IN (0,1,1300000000001)
+  AND IFNULL(`po_no`,'')   <> ''
+  AND IFNULL(`po_line`,'') <> '';
+
+SELECT COUNT(*) AS dwd_rows_after_restore FROM `dwd_process_outsource_delivery`;
+SELECT `estimated_rows` AS rows_before_cleanup FROM `aidop_migration_derived_cleanup_log`
+WHERE `migration_version`='WIP-P0A' AND `table_name`='dwd_process_outsource_delivery';
+
+-- ===========================================================================
+-- 回滚（注释，非可执行；执行前请确认无消费者）
+--   std：INSERT IGNORE INTO mdp_std_process_outsource_order
+--          SELECT * FROM mdp_std_process_outsource_order_bak_p0a;
+--   dwd：错误派生数据不可回滚也不需要回滚（零消费者、可由 std 重新生成）；
+--        真实历史由 dwd_process_outsource_delivery_keep_p0a 保全，可再次回插。
+--   schema：ALTER ... MODIFY po_no/po_line 恢复 nullable；
+--           DROP INDEX uk_po_line_stat; ADD UNIQUE uk_work_op_stat(...原列序...)。
+--   ⚠️ 回滚 schema 会让 ODKU 重新失效、膨胀复发，仅在确认整改方案作废时执行。
+-- ===========================================================================

server/Admin.NET.Web.Entry/UpdateScripts/WIP-P0A-CLEANUP.verify.sql

@@ -0,0 +1,127 @@
+-- WIP-P0A-CLEANUP.verify.sql
+--   P0-A 数据纠偏的执行后校验。返回单列 ok：1 = 全部断言通过，0 = 任一失败。
+--   提交前随主脚本一并 rename 为 1.0.<next>.verify.sql 并注册 csproj Copy 条目。
+--
+--   V1 std 种群 A 归零且仅剩真实 PW
+--   V2 DWD 行数等于保全子集行数
+--   V3 po_no / po_line 无 NULL 且无空串
+--   V4 新唯一键下无重复
+--   V5 唯一键 DDL 真实存在且列序正确
+--   V6 租户守卫 CHECK 恰好一套、无语义重复
+--   V7 保全历史与迁移前子集完全一致
+--   V8 非法租户未混入
+SELECT CASE WHEN
+
+-- V1 · std：种群 A 归零
+    (
+      SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
+      WHERE `sync_batch_id` LIKE 'S3\_MDP\_%' AND `source_biz_key` NOT LIKE '%|%'
+        AND `work_order` = '' AND `po_no` IS NULL AND `po_line` IS NULL
+    ) = 0
+
+-- V1b · std：幸存者全部是真实 PW（biz_key 含管道符且 po_no 非空）
+AND (
+      SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
+      WHERE `source_biz_key` NOT LIKE '%|%' OR IFNULL(`po_no`,'') = ''
+    ) = 0
+
+-- V1c · std：幸存者数量必须大于 0（防止误把整表清空）
+AND (SELECT COUNT(*) FROM `mdp_std_process_outsource_order`) > 0
+
+-- V2 · DWD 行数 == 保全子集行数（迁移刚完成时，尚未跑新批次）
+AND (SELECT COUNT(*) FROM `dwd_process_outsource_delivery`)
+  = (SELECT COUNT(*) FROM `dwd_process_outsource_delivery_keep_p0a`
+     WHERE `tenant_id` NOT IN (0,1,1300000000001)
+       AND IFNULL(`po_no`,'') <> '' AND IFNULL(`po_line`,'') <> '')
+
+-- V3 · NULL / 空串：收口后 po_no / po_line 既不为 NULL 也不为空串
+AND (SELECT COUNT(*) FROM `dwd_process_outsource_delivery` WHERE `po_no`   IS NULL) = 0
+AND (SELECT COUNT(*) FROM `dwd_process_outsource_delivery` WHERE `po_line` IS NULL) = 0
+AND (SELECT COUNT(*) FROM `dwd_process_outsource_delivery` WHERE `po_no`   = '')    = 0
+AND (SELECT COUNT(*) FROM `dwd_process_outsource_delivery` WHERE `po_line` = '')    = 0
+
+-- V3b · 列定义确已收口为 NOT NULL
+AND (
+      SELECT COUNT(*) FROM information_schema.COLUMNS
+      WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
+        AND COLUMN_NAME IN ('po_no','po_line') AND IS_NULLABLE='YES'
+    ) = 0
+
+-- V4 · 新唯一键下无重复
+AND (
+      SELECT COUNT(*) FROM (
+        SELECT 1 FROM `dwd_process_outsource_delivery`
+        GROUP BY `tenant_id`,`factory_id`,`stat_date`,`po_no`,`po_line`
+        HAVING COUNT(*) > 1
+      ) dup
+    ) = 0
+
+-- V5 · 唯一键 DDL 存在、恰好 5 列、且列序完全正确
+AND (
+      SELECT COUNT(*) FROM information_schema.STATISTICS
+      WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
+        AND INDEX_NAME='uk_po_line_stat' AND NON_UNIQUE=0
+    ) = 5
+AND (
+      SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX SEPARATOR ',')
+      FROM information_schema.STATISTICS
+      WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
+        AND INDEX_NAME='uk_po_line_stat'
+    ) = 'tenant_id,factory_id,stat_date,po_no,po_line'
+
+-- V5b · 旧唯一键已移除（否则膨胀语义仍在）
+AND (
+      SELECT COUNT(*) FROM information_schema.STATISTICS
+      WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
+        AND INDEX_NAME='uk_work_op_stat'
+    ) = 0
+
+-- V6 · 租户守卫 CHECK 恰好一套，且名字是稳定名（无语义重复）
+AND (
+      SELECT COUNT(*) FROM information_schema.CHECK_CONSTRAINTS cc
+      JOIN information_schema.TABLE_CONSTRAINTS tc
+        ON tc.CONSTRAINT_SCHEMA=cc.CONSTRAINT_SCHEMA AND tc.CONSTRAINT_NAME=cc.CONSTRAINT_NAME
+      WHERE tc.TABLE_SCHEMA=DATABASE() AND tc.TABLE_NAME='dwd_process_outsource_delivery'
+        AND tc.CONSTRAINT_TYPE='CHECK'
+        AND cc.CHECK_CLAUSE LIKE '%tenant_id%' AND cc.CHECK_CLAUSE LIKE '%1300000000001%'
+    ) = 1
+AND (
+      SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS
+      WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
+        AND CONSTRAINT_TYPE='CHECK' AND CONSTRAINT_NAME='ck_dwd_process_outsource_valid_tenant'
+    ) = 1
+
+-- V7 · 保全历史与迁移前子集完全一致（按业务键集合双向比对，差集为 0）
+AND (
+      SELECT COUNT(*) FROM `dwd_process_outsource_delivery_keep_p0a` k
+      WHERE k.`tenant_id` NOT IN (0,1,1300000000001)
+        AND IFNULL(k.`po_no`,'') <> '' AND IFNULL(k.`po_line`,'') <> ''
+        AND NOT EXISTS (
+          SELECT 1 FROM `dwd_process_outsource_delivery` d
+          WHERE d.`tenant_id`=k.`tenant_id` AND d.`factory_id`=k.`factory_id`
+            AND d.`stat_date`=k.`stat_date` AND d.`po_no`=k.`po_no` AND d.`po_line`=k.`po_line`
+        )
+    ) = 0
+AND (
+      SELECT COUNT(*) FROM `dwd_process_outsource_delivery` d
+      WHERE NOT EXISTS (
+          SELECT 1 FROM `dwd_process_outsource_delivery_keep_p0a` k
+          WHERE k.`tenant_id`=d.`tenant_id` AND k.`factory_id`=d.`factory_id`
+            AND k.`stat_date`=d.`stat_date` AND k.`po_no`=d.`po_no` AND k.`po_line`=d.`po_line`
+        )
+    ) = 0
+
+-- V8 · 非法租户未混入 DWD
+AND (SELECT COUNT(*) FROM `dwd_process_outsource_delivery` WHERE `tenant_id` IN (0,1,1300000000001)) = 0
+
+-- V9 · 元数据留证已写入，且记录了清理前的量级
+AND (
+      SELECT COUNT(*) FROM `aidop_migration_derived_cleanup_log`
+      WHERE `migration_version`='WIP-P0A' AND `table_name`='dwd_process_outsource_delivery'
+        AND `estimated_rows` > 1000000
+    ) = 1
+
+-- V10 · std 备份表存在且非空（回滚路径可用）
+AND (SELECT COUNT(*) FROM `mdp_std_process_outsource_order_bak_p0a`) > 0
+
+THEN 1 ELSE 0 END AS ok;

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

@@ -0,0 +1,228 @@
+using System;
+using System.IO;
+using System.Linq;
+using Xunit;
+
+namespace Admin.NET.Plugin.AiDOP.Tests.DataPlatform;
+
+/// <summary>
+/// P0-A 数据纠偏 migration 的结构契约测试。
+///
+/// <para><b>这些断言能证明什么、不能证明什么（如实说明，不要高估）</b>：
+/// 它们对 migration 脚本文本做结构断言，能够拦住「谓词被改弱」「保全谓词退回
+/// po_no IS NOT NULL」「漏掉租户排除」「先改 schema 后清数据」这类**致命且静默**的回归。
+/// 它们**不能**证明脚本在真实数据上的执行结果 —— 那已由对 aidopdev 的只读 dry-run 覆盖
+/// （G1 117154+4=117158、G2 双向差集 0、G5 保全 31 行），本仓无 DB 集成测试基础设施。</para>
+/// </summary>
+public class P0aCleanupMigrationContractTests
+{
+    private const StringComparison Ord = StringComparison.Ordinal;
+
+    private static string RepoRoot()
+    {
+        var dir = new DirectoryInfo(AppContext.BaseDirectory);
+        while (dir != null && !Directory.Exists(Path.Combine(dir.FullName, "server", "Admin.NET.Web.Entry")))
+            dir = dir.Parent;
+        Assert.NotNull(dir);
+        return dir!.FullName;
+    }
+
+    private static string ScriptsDir() =>
+        Path.Combine(RepoRoot(), "server", "Admin.NET.Web.Entry", "UpdateScripts");
+
+    /// <summary>主脚本文本。占位名或已 rename 的正式名都能找到。</summary>
+    private static string Migration()
+    {
+        var dir = ScriptsDir();
+        var f = Directory.GetFiles(dir, "WIP-P0A-CLEANUP.sql").FirstOrDefault()
+             ?? Directory.GetFiles(dir, "*.sql")
+                    .FirstOrDefault(p => !p.EndsWith(".verify.sql", Ord)
+                                      && File.ReadAllText(p).Contains("P0A-G1", Ord));
+        Assert.True(f != null, "找不到 P0-A cleanup migration（占位名 WIP-P0A-CLEANUP.sql 或已 rename 的正式号）");
+        return File.ReadAllText(f!);
+    }
+
+    private static string Verify()
+    {
+        var dir = ScriptsDir();
+        var f = Directory.GetFiles(dir, "WIP-P0A-CLEANUP.verify.sql").FirstOrDefault()
+             ?? Directory.GetFiles(dir, "*.verify.sql")
+                    .FirstOrDefault(p => File.ReadAllText(p).Contains("uk_po_line_stat", Ord));
+        Assert.True(f != null, "找不到 P0-A cleanup 的 verify 脚本");
+        return File.ReadAllText(f!);
+    }
+
+    // ── T1 · 种群 A 谓词：三重冗余缺一不可 ───────────────────────────────────────
+    // 只用 po_no IS NULL 是不够的（未来可能出现合法空 po_no）；必须同时带
+    // 批次来源(sync_batch_id) + biz_key 形态(NOT LIKE '%|%') + 结构特征(work_order='')。
+    [Fact]
+    public void T1_PopulationA_predicate_keeps_all_three_independent_conditions()
+    {
+        var sql = Migration();
+        Assert.Contains(@"`sync_batch_id`  LIKE 'S3\_MDP\_%'", sql, Ord);
+        Assert.Contains(@"`source_biz_key` NOT LIKE '%|%'", sql, Ord);
+        Assert.Contains("`work_order` = ''", sql, Ord);
+        Assert.Contains("`po_no`   IS NULL", sql, Ord);
+        Assert.Contains("`po_line` IS NULL", sql, Ord);
+    }
+
+    // ── T1b · 血缘交叉验证必须存在，且不得附加 source_system 等值条件 ──────────
+    // stg 侧两代分属 AIDOP(colon) / AIDOPDEV_MYSQL(digits)，std 侧被硬编码为 'AIDOP'，
+    // 加上 source_system 等值会假阴约 7.6 万行 —— 这是实测过的陷阱。
+    [Fact]
+    public void T1b_Provenance_crosscheck_must_not_join_on_source_system()
+    {
+        var sql = Migration();
+        Assert.Contains("s.`source_table` = 'RoutingOpDetail'", sql, Ord);
+        Assert.Contains("s.`source_biz_key` = o.`source_biz_key`", sql, Ord);
+        Assert.DoesNotContain("s.`source_system` = o.`source_system`", sql, Ord);
+        Assert.DoesNotContain("s.source_system = o.source_system", sql, Ord);
+    }
+
+    // ── T2 · 幸存者：必须断言 std 不被清空，且 DELETE 有精确 WHERE ──────────────
+    [Fact]
+    public void T2_Std_cleanup_is_scoped_and_never_truncates_the_table()
+    {
+        var sql = Migration();
+        Assert.Contains("DELETE FROM `mdp_std_process_outsource_order`", sql, Ord);
+        Assert.DoesNotContain("TRUNCATE TABLE `mdp_std_process_outsource_order`", sql, Ord);
+
+        // 清理后必须立即断言幸存者数量与门禁阶段一致
+        Assert.Contains("@std_after <> @survivors", sql, Ord);
+
+        // verify 侧必须防「整表被清空」
+        Assert.Contains("FROM `mdp_std_process_outsource_order`) > 0", Verify(), Ord);
+    }
+
+    // ── T3 · 新唯一键：Option A 粒度，且旧键必须被移除 ──────────────────────────
+    [Fact]
+    public void T3_New_unique_key_matches_option_a_grain_and_old_key_is_dropped()
+    {
+        var sql = Migration();
+        Assert.Contains(
+            "ADD UNIQUE INDEX `uk_po_line_stat` (`tenant_id`,`factory_id`,`stat_date`,`po_no`,`po_line`)",
+            sql, Ord);
+        Assert.Contains("DROP INDEX `uk_work_op_stat`", sql, Ord);
+
+        // verify 必须钉死列序，而不只是「索引存在」
+        Assert.Contains("'tenant_id,factory_id,stat_date,po_no,po_line'", Verify(), Ord);
+        // 旧键必须已消失
+        Assert.Contains("INDEX_NAME='uk_work_op_stat'", Verify(), Ord);
+    }
+
+    // ── T4 · NULL 收口：两列都必须 NOT NULL DEFAULT ''，且 verify 连空串也拦 ────
+    [Fact]
+    public void T4_Both_key_columns_are_hardened_to_not_null()
+    {
+        var sql = Migration();
+        Assert.Contains("MODIFY COLUMN `po_no` VARCHAR(100) NOT NULL DEFAULT ''", sql, Ord);
+        Assert.Contains("MODIFY COLUMN `po_line` VARCHAR(50) NOT NULL DEFAULT ''", sql, Ord);
+
+        var v = Verify();
+        Assert.Contains("WHERE `po_no`   IS NULL) = 0", v, Ord);
+        Assert.Contains("WHERE `po_line` IS NULL) = 0", v, Ord);
+        Assert.Contains("WHERE `po_no`   = '')    = 0", v, Ord);
+        Assert.Contains("WHERE `po_line` = '')    = 0", v, Ord);
+        Assert.Contains("IS_NULLABLE='YES'", v, Ord);
+    }
+
+    // ── T5 · 非法租户：回插必须显式排除，不依赖隐式 filter ─────────────────────
+    [Fact]
+    public void T5_Restore_explicitly_excludes_illegal_tenants()
+    {
+        var sql = Migration();
+        Assert.Contains("`tenant_id` NOT IN (0,1,1300000000001)", sql, Ord);
+        Assert.Contains("WHERE `tenant_id` IN (0,1,1300000000001)) = 0", Verify(), Ord);
+    }
+
+    // ── T6 · 31 行保全往返：谓词必须是 IFNULL(po_no,'')<>''，不能退回 IS NOT NULL ──
+    // 这是本批最容易静默出错的一处：代码侧 IFNULL 补丁上线后，种群 A 的 po_no
+    // 变成**空串**而非 NULL；用 po_no IS NOT NULL 会把垃圾一并当成不可再生历史保下来
+    // （实测当时会多保 56 行）。
+    [Fact]
+    public void T6_Preservation_predicate_excludes_empty_string_not_just_null()
+    {
+        var sql = Migration();
+        Assert.Contains("WHERE IFNULL(`po_no`,'') <> ''", sql, Ord);
+
+        // 保全 / 回插 / verify 三处口径必须一致，且都不得出现裸 IS NOT NULL
+        Assert.DoesNotContain("WHERE `po_no` IS NOT NULL", sql, Ord);
+        Assert.DoesNotContain("WHERE po_no IS NOT NULL", sql, Ord);
+
+        Assert.Contains("CREATE TABLE IF NOT EXISTS `dwd_process_outsource_delivery_keep_p0a`", sql, Ord);
+        Assert.Contains("INSERT INTO `dwd_process_outsource_delivery`", sql, Ord);
+
+        // verify 必须做双向差集，而不是只比行数
+        var v = Verify();
+        Assert.Contains("NOT EXISTS", v, Ord);
+        Assert.Contains("dwd_process_outsource_delivery_keep_p0a", v, Ord);
+    }
+
+    // ── T7 · 执行顺序铁律：清数据必须在改 schema 之前 ──────────────────────────
+    // 反序（先 NOT NULL / 先换 UK，再清种群 A）会把大量不同主数据压进少数键，
+    // 把「膨胀」换成更坏的「静默丢数」。
+    [Fact]
+    public void T7_Data_cleanup_must_precede_schema_hardening()
+    {
+        var sql = Migration();
+        var stdDelete = sql.IndexOf("DELETE FROM `mdp_std_process_outsource_order`", Ord);
+        var truncate  = sql.IndexOf("TRUNCATE TABLE `dwd_process_outsource_delivery`", Ord);
+        var notNull   = sql.IndexOf("MODIFY COLUMN `po_no`", Ord);
+        var ukSwap    = sql.IndexOf("DROP INDEX `uk_work_op_stat`", Ord);
+        var restore   = sql.IndexOf("SELECT * FROM `dwd_process_outsource_delivery_keep_p0a`", Ord);
+
+        Assert.True(stdDelete > 0 && truncate > 0 && notNull > 0 && ukSwap > 0 && restore > 0);
+        Assert.True(stdDelete < truncate, "std 清理必须先于 DWD TRUNCATE");
+        Assert.True(truncate < notNull,   "schema 收口必须在空表期（TRUNCATE 之后）");
+        Assert.True(notNull  < restore,   "回插必须在 NOT NULL 收口之后");
+        Assert.True(ukSwap   < restore,   "回插必须在唯一键切换之后");
+    }
+
+    // ── T8 · 保全必须先于 TRUNCATE，且有完整性断言 ─────────────────────────────
+    [Fact]
+    public void T8_Preservation_happens_before_truncate_with_a_count_assertion()
+    {
+        var sql = Migration();
+        var backup   = sql.IndexOf("INSERT IGNORE INTO `dwd_process_outsource_delivery_keep_p0a`", Ord);
+        var assertion = sql.IndexOf("@kept <> @preserve_cnt", Ord);
+        var truncate = sql.IndexOf("TRUNCATE TABLE `dwd_process_outsource_delivery`", Ord);
+
+        Assert.True(backup > 0 && assertion > 0 && truncate > 0);
+        Assert.True(backup < assertion, "必须先备份再断言");
+        Assert.True(assertion < truncate, "行数一致性断言必须在 TRUNCATE 之前");
+    }
+
+    // ── T9 · CHECK 约束：不得硬编码假定旧名，且必须防重复 ──────────────────────
+    [Fact]
+    public void T9_Tenant_check_constraint_is_resolved_dynamically_and_deduped()
+    {
+        var sql = Migration();
+        // 按语义定位，而不是假定名字存在
+        Assert.Contains("cc.CHECK_CLAUSE LIKE '%tenant_id%'", sql, Ord);
+        Assert.Contains("@tenant_chk_cnt > 1", sql, Ord);   // 发现重复即中止
+        Assert.Contains("ck_dwd_process_outsource_valid_tenant", sql, Ord); // 收敛到稳定名
+
+        // verify 必须断言「恰好一套」
+        Assert.Contains("AND cc.CHECK_CLAUSE LIKE '%1300000000001%'", Verify(), Ord);
+    }
+
+    // ── T10 · 门禁必须 fail-fast，且在任何破坏性语句之前 ───────────────────────
+    [Fact]
+    public void T10_Guards_abort_before_any_destructive_statement()
+    {
+        var sql = Migration();
+        foreach (var g in new[] { "P0A-G1", "P0A-G2", "P0A-G3", "P0A-G4", "P0A-G5" })
+            Assert.Contains(g, sql, Ord);
+
+        var lastGuard = sql.IndexOf("P0A-G5", Ord);
+        var firstDestructive = new[]
+        {
+            sql.IndexOf("DELETE FROM `mdp_std_process_outsource_order`", Ord),
+            sql.IndexOf("TRUNCATE TABLE `dwd_process_outsource_delivery`", Ord),
+            sql.IndexOf("MODIFY COLUMN `po_no`", Ord),
+        }.Where(i => i > 0).Min();
+
+        Assert.True(lastGuard < firstDestructive,
+            "全部门禁必须在第一条破坏性语句之前完成");
+    }
+}