feat(ac15): 阶段2前置——core 结构补齐 + 资金域4表映射器 + 最小 fixture 逐列核对

本批只证明「一条链路能跑通」，不做 8 个写 SP 的比对（下一批）。

一、结构补齐（internal/academic/dualrun/fixture/ddl.go + cmd/ac15fixture ddl）
- 读 02-data-model/core-model-ddl.sql，只执行 CREATE TABLE：
  DROP TABLE 语句全部丢弃（13 条），CREATE TABLE 改写成 IF NOT EXISTS，已存在表跳过
- yuansiwei_2026 表数 28 → 40（新增 12：t_institution/t_user/t_identity/t_wechat_union/
  t_membership/t_role/t_guardian_link/t_permission/t_role_permission/casbin_rule/
  t_legacy_mapping/t_audit_log；跳过已存在 t_campus）
- 未 ALTER / TRUNCATE / DROP 任何已有表；未触碰 e2e_test

二、映射器（资金域 4 表）
- 用户 id：t_legacy_mapping，legacy int → VARCHAR(26) ULID；
  fixture ULID 确定性派生（固定时间戳 + sha256(legacy_id) 前 10 字节），保证 map/verify/clean 幂等
- 主键：t_user_expense 自然键 user_id；t_user_expense_school 靠 uk_user_inst_campus 幂等（不保留 legacy id）；
  t_user_expense_log 用 D1——显式插 legacy id
- 金额：DECIMAL(12,2)，逐行 ROUND(x,2)；读侧 CAST(... AS DECIMAL(20,3)) 取十进制串，
  写侧整数分 → 十进制串，全程不经 float64（decimal.go 为唯一换算通道）
- 时间：秒级原样保留，DATETIME(3) 补 .000，不使用当前时间
- 空值：create_user_id 0/系统 → NULL；request_id / balance_after → NULL；legacy 金额 NULL → 0.00 且告警

三、最小 fixture（userid=53691：1 账户 + 1 校区账户 + 6 条流水）
- 84 项逐列核对 + 3 条语义断言全绿；核对期望值由 MySQL 服务端 CAST/DATE_FORMAT 产生，
  不依赖本包映射代码（否则映射写错时两边一起错）
- 验证完 DELETE 清理，4 张表回 0 行

四、顺手挖到的四个坑（已固化）
- information_schema.tables.AUTO_INCREMENT 是缓存值（stats_expiry=86400，读出来是 1），
  必须同连接 SET information_schema_stats_expiry=0
- DELETE 不回退 AUTO_INCREMENT，真实表「恰好 max+1」只有首次可观测；
  另用 CREATE TEMPORARY TABLE ... LIKE 镜像可重复演示 100 → 101
- SHOW TABLE STATUS LIKE ? 不支持占位符（Error 1064）
- 若 request_id 落空串而非 NULL，同 type 历史流水会撞 uk_request_type（Error 1062）

五、变异实验（3 组，均已还原）
- M1 ROUND → 截断：6 列金额红（0.79→0.78、333.33→333.32 …）
- M2 balance_after 落 0.00：6 列 + 1 断言红
- M3 create_user_id 0 落 "0"：单测红（legacy 真实数据 create_userid=0 的行数为 0，
  该规则只能靠单测证明）

未改动 internal/academic/{service,repository,handler}；finance_bk 全程只读
（legacy 连接 SET SESSION TRANSACTION READ ONLY）。

cmd/ac15fixture/main.go

@@ -0,0 +1,190 @@
+// Command ac15fixture 是 AC-15 阶段 2 前置件的 CLI 入口。
+//
+// 用法（在仓库根目录执行）：
+//
+//	go run ./cmd/ac15fixture ddl                     # 用 core-model-ddl.sql 补齐目标库结构（只 CREATE，不 DROP）
+//	go run ./cmd/ac15fixture count                   # 打印 5 张相关表当前行数
+//	go run ./cmd/ac15fixture map    --user 53691     # 抽样搬运（不改任何已有 target 业务代码）
+//	go run ./cmd/ac15fixture verify --user 53691     # 逐列核对（退出码 0=PASS，1=FAIL）
+//	go run ./cmd/ac15fixture run    --user 53691     # map + verify
+//	go run ./cmd/ac15fixture clean  --user 53691     # 删除 fixture 数据并复核回 0 行
+//
+// 环境变量（口令不落盘、不打印）：
+//
+//	AC15_DB_HOST / AC15_DB_PORT / AC15_DB_USER / AC15_DB_PASS / AC15_TARGET_DB
+//	AC15_LEGACY_USER / AC15_LEGACY_PASS / AC15_LEGACY_DB
+//	AC15_CORE_DDL   core DDL 路径（默认取仓库外部的 02-data-model/core-model-ddl.sql）
+//
+// 退出码：0 = 成功/PASS；1 = 核对 FAIL；2 = 参数错误；3 = 环境或守卫失败。
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+
+	"yuansiwei/unified-backend/internal/academic/dualrun/fixture"
+)
+
+func main() {
+	if len(os.Args) < 2 {
+		usage()
+		os.Exit(2)
+	}
+	cmd := os.Args[1]
+	args := os.Args[2:]
+	switch cmd {
+	case "ddl":
+		os.Exit(runDDL(args))
+	case "map":
+		os.Exit(runMap(args))
+	case "verify":
+		os.Exit(runVerify(args))
+	case "run":
+		os.Exit(runAll(args))
+	case "clean":
+		os.Exit(runClean(args))
+	case "count":
+		os.Exit(runCount(args))
+	case "-h", "--help", "help":
+		usage()
+		os.Exit(0)
+	default:
+		fmt.Fprintf(os.Stderr, "未知子命令：%s\n\n", cmd)
+		usage()
+		os.Exit(2)
+	}
+}
+
+func usage() {
+	fmt.Fprint(os.Stderr, `ac15fixture —— AC-15 阶段 2 前置件（资金域 4 表映射 + 逐列核对）
+
+子命令：
+  ddl      补齐 core-model-ddl.sql 的 13 张表（只 CREATE TABLE IF NOT EXISTS，禁止 DROP）
+  map      抽样搬运一个 legacy 用户（账户 + 校区账户 + 流水）
+  verify   逐列核对（4 张表 + AUTO_INCREMENT + 空值语义 + float 陷阱）
+  run      map + verify
+  clean    删除 fixture 数据并复核 4 张表回 0 行
+  count    打印相关表当前行数
+
+通用参数：--user <legacy userid>（默认 53691）, --ddl <path>
+`)
+}
+
+// bootstrap 解析公共参数、加载配置并跑守卫。
+func bootstrap(args []string, withUser bool) (fixture.Config, int64, string, int) {
+	fs := flag.NewFlagSet("ac15fixture", flag.ContinueOnError)
+	fs.SetOutput(os.Stderr)
+	user := fs.Int64("user", fixture.DefaultLegacyUserID, "legacy userid")
+	ddl := fs.String("ddl", "", "core DDL 路径（默认 AC15_CORE_DDL 或仓库外部默认路径）")
+	if err := fs.Parse(args); err != nil {
+		return fixture.Config{}, 0, "", 2
+	}
+	if !withUser {
+		*user = fixture.DefaultLegacyUserID
+	}
+
+	repoRoot, err := os.Getwd()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "FATAL: 取工作目录失败：%v\n", err)
+		return fixture.Config{}, 0, "", 3
+	}
+	cfg, err := fixture.LoadConfig(repoRoot)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
+		return fixture.Config{}, 0, "", 3
+	}
+	if err := cfg.Guard(); err != nil {
+		fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
+		return fixture.Config{}, 0, "", 3
+	}
+	fmt.Printf("target=%s  legacy=%s\n\n", cfg.TargetMasked(), cfg.LegacyMasked())
+	return cfg, *user, *ddl, 0
+}
+
+func runDDL(args []string) int {
+	cfg, _, ddlPath, code := bootstrap(args, false)
+	if code != 0 {
+		return code
+	}
+	rep, err := fixture.ApplyCoreDDL(context.Background(), cfg, ddlPath)
+	if rep != nil {
+		fixture.PrintDDLReport(os.Stdout, rep)
+	}
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "\nFATAL: %v\n", err)
+		return 3
+	}
+	return 0
+}
+
+func runMap(args []string) int {
+	cfg, user, _, code := bootstrap(args, true)
+	if code != 0 {
+		return code
+	}
+	rep, err := fixture.PrepareAndLoad(context.Background(), cfg, user)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
+		return 3
+	}
+	fixture.PrintMapReport(os.Stdout, rep)
+	return 0
+}
+
+func runVerify(args []string) int {
+	cfg, user, _, code := bootstrap(args, true)
+	if code != 0 {
+		return code
+	}
+	rep, err := fixture.Verify(context.Background(), cfg, user)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
+		return 3
+	}
+	fixture.PrintVerifyReport(os.Stdout, rep)
+	if !rep.OK() {
+		return 1
+	}
+	return 0
+}
+
+func runAll(args []string) int {
+	if code := runMap(args); code != 0 {
+		return code
+	}
+	fmt.Println()
+	return runVerify(args)
+}
+
+func runClean(args []string) int {
+	cfg, user, _, code := bootstrap(args, true)
+	if code != 0 {
+		return code
+	}
+	rep, err := fixture.Cleanup(context.Background(), cfg, user)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
+		return 3
+	}
+	fixture.PrintCleanReport(os.Stdout, rep)
+	return 0
+}
+
+func runCount(args []string) int {
+	cfg, _, _, code := bootstrap(args, false)
+	if code != 0 {
+		return code
+	}
+	counts, err := fixture.CountTargetTables(context.Background(), cfg)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
+		return 3
+	}
+	fmt.Println("== 目标表行数 ==")
+	for _, t := range []string{"t_user", "t_legacy_mapping", "t_user_expense", "t_user_expense_school", "t_user_expense_log"} {
+		fmt.Printf("  %-24s %d\n", t, counts[t])
+	}
+	return 0
+}

internal/academic/dualrun/fixture/README.md

@@ -0,0 +1,109 @@
+# `internal/academic/dualrun/fixture` —— AC-15 阶段 2 前置件（资金域 4 表映射器）
+
+本包做一件事：**把 legacy 的一个真实用户，按目标态口径搬到 `yuansiwei_2026`，再逐列核对、清理干净**。
+
+它是「阶段 2 双跑」的**前置件**——先证明「结构 + 映射 + 取整口径」这条链能跑通，再去比 8 个写 SP 的产出。
+
+| 项 | 范围 |
+|---|---|
+| 表 | 资金域 4 表：`t_user_expense` / `t_user_expense_school` / `t_user_expense_log` + 解析用户所需的 `t_user` / `t_legacy_mapping` |
+| 不做的 | 其余 8 张资金/活动表、课程链、8 个写 SP 的双跑比对（下一批） |
+| 不碰的 | `internal/academic/{service,repository,handler}` 任何文件；`e2e_test` 库；`finance_bk` 只 SELECT |
+
+---
+
+## 1. 怎么跑
+
+```bash
+# 1) 补齐 core 结构（只 CREATE TABLE IF NOT EXISTS，DDL 里的 DROP TABLE 全部丢弃）
+go run ./cmd/ac15fixture ddl
+
+# 2) 抽样搬运（默认 userid=53691）
+go run ./cmd/ac15fixture map    --user 53691
+
+# 3) 逐列核对（0=PASS，1=FAIL）
+go run ./cmd/ac15fixture verify --user 53691
+
+# 4) 清理并复核 4 张表回 0 行
+go run ./cmd/ac15fixture clean  --user 53691
+
+# 单测（纯映射规则，不连库）
+go test ./internal/academic/dualrun/fixture/
+```
+
+环境变量（口令不落盘、不打印）：`AC15_DB_HOST/PORT/USER/PASS`、`AC15_TARGET_DB`（默认 `yuansiwei_2026`）、
+`AC15_LEGACY_DB`（默认 `finance_bk`）、`AC15_LEGACY_USER/PASS`、`AC15_CORE_DDL`。
+
+未设 `AC15_DB_PASS` 时回退读仓库既有 `config/dev.json` 的 `db.dsn`（只读，不新增口令副本）。
+
+守卫：host 必须本机；target 库必须在允许清单内且**只允许写 `yuansiwei_2026`**；
+legacy 连接建立后立刻 `SET SESSION TRANSACTION READ ONLY`，写语句会被服务端拒绝。
+
+## 2. 文件
+
+| 文件 | 职责 |
+|---|---|
+| `config.go` | 连接参数、DSN、白名单守卫、脱敏 |
+| `decimal.go` | 十进制串 ↔ 整数分（**唯一金额换算通道**，全程不经过 float64） |
+| `model.go` | legacy 行 / 目标行 / 报告结构 |
+| `mapping.go` | **纯映射规则**（取整、空值语义、时间、D1 id、确定性 ULID），可单测 |
+| `extract.go` | legacy 只读查询（金额一律 `CAST(... AS DECIMAL(20,3))`） |
+| `load.go` | 幂等写入（`ON DUPLICATE KEY UPDATE`） |
+| `verify.go` | 逐列核对 + 语义断言 + float 陷阱 + AUTO_INCREMENT |
+| `ddl.go` | 把 `core-model-ddl.sql` 安全套到目标库（丢弃 DROP、改写 IF NOT EXISTS） |
+| `run.go` | 流程编排（PrepareAndLoad / Cleanup） |
+| `report.go` | 终端报告排版 |
+| `mapping_test.go` | 纯规则单测（含变异实验的红/绿基线） |
+| `cmd/ac15fixture` | CLI 入口 |
+
+## 3. 映射规则（已拍板，代码即口径）
+
+1. **用户 id 走 `t_legacy_mapping`**：`legacy_academic_user_id`(int) → `canonical_user_id`(VARCHAR(26) ULID)。
+   fixture 的 ULID 由 `FixtureCanonicalUserID(legacyID)` **确定性**派生
+   （时间戳固定 2026-01-01 + 熵 = `sha256("ac15-fixture:user:<id>")[:10]`）：
+   同一 legacy id 每次得到同一 ULID，于是 map/verify/clean 三步可独立执行、天然幂等。
+   **生产迁移的 canonical_user_id 由归一 pipeline 的种子脚本发放，与此无关。**
+   注意：流水里出现的 4 个「创建人」也各插 1 行映射 —— 否则 `create_user_id` 会全部退化成 NULL，
+   「解析成功」那一半就没被验证到。
+2. **主键**：`t_user_expense` 用自然键 `user_id`（无自增 id）；`t_user_expense_school` 用 `uk_user_inst_campus`
+   做幂等，**不保留** legacy id（对齐键是业务复合键）；`t_user_expense_log` 用 **D1**。
+3. **D1（流水表）**：`INSERT ... (id, ...) VALUES (<legacy_id>, ...)`，同一 legacy 流水在目标侧 `id` 完全相同。
+   **不从 legacy id 反算 target id**。
+4. **金额 `DECIMAL(12,2)`**：**逐行 `ROUND(x,2)`**，不是 `ROUND(SUM,2)`；
+   读侧 `CAST(x AS DECIMAL(20,3))` 拿十进制串，写侧把整数分格式化成 `"-2160.49"` 交给驱动，全程无 float。
+5. **时间 `DATETIME(3)`**：legacy `datetime(0)` 秒级**原样**保留，毫秒补 `.000`，**不使用当前时间**。
+6. **空值语义（逐列判断）**：
+   - `create_user_id`：legacy `0` / NULL / 无映射 → **NULL**（不是 0、不是空串）
+   - `request_id`：legacy 无此列 → **NULL**
+   - `balance_after`：legacy 无快照 → **NULL**
+   - legacy 金额列 NULL（目标 NOT NULL）→ 落 `0.00`，但**必须产生告警**（NULL ≠ 0，业务上是两件事）
+7. **`remark` text → varchar(1000)**：按**字符**截断（不切半个汉字），超长产生告警。本 fixture 最长 271 字符，未触发。
+
+## 4. 核对为什么可信（期望值的来源）
+
+核对**不用本包的映射代码产生期望值**——否则映射写错时两边一起错，核对永远是绿的。
+
+期望值一律由 **MySQL 服务端**算：`CAST(x AS DECIMAL(12,2))`、`DATE_FORMAT(t,'%Y-%m-%d %H:%i:%s.%f')`。
+于是 Go 侧的取整/换算只要与 DECIMAL 语义不一致，核对立刻红；变异实验（故意改错映射）必然红。
+
+## 5. 踩到的四个真坑（都已固化在代码里，别再踩）
+
+1. **`information_schema.tables.AUTO_INCREMENT` 是缓存值**：MySQL 8.0 默认
+   `information_schema_stats_expiry = 86400`，读出来是 `1`。必须
+   `SET SESSION information_schema_stats_expiry = 0` 且 SET 与 SELECT 走**同一条连接**（连接池会换连接）。
+2. **`DELETE` 不回退 AUTO_INCREMENT 计数器**：fixture 反复跑时计数器单调增长，
+   真实表上「恰好 max+1」只有首次能观测到；因此另配 `CREATE TEMPORARY TABLE ... LIKE` 镜像，
+   每次都能可重复地演示「显式插 id=100 → 下一条 101」。
+3. **`SHOW TABLE STATUS LIKE ?` 不支持占位符**（Error 1064），表名只能内联（已加字符白名单校验）。
+4. **`t_user_expense_log.uk_request_type(request_id, type)`**：若把 `request_id` 落成空串而不是 NULL，
+   同 `type` 的多条历史流水会撞唯一键（Error 1062）。NULL 不参与唯一键比较，这才是历史数据的正确落法。
+
+## 6. 已知事实（不是缺陷，是数据现状）
+
+- legacy `t_user_expense_log` 里 `create_userid = 0` 或 NULL 的行数 = **0**（90007 行全都有真实创建人）。
+  所以「0 → NULL」这条规则**无法用真实数据证明**，只能由 `mapping_test.go` 单测 + 变异实验 M3 证明。
+- legacy `t_user_expense.balance/spend/total_income/total_expenditure` 是 `float(10,3)`，
+  `t_user_expense_school.*` 与 `t_user_expense_log.money` 是 `decimal(10,3)`——**同一业务列两边类型不同**，
+  所以读侧统一 `CAST(... AS DECIMAL(20,3))`，不按列假设类型。
+- `t_user_expense_school.schoolid` 直迁为 `institution_id`（本 fixture 取到 63317），
+  org 域（机构/校区）映射属另一条链路，本批不涉及。

internal/academic/dualrun/fixture/config.go

@@ -0,0 +1,298 @@
+// Package fixture 实现 AC-15 阶段 2 的**前置件**：把 legacy 资金域 4 张表的一条链路
+// 搬到目标库 yuansiwei_2026，并逐列核对。
+//
+// 定位与边界（务必先读）：
+//   - 本包只做**资金域 4 表**：t_user_expense / t_user_expense_school / t_user_expense_log，
+//     以及解析用户所需的 t_user + t_legacy_mapping。其余 7 张资金/活动表与 8 个写 SP 的
+//     双跑比对**不在本包范围**（下一批）。
+//   - 本包**不改** internal/academic/{service,repository,handler}：映射规则全部在此独立实现，
+//     避免为了「能跑通」去污染生产路径。
+//   - 本包唯一的写目标是 yuansiwei_2026（fixture 数据，用完 DELETE）与 legacy 侧只读连接。
+//     finance_bk **只 SELECT**。
+//
+// 三条落地口径（来自已拍板结论，不再讨论）：
+//  1. 用户 id 走 t_legacy_mapping：legacy int user_id → canonical_user_id(VARCHAR(26) ULID)。
+//  2. 主键：t_user_expense 的键是 user_id（无自增 id）；t_user_expense_log 用 D1
+//     ——**显式插入 legacy id**，双跑按 id 直接对齐。不试图从 legacy id 反算 target id。
+//  3. 金额 DECIMAL(12,2)：**逐行 ROUND(x,2)**，不是 ROUND(SUM,2)；且不用 float 运算，
+//     用「十进制字符串 → 整数分」精确换算（见 decimal.go）。
+package fixture
+
+import (
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"math"
+	"os"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	"gorm.io/driver/mysql"
+	"gorm.io/gorm"
+	"gorm.io/gorm/logger"
+)
+
+// 允许连接的 host（allow-list，非 deny-list）：只允许本机。
+var allowedHosts = map[string]bool{
+	"127.0.0.1": true,
+	"localhost": true,
+	"::1":       true,
+}
+
+// 允许连接的库。finance_bk 只读；yuansiwei_2026 是本包唯一写目标。
+var (
+	allowedDatabases = map[string]bool{
+		"finance_bk":     true,
+		"legacy_oracle":  true,
+		"yuansiwei_2026": true,
+	}
+	// writableDatabases 明确「哪些库允许写」：目标库只白名单 yuansiwei_2026。
+	writableDatabases = map[string]bool{
+		"yuansiwei_2026": true,
+	}
+	// readOnlyDatabases 明确「哪些库绝不允许写」——写语句一律在 guard 层拒绝。
+	readOnlyDatabases = map[string]bool{
+		"finance_bk":    true,
+		"legacy_oracle": true,
+		// e2e_test 有独立的建表逻辑，本包不碰。
+		"e2e_test": true,
+	}
+)
+
+// DefaultLegacyUserID 是 fixture 选中的 legacy 用户（选择依据见 README/报告）：
+// t_user_expense 1 行 + t_user_expense_school 1 行 + t_user_expense_log 6 条，
+// 且 balance=0.788 / spend=4999.212 带 3 位小数——覆盖「逐行 ROUND 到分」的取值边界。
+const DefaultLegacyUserID int64 = 53691
+
+// Config 是 fixture 的两条连接参数。
+//
+// 口令**只经环境变量或已有配置文件读取，不打印、不落盘、不进报告**。
+type Config struct {
+	Host string
+	Port int
+
+	TargetDB   string
+	TargetUser string
+	TargetPass string
+
+	LegacyDB   string
+	LegacyUser string
+	LegacyPass string
+}
+
+// dsnFile 是仓库里已存在的 dev 配置（口令已在磁盘上，本包只读取、不新增副本）。
+type dsnFile struct {
+	DB struct {
+		DSN string `json:"dsn"`
+	} `json:"db"`
+}
+
+var dsnPattern = regexp.MustCompile(`^([^:]+):([^@]+)@tcp\(([^:]+):(\d+)\)/([^?]+)`)
+
+// distMasked 返回不含口令的 DSN 形态，用于日志与报告。
+func (c Config) TargetMasked() string {
+	return fmt.Sprintf("%s@tcp(%s:%d)/%s", c.TargetUser, c.Host, c.Port, c.TargetDB)
+}
+
+// LegacyMasked 返回 legacy 侧不含口令的 DSN 形态。
+func (c Config) LegacyMasked() string {
+	return fmt.Sprintf("%s@tcp(%s:%d)/%s", c.LegacyUser, c.Host, c.Port, c.LegacyDB)
+}
+
+// LoadConfig 读取连接参数。
+//
+// 优先级：环境变量（AC15_*）> 仓库内 config/dev.json 的 db.dsn。
+// 之所以保留文件回退：dev 环境里 AC15_DB_PASS 未必已导出，而 dev.json 是仓库既有的
+// 配置载体（不新增口令副本、不打印口令）。生产/CI 走环境变量即可。
+func LoadConfig(repoRoot string) (Config, error) {
+	cfg := Config{
+		Host:       envOr("AC15_DB_HOST", "127.0.0.1"),
+		Port:       envIntOr("AC15_DB_PORT", 3306),
+		TargetDB:   envOr("AC15_TARGET_DB", "yuansiwei_2026"),
+		TargetUser: envOr("AC15_DB_USER", "dev"),
+		TargetPass: os.Getenv("AC15_DB_PASS"),
+		LegacyDB:   envOr("AC15_LEGACY_DB", "finance_bk"),
+		LegacyUser: envOr("AC15_LEGACY_USER", ""),
+		LegacyPass: os.Getenv("AC15_LEGACY_PASS"),
+	}
+
+	// 目标库口令缺失时回退到仓库既有 dev.json（只读其 db.dsn，不写回）。
+	if cfg.TargetPass == "" || cfg.TargetUser == "dev" {
+		path := filepath.Join(repoRoot, "config", "dev.json")
+		if raw, err := os.ReadFile(path); err == nil {
+			var f dsnFile
+			if err := json.Unmarshal(raw, &f); err == nil {
+				if m := dsnPattern.FindStringSubmatch(f.DB.DSN); m != nil {
+					if cfg.TargetPass == "" {
+						cfg.TargetPass = m[2]
+					}
+					if os.Getenv("AC15_DB_USER") == "" {
+						cfg.TargetUser = m[1]
+					}
+				}
+			}
+		}
+	}
+
+	// legacy 侧默认复用同一账号（本机 dev 账号对 finance_bk 有只读权限）；
+	// 也可用 AC15_LEGACY_USER / AC15_LEGACY_PASS 单独指定。
+	if cfg.LegacyUser == "" {
+		cfg.LegacyUser = cfg.TargetUser
+	}
+	if cfg.LegacyPass == "" {
+		cfg.LegacyPass = cfg.TargetPass
+	}
+
+	if cfg.TargetPass == "" {
+		return Config{}, fmt.Errorf("目标库口令不可得：请设置 AC15_DB_PASS 或在 %s/config/dev.json 提供 db.dsn", repoRoot)
+	}
+	return cfg, nil
+}
+
+func envOr(key, def string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return def
+}
+
+func envIntOr(key string, def int) int {
+	if v := os.Getenv(key); v != "" {
+		if n, err := strconv.Atoi(v); err == nil && n > 0 {
+			return n
+		}
+	}
+	return def
+}
+
+// Guard 校验 host 与库名白名单，并对**写目标**做更严的判定。
+//
+// 返回 error 而非 panic：本包由 CLI 串成多步，单步失败要能定位到是「守卫」还是「数据」，
+// 因此这里不做 fail-fast panic（与 dualrun 包阶段 1 的 panic 语义不同，后者是单次运行）。
+func (c Config) Guard() error {
+	if !allowedHosts[c.Host] {
+		return fmt.Errorf("数据源守卫失败：host %q 不在允许清单内（只允许本机）", maskHost(c.Host))
+	}
+	if !allowedDatabases[c.TargetDB] {
+		return fmt.Errorf("数据源守卫失败：target 库 %q 不在允许清单内", c.TargetDB)
+	}
+	if !writableDatabases[c.TargetDB] {
+		return fmt.Errorf("数据源守卫失败：%q 不是本包允许写的库（只允许 yuansiwei_2026）", c.TargetDB)
+	}
+	if !allowedDatabases[c.LegacyDB] {
+		return fmt.Errorf("数据源守卫失败：legacy 库 %q 不在允许清单内", c.LegacyDB)
+	}
+	if readOnlyDatabases[c.LegacyDB] == false {
+		// legacy_oracle 虽在 allowedDatabases 内，但本包对 legacy 侧一律只读；
+		// 这里只做提示性约束：真正保证由 extract.go 只发 SELECT 实现。
+		_ = c.LegacyDB
+	}
+	return nil
+}
+
+// maskHost 把 host 后两段遮掉，保证日志/报告不出现完整地址。
+func maskHost(host string) string {
+	parts := strings.Split(host, ".")
+	if len(parts) == 4 {
+		return parts[0] + "." + parts[1] + ".***.**"
+	}
+	return "<masked>"
+}
+
+func (c Config) targetDSN() string {
+	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=true&loc=Local&timeout=5s",
+		c.TargetUser, c.TargetPass, c.Host, c.Port, c.TargetDB)
+}
+
+func (c Config) legacyDSN() string {
+	return fmt.Sprintf("%s:%s@tcp(%s:%d)/%s?charset=utf8mb4&parseTime=true&loc=Local&timeout=5s",
+		c.LegacyUser, c.LegacyPass, c.Host, c.Port, c.LegacyDB)
+}
+
+// openGorm 用 gorm 的 mysql driver 打开连接（复用仓库既有依赖，不新增 driver）。
+func openGorm(dsn string) (*sql.DB, error) {
+	gdb, err := gorm.Open(mysql.Open(dsn), &gorm.Config{
+		Logger:               logger.Default.LogMode(logger.Silent),
+		DisableAutomaticPing: false,
+	})
+	if err != nil {
+		return nil, err
+	}
+	sqlDB, err := gdb.DB()
+	if err != nil {
+		return nil, err
+	}
+	sqlDB.SetMaxOpenConns(2)
+	sqlDB.SetMaxIdleConns(2)
+	sqlDB.SetConnMaxLifetime(5 * time.Minute)
+	return sqlDB, nil
+}
+
+// OpenTarget 打开目标库连接，并二次核对 SELECT DATABASE()。
+func (c Config) OpenTarget() (*sql.DB, error) {
+	db, err := openGorm(c.targetDSN())
+	if err != nil {
+		return nil, fmt.Errorf("连接目标库失败：%w", err)
+	}
+	if err := assertDatabase(db, c.TargetDB); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	return db, nil
+}
+
+// OpenLegacy 打开 legacy **只读**连接。
+//
+// 只读不是口头纪律：连接建立后立刻 `SET SESSION TRANSACTION READ ONLY`，
+// 服务端会把本会话的写语句直接拒绝（MySQL 5.6+ 支持 session 级 read only）。
+func (c Config) OpenLegacy() (*sql.DB, error) {
+	db, err := openGorm(c.legacyDSN())
+	if err != nil {
+		return nil, fmt.Errorf("连接 legacy 库失败：%w", err)
+	}
+	if err := assertDatabase(db, c.LegacyDB); err != nil {
+		_ = db.Close()
+		return nil, err
+	}
+	if _, err := db.Exec("SET SESSION TRANSACTION READ ONLY"); err != nil {
+		_ = db.Close()
+		return nil, fmt.Errorf("legacy 会话只读设置失败（拒绝在可写会话上跑比对）：%w", err)
+	}
+	return db, nil
+}
+
+// assertDatabase 让服务器自己报出当前库再比对（只写 DSN 库名不够）。
+func assertDatabase(db *sql.DB, want string) error {
+	var got sql.NullString
+	if err := db.QueryRow("SELECT DATABASE()").Scan(&got); err != nil {
+		return err
+	}
+	if !got.Valid || got.String != want {
+		return fmt.Errorf("当前库 %q 与期望 %q 不一致", got.String, want)
+	}
+	return nil
+}
+
+// moneyToleranceCents 是金额比对的硬容差：0 分——逐分必须相等。
+const moneyToleranceCents int64 = 0
+
+// absInt64 取绝对值（仅用于报告展示差值）。
+func absInt64(v int64) int64 {
+	if v < 0 {
+		return -v
+	}
+	return v
+}
+
+// float64Of 把十进制字符串解析成 float64——**仅用于「float 陷阱」演示**，
+// 生产映射路径禁止用 float 做金额运算（见 decimal.go 的整数分换算）。
+func float64Of(s string) float64 {
+	v, err := strconv.ParseFloat(strings.TrimSpace(s), 64)
+	if err != nil {
+		return math.NaN()
+	}
+	return v
+}

internal/academic/dualrun/fixture/ddl.go

@@ -0,0 +1,194 @@
+package fixture
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"os"
+	"regexp"
+	"strings"
+)
+
+// 本文件把 02-data-model/core-model-ddl.sql 安全地套到目标库上，规则只有三条：
+//
+//  1. **只执行 CREATE TABLE**（外加 DDL 自带的 INSERT IGNORE 种子与两个 SET）。
+//     `DROP TABLE IF EXISTS` 一律**丢弃**——DDL 文件是「干净起步」口径写的，
+//     而目标库已经有 28 张表（27 academic + t_campus），照抄会删库。
+//  2. `CREATE TABLE x` 一律改写成 `CREATE TABLE IF NOT EXISTS x`，
+//     已存在的表（如 t_campus）跳过而不报错中断。
+//  3. 绝不 ALTER / TRUNCATE / DROP 任何已有表；不碰 e2e_test。
+
+// defaultCoreDDLPath 是 core DDL 的默认位置（可用 AC15_CORE_DDL 覆盖）。
+const defaultCoreDDLPath = `C:\Users\Run\WorkBuddy\项目合并\unified\02-data-model\core-model-ddl.sql`
+
+// DDLAction 描述一条语句的实际处置结果。
+type DDLAction struct {
+	Kind          string // create / seed / set / skipped-drop
+	Target        string // 表名或说明
+	Executed      bool
+	ExistedBefore bool
+	Err           string
+}
+
+// DDLReport 是一次补齐操作的结果。
+type DDLReport struct {
+	Path         string
+	Actions      []DDLAction
+	TablesBefore int
+	TablesAfter  int
+	Created      []string
+	Skipped      []string
+}
+
+var (
+	createTableRe       = regexp.MustCompile("(?i)^CREATE\\s+TABLE\\s+(?:IF\\s+NOT\\s+EXISTS\\s+)?`?([A-Za-z0-9_]+)`?")
+	insertIntoRe        = regexp.MustCompile("(?i)^INSERT\\s+(?:IGNORE\\s+)?INTO\\s+`?([A-Za-z0-9_]+)`?")
+	backtickAfterCreate = regexp.MustCompile("(?i)^CREATE\\s+TABLE\\s+`")
+)
+
+// SplitStatements 去掉纯注释行后按 `;` 切分成语句列表。
+//
+// DDL 文件里的注释全部是**整行注释**（行首 `--`），没有行尾注释，
+// 因此「先剔行再切分」是安全的；一旦将来出现行尾注释，这里会立刻在解析阶段暴露（语句变非法）而不是悄悄漏建表。
+func SplitStatements(src string) []string {
+	var body strings.Builder
+	for _, line := range strings.Split(src, "\n") {
+		t := strings.TrimSpace(line)
+		if t == "" || strings.HasPrefix(t, "--") {
+			continue
+		}
+		body.WriteString(line)
+		body.WriteString("\n")
+	}
+	var out []string
+	for _, stmt := range strings.Split(body.String(), ";") {
+		if s := strings.TrimSpace(stmt); s != "" {
+			out = append(out, s)
+		}
+	}
+	return out
+}
+
+// countTables 统计某个 schema 的表数。
+func countTables(ctx context.Context, db *sql.DB, schema string) (int, error) {
+	var n int
+	err := db.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = ?`, schema).Scan(&n)
+	return n, err
+}
+
+// tableExists 判断表是否已存在。
+func tableExists(ctx context.Context, db *sql.DB, schema, table string) (bool, error) {
+	var n int
+	err := db.QueryRowContext(ctx,
+		`SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = ? AND table_name = ?`,
+		schema, table).Scan(&n)
+	return n > 0, err
+}
+
+// ApplyCoreDDL 把 core DDL 补齐到目标库。
+//
+// ddlPath 为空时用 defaultCoreDDLPath（环境变量 AC15_CORE_DDL 优先）。
+func ApplyCoreDDL(ctx context.Context, cfg Config, ddlPath string) (*DDLReport, error) {
+	if ddlPath == "" {
+		ddlPath = os.Getenv("AC15_CORE_DDL")
+	}
+	if ddlPath == "" {
+		ddlPath = defaultCoreDDLPath
+	}
+	raw, err := os.ReadFile(ddlPath)
+	if err != nil {
+		return nil, fmt.Errorf("读取 core DDL 失败（%s）：%w", ddlPath, err)
+	}
+	if !strings.Contains(strings.ToUpper(string(raw)), "CREATE TABLE") {
+		return nil, fmt.Errorf("core DDL 文件内容异常（无 CREATE TABLE）：%s", ddlPath)
+	}
+
+	db, err := cfg.OpenTarget()
+	if err != nil {
+		return nil, err
+	}
+	defer db.Close()
+
+	before, err := countTables(ctx, db, cfg.TargetDB)
+	if err != nil {
+		return nil, err
+	}
+	rep := &DDLReport{Path: ddlPath, TablesBefore: before}
+
+	for _, stmt := range SplitStatements(string(raw)) {
+		upper := strings.ToUpper(stmt)
+
+		// 1) DROP TABLE —— 一律跳过（目标是「补齐」，不是「重建」）。
+		if strings.HasPrefix(upper, "DROP TABLE") {
+			rep.Actions = append(rep.Actions, DDLAction{Kind: "skipped-drop", Target: firstLine(stmt), Executed: false})
+			continue
+		}
+
+		// 2) SET ... —— 会话级开关，照执行（不改变任何数据）。
+		if strings.HasPrefix(upper, "SET ") {
+			if _, err := db.ExecContext(ctx, stmt); err != nil {
+				rep.Actions = append(rep.Actions, DDLAction{Kind: "set", Target: firstLine(stmt), Err: err.Error()})
+				return rep, fmt.Errorf("执行 SET 失败：%w", err)
+			}
+			rep.Actions = append(rep.Actions, DDLAction{Kind: "set", Target: firstLine(stmt), Executed: true})
+			continue
+		}
+
+		// 3) CREATE TABLE —— 改写成 IF NOT EXISTS，已存在则跳过。
+		if m := createTableRe.FindStringSubmatch(stmt); m != nil {
+			table := m[1]
+			exists, err := tableExists(ctx, db, cfg.TargetDB, table)
+			if err != nil {
+				return rep, err
+			}
+			if exists {
+				rep.Skipped = append(rep.Skipped, table)
+				rep.Actions = append(rep.Actions, DDLAction{Kind: "create", Target: table, Executed: false, ExistedBefore: true})
+				continue
+			}
+			fixed := stmt
+			if backtickAfterCreate.MatchString(fixed) {
+				idx := strings.Index(fixed, "`")
+				fixed = fixed[:idx] + "IF NOT EXISTS " + fixed[idx:]
+			}
+			if _, err := db.ExecContext(ctx, fixed); err != nil {
+				rep.Actions = append(rep.Actions, DDLAction{Kind: "create", Target: table, Err: err.Error()})
+				return rep, fmt.Errorf("建表 %s 失败：%w", table, err)
+			}
+			rep.Created = append(rep.Created, table)
+			rep.Actions = append(rep.Actions, DDLAction{Kind: "create", Target: table, Executed: true})
+			continue
+		}
+
+		// 4) INSERT IGNORE 种子 —— 幂等（IGNORE + 主键冲突不动），照执行。
+		if m := insertIntoRe.FindStringSubmatch(stmt); m != nil {
+			if _, err := db.ExecContext(ctx, stmt); err != nil {
+				rep.Actions = append(rep.Actions, DDLAction{Kind: "seed", Target: m[1], Err: err.Error()})
+				return rep, fmt.Errorf("插入种子 %s 失败：%w", m[1], err)
+			}
+			rep.Actions = append(rep.Actions, DDLAction{Kind: "seed", Target: m[1], Executed: true})
+			continue
+		}
+
+		// 5) 其它语句一律拒绝执行（白名单策略：没预期到的语句不跑）。
+		rep.Actions = append(rep.Actions, DDLAction{Kind: "rejected", Target: firstLine(stmt), Executed: false,
+			Err: "不在允许的语句类型（CREATE TABLE / INSERT IGNORE / SET）内"})
+		return rep, fmt.Errorf("DDL 中出现未预期的语句，拒绝执行：%s", firstLine(stmt))
+	}
+
+	after, err := countTables(ctx, db, cfg.TargetDB)
+	if err != nil {
+		return rep, err
+	}
+	rep.TablesAfter = after
+	return rep, nil
+}
+
+// firstLine 取语句首行（多为 60~80 字符），仅用于报告展示。
+func firstLine(s string) string {
+	if i := strings.IndexByte(s, '\n'); i >= 0 {
+		return strings.TrimSpace(s[:i])
+	}
+	return strings.TrimSpace(s)
+}

internal/academic/dualrun/fixture/decimal.go

@@ -0,0 +1,117 @@
+package fixture
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+)
+
+// 本文件是金额换算的唯一通道。
+//
+// 为什么不用 float：
+//   legacy 的 t_user_expense.balance 是 `float(10,3)`（二进制浮点），
+//   legacy 的 t_user_expense_log.money 是 `decimal(10,3)`（精确小数），
+//   目标态两列都是 `decimal(12,2)`。
+//
+// 若用 float64 读 —— 0.788 读进来是 0.78800000000000003，乘 100 再 Round 会
+// 在恰好 x.xx5 的边界上随机偏向（12.015*100 = 1201.4999999999998 → 落成 12.01，
+// 而 MySQL 的 DECIMAL ROUND 给 12.02）。**双跑对账不能用「大概相等」蒙过去**，
+// 所以这里全部走「十进制字符串 → 整数分」的精确整数运算：
+//   - 读取侧一律 `CAST(col AS DECIMAL(20,3))`，让 MySQL 先把浮点按 3 位小数的
+//     十进制表示交给我们（该列定义本来就是 3 位），再用整数运算降到 2 位。
+//   - 写入侧一律把「整数分」格式化成 "-2160.49" 这样的十进制字符串交给驱动，
+//     让 DECIMAL 列精确接收，全程不经过 float64。
+
+// DecimalToCents 把十进制字符串精确换算成「分」，小数第 3 位四舍五入（half away from zero）。
+//
+// 入参形如 "0.788" / "-2160.489" / "5000.000" / "0" / "12"。
+// 约束：最多 3 位小数（legacy 精度就是 3 位）；超过 3 位视为调用方错误，
+// 直接报错而不是静默截断——静默截断正是「看起来对、其实差一分」的来源。
+func DecimalToCents(s string) (int64, error) {
+	raw := s
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return 0, fmt.Errorf("DecimalToCents: 空字符串（raw=%q）", raw)
+	}
+
+	neg := false
+	switch s[0] {
+	case '-':
+		neg = true
+		s = s[1:]
+	case '+':
+		s = s[1:]
+	}
+	if s == "" {
+		return 0, fmt.Errorf("DecimalToCents: 只有符号（raw=%q）", raw)
+	}
+
+	intPart := s
+	fracPart := ""
+	if i := strings.IndexByte(s, '.'); i >= 0 {
+		intPart, fracPart = s[:i], s[i+1:]
+	}
+	if len(fracPart) > 3 {
+		return 0, fmt.Errorf("DecimalToCents: 小数位超过 3 位（raw=%q），拒绝静默截断", raw)
+	}
+	// 右补 0 到 3 位，便于统一按「分 + 千分位」处理。
+	for len(fracPart) < 3 {
+		fracPart += "0"
+	}
+
+	var intVal int64
+	if intPart != "" {
+		v, err := strconv.ParseInt(intPart, 10, 64)
+		if err != nil {
+			return 0, fmt.Errorf("DecimalToCents: 整数部分非法（raw=%q）：%w", raw, err)
+		}
+		intVal = v
+	}
+	frac3, err := strconv.Atoi(fracPart)
+	if err != nil {
+		return 0, fmt.Errorf("DecimalToCents: 小数部分非法（raw=%q）：%w", raw, err)
+	}
+
+	cents := intVal*100 + int64(frac3/10)
+	// 第 3 位小数 >= 5 进位（对绝对值进位，负数同向，即 half away from zero，
+	// 与 MySQL DECIMAL 的 ROUND 行为一致）。
+	if frac3%10 >= 5 {
+		cents++
+	}
+	if neg {
+		cents = -cents
+	}
+	return cents, nil
+}
+
+// CentsToDecimal 把「分」格式化成 2 位小数的十进制字符串（写入 DECIMAL(12,2) 用）。
+func CentsToDecimal(cents int64) string {
+	neg := cents < 0
+	if neg {
+		cents = -cents
+	}
+	sign := ""
+	if neg {
+		sign = "-"
+	}
+	return fmt.Sprintf("%s%d.%02d", sign, cents/100, cents%100)
+}
+
+// MustCents 用于测试与 fixture 内部常量，出错即 panic。
+func MustCents(s string) int64 {
+	c, err := DecimalToCents(s)
+	if err != nil {
+		panic(err)
+	}
+	return c
+}
+
+// RoundDecimalString 把十进制字符串四舍五入到 2 位并返回 2 位小数字符串。
+// 等价于 SQL 的 `ROUND(x, 2)`，用于核对表里「legacy 值 → 迁移后期望值」这一列。
+func RoundDecimalString(s string) (string, error) {
+	c, err := DecimalToCents(s)
+	if err != nil {
+		return "", err
+	}
+	return CentsToDecimal(c), nil
+}

internal/academic/dualrun/fixture/extract.go

@@ -0,0 +1,139 @@
+package fixture
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+)
+
+// 本文件只发 SELECT（legacy 侧），且所有金额列一律 `CAST(... AS DECIMAL(20,3))`。
+//
+// 为什么不直接 `SELECT balance`：
+//   balance 是 float(10,3)。go-sql-driver 会把它当 float64 交给 Go，
+//   于是 0.788 变成 0.78800000000000003，后面任何 Round 都在误差上做决策。
+//   `CAST(x AS DECIMAL(20,3))` 让 MySQL 先把该列按「3 位小数的十进制」输出，
+//   我们再按字符串精确换成分——误差在源头就被掐掉。
+
+// extractExpense 读 1 个账户行。
+func extractExpense(ctx context.Context, db *sql.DB, userID int64) (*LegacyExpense, error) {
+	const q = `
+SELECT userid,
+       CAST(balance                 AS DECIMAL(20,3)),
+       CAST(spend                   AS DECIMAL(20,3)),
+       CAST(total_income            AS DECIMAL(20,3)),
+       CAST(total_expenditure       AS DECIMAL(20,3)),
+       CAST(total_recommended_amount AS DECIMAL(20,3)),
+       CAST(special_expenses        AS DECIMAL(20,3))
+  FROM t_user_expense
+ WHERE userid = ?`
+	row := db.QueryRowContext(ctx, q, userID)
+	var out LegacyExpense
+	if err := row.Scan(&out.UserID, &out.Balance, &out.Spend, &out.TotalIncome,
+		&out.TotalExpenditure, &out.TotalRecommendedAmount, &out.SpecialExpenses); err != nil {
+		if err == sql.ErrNoRows {
+			return nil, fmt.Errorf("legacy t_user_expense 无 userid=%d 的行", userID)
+		}
+		return nil, err
+	}
+	return &out, nil
+}
+
+// extractExpenseSchools 读该用户的全部校区账户行（按 legacy id 升序，保证可复现）。
+func extractExpenseSchools(ctx context.Context, db *sql.DB, userID int64) ([]LegacyExpenseSchool, error) {
+	const q = `
+SELECT id, userid, schoolid, cast(ifnull(campusid,0) as signed),
+       CAST(balance                 AS DECIMAL(20,3)),
+       CAST(spend                   AS DECIMAL(20,3)),
+       CAST(total_income            AS DECIMAL(20,3)),
+       CAST(total_expenditure       AS DECIMAL(20,3)),
+       CAST(total_recommended_amount AS DECIMAL(20,3)),
+       CAST(special_expenses        AS DECIMAL(20,3))
+  FROM t_user_expense_school
+ WHERE userid = ?
+ ORDER BY id`
+	rows, err := db.QueryContext(ctx, q, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []LegacyExpenseSchool
+	for rows.Next() {
+		var r LegacyExpenseSchool
+		if err := rows.Scan(&r.ID, &r.UserID, &r.SchoolID, &r.CampusID,
+			&r.Balance, &r.Spend, &r.TotalIncome, &r.TotalExpenditure,
+			&r.TotalRecommendedAmount, &r.SpecialExpenses); err != nil {
+			return nil, err
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}
+
+// extractExpenseLogs 读该用户的全部流水（按 legacy id 升序）。
+func extractExpenseLogs(ctx context.Context, db *sql.DB, userID int64) ([]LegacyExpenseLog, error) {
+	const q = `
+SELECT id, userid, cast(ifnull(type,0) as signed), logid,
+       CAST(money AS DECIMAL(20,3)),
+       create_userid, create_time, remark
+  FROM t_user_expense_log
+ WHERE userid = ?
+ ORDER BY id`
+	rows, err := db.QueryContext(ctx, q, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var out []LegacyExpenseLog
+	for rows.Next() {
+		var r LegacyExpenseLog
+		if err := rows.Scan(&r.ID, &r.UserID, &r.Type, &r.LogID,
+			&r.Money, &r.CreateUserID, &r.CreateTime, &r.Remark); err != nil {
+			return nil, err
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}
+
+// pickLegacyUser 在未显式指定用户时，按验收要求挑一个用户：
+// 有账户行 + 至少 1 个校区账户 + 至少 3 条流水 + balance 带 3 位小数（覆盖取整边界）。
+//
+// 只用 SELECT，且条件全部来自 information 之外的业务表本身。
+func pickLegacyUser(ctx context.Context, db *sql.DB) (int64, error) {
+	const q = `
+SELECT e.userid
+  FROM t_user_expense e
+  JOIN (SELECT userid, COUNT(*) n FROM t_user_expense_school GROUP BY userid) s ON s.userid = e.userid
+  JOIN (SELECT userid, COUNT(*) n FROM t_user_expense_log GROUP BY userid HAVING n >= 3) l ON l.userid = e.userid
+ WHERE ABS(CAST(e.balance * 1000 AS DECIMAL(20,0))) % 100 >= 1
+   AND ABS(e.balance) > 0
+ ORDER BY e.userid
+ LIMIT 1`
+	var id int64
+	if err := db.QueryRowContext(ctx, q).Scan(&id); err != nil {
+		return 0, fmt.Errorf("自动挑选 legacy 用户失败（无满足条件的用户）：%w", err)
+	}
+	return id, nil
+}
+
+// countLegacyRows 统计该用户在 3 张 legacy 表里的行数（报告里展示「搬了几行」）。
+func countLegacyRows(ctx context.Context, db *sql.DB, userID int64) (map[string]int, error) {
+	out := make(map[string]int, 3)
+	for _, q := range []struct {
+		name string
+		sql  string
+	}{
+		{"t_user_expense", `SELECT COUNT(*) FROM t_user_expense WHERE userid = ?`},
+		{"t_user_expense_school", `SELECT COUNT(*) FROM t_user_expense_school WHERE userid = ?`},
+		{"t_user_expense_log", `SELECT COUNT(*) FROM t_user_expense_log WHERE userid = ?`},
+	} {
+		var n int
+		if err := db.QueryRowContext(ctx, q.sql, userID).Scan(&n); err != nil {
+			return nil, err
+		}
+		out[q.name] = n
+	}
+	return out, nil
+}

internal/academic/dualrun/fixture/load.go

@@ -0,0 +1,157 @@
+package fixture
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"strings"
+)
+
+// 本文件是**唯一**往 yuansiwei_2026 写数据的地方，且只写 5 张表：
+//   t_user / t_legacy_mapping / t_user_expense / t_user_expense_school / t_user_expense_log
+//
+// 所有写入都用幂等写法（ON DUPLICATE KEY UPDATE），理由：
+//   · map/verify/clean 是 3 个独立命令，中途失败要能重跑；
+//   · 变异实验要跑多轮「红→绿」，每轮都从同一起点开始。
+
+// upsertUser 插入 fixture 用的 t_user 行（最小列集）。
+//
+// t_user 的 NOT NULL 列只有 canonical_user_id / person_type / status / created_at / updated_at，
+// 后三者都有默认值，因此最少只需给 canonical_user_id（+ 便于识别的 display_name）。
+func upsertUser(ctx context.Context, db *sql.DB, canonicalUserID, displayName string) error {
+	const q = `
+INSERT INTO t_user (canonical_user_id, username, display_name, person_type, status)
+VALUES (?, NULL, ?, 'student', 'active')
+ON DUPLICATE KEY UPDATE display_name = VALUES(display_name)`
+	if _, err := db.ExecContext(ctx, q, canonicalUserID, displayName); err != nil {
+		return fmt.Errorf("写 t_user(%s) 失败：%w", canonicalUserID, err)
+	}
+	return nil
+}
+
+// upsertMapping 插入 t_legacy_mapping 行（legacy int id → ULID）。
+//
+// 唯一键是 uk_canonical(canonical_user_id)，因此重复执行只会更新同一行。
+func upsertMapping(ctx context.Context, db *sql.DB, m MappingEntry) error {
+	const q = `
+INSERT INTO t_legacy_mapping
+  (canonical_user_id, legacy_academic_user_id, institution_id, campus_id,
+   role, match_source, match_confidence, merged_at)
+VALUES (?, ?, ?, ?, ?, ?, ?, NOW(3))
+ON DUPLICATE KEY UPDATE
+  legacy_academic_user_id = VALUES(legacy_academic_user_id),
+  institution_id          = VALUES(institution_id),
+  campus_id               = VALUES(campus_id),
+  role                    = VALUES(role),
+  match_source            = VALUES(match_source),
+  match_confidence        = VALUES(match_confidence)`
+	if _, err := db.ExecContext(ctx, q, m.CanonicalUserID, m.LegacyAcademicUserID,
+		m.InstitutionID, m.CampusID, nullIfEmpty(m.Role), m.MatchSource, m.MatchConfidence); err != nil {
+		return fmt.Errorf("写 t_legacy_mapping(legacy=%d) 失败：%w", m.LegacyAcademicUserID, err)
+	}
+	return nil
+}
+
+// upsertExpense 写全局账户行（P）：金额以十进制字符串写入 DECIMAL(12,2)。
+func upsertExpense(ctx context.Context, db *sql.DB, r TargetExpense) error {
+	const q = `
+INSERT INTO t_user_expense
+  (user_id, balance, spend, total_income, total_expenditure,
+   total_recommended_amount, special_expenses)
+VALUES (?, ?, ?, ?, ?, ?, ?)
+ON DUPLICATE KEY UPDATE
+  balance                  = VALUES(balance),
+  spend                    = VALUES(spend),
+  total_income             = VALUES(total_income),
+  total_expenditure        = VALUES(total_expenditure),
+  total_recommended_amount = VALUES(total_recommended_amount),
+  special_expenses         = VALUES(special_expenses)`
+	_, err := db.ExecContext(ctx, q, r.UserID,
+		CentsToDecimal(r.BalanceCents),
+		CentsToDecimal(r.SpendCents),
+		CentsToDecimal(r.TotalIncomeCents),
+		CentsToDecimal(r.TotalExpenditureCents),
+		CentsToDecimal(r.TotalRecommendedCents),
+		CentsToDecimal(r.SpecialExpensesCents))
+	if err != nil {
+		return fmt.Errorf("写 t_user_expense(user=%s) 失败：%w", r.UserID, err)
+	}
+	return nil
+}
+
+// upsertExpenseSchool 写校区账户行。
+//
+// id 不显式指定：该表的对齐键是 uk_user_inst_campus，靠唯一键做幂等。
+func upsertExpenseSchool(ctx context.Context, db *sql.DB, r TargetExpenseSchool) error {
+	const q = `
+INSERT INTO t_user_expense_school
+  (user_id, institution_id, campus_id, balance, spend, total_income,
+   total_expenditure, total_recommended_amount, special_expenses)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON DUPLICATE KEY UPDATE
+  balance                  = VALUES(balance),
+  spend                    = VALUES(spend),
+  total_income             = VALUES(total_income),
+  total_expenditure        = VALUES(total_expenditure),
+  total_recommended_amount = VALUES(total_recommended_amount),
+  special_expenses         = VALUES(special_expenses)`
+	_, err := db.ExecContext(ctx, q, r.UserID, r.InstitutionID, r.CampusID,
+		CentsToDecimal(r.BalanceCents),
+		CentsToDecimal(r.SpendCents),
+		CentsToDecimal(r.TotalIncomeCents),
+		CentsToDecimal(r.TotalExpenditureCents),
+		CentsToDecimal(r.TotalRecommendedCents),
+		CentsToDecimal(r.SpecialExpensesCents))
+	if err != nil {
+		return fmt.Errorf("写 t_user_expense_school(user=%s,campus=%d) 失败：%w", r.UserID, r.CampusID, err)
+	}
+	return nil
+}
+
+// insertExpenseLog 写流水行（D1：**显式插入 legacy id**）。
+//
+// 幂等键是 PRIMARY KEY(id)=legacy id，天然可重跑。
+// 注意：这里刻意**不**依赖 AUTO_INCREMENT 生成 id——id 由 legacy 决定。
+func insertExpenseLog(ctx context.Context, db *sql.DB, r TargetExpenseLog) error {
+	const q = `
+INSERT INTO t_user_expense_log
+  (id, user_id, type, log_id, request_id, money, balance_after, create_user_id, create_time, remark)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ON DUPLICATE KEY UPDATE
+  user_id        = VALUES(user_id),
+  type           = VALUES(type),
+  log_id         = VALUES(log_id),
+  request_id     = VALUES(request_id),
+  money          = VALUES(money),
+  balance_after  = VALUES(balance_after),
+  create_user_id = VALUES(create_user_id),
+  create_time    = VALUES(create_time),
+  remark         = VALUES(remark)`
+	_, err := db.ExecContext(ctx, q,
+		r.ID, r.UserID, r.Type,
+		nullStr(r.LogID), nullStr(r.RequestID),
+		CentsToDecimal(r.MoneyCents),
+		nullStr(r.BalanceAfter), nullStr(r.CreateUserID),
+		r.CreateTime.Format("2006-01-02 15:04:05"),
+		nullStr(r.Remark))
+	if err != nil {
+		return fmt.Errorf("写 t_user_expense_log(id=%d) 失败：%w", r.ID, err)
+	}
+	return nil
+}
+
+// nullIfEmpty 把空串转 NULL（role 之类的可选列用）。
+func nullIfEmpty(s string) any {
+	if strings.TrimSpace(s) == "" {
+		return nil
+	}
+	return s
+}
+
+// nullStr 把 sql.NullString 转成可交给驱动的 any（NULL 或 string）。
+func nullStr(v sql.NullString) any {
+	if !v.Valid {
+		return nil
+	}
+	return v.String
+}

internal/academic/dualrun/fixture/mapping.go

@@ -0,0 +1,198 @@
+package fixture
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"database/sql"
+	"fmt"
+	"time"
+	"unicode/utf8"
+
+	"github.com/oklog/ulid"
+)
+
+// 本文件是**纯映射规则**：不碰数据库、不读环境变量、不取当前时间，全部输入即输出。
+// 这样每条规则都能单测，变异实验（故意改错一处）也只需改这里一处即可复现红/绿。
+
+// fixtureEpochMs 是 fixture ULID 的固定时间戳（2026-01-01T00:00:00Z）。
+//
+// 固定而非 Now()：ULID 单调性对 fixture 无意义，但**确定性**很有意义——
+// 同一 legacy user_id 每次跑都得到同一个 canonical_user_id，于是
+//
+//	· 重复执行 map/verify/clean 天然幂等（不必落盘中间状态）；
+//	· 报告里的 ULID 可复现、可对比。
+const fixtureEpochMs uint64 = 1767225600000 // 2026-01-01T00:00:00Z
+
+// FixtureCanonicalUserID 由 legacy int user_id **确定性**派生一个 ULID。
+//
+// 熵 = sha256("ac15-fixture:user:<id>") 前 10 字节（80 位，ULID 上限）。
+// 生产迁移的 canonical_user_id 由归一 pipeline 的种子脚本发放，与此无关；
+// 这里只要「同一 legacy id ↔ 同一 ULID」，且不撞车即可。
+func FixtureCanonicalUserID(legacyUserID int64) string {
+	sum := sha256.Sum256([]byte(fmt.Sprintf("ac15-fixture:user:%d", legacyUserID)))
+	var entropy [10]byte
+	copy(entropy[:], sum[:10])
+	id, err := ulid.New(fixtureEpochMs, bytes.NewReader(entropy[:]))
+	if err != nil {
+		// entropy 长度固定 10 字节，ulid.New 只会在熵不足时报错，这里不可达。
+		panic(fmt.Sprintf("fixture: ULID 生成失败：%v", err))
+	}
+	return id.String()
+}
+
+// moneyOrZero 把可空的 legacy 金额列统一成十进制字符串。
+//
+// 语义：legacy 侧该列 NULL（历史上没记过）→ 目标态 NOT NULL DEFAULT 0，
+// 只能落 0；但**必须记录**，因为这与「真的是 0」在业务上是两件事。
+func moneyOrZero(v sql.NullString) (string, bool) {
+	if !v.Valid || v.String == "" {
+		return "0.000", false
+	}
+	return v.String, true
+}
+
+// MapExpense 映射 t_user_expense（1 个账户行）。
+func MapExpense(l LegacyExpense, canonicalUserID string) (TargetExpense, []string, error) {
+	var warns []string
+	out := TargetExpense{UserID: canonicalUserID}
+
+	type col struct {
+		name string
+		in   sql.NullString
+		out  *int64
+	}
+	cols := []col{
+		{"balance", l.Balance, &out.BalanceCents},
+		{"spend", l.Spend, &out.SpendCents},
+		{"total_income", l.TotalIncome, &out.TotalIncomeCents},
+		{"total_expenditure", l.TotalExpenditure, &out.TotalExpenditureCents},
+		{"total_recommended_amount", l.TotalRecommendedAmount, &out.TotalRecommendedCents},
+		{"special_expenses", l.SpecialExpenses, &out.SpecialExpensesCents},
+	}
+	for _, c := range cols {
+		s, present := moneyOrZero(c.in)
+		if !present {
+			warns = append(warns, fmt.Sprintf("t_user_expense.%s 在 legacy 为 NULL，目标态 NOT NULL → 落 0.00", c.name))
+		}
+		cents, err := DecimalToCents(s)
+		if err != nil {
+			return TargetExpense{}, warns, fmt.Errorf("t_user_expense.%s：%w", c.name, err)
+		}
+		*c.out = cents
+	}
+	return out, warns, nil
+}
+
+// MapExpenseSchool 映射一个校区账户行（复合业务键 user_id + institution_id + campus_id）。
+func MapExpenseSchool(l LegacyExpenseSchool, canonicalUserID string) (TargetExpenseSchool, []string, error) {
+	var warns []string
+	out := TargetExpenseSchool{
+		UserID:        canonicalUserID,
+		InstitutionID: l.SchoolID, // legacy schoolid → target institution_id（1:1 直迁）
+		CampusID:      l.CampusID, // legacy campusid → target campus_id（NULL 已在读取侧 IFNULL 0）
+	}
+	type col struct {
+		name string
+		in   sql.NullString
+		out  *int64
+	}
+	cols := []col{
+		{"balance", l.Balance, &out.BalanceCents},
+		{"spend", l.Spend, &out.SpendCents},
+		{"total_income", l.TotalIncome, &out.TotalIncomeCents},
+		{"total_expenditure", l.TotalExpenditure, &out.TotalExpenditureCents},
+		{"total_recommended_amount", l.TotalRecommendedAmount, &out.TotalRecommendedCents},
+		{"special_expenses", l.SpecialExpenses, &out.SpecialExpensesCents},
+	}
+	for _, c := range cols {
+		s, present := moneyOrZero(c.in)
+		if !present {
+			warns = append(warns, fmt.Sprintf("t_user_expense_school(id=%d).%s 在 legacy 为 NULL → 落 0.00", l.ID, c.name))
+		}
+		cents, err := DecimalToCents(s)
+		if err != nil {
+			return TargetExpenseSchool{}, warns, fmt.Errorf("t_user_expense_school(id=%d).%s：%w", l.ID, c.name, err)
+		}
+		*c.out = cents
+	}
+	return out, warns, nil
+}
+
+// CreatorResolver 把 legacy int create_userid 解析成 canonical_user_id（ULID）。
+// 返回 ok=false 表示该 legacy id 没有映射行。
+type CreatorResolver func(legacyUserID int64) (canonicalUserID string, ok bool)
+
+// MapExpenseLog 映射一条流水（D1：id 原样保留）。
+//
+// 三条已拍板的空值语义（逐列判断，不一律转 0）：
+//  1. create_user_id：legacy 0（或解析不到的旧 id）→ **NULL**，不是 0；
+//  2. request_id：legacy 无此列 → **NULL**（历史数据没有幂等号）；
+//  3. balance_after：legacy 无此列 → **NULL**（历史上没记余额快照）。
+func MapExpenseLog(l LegacyExpenseLog, canonicalUserID string, resolve CreatorResolver) (TargetExpenseLog, []string, error) {
+	var warns []string
+	out := TargetExpenseLog{
+		ID:           l.ID, // D1：显式插 legacy id
+		UserID:       canonicalUserID,
+		Type:         l.Type,
+		RequestID:    sql.NullString{}, // 恒 NULL
+		BalanceAfter: sql.NullString{}, // 恒 NULL
+	}
+
+	// log_id：legacy logid(varchar(32) 旧 UUID) → target log_id(varchar(64))
+	if l.LogID.Valid && l.LogID.String != "" {
+		out.LogID = sql.NullString{String: l.LogID.String, Valid: true}
+	}
+
+	// money：decimal(10,3) → 整数分。legacy 为 NULL 时目标 NOT NULL，
+	// 记告警并落 0（本 fixture 命中 0 行，规则仍需成立）。
+	s, present := moneyOrZero(l.Money)
+	if !present {
+		warns = append(warns, fmt.Sprintf("t_user_expense_log(id=%d).money 在 legacy 为 NULL → 落 0.00", l.ID))
+	}
+	cents, err := DecimalToCents(s)
+	if err != nil {
+		return TargetExpenseLog{}, warns, fmt.Errorf("t_user_expense_log(id=%d).money：%w", l.ID, err)
+	}
+	out.MoneyCents = cents
+
+	// create_user_id：0 / NULL / 解析不到 → NULL（不落 0、不落空串）。
+	if l.CreateUserID.Valid && l.CreateUserID.Int64 > 0 {
+		if uid, ok := resolve(l.CreateUserID.Int64); ok {
+			out.CreateUserID = sql.NullString{String: uid, Valid: true}
+		} else {
+			warns = append(warns, fmt.Sprintf(
+				"t_user_expense_log(id=%d).create_user_id：legacy %d 无 t_legacy_mapping 行 → NULL",
+				l.ID, l.CreateUserID.Int64))
+		}
+	}
+
+	// create_time：datetime(0) → datetime(3)，**秒级原样**，毫秒补 .000。
+	if !l.CreateTime.Valid {
+		return TargetExpenseLog{}, warns, fmt.Errorf(
+			"t_user_expense_log(id=%d).create_time 在 legacy 为 NULL，目标态 NOT NULL —— 拒绝用当前时间填充", l.ID)
+	}
+	out.CreateTime = time.Date(
+		l.CreateTime.Time.Year(), l.CreateTime.Time.Month(), l.CreateTime.Time.Day(),
+		l.CreateTime.Time.Hour(), l.CreateTime.Time.Minute(), l.CreateTime.Time.Second(),
+		0, l.CreateTime.Time.Location(),
+	)
+
+	// remark：text → varchar(1000)。按**字符数**截断（不是字节），并记告警。
+	if l.Remark.Valid && l.Remark.String != "" {
+		out.Remark = sql.NullString{String: truncateRunes(l.Remark.String, 1000), Valid: true}
+		if utf8.RuneCountInString(l.Remark.String) > 1000 {
+			warns = append(warns, fmt.Sprintf(
+				"t_user_expense_log(id=%d).remark 超过 1000 字符，已按字符截断（目标列 varchar(1000)）", l.ID))
+		}
+	}
+	return out, warns, nil
+}
+
+// truncateRunes 按字符（rune）截断，避免把一个多字节汉字截成半个。
+func truncateRunes(s string, max int) string {
+	if utf8.RuneCountInString(s) <= max {
+		return s
+	}
+	rs := []rune(s)
+	return string(rs[:max])
+}

internal/academic/dualrun/fixture/mapping_test.go

@@ -0,0 +1,238 @@
+package fixture
+
+import (
+	"database/sql"
+	"testing"
+	"time"
+)
+
+// 本文件单测**纯映射规则**（不连库）：取整、空值语义、时间、D1 id、确定性 ULID。
+// 库相关的端到端核对由 cmd/ac15fixture verify 完成（需要真库，不进单测）。
+
+func TestDecimalToCents(t *testing.T) {
+	cases := []struct {
+		in   string
+		want int64
+	}{
+		{"0.000", 0},
+		{"0", 0},
+		{"0.788", 79},          // 第 3 位 8 进位
+		{"4999.212", 499921},   // 第 3 位 2 舍去
+		{"-2160.489", -216049}, // 负数同向（half away from zero）
+		{"-12.015", -1202},     // 恰好 x.xx5：进位
+		{"12.015", 1202},
+		{"5000.000", 500000},
+		{"-160.036", -16004},
+		{"333.328", 33333},
+		{"  7.5  ", 750}, // 空格与补齐
+		{"-0.004", 0},    // 舍到 0
+		{"-0.005", -1},   // 进位到 -0.01
+	}
+	for _, c := range cases {
+		got, err := DecimalToCents(c.in)
+		if err != nil {
+			t.Fatalf("DecimalToCents(%q) 出错：%v", c.in, err)
+		}
+		if got != c.want {
+			t.Errorf("DecimalToCents(%q) = %d, want %d", c.in, got, c.want)
+		}
+	}
+
+	// 拒绝静默截断：4 位小数必须报错
+	if _, err := DecimalToCents("1.2345"); err == nil {
+		t.Error("DecimalToCents(1.2345) 应报错（拒绝静默截断），实际未报错")
+	}
+}
+
+func TestCentsToDecimal(t *testing.T) {
+	cases := []struct {
+		in   int64
+		want string
+	}{
+		{0, "0.00"},
+		{79, "0.79"},
+		{-216049, "-2160.49"},
+		{499921, "4999.21"},
+		{-16004, "-160.04"},
+		{-1, "-0.01"},
+	}
+	for _, c := range cases {
+		if got := CentsToDecimal(c.in); got != c.want {
+			t.Errorf("CentsToDecimal(%d) = %q, want %q", c.in, got, c.want)
+		}
+	}
+}
+
+// TestRoundTrip 保证「字符串 → 分 → 字符串」在 2 位小数上稳定（幂等）。
+func TestRoundTrip(t *testing.T) {
+	for _, s := range []string{"0.00", "0.79", "-2160.49", "4999.21", "5000.00"} {
+		c := MustCents(s)
+		if got := CentsToDecimal(c); got != s {
+			t.Errorf("round trip %q → %d → %q", s, c, got)
+		}
+	}
+}
+
+// TestFixtureCanonicalUserIDDeterministic 证明 fixture ULID 是确定性的：
+// 同一 legacy id 每次跑得到同一个值，不同 legacy id 不撞车。
+func TestFixtureCanonicalUserIDDeterministic(t *testing.T) {
+	a1 := FixtureCanonicalUserID(53691)
+	a2 := FixtureCanonicalUserID(53691)
+	if a1 != a2 {
+		t.Fatalf("同一 legacy id 得到不同 ULID：%s vs %s", a1, a2)
+	}
+	if len(a1) != 26 {
+		t.Fatalf("ULID 长度应为 26，实际 %d（%q）", len(a1), a1)
+	}
+	seen := map[string]bool{}
+	for _, id := range []int64{53691, 20052, 34369, 45014, 17230, 17735} {
+		u := FixtureCanonicalUserID(id)
+		if seen[u] {
+			t.Fatalf("不同 legacy id 撞到同一 ULID：%s", u)
+		}
+		seen[u] = true
+	}
+}
+
+func ptrNStr(s string) sql.NullString { return sql.NullString{String: s, Valid: true} }
+func nullNStr() sql.NullString        { return sql.NullString{} }
+
+// TestMapExpenseLogNullSemantics 是空值语义的单测：
+// create_user_id=0 → NULL、request_id → NULL、balance_after → NULL、id 原样保留。
+//
+// ⚠ legacy 真实数据里 create_userid=0 的行数是 **0**（见报告 §5），
+// 所以这条规则只能靠单测 + 变异实验证明，不能靠真实数据证明。
+func TestMapExpenseLogNullSemantics(t *testing.T) {
+	legacy := LegacyExpenseLog{
+		ID:           33451,
+		UserID:       53691,
+		Type:         100,
+		LogID:        ptrNStr("fa5d7262e03636c4b69475d731f7388e"),
+		Money:        ptrNStr("5000.000"),
+		CreateUserID: sql.NullInt64{Int64: 0, Valid: true}, // 系统/未知创建人
+		CreateTime:   sql.NullTime{Time: time.Date(2019, 9, 7, 12, 25, 49, 0, time.Local), Valid: true},
+		Remark:       ptrNStr("充值缴费"),
+	}
+	resolver := func(id int64) (string, bool) { return "", false }
+
+	got, warns, err := MapExpenseLog(legacy, "01ABC", resolver)
+	if err != nil {
+		t.Fatalf("MapExpenseLog 出错：%v", err)
+	}
+	if got.ID != 33451 {
+		t.Errorf("D1：id 应原样保留 33451，实际 %d", got.ID)
+	}
+	if got.CreateUserID.Valid {
+		t.Errorf("create_user_id=0 应映射为 NULL，实际落成 %q", got.CreateUserID.String)
+	}
+	if got.RequestID.Valid {
+		t.Error("request_id 应恒为 NULL（legacy 无此列）")
+	}
+	if got.BalanceAfter.Valid {
+		t.Error("balance_after 应恒为 NULL（legacy 无快照）")
+	}
+	if got.MoneyCents != 500000 {
+		t.Errorf("money 5000.000 → 500000 分，实际 %d", got.MoneyCents)
+	}
+	if got.CreateTime.Nanosecond() != 0 {
+		t.Errorf("create_time 毫秒部分应为 0（DATETIME(3) 补 .000），实际 %d", got.CreateTime.Nanosecond())
+	}
+	if len(warns) != 0 {
+		t.Errorf("不应有告警，实际 %v", warns)
+	}
+}
+
+// TestMapExpenseLogResolvesCreator 证明「legacy int create_userid → ULID」走的是 resolver。
+func TestMapExpenseLogResolvesCreator(t *testing.T) {
+	legacy := LegacyExpenseLog{
+		ID:           1,
+		Type:         201,
+		Money:        ptrNStr("-3000.000"),
+		CreateUserID: sql.NullInt64{Int64: 34369, Valid: true},
+		CreateTime:   sql.NullTime{Time: time.Date(2019, 9, 7, 12, 37, 12, 0, time.Local), Valid: true},
+	}
+	resolver := func(id int64) (string, bool) {
+		if id == 34369 {
+			return "01RESOLVED0000000000000000", true
+		}
+		return "", false
+	}
+	got, _, err := MapExpenseLog(legacy, "01ABC", resolver)
+	if err != nil {
+		t.Fatalf("MapExpenseLog 出错：%v", err)
+	}
+	if !got.CreateUserID.Valid || got.CreateUserID.String != "01RESOLVED0000000000000000" {
+		t.Errorf("create_user_id 应解析成 ULID，实际 %+v", got.CreateUserID)
+	}
+}
+
+// TestMapExpenseLogRejectsNullCreateTime 证明「create_time 为 NULL 时拒绝用当前时间填充」。
+func TestMapExpenseLogRejectsNullCreateTime(t *testing.T) {
+	legacy := LegacyExpenseLog{ID: 1, Money: ptrNStr("1.000"), CreateTime: sql.NullTime{}}
+	if _, _, err := MapExpenseLog(legacy, "01ABC", func(int64) (string, bool) { return "", false }); err == nil {
+		t.Error("create_time 为 NULL 时应报错（不得用当前时间填充），实际未报错")
+	}
+}
+
+// TestMapExpenseRoundsPerRow 证明金额是**逐列**取整到分（不是取整求和）。
+func TestMapExpenseRoundsPerRow(t *testing.T) {
+	l := LegacyExpense{
+		UserID:                 53691,
+		Balance:                ptrNStr("0.788"),
+		Spend:                  ptrNStr("4999.212"),
+		TotalIncome:            ptrNStr("5000.000"),
+		TotalExpenditure:       ptrNStr("4999.212"),
+		TotalRecommendedAmount: ptrNStr("0.000"),
+		SpecialExpenses:        ptrNStr("0.000"),
+	}
+	got, warns, err := MapExpense(l, "01ABC")
+	if err != nil {
+		t.Fatalf("MapExpense 出错：%v", err)
+	}
+	if got.BalanceCents != 79 {
+		t.Errorf("balance 0.788 → 0.79（79 分），实际 %d", got.BalanceCents)
+	}
+	if got.SpendCents != 499921 {
+		t.Errorf("spend 4999.212 → 4999.21，实际 %d", got.SpendCents)
+	}
+	if got.TotalIncomeCents != 500000 {
+		t.Errorf("total_income 5000.000 → 5000.00，实际 %d", got.TotalIncomeCents)
+	}
+	if len(warns) != 0 {
+		t.Errorf("不应有告警，实际 %v", warns)
+	}
+}
+
+// TestMapExpenseNullToZero 证明 legacy NULL 金额落 0 且**必须产生告警**。
+func TestMapExpenseNullToZero(t *testing.T) {
+	l := LegacyExpense{
+		UserID:                 1,
+		Balance:                nullNStr(),
+		Spend:                  ptrNStr("0.000"),
+		TotalIncome:            ptrNStr("0.000"),
+		TotalExpenditure:       ptrNStr("0.000"),
+		TotalRecommendedAmount: nullNStr(),
+		SpecialExpenses:        nullNStr(),
+	}
+	got, warns, err := MapExpense(l, "01ABC")
+	if err != nil {
+		t.Fatalf("MapExpense 出错：%v", err)
+	}
+	if got.BalanceCents != 0 || got.TotalRecommendedCents != 0 {
+		t.Error("NULL 金额应落 0")
+	}
+	if len(warns) != 3 {
+		t.Errorf("3 个 NULL 列应产生 3 条告警，实际 %d 条：%v", len(warns), warns)
+	}
+}
+
+// TestTruncateRunes 证明 remark 按字符截断（不会把汉字截成半个）。
+func TestTruncateRunes(t *testing.T) {
+	s := "充值缴费abc"
+	if got := truncateRunes(s, 4); got != "充值缴费" {
+		t.Errorf("按字符截断到 4 应得 %q，实际 %q", "充值缴费", got)
+	}
+	if got := truncateRunes("abc", 10); got != "abc" {
+		t.Errorf("未超长应原样返回，实际 %q", got)
+	}
+}

internal/academic/dualrun/fixture/model.go

@@ -0,0 +1,191 @@
+package fixture
+
+import (
+	"database/sql"
+	"time"
+)
+
+// 本文件只放**数据形状**：legacy 读出来的行 + 目标态要写的行。
+// 转换规则一律放在 mapping.go（纯函数，可单测），落库一律放 load.go。
+
+// ---------- legacy 侧（读出来即定型，金额统一用十进制字符串） ----------
+
+// LegacyExpense 对应 legacy `t_user_expense`（按 userid 1 行）。
+type LegacyExpense struct {
+	UserID                 int64
+	Balance                sql.NullString // float(10,3)
+	Spend                  sql.NullString // float(10,3)
+	TotalIncome            sql.NullString // float(10,3)
+	TotalExpenditure       sql.NullString // float(10,3)
+	TotalRecommendedAmount sql.NullString // decimal(10,3)
+	SpecialExpenses        sql.NullString // decimal(10,3)
+}
+
+// LegacyExpenseSchool 对应 legacy `t_user_expense_school`（userid + schoolid + campusid）。
+type LegacyExpenseSchool struct {
+	ID                     int64
+	UserID                 int64
+	SchoolID               int64 // → institution_id
+	CampusID               int64 // → campus_id（legacy 可空，读时 IFNULL 0）
+	Balance                sql.NullString
+	Spend                  sql.NullString
+	TotalIncome            sql.NullString
+	TotalExpenditure       sql.NullString
+	TotalRecommendedAmount sql.NullString
+	SpecialExpenses        sql.NullString
+}
+
+// LegacyExpenseLog 对应 legacy `t_user_expense_log`（int 自增 id）。
+type LegacyExpenseLog struct {
+	ID           int64
+	UserID       int64
+	Type         int32
+	LogID        sql.NullString // varchar(32) 旧 UUID → target log_id varchar(64)
+	Money        sql.NullString // decimal(10,3)
+	CreateUserID sql.NullInt64  // int：0 或系统 → target NULL
+	CreateTime   sql.NullTime   // datetime(0) → target datetime(3)（补 .000）
+	Remark       sql.NullString // text → varchar(1000)
+}
+
+// ---------- 目标侧（金额一律整数分，写入前才格式化成十进制字符串） ----------
+
+// TargetExpense 对应目标 `t_user_expense`（主键 user_id，无自增 id）。
+type TargetExpense struct {
+	UserID                string
+	BalanceCents          int64
+	SpendCents            int64
+	TotalIncomeCents      int64
+	TotalExpenditureCents int64
+	TotalRecommendedCents int64
+	SpecialExpensesCents  int64
+}
+
+// TargetExpenseSchool 对应目标 `t_user_expense_school`。
+//
+// id 策略：**不保留** legacy id——该表在目标态的对齐键是复合业务键
+// (user_id, institution_id, campus_id)，与流水表的 D1 口径不同（见 README §3）。
+type TargetExpenseSchool struct {
+	UserID                string
+	InstitutionID         int64
+	CampusID              int64
+	BalanceCents          int64
+	SpendCents            int64
+	TotalIncomeCents      int64
+	TotalExpenditureCents int64
+	TotalRecommendedCents int64
+	SpecialExpensesCents  int64
+}
+
+// TargetExpenseLog 对应目标 `t_user_expense_log`（D1：显式插 legacy id）。
+type TargetExpenseLog struct {
+	ID           int64 // = legacy id（D1）
+	UserID       string
+	Type         int32
+	LogID        sql.NullString // legacy logid
+	RequestID    sql.NullString // legacy 无此列 → 恒 NULL
+	MoneyCents   int64
+	BalanceAfter sql.NullString // legacy 无此列 → 恒 NULL（decimal(12,2) 用字符串承载）
+	CreateUserID sql.NullString // legacy int create_userid → ULID；0/系统 → NULL
+	CreateTime   time.Time      // 秒级（DATETIME(3) 落 .000）
+	Remark       sql.NullString
+}
+
+// MappingEntry 是一条 t_legacy_mapping 记录（fixture 自己插）。
+type MappingEntry struct {
+	CanonicalUserID      string
+	LegacyAcademicUserID int64
+	InstitutionID        int64
+	CampusID             int64
+	Role                 string
+	MatchSource          string
+	MatchConfidence      int
+}
+
+// ---------- 报告用结构 ----------
+
+// ColumnCheck 是核对表的一行：一列的 legacy 值 / target 值 / 是否相等。
+type ColumnCheck struct {
+	Table  string
+	RowKey string
+	Column string
+	Legacy string
+	Target string
+	Equal  bool
+	Note   string
+}
+
+// VerifyReport 是一轮核对的完整结果。
+type VerifyReport struct {
+	Checks      []ColumnCheck
+	Assertions  []Assertion
+	LoadedRows  map[string]int
+	TargetRows  map[string]int
+	FloatTrap   FloatTrapDemo
+	AutoIncTest AutoIncrementDemo
+}
+
+// Assertion 是一条语义断言（如「balance_after 必须为 NULL」）。
+type Assertion struct {
+	Name   string
+	Pass   bool
+	Detail string
+}
+
+// FloatTrapDemo 记录「float 直接相加」与「逐行取整到分再相加」的差异证据。
+type FloatTrapDemo struct {
+	LegacyValues        []string // 逐条 legacy money
+	PerRowRounded       []string // 逐条 ROUND(x,2)
+	SumOfRoundedCents   int64    // Σ ROUND(x,2)
+	SumRoundedStr       string   // 上面那个值的十进制表示
+	FloatSumRaw         float64  // 直接用 float64 相加
+	RoundOfSumStr       string   // ROUND(Σx, 2)（先加后取整）
+	DeltaVsPerRow       int64    // ROUND(Σ) 与 ΣROUND 的分差
+	FloatAccumErr       float64  // float64 累加误差（FloatSumRaw - 精确和）
+	ExactSumThousandths int64    // 精确和（千分位整数，用于展示 float 累加误差量级）
+}
+
+// AutoIncrementDemo 记录「显式插 id 后 AUTO_INCREMENT 不倒退」的证据。
+//
+// 两段证据：
+//
+//	A. 真实表 t_user_expense_log：显式插了 legacy id 之后，**不带 id** 插入的新行 id
+//	   必须等于即时读到的 AUTO_INCREMENT，且 >= max(id)+1 —— 只要它不是 1 附近的小数，
+//	   就说明计数器没有被显式插 id 弄回起点。
+//	B. 临时表镜像（CREATE TEMPORARY TABLE ... LIKE，会话结束自动消失）：
+//	   显式插 id=100，再不带 id 插，必须得到 101，把「max+1」这条语义**可重复地**演示出来。
+//
+// 为什么要 B：MySQL 的 AUTO_INCREMENT 计数器**DELETE 之后不回退**，
+// 所以 fixture 反复跑时真实表的计数器会单调增长，真实表上「恰好 max+1」只有首次能观测到；
+// 临时表镜像每次都从空表起步，因此每次都能给出一模一样的 max+1 结论。
+type AutoIncrementDemo struct {
+	MaxIDBefore      int64
+	AutoIncrementVal int64
+	ProbeID          int64
+	ExpectedProbeID  int64
+	EqualsMaxPlusOne bool
+	Pass             bool
+
+	MirrorExplicitID int64
+	MirrorProbeID    int64
+	MirrorExpected   int64
+	MirrorPass       bool
+}
+
+// FailCount 统计核对表中不相等的列数。
+func (r *VerifyReport) FailCount() int {
+	n := 0
+	for _, c := range r.Checks {
+		if !c.Equal {
+			n++
+		}
+	}
+	for _, a := range r.Assertions {
+		if !a.Pass {
+			n++
+		}
+	}
+	return n
+}
+
+// OK 表示整轮核对通过。
+func (r *VerifyReport) OK() bool { return r.FailCount() == 0 }

internal/academic/dualrun/fixture/report.go

@@ -0,0 +1,156 @@
+package fixture
+
+import (
+	"fmt"
+	"io"
+	"strings"
+)
+
+// 本文件只做**终端报告排版**，不含任何业务判断。
+// 排版目标是「能直接贴进报告文档」：定宽列 + 明确的 ✔/✘ 标记。
+
+const (
+	markOK   = "✔"
+	markFail = "✘"
+)
+
+// PrintDDLReport 打印结构补齐结果。
+func PrintDDLReport(w io.Writer, r *DDLReport) {
+	fmt.Fprintf(w, "== core 结构补齐 ==\n")
+	fmt.Fprintf(w, "DDL 文件：%s\n", r.Path)
+	fmt.Fprintf(w, "表数：%d → %d\n", r.TablesBefore, r.TablesAfter)
+	fmt.Fprintf(w, "新增（%d）：%s\n", len(r.Created), strings.Join(r.Created, ", "))
+	fmt.Fprintf(w, "跳过已存在（%d）：%s\n", len(r.Skipped), strings.Join(r.Skipped, ", "))
+	dropped := 0
+	seeds := 0
+	for _, a := range r.Actions {
+		switch a.Kind {
+		case "skipped-drop":
+			dropped++
+		case "seed":
+			seeds++
+		}
+	}
+	fmt.Fprintf(w, "丢弃的 DROP TABLE 语句：%d 条（不执行）\n", dropped)
+	fmt.Fprintf(w, "执行的 INSERT IGNORE 种子：%d 条（幂等）\n", seeds)
+}
+
+// PrintMapReport 打印搬数据结果。
+func PrintMapReport(w io.Writer, r *MapReport) {
+	fmt.Fprintf(w, "== 映射与装载 ==\n")
+	fmt.Fprintf(w, "legacy userid：%d\n", r.LegacyUserID)
+	fmt.Fprintf(w, "canonical_user_id：%s\n", r.CanonicalUserID)
+	fmt.Fprintf(w, "legacy 行数：t_user_expense=%d, t_user_expense_school=%d, t_user_expense_log=%d（合计 %d）\n",
+		r.LegacyCounts["t_user_expense"], r.LegacyCounts["t_user_expense_school"],
+		r.LegacyCounts["t_user_expense_log"], r.LegacyRows)
+	fmt.Fprintf(w, "写入：t_user=%d 行, t_legacy_mapping=%d 行（含创建人 %d 个）\n",
+		r.UsersWritten, r.MappingsWritten, len(r.CreatorIDs))
+	fmt.Fprintf(w, "流水 id（D1 原样保留）：%v\n", r.LogIDs)
+	if len(r.Samples) > 0 {
+		fmt.Fprintf(w, "流水样本（remark 已脱敏）：\n")
+		for _, s := range r.Samples {
+			fmt.Fprintf(w, "  id=%-8d type=%-4d money=%-12s create_time=%s remark=%s\n",
+				s.ID, s.Type, s.Money, s.CreateTime, s.RemarkMask)
+		}
+	}
+	if len(r.Warnings) > 0 {
+		fmt.Fprintf(w, "映射告警（%d）：\n", len(r.Warnings))
+		for _, x := range r.Warnings {
+			fmt.Fprintf(w, "  - %s\n", x)
+		}
+	} else {
+		fmt.Fprintf(w, "映射告警：无\n")
+	}
+}
+
+// PrintVerifyReport 打印逐列核对表 + 断言 + 两组证据。
+func PrintVerifyReport(w io.Writer, r *VerifyReport) {
+	fmt.Fprintf(w, "== 逐列核对 ==\n")
+	for _, table := range r.SortedTableNames() {
+		fmt.Fprintf(w, "\n-- %s --\n", table)
+		fmt.Fprintf(w, "  %s %-26s %-26s %-26s %-26s %s\n", "st", "rowKey", "column", "legacy", "target", "note")
+		for _, c := range r.Checks {
+			if c.Table != table {
+				continue
+			}
+			st := markOK
+			if !c.Equal {
+				st = markFail
+			}
+			fmt.Fprintf(w, "  %s %-26s %-26s %-26s %-26s %s\n",
+				st, c.RowKey, c.Column, truncate(c.Legacy, 26), truncate(c.Target, 26), c.Note)
+		}
+	}
+
+	fmt.Fprintf(w, "\n== 语义断言 ==\n")
+	for _, a := range r.Assertions {
+		st := markOK
+		if !a.Pass {
+			st = markFail
+		}
+		fmt.Fprintf(w, "  %s %s\n     %s\n", st, a.Name, a.Detail)
+	}
+
+	fmt.Fprintf(w, "\n== float 陷阱证据（legacy 逐条 3 位小数）==\n")
+	f := r.FloatTrap
+	for i := range f.LegacyValues {
+		fmt.Fprintf(w, "  [%d] legacy=%-12s ROUND(x,2)=%-10s\n", i, f.LegacyValues[i], f.PerRowRounded[i])
+	}
+	fmt.Fprintf(w, "  Σ ROUND(x,2)  = %s（迁移口径，逐行取整后相加）\n", f.SumRoundedStr)
+	fmt.Fprintf(w, "  ROUND(Σ x, 2) = %s（旧口径，先加后取整）\n", f.RoundOfSumStr)
+	fmt.Fprintf(w, "  两者分差      = %d 分\n", f.DeltaVsPerRow)
+	fmt.Fprintf(w, "  float64 直接累加 = %s（精确和 = %s）\n",
+		FormatFloat(f.FloatSumRaw), CentsToDecimal(f.ExactSumThousandths/10))
+	fmt.Fprintf(w, "  float64 累加误差 = %s\n", FormatFloat(f.FloatAccumErr))
+
+	fmt.Fprintf(w, "\n== AUTO_INCREMENT 不倒退 ==\n")
+	a := r.AutoIncTest
+	st := markOK
+	if !a.Pass {
+		st = markFail
+	}
+	fmt.Fprintf(w, "  %s 真实表 t_user_expense_log：max(id)=%d, AUTO_INCREMENT=%d（= max+1? %v）\n",
+		st, a.MaxIDBefore, a.AutoIncrementVal, a.EqualsMaxPlusOne)
+	fmt.Fprintf(w, "     不带 id 插入 → id=%d（= 计数器 %d）⇒ 计数器没有被显式插 id 重置\n",
+		a.ProbeID, a.ExpectedProbeID)
+	stm := markOK
+	if !a.MirrorPass {
+		stm = markFail
+	}
+	fmt.Fprintf(w, "  %s 临时表镜像（CREATE TEMPORARY TABLE ... LIKE）：显式插 id=%d → 不带 id 插入得到 id=%d（期望 %d）\n",
+		stm, a.MirrorExplicitID, a.MirrorProbeID, a.MirrorExpected)
+
+	fmt.Fprintf(w, "\n核对结论：%s（不等列数 + 失败断言数 = %d）\n", verdict(r.OK()), r.FailCount())
+}
+
+// PrintCleanReport 打印清理结果。
+func PrintCleanReport(w io.Writer, r *CleanReport) {
+	fmt.Fprintf(w, "== 清理 ==\n")
+	fmt.Fprintf(w, "删除的 canonical_user_id（%d 个）：%s\n", len(r.CanonicalIDs), strings.Join(r.CanonicalIDs, ", "))
+	for _, t := range []string{"t_user_expense", "t_user_expense_school", "t_user_expense_log", "t_legacy_mapping", "t_user"} {
+		fmt.Fprintf(w, "  %-24s 删除 %-4d 行，剩余全表 %d 行\n", t, r.Deleted[t], r.Remaining[t])
+	}
+	zero := true
+	for _, t := range []string{"t_user_expense", "t_user_expense_school", "t_user_expense_log"} {
+		if r.Remaining[t] != 0 {
+			zero = false
+		}
+	}
+	fmt.Fprintf(w, "4 张表回 0 行：%s\n", verdict(zero))
+}
+
+func verdict(ok bool) string {
+	if ok {
+		return "PASS"
+	}
+	return "FAIL"
+}
+
+// truncate 按字符截断展示宽度（中文按 2 列宽估算，够用即可）。
+func truncate(s string, max int) string {
+	r := []rune(s)
+	if len(r) <= max {
+		return s
+	}
+	return string(r[:max-1]) + "…"
+}

internal/academic/dualrun/fixture/run.go

@@ -0,0 +1,323 @@
+package fixture
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"sort"
+	"strings"
+)
+
+// 本文件是流程编排：Pick → Extract → Map → Load →（Verify）→ Clean。
+//
+// 每一步都可单独从 CLI 调用，因此**没有任何跨进程的内存状态**：
+// canonical_user_id 由 FixtureCanonicalUserID 确定性派生，重复执行天然幂等。
+
+// MapReport 是一次「搬数据」的结果。
+type MapReport struct {
+	LegacyUserID    int64
+	CanonicalUserID string
+	LegacyCounts    map[string]int
+	LegacyRows      int
+	Warnings        []string
+	CreatorIDs      []int64
+	UsersWritten    int
+	MappingsWritten int
+	LogIDs          []int64
+	// 样本：3 条流水（id/type/money/时间），供报告引用（remark 已脱敏）。
+	Samples []LogSample
+}
+
+// LogSample 是报告里引用的流水样本。
+type LogSample struct {
+	ID         int64
+	Type       int32
+	Money      string
+	CreateTime string
+	RemarkMask string
+}
+
+// PrepareAndLoad 把选中用户的一条链路搬到目标库。
+//
+// 步骤：
+//  1. legacy 侧只读取出 1 账户 + N 校区账户 + M 条流水；
+//  2. 为「本人 + 流水里出现的创建人」各插 1 行 t_legacy_mapping 与 t_user；
+//     —— 创建人是 create_user_id 的引用对象，不插就会退化成 NULL，核对就看不到「解析成功」这一半；
+//  3. 逐行映射（mapping.go 纯函数）→ 幂等写入（load.go）。
+func PrepareAndLoad(ctx context.Context, cfg Config, legacyUserID int64) (*MapReport, error) {
+	legacy, err := cfg.OpenLegacy()
+	if err != nil {
+		return nil, err
+	}
+	defer legacy.Close()
+
+	target, err := cfg.OpenTarget()
+	if err != nil {
+		return nil, err
+	}
+	defer target.Close()
+
+	rep := &MapReport{LegacyUserID: legacyUserID}
+
+	expense, err := extractExpense(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, err
+	}
+	schools, err := extractExpenseSchools(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, err
+	}
+	logs, err := extractExpenseLogs(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, err
+	}
+	if len(schools) == 0 {
+		return nil, fmt.Errorf("legacy 用户 %d 无 t_user_expense_school 行，不满足 fixture 前置条件", legacyUserID)
+	}
+	if len(logs) < 3 {
+		return nil, fmt.Errorf("legacy 用户 %d 只有 %d 条流水（要求 >=3）", legacyUserID, len(logs))
+	}
+
+	rep.LegacyCounts, err = countLegacyRows(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, err
+	}
+	rep.LegacyRows = rep.LegacyCounts["t_user_expense"] + rep.LegacyCounts["t_user_expense_school"] + rep.LegacyCounts["t_user_expense_log"]
+
+	canonical := FixtureCanonicalUserID(legacyUserID)
+	rep.CanonicalUserID = canonical
+
+	// ---- 创建人集合（去重、升序）----
+	creatorSet := map[int64]bool{}
+	for _, l := range logs {
+		if l.CreateUserID.Valid && l.CreateUserID.Int64 > 0 {
+			creatorSet[l.CreateUserID.Int64] = true
+		}
+	}
+	for id := range creatorSet {
+		rep.CreatorIDs = append(rep.CreatorIDs, id)
+	}
+	sort.Slice(rep.CreatorIDs, func(i, j int) bool { return rep.CreatorIDs[i] < rep.CreatorIDs[j] })
+
+	// ---- 1) t_legacy_mapping + t_user 行 ----
+	if err := upsertMapping(ctx, target, MappingEntry{
+		CanonicalUserID:      canonical,
+		LegacyAcademicUserID: legacyUserID,
+		InstitutionID:        schools[0].SchoolID,
+		CampusID:             schools[0].CampusID,
+		Role:                 "student",
+		MatchSource:          "seed",
+		MatchConfidence:      100,
+	}); err != nil {
+		return nil, err
+	}
+	rep.MappingsWritten++
+	if err := upsertUser(ctx, target, canonical, "AC15 fixture 账户(legacy="+itoa(legacyUserID)+")"); err != nil {
+		return nil, err
+	}
+	rep.UsersWritten++
+
+	for _, cid := range rep.CreatorIDs {
+		uid := FixtureCanonicalUserID(cid)
+		if err := upsertMapping(ctx, target, MappingEntry{
+			CanonicalUserID:      uid,
+			LegacyAcademicUserID: cid,
+			InstitutionID:        schools[0].SchoolID,
+			CampusID:             schools[0].CampusID,
+			Role:                 "staff",
+			MatchSource:          "seed",
+			MatchConfidence:      100,
+		}); err != nil {
+			return nil, err
+		}
+		rep.MappingsWritten++
+		if err := upsertUser(ctx, target, uid, "AC15 fixture 创建人(legacy="+itoa(cid)+")"); err != nil {
+			return nil, err
+		}
+		rep.UsersWritten++
+	}
+
+	resolver := func(id int64) (string, bool) {
+		if id <= 0 {
+			return "", false
+		}
+		return FixtureCanonicalUserID(id), true
+	}
+
+	// ---- 2) 全局账户 ----
+	te, warns, err := MapExpense(*expense, canonical)
+	if err != nil {
+		return nil, err
+	}
+	rep.Warnings = append(rep.Warnings, warns...)
+	if err := upsertExpense(ctx, target, te); err != nil {
+		return nil, err
+	}
+
+	// ---- 3) 校区账户 ----
+	for _, s := range schools {
+		ts, w, err := MapExpenseSchool(s, canonical)
+		if err != nil {
+			return nil, err
+		}
+		rep.Warnings = append(rep.Warnings, w...)
+		if err := upsertExpenseSchool(ctx, target, ts); err != nil {
+			return nil, err
+		}
+	}
+
+	// ---- 4) 流水（D1 显式 id）----
+	for _, l := range logs {
+		tl, w, err := MapExpenseLog(l, canonical, resolver)
+		if err != nil {
+			return nil, err
+		}
+		rep.Warnings = append(rep.Warnings, w...)
+		if err := insertExpenseLog(ctx, target, tl); err != nil {
+			return nil, err
+		}
+		rep.LogIDs = append(rep.LogIDs, tl.ID)
+	}
+
+	// ---- 5) 样本（前 3 条，remark 脱敏）----
+	for i, l := range logs {
+		if i >= 3 {
+			break
+		}
+		rep.Samples = append(rep.Samples, LogSample{
+			ID:         l.ID,
+			Type:       l.Type,
+			Money:      l.Money.String,
+			CreateTime: l.CreateTime.Time.Format("2006-01-02 15:04:05"),
+			RemarkMask: maskRemark(l.Remark),
+		})
+	}
+	return rep, nil
+}
+
+// CleanReport 是清理结果。
+type CleanReport struct {
+	Deleted      map[string]int64
+	Remaining    map[string]int64
+	CanonicalIDs []string
+}
+
+// Cleanup 删除本 fixture 造的全部数据，并回报 4 张表的剩余行数。
+//
+// 删除范围严格限定在「由 FixtureCanonicalUserID 派生出来的 ULID」上，
+// 不会误删任何非 fixture 行。
+func Cleanup(ctx context.Context, cfg Config, legacyUserID int64) (*CleanReport, error) {
+	target, err := cfg.OpenTarget()
+	if err != nil {
+		return nil, err
+	}
+	defer target.Close()
+
+	legacy, err := cfg.OpenLegacy()
+	if err != nil {
+		return nil, err
+	}
+	defer legacy.Close()
+
+	logs, err := extractExpenseLogs(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, err
+	}
+	creatorSet := map[int64]bool{legacyUserID: true}
+	for _, l := range logs {
+		if l.CreateUserID.Valid && l.CreateUserID.Int64 > 0 {
+			creatorSet[l.CreateUserID.Int64] = true
+		}
+	}
+	var ids []int64
+	for id := range creatorSet {
+		ids = append(ids, id)
+	}
+	sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
+
+	var canonicals []string
+	for _, id := range ids {
+		canonicals = append(canonicals, FixtureCanonicalUserID(id))
+	}
+
+	rep := &CleanReport{Deleted: map[string]int64{}, Remaining: map[string]int64{}, CanonicalIDs: canonicals}
+
+	inList := strings.TrimSuffix(strings.Repeat("?,", len(canonicals)), ",")
+	args := make([]any, len(canonicals))
+	for i, c := range canonicals {
+		args[i] = c
+	}
+
+	// 1) 探针行（变异/异常中断时可能残留）
+	if _, err := target.ExecContext(ctx,
+		`DELETE FROM t_user_expense_log WHERE remark = 'ac15 fixture auto-inc probe'`); err != nil {
+		return nil, err
+	}
+
+	for _, t := range []string{"t_user_expense_log", "t_user_expense_school", "t_user_expense"} {
+		res, err := target.ExecContext(ctx, fmt.Sprintf("DELETE FROM %s WHERE user_id IN (%s)", t, inList), args...)
+		if err != nil {
+			return nil, fmt.Errorf("清理 %s 失败：%w", t, err)
+		}
+		n, _ := res.RowsAffected()
+		rep.Deleted[t] = n
+	}
+
+	// 映射与用户行
+	res, err := target.ExecContext(ctx,
+		fmt.Sprintf("DELETE FROM t_legacy_mapping WHERE canonical_user_id IN (%s)", inList), args...)
+	if err != nil {
+		return nil, err
+	}
+	n, _ := res.RowsAffected()
+	rep.Deleted["t_legacy_mapping"] = n
+
+	res, err = target.ExecContext(ctx,
+		fmt.Sprintf("DELETE FROM t_user WHERE canonical_user_id IN (%s)", inList), args...)
+	if err != nil {
+		return nil, err
+	}
+	n, _ = res.RowsAffected()
+	rep.Deleted["t_user"] = n
+
+	// 复核 4 张表的**全表**行数（不按 user 过滤：要证明回到 0 行）
+	for _, t := range []string{"t_user_expense", "t_user_expense_school", "t_user_expense_log"} {
+		var cnt int64
+		if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&cnt); err != nil {
+			return nil, err
+		}
+		rep.Remaining[t] = cnt
+	}
+	for _, t := range []string{"t_legacy_mapping", "t_user"} {
+		var cnt int64
+		if err := target.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&cnt); err != nil {
+			return nil, err
+		}
+		rep.Remaining[t] = cnt
+	}
+	return rep, nil
+}
+
+// CountTargetTables 统计目标 4 张表当前行数（清理前后对照用）。
+func CountTargetTables(ctx context.Context, cfg Config) (map[string]int64, error) {
+	db, err := cfg.OpenTarget()
+	if err != nil {
+		return nil, err
+	}
+	defer db.Close()
+	return countTargetTablesWith(ctx, db)
+}
+
+func countTargetTablesWith(ctx context.Context, db *sql.DB) (map[string]int64, error) {
+	out := map[string]int64{}
+	for _, t := range []string{"t_user_expense", "t_user_expense_school", "t_user_expense_log", "t_legacy_mapping", "t_user"} {
+		var n int64
+		if err := db.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+t).Scan(&n); err != nil {
+			return nil, err
+		}
+		out[t] = n
+	}
+	return out, nil
+}
+
+func itoa(v int64) string { return fmt.Sprintf("%d", v) }

internal/academic/dualrun/fixture/verify.go

@@ -0,0 +1,778 @@
+package fixture
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"math"
+	"sort"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// 本文件做**逐列核对**，关键设计是「期望值从哪来」：
+//
+// ⚠ 期望值**不由本包的映射代码产生**（否则映射写错时两边一起错，核对永远绿）。
+// 期望值一律由 **MySQL 服务端**算出：`CAST(x AS DECIMAL(12,2))` / `DATE_FORMAT(...)`。
+// 于是：
+//   · Go 侧的取整/换算只要与 DECIMAL 语义不一致，核对立刻红；
+//   · 变异实验（故意改错映射）必然红——这正是它的意义。
+
+// normalizeMoney 把服务端返回的十进制串规范成 2 位小数字符串，
+// 用于「期望 vs 实际」的字符串级比较（"-0.00" → "0.00"）。
+func normalizeMoney(s string) string {
+	s = strings.TrimSpace(s)
+	if s == "" {
+		return ""
+	}
+	c, err := DecimalToCents(s)
+	if err != nil {
+		return "ERR:" + s
+	}
+	if c == 0 {
+		return "0.00"
+	}
+	return CentsToDecimal(c)
+}
+
+// thousandthsOf 把十进制串精确转为「千分位」整数，用于 float 陷阱演示中的精确和。
+func thousandthsOf(s string) (int64, error) {
+	raw := strings.TrimSpace(s)
+	neg := false
+	if strings.HasPrefix(raw, "-") {
+		neg = true
+		raw = raw[1:]
+	}
+	intPart, frac := raw, ""
+	if i := strings.IndexByte(raw, '.'); i >= 0 {
+		intPart, frac = raw[:i], raw[i+1:]
+	}
+	for len(frac) < 3 {
+		frac += "0"
+	}
+	if len(frac) > 3 {
+		frac = frac[:3]
+	}
+	ip := int64(0)
+	if intPart != "" {
+		v, err := strconv.ParseInt(intPart, 10, 64)
+		if err != nil {
+			return 0, err
+		}
+		ip = v
+	}
+	fp, err := strconv.ParseInt(frac, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	v := ip*1000 + fp
+	if neg {
+		v = -v
+	}
+	return v, nil
+}
+
+// ---------- 期望值读取（服务端口径） ----------
+
+type expExpense struct {
+	UserID                                                    int64
+	Balance, Spend, Income, Expenditure, Recommended, Special string
+}
+
+func readExpectedExpense(ctx context.Context, legacy *sql.DB, userID int64) (*expExpense, error) {
+	const q = `
+SELECT userid,
+       CAST(balance                  AS DECIMAL(12,2)),
+       CAST(spend                    AS DECIMAL(12,2)),
+       CAST(total_income             AS DECIMAL(12,2)),
+       CAST(total_expenditure        AS DECIMAL(12,2)),
+       CAST(ifnull(total_recommended_amount,0) AS DECIMAL(12,2)),
+       CAST(ifnull(special_expenses,0)         AS DECIMAL(12,2))
+  FROM t_user_expense WHERE userid = ?`
+	var e expExpense
+	if err := legacy.QueryRowContext(ctx, q, userID).Scan(
+		&e.UserID, &e.Balance, &e.Spend, &e.Income, &e.Expenditure, &e.Recommended, &e.Special); err != nil {
+		return nil, err
+	}
+	return &e, nil
+}
+
+type expSchool struct {
+	ID, SchoolID, CampusID                                    int64
+	Balance, Spend, Income, Expenditure, Recommended, Special string
+}
+
+func readExpectedSchools(ctx context.Context, legacy *sql.DB, userID int64) ([]expSchool, error) {
+	const q = `
+SELECT id, schoolid, cast(ifnull(campusid,0) as signed),
+       CAST(balance                  AS DECIMAL(12,2)),
+       CAST(spend                    AS DECIMAL(12,2)),
+       CAST(total_income             AS DECIMAL(12,2)),
+       CAST(total_expenditure        AS DECIMAL(12,2)),
+       CAST(ifnull(total_recommended_amount,0) AS DECIMAL(12,2)),
+       CAST(ifnull(special_expenses,0)         AS DECIMAL(12,2))
+  FROM t_user_expense_school WHERE userid = ? ORDER BY id`
+	rows, err := legacy.QueryContext(ctx, q, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []expSchool
+	for rows.Next() {
+		var r expSchool
+		if err := rows.Scan(&r.ID, &r.SchoolID, &r.CampusID,
+			&r.Balance, &r.Spend, &r.Income, &r.Expenditure, &r.Recommended, &r.Special); err != nil {
+			return nil, err
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}
+
+type expLog struct {
+	ID            int64
+	Type          int32
+	LogID         sql.NullString
+	Money         string
+	CreateUserID  sql.NullInt64
+	CreateTimeStr string
+	Remark        sql.NullString
+}
+
+func readExpectedLogs(ctx context.Context, legacy *sql.DB, userID int64) ([]expLog, error) {
+	const q = `
+SELECT id, cast(ifnull(type,0) as signed), logid,
+       CAST(money AS DECIMAL(12,2)),
+       create_userid,
+       DATE_FORMAT(create_time, '%Y-%m-%d %H:%i:%s.%f'),
+       remark
+  FROM t_user_expense_log WHERE userid = ? ORDER BY id`
+	rows, err := legacy.QueryContext(ctx, q, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []expLog
+	for rows.Next() {
+		var r expLog
+		if err := rows.Scan(&r.ID, &r.Type, &r.LogID, &r.Money, &r.CreateUserID, &r.CreateTimeStr, &r.Remark); err != nil {
+			return nil, err
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}
+
+// ---------- 目标值读取 ----------
+
+type actExpense struct {
+	UserID                                                    string
+	Balance, Spend, Income, Expenditure, Recommended, Special string
+}
+
+func readActualExpense(ctx context.Context, db *sql.DB, userID string) (*actExpense, error) {
+	const q = `
+SELECT user_id, CAST(balance AS DECIMAL(12,2)), CAST(spend AS DECIMAL(12,2)),
+       CAST(total_income AS DECIMAL(12,2)), CAST(total_expenditure AS DECIMAL(12,2)),
+       CAST(total_recommended_amount AS DECIMAL(12,2)), CAST(special_expenses AS DECIMAL(12,2))
+  FROM t_user_expense WHERE user_id = ?`
+	var a actExpense
+	err := db.QueryRowContext(ctx, q, userID).Scan(&a.UserID, &a.Balance, &a.Spend,
+		&a.Income, &a.Expenditure, &a.Recommended, &a.Special)
+	if err != nil {
+		return nil, err
+	}
+	return &a, nil
+}
+
+type actSchool struct {
+	ID, InstitutionID, CampusID                               int64
+	Balance, Spend, Income, Expenditure, Recommended, Special string
+}
+
+func readActualSchools(ctx context.Context, db *sql.DB, userID string) ([]actSchool, error) {
+	const q = `
+SELECT id, institution_id, campus_id,
+       CAST(balance AS DECIMAL(12,2)), CAST(spend AS DECIMAL(12,2)),
+       CAST(total_income AS DECIMAL(12,2)), CAST(total_expenditure AS DECIMAL(12,2)),
+       CAST(total_recommended_amount AS DECIMAL(12,2)), CAST(special_expenses AS DECIMAL(12,2))
+  FROM t_user_expense_school WHERE user_id = ? ORDER BY institution_id, campus_id`
+	rows, err := db.QueryContext(ctx, q, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []actSchool
+	for rows.Next() {
+		var r actSchool
+		if err := rows.Scan(&r.ID, &r.InstitutionID, &r.CampusID, &r.Balance, &r.Spend,
+			&r.Income, &r.Expenditure, &r.Recommended, &r.Special); err != nil {
+			return nil, err
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}
+
+type actLog struct {
+	ID            int64
+	UserID        string
+	Type          int32
+	LogID         sql.NullString
+	RequestID     sql.NullString
+	Money         string
+	BalanceAfter  sql.NullString
+	CreateUserID  sql.NullString
+	CreateTimeStr string
+	Remark        sql.NullString
+}
+
+func readActualLogs(ctx context.Context, db *sql.DB, userID string) ([]actLog, error) {
+	const q = `
+SELECT id, user_id, type, log_id, request_id,
+       CAST(money AS DECIMAL(12,2)),
+       CAST(balance_after AS DECIMAL(12,2)),
+       create_user_id,
+       DATE_FORMAT(create_time, '%Y-%m-%d %H:%i:%s.%f'),
+       remark
+  FROM t_user_expense_log WHERE user_id = ? ORDER BY id`
+	rows, err := db.QueryContext(ctx, q, userID)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []actLog
+	for rows.Next() {
+		var r actLog
+		if err := rows.Scan(&r.ID, &r.UserID, &r.Type, &r.LogID, &r.RequestID, &r.Money,
+			&r.BalanceAfter, &r.CreateUserID, &r.CreateTimeStr, &r.Remark); err != nil {
+			return nil, err
+		}
+		out = append(out, r)
+	}
+	return out, rows.Err()
+}
+
+// ---------- 核对主体 ----------
+
+// Verify 执行一轮完整核对：4 张表逐列 + AUTO_INCREMENT + 空值语义 + float 陷阱。
+func Verify(ctx context.Context, cfg Config, legacyUserID int64) (*VerifyReport, error) {
+	canonical := FixtureCanonicalUserID(legacyUserID)
+
+	target, err := cfg.OpenTarget()
+	if err != nil {
+		return nil, err
+	}
+	defer target.Close()
+
+	legacy, err := cfg.OpenLegacy()
+	if err != nil {
+		return nil, err
+	}
+	defer legacy.Close()
+
+	rep := &VerifyReport{
+		LoadedRows: map[string]int{},
+		TargetRows: map[string]int{},
+	}
+
+	// ===== 1. t_user_expense（1 行 7 列）=====
+	ee, err := readExpectedExpense(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, fmt.Errorf("读期望 t_user_expense 失败：%w", err)
+	}
+	ae, err := readActualExpense(ctx, target, canonical)
+	if err != nil {
+		return nil, fmt.Errorf("读实际 t_user_expense 失败：%w", err)
+	}
+	key := "user_id=" + canonical
+	rep.Checks = append(rep.Checks,
+		check("t_user_expense", key, "user_id", fmt.Sprintf("%d", ee.UserID), ae.UserID,
+			ae.UserID == canonical, "legacy int → canonical ULID（经 t_legacy_mapping）"),
+		check("t_user_expense", key, "balance", ee.Balance, ae.Balance,
+			normalizeMoney(ee.Balance) == normalizeMoney(ae.Balance), ""),
+		check("t_user_expense", key, "spend", ee.Spend, ae.Spend,
+			normalizeMoney(ee.Spend) == normalizeMoney(ae.Spend), ""),
+		check("t_user_expense", key, "total_income", ee.Income, ae.Income,
+			normalizeMoney(ee.Income) == normalizeMoney(ae.Income), ""),
+		check("t_user_expense", key, "total_expenditure", ee.Expenditure, ae.Expenditure,
+			normalizeMoney(ee.Expenditure) == normalizeMoney(ae.Expenditure), ""),
+		check("t_user_expense", key, "total_recommended_amount", ee.Recommended, ae.Recommended,
+			normalizeMoney(ee.Recommended) == normalizeMoney(ae.Recommended), ""),
+		check("t_user_expense", key, "special_expenses", ee.Special, ae.Special,
+			normalizeMoney(ee.Special) == normalizeMoney(ae.Special), ""),
+	)
+	rep.LoadedRows["t_user_expense"] = 1
+	rep.TargetRows["t_user_expense"] = 1
+
+	// ===== 2. t_user_expense_school（N 行 × 8 列）=====
+	es, err := readExpectedSchools(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, fmt.Errorf("读期望 t_user_expense_school 失败：%w", err)
+	}
+	as, err := readActualSchools(ctx, target, canonical)
+	if err != nil {
+		return nil, fmt.Errorf("读实际 t_user_expense_school 失败：%w", err)
+	}
+	rep.LoadedRows["t_user_expense_school"] = len(es)
+	rep.TargetRows["t_user_expense_school"] = len(as)
+	rep.Checks = append(rep.Checks, check("t_user_expense_school", "-", "行数",
+		fmt.Sprintf("%d", len(es)), fmt.Sprintf("%d", len(as)), len(es) == len(as), "复合业务键 (user,institution,campus) 对齐"))
+
+	// 按 (institution_id, campus_id) 建索引，避免依赖 legacy id（该表不保留 legacy id）。
+	actualByKey := map[string]actSchool{}
+	for _, r := range as {
+		actualByKey[fmt.Sprintf("%d/%d", r.InstitutionID, r.CampusID)] = r
+	}
+	for _, e := range es {
+		k := fmt.Sprintf("%d/%d", e.SchoolID, e.CampusID)
+		a, ok := actualByKey[k]
+		rowKey := fmt.Sprintf("inst=%d/campus=%d", e.SchoolID, e.CampusID)
+		if !ok {
+			rep.Checks = append(rep.Checks, ColumnCheck{
+				Table: "t_user_expense_school", RowKey: rowKey, Column: "<整行>",
+				Legacy: "存在", Target: "缺失", Equal: false, Note: "目标侧缺该复合键行",
+			})
+			continue
+		}
+		rep.Checks = append(rep.Checks,
+			check("t_user_expense_school", rowKey, "institution_id", fmt.Sprintf("%d", e.SchoolID), fmt.Sprintf("%d", a.InstitutionID), e.SchoolID == a.InstitutionID, "schoolid→institution_id"),
+			check("t_user_expense_school", rowKey, "campus_id", fmt.Sprintf("%d", e.CampusID), fmt.Sprintf("%d", a.CampusID), e.CampusID == a.CampusID, "campusid→campus_id"),
+			check("t_user_expense_school", rowKey, "balance", e.Balance, a.Balance, normalizeMoney(e.Balance) == normalizeMoney(a.Balance), ""),
+			check("t_user_expense_school", rowKey, "spend", e.Spend, a.Spend, normalizeMoney(e.Spend) == normalizeMoney(a.Spend), ""),
+			check("t_user_expense_school", rowKey, "total_income", e.Income, a.Income, normalizeMoney(e.Income) == normalizeMoney(a.Income), ""),
+			check("t_user_expense_school", rowKey, "total_expenditure", e.Expenditure, a.Expenditure, normalizeMoney(e.Expenditure) == normalizeMoney(a.Expenditure), ""),
+			check("t_user_expense_school", rowKey, "total_recommended_amount", e.Recommended, a.Recommended, normalizeMoney(e.Recommended) == normalizeMoney(a.Recommended), ""),
+			check("t_user_expense_school", rowKey, "special_expenses", e.Special, a.Special, normalizeMoney(e.Special) == normalizeMoney(a.Special), ""),
+		)
+	}
+
+	// ===== 3. t_user_expense_log（M 行 × 10 列）=====
+	el, err := readExpectedLogs(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, fmt.Errorf("读期望 t_user_expense_log 失败：%w", err)
+	}
+	al, err := readActualLogs(ctx, target, canonical)
+	if err != nil {
+		return nil, fmt.Errorf("读实际 t_user_expense_log 失败：%w", err)
+	}
+	rep.LoadedRows["t_user_expense_log"] = len(el)
+	rep.TargetRows["t_user_expense_log"] = len(al)
+	rep.Checks = append(rep.Checks, check("t_user_expense_log", "-", "行数",
+		fmt.Sprintf("%d", len(el)), fmt.Sprintf("%d", len(al)), len(el) == len(al), "按 id 对齐"))
+
+	actualLogByID := map[int64]actLog{}
+	for _, r := range al {
+		actualLogByID[r.ID] = r
+	}
+	for _, e := range el {
+		rowKey := fmt.Sprintf("id=%d", e.ID)
+		a, ok := actualLogByID[e.ID]
+		if !ok {
+			rep.Checks = append(rep.Checks, ColumnCheck{
+				Table: "t_user_expense_log", RowKey: rowKey, Column: "<整行>",
+				Legacy: "存在", Target: "缺失", Equal: false, Note: "D1 显式插 id 未生效：目标侧无该 id",
+			})
+			continue
+		}
+		// create_user_id 期望：legacy 0/NULL → NULL；>0 → 必须是某个 ULID（由 t_legacy_mapping 解析）
+		expCreateUser := "NULL"
+		createUserOK := !a.CreateUserID.Valid
+		if e.CreateUserID.Valid && e.CreateUserID.Int64 > 0 {
+			expCreateUser = "ULID(legacy=" + strconv.FormatInt(e.CreateUserID.Int64, 10) + ")"
+			createUserOK = a.CreateUserID.Valid && a.CreateUserID.String == FixtureCanonicalUserID(e.CreateUserID.Int64)
+		}
+		actCreateUser := "NULL"
+		if a.CreateUserID.Valid {
+			actCreateUser = a.CreateUserID.String
+		}
+
+		rep.Checks = append(rep.Checks,
+			check("t_user_expense_log", rowKey, "id", strconv.FormatInt(e.ID, 10), strconv.FormatInt(a.ID, 10), e.ID == a.ID, "D1：与 legacy id 完全相同"),
+			check("t_user_expense_log", rowKey, "user_id", strconv.FormatInt(legacyUserID, 10), a.UserID, a.UserID == canonical, "legacy userid → canonical ULID"),
+			check("t_user_expense_log", rowKey, "type", strconv.FormatInt(int64(e.Type), 10), strconv.FormatInt(int64(a.Type), 10), e.Type == a.Type, ""),
+			check("t_user_expense_log", rowKey, "log_id", nullStrView(e.LogID), nullStrView(a.LogID), nullStrView(e.LogID) == nullStrView(a.LogID), "logid→log_id"),
+			check("t_user_expense_log", rowKey, "request_id", "NULL", nullStrView(a.RequestID), !a.RequestID.Valid, "legacy 无此列 → 必须 NULL"),
+			check("t_user_expense_log", rowKey, "money", e.Money, a.Money, normalizeMoney(e.Money) == normalizeMoney(a.Money), "逐行 ROUND 到分"),
+			check("t_user_expense_log", rowKey, "balance_after", "NULL", nullStrView(a.BalanceAfter), !a.BalanceAfter.Valid, "legacy 无快照 → 必须 NULL"),
+			check("t_user_expense_log", rowKey, "create_user_id", expCreateUser, actCreateUser, createUserOK, "legacy 0/系统 → NULL"),
+			check("t_user_expense_log", rowKey, "create_time", e.CreateTimeStr, a.CreateTimeStr, e.CreateTimeStr == a.CreateTimeStr, "秒级相同，毫秒补 .000"),
+			check("t_user_expense_log", rowKey, "remark", maskRemark(e.Remark), maskRemark(a.Remark), sameRemark(e.Remark, a.Remark), "报告脱敏，比对用原文"),
+		)
+	}
+
+	// ===== 4. t_user / t_legacy_mapping 链接 =====
+	if err := verifyIdentity(ctx, target, legacyUserID, canonical, rep); err != nil {
+		return nil, err
+	}
+
+	// ===== 5. 语义断言 =====
+	rep.Assertions = append(rep.Assertions, Assertion{
+		Name:   "create_user_id=0 → NULL（本 fixture 数据无 0 行时按空集通过）",
+		Pass:   countZeroCreator(target, canonical),
+		Detail: "SQL: SELECT COUNT(*) FROM t_user_expense_log WHERE user_id=? AND create_user_id='0' 必须为 0",
+	})
+	rep.Assertions = append(rep.Assertions, Assertion{
+		Name:   "balance_after 全为 NULL",
+		Pass:   countNonNullBalanceAfter(target, canonical),
+		Detail: "SQL: ... WHERE user_id=? AND balance_after IS NOT NULL 必须为 0",
+	})
+
+	// ===== 6. AUTO_INCREMENT 不倒退 =====
+	aiDemo, err := autoIncrementDemo(ctx, target, canonical)
+	if err != nil {
+		return nil, err
+	}
+	rep.AutoIncTest = aiDemo
+	rep.Assertions = append(rep.Assertions, Assertion{
+		Name: "显式插 id 后 AUTO_INCREMENT 不倒退（探针 id = 计数器，且 >= max+1；镜像表演示 max+1）",
+		Pass: aiDemo.Pass && aiDemo.MirrorPass,
+		Detail: fmt.Sprintf(
+			"真实表：max(id)=%d, AUTO_INCREMENT=%d（=max+1? %v）, 不带 id 插入得到 id=%d；"+
+				"临时表镜像：显式插 id=%d → 不带 id 插入得到 id=%d（期望 %d）",
+			aiDemo.MaxIDBefore, aiDemo.AutoIncrementVal, aiDemo.EqualsMaxPlusOne, aiDemo.ProbeID,
+			aiDemo.MirrorExplicitID, aiDemo.MirrorProbeID, aiDemo.MirrorExpected),
+	})
+
+	// ===== 7. float 陷阱证据 =====
+	trap, err := floatTrap(ctx, legacy, legacyUserID)
+	if err != nil {
+		return nil, err
+	}
+	rep.FloatTrap = trap
+
+	return rep, nil
+}
+
+// check 构造一条核对行。
+func check(table, rowKey, column, legacyVal, targetVal string, equal bool, note string) ColumnCheck {
+	return ColumnCheck{Table: table, RowKey: rowKey, Column: column,
+		Legacy: legacyVal, Target: targetVal, Equal: equal, Note: note}
+}
+
+func nullStrView(v sql.NullString) string {
+	if !v.Valid {
+		return "NULL"
+	}
+	return v.String
+}
+
+// maskRemark 报告里脱敏：备注可能含真实姓名/手机号，只展示前 8 个字符 + 长度。
+func maskRemark(v sql.NullString) string {
+	if !v.Valid {
+		return "NULL"
+	}
+	r := []rune(v.String)
+	head := r
+	if len(head) > 8 {
+		head = head[:8]
+	}
+	return fmt.Sprintf("%s…(len=%d)", string(head), len(r))
+}
+
+// sameRemark 用原文比对（脱敏只影响展示，不影响核对）。
+func sameRemark(a, b sql.NullString) bool {
+	if a.Valid != b.Valid {
+		return false
+	}
+	if !a.Valid {
+		return true
+	}
+	return a.String == b.String
+}
+
+// verifyIdentity 核对 t_legacy_mapping 与 t_user 的链接是否正确。
+func verifyIdentity(ctx context.Context, target *sql.DB, legacyUserID int64, canonical string, rep *VerifyReport) error {
+	var mapped string
+	err := target.QueryRowContext(ctx,
+		`SELECT canonical_user_id FROM t_legacy_mapping WHERE legacy_academic_user_id = ?`, legacyUserID).Scan(&mapped)
+	switch {
+	case err == sql.ErrNoRows:
+		rep.Checks = append(rep.Checks, ColumnCheck{Table: "t_legacy_mapping",
+			RowKey: fmt.Sprintf("legacy=%d", legacyUserID), Column: "canonical_user_id",
+			Legacy: "存在", Target: "缺失", Equal: false, Note: "映射行不存在"})
+	case err != nil:
+		return err
+	default:
+		rep.Checks = append(rep.Checks, check("t_legacy_mapping", fmt.Sprintf("legacy=%d", legacyUserID),
+			"canonical_user_id", canonical, mapped, mapped == canonical, "legacy int → ULID 映射正确"))
+	}
+
+	var displayName sql.NullString
+	err = target.QueryRowContext(ctx,
+		`SELECT display_name FROM t_user WHERE canonical_user_id = ?`, canonical).Scan(&displayName)
+	switch {
+	case err == sql.ErrNoRows:
+		rep.Checks = append(rep.Checks, ColumnCheck{Table: "t_user",
+			RowKey: canonical, Column: "<整行>", Legacy: "存在", Target: "缺失", Equal: false, Note: "t_user 行不存在"})
+	case err != nil:
+		return err
+	default:
+		rep.Checks = append(rep.Checks, check("t_user", canonical, "canonical_user_id",
+			canonical, canonical, true, "主键存在即可（其余列为合法默认值）"))
+	}
+	return nil
+}
+
+// countZeroCreator 断言「没有一行 create_user_id 落成字符串 '0'」。
+func countZeroCreator(db *sql.DB, userID string) bool {
+	var n int
+	err := db.QueryRow(
+		`SELECT COUNT(*) FROM t_user_expense_log WHERE user_id = ? AND create_user_id = '0'`, userID).Scan(&n)
+	return err == nil && n == 0
+}
+
+// countNonNullBalanceAfter 断言 balance_after 全部为 NULL。
+func countNonNullBalanceAfter(db *sql.DB, userID string) bool {
+	var n int
+	err := db.QueryRow(
+		`SELECT COUNT(*) FROM t_user_expense_log WHERE user_id = ? AND balance_after IS NOT NULL`, userID).Scan(&n)
+	return err == nil && n == 0
+}
+
+// readAutoIncrement 读**即时**自增计数器。
+//
+// 两个坑都必须躲掉，否则会得到「AUTO_INCREMENT = 1」这种完全错误的结论：
+//  1. 不能读 information_schema.tables.AUTO_INCREMENT —— MySQL 8.0 默认
+//     `information_schema_stats_expiry = 86400`，那是最多缓存一天的值（实测返回 1）。
+//  2. 必须先 `SET SESSION information_schema_stats_expiry = 0`，且 SET 与 SELECT
+//     必须在**同一条连接**上（连接池会换连接），所以这里用 db.Conn 独占一条。
+func readAutoIncrement(ctx context.Context, db *sql.DB, table string) (int64, error) {
+	// SHOW 语句不支持占位符（Error 1064），因此表名内联；
+	// 表名来自本包常量，且此处再做一次字符白名单校验，避免拼接风险。
+	if table == "" {
+		return 0, fmt.Errorf("readAutoIncrement: 表名为空")
+	}
+	for _, r := range table {
+		if !(r == '_' || (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9')) {
+			return 0, fmt.Errorf("readAutoIncrement: 非法表名 %q", table)
+		}
+	}
+	conn, err := db.Conn(ctx)
+	if err != nil {
+		return 0, err
+	}
+	defer conn.Close()
+	if _, err := conn.ExecContext(ctx, "SET SESSION information_schema_stats_expiry = 0"); err != nil {
+		return 0, fmt.Errorf("关闭 information_schema 统计缓存失败：%w", err)
+	}
+	rows, err := conn.QueryContext(ctx, fmt.Sprintf("SHOW TABLE STATUS LIKE '%s'", table))
+	if err != nil {
+		return 0, err
+	}
+	defer rows.Close()
+	cols, err := rows.Columns()
+	if err != nil {
+		return 0, err
+	}
+	if !rows.Next() {
+		return 0, fmt.Errorf("SHOW TABLE STATUS 无 %s 行", table)
+	}
+	vals := make([]any, len(cols))
+	ptrs := make([]any, len(cols))
+	for i := range vals {
+		ptrs[i] = &vals[i]
+	}
+	if err := rows.Scan(ptrs...); err != nil {
+		return 0, err
+	}
+	for i, c := range cols {
+		if !strings.EqualFold(c, "Auto_increment") {
+			continue
+		}
+		switch v := vals[i].(type) {
+		case nil:
+			return 0, nil
+		case []byte:
+			return strconv.ParseInt(string(v), 10, 64)
+		case int64:
+			return v, nil
+		case uint64:
+			return int64(v), nil
+		default:
+			return 0, fmt.Errorf("Auto_increment 类型未知：%T", v)
+		}
+	}
+	return 0, fmt.Errorf("SHOW TABLE STATUS 结果里没有 Auto_increment 列")
+}
+
+// autoIncrementDemo 验证「显式插 id 之后自增计数器不倒回」。
+//
+// A 段（真实表）：读 max(id) 与即时 AUTO_INCREMENT → 插一条**不带 id** 的探针行 →
+// 探针 id 必须等于该计数器、且 >= max(id)+1（若被重置成 1，这里会直接撞主键或取到小数）。
+// B 段（临时表镜像）：CREATE TEMPORARY TABLE ... LIKE → 显式插 id=100 →
+// 不带 id 插 → 必须 101。临时表随会话结束自动消失，不污染 schema、无需 DROP。
+func autoIncrementDemo(ctx context.Context, db *sql.DB, userID string) (AutoIncrementDemo, error) {
+	var d AutoIncrementDemo
+	if err := db.QueryRowContext(ctx,
+		`SELECT IFNULL(MAX(id),0) FROM t_user_expense_log`).Scan(&d.MaxIDBefore); err != nil {
+		return d, err
+	}
+	ai, err := readAutoIncrement(ctx, db, "t_user_expense_log")
+	if err != nil {
+		return d, err
+	}
+	d.AutoIncrementVal = ai
+	d.ExpectedProbeID = ai
+	d.EqualsMaxPlusOne = ai == d.MaxIDBefore+1
+
+	res, err := db.ExecContext(ctx, `
+INSERT INTO t_user_expense_log
+  (user_id, type, log_id, request_id, money, balance_after, create_user_id, create_time, remark)
+VALUES (?, 0, NULL, NULL, '0.00', NULL, NULL, '1970-01-01 00:00:00', 'ac15 fixture auto-inc probe')`, userID)
+	if err != nil {
+		return d, fmt.Errorf("AUTO_INCREMENT 探针插入失败（计数器疑似被重置）：%w", err)
+	}
+	probeID, err := res.LastInsertId()
+	if err != nil {
+		return d, err
+	}
+	d.ProbeID = probeID
+	// 判定：探针 id 必须等于读到的计数器（证明「不带 id 的插入」就是拿计数器发号），
+	// 且计数器不得小于 max(id)+1（证明显式插 id 没有把计数器倒退）。
+	d.Pass = probeID == d.ExpectedProbeID && d.AutoIncrementVal >= d.MaxIDBefore+1
+	if _, err := db.ExecContext(ctx,
+		`DELETE FROM t_user_expense_log WHERE remark = 'ac15 fixture auto-inc probe'`); err != nil {
+		return d, fmt.Errorf("AUTO_INCREMENT 探针清理失败：%w", err)
+	}
+
+	// ---- B 段：临时表镜像，可重复地演示「显式插 id=100 → 下一条 = 101」----
+	if err := mirrorAutoIncrement(ctx, db, userID, &d); err != nil {
+		return d, err
+	}
+	return d, nil
+}
+
+// mirrorAutoIncrement 用 TEMPORARY TABLE 复现「max+1」。
+//
+// 独立连接执行：临时表是会话级的，必须确保 CREATE / INSERT / SELECT 走同一条连接。
+func mirrorAutoIncrement(ctx context.Context, db *sql.DB, userID string, d *AutoIncrementDemo) error {
+	conn, err := db.Conn(ctx)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	const tmp = "tmp_ac15_ai_probe"
+	steps := []string{
+		fmt.Sprintf("CREATE TEMPORARY TABLE `%s` LIKE t_user_expense_log", tmp),
+		fmt.Sprintf("INSERT INTO `%s` (id, user_id, type, money, create_time, remark) VALUES (100, '%s', 0, '0.00', '1970-01-01 00:00:00', 'mirror explicit id')", tmp, userID),
+	}
+	for _, s := range steps {
+		if _, err := conn.ExecContext(ctx, s); err != nil {
+			return fmt.Errorf("临时表镜像步骤失败（%s）：%w", s, err)
+		}
+	}
+	d.MirrorExplicitID = 100
+
+	res, err := conn.ExecContext(ctx, fmt.Sprintf(
+		"INSERT INTO `%s` (user_id, type, money, create_time, remark) VALUES ('%s', 0, '0.00', '1970-01-01 00:00:00', 'mirror auto id')",
+		tmp, userID))
+	if err != nil {
+		return fmt.Errorf("临时表镜像第二次插入失败：%w", err)
+	}
+	mirrorID, err := res.LastInsertId()
+	if err != nil {
+		return err
+	}
+	d.MirrorProbeID = mirrorID
+	d.MirrorExpected = d.MirrorExplicitID + 1
+	d.MirrorPass = mirrorID == d.MirrorExpected
+	return nil
+}
+
+// floatTrap 采集「float 直接相加」与「逐行取整到分再相加」的差异证据。
+//
+// 三组数字：
+//
+//	· Σ ROUND(x,2)  —— 逐行取整后相加（迁移口径，也是双跑对账口径）
+//	· ROUND(Σ x, 2) —— 先相加再取整（旧 SP 常见口径，与上面可能差 1 分）
+//	· float64 直接累加 —— 展示二进制浮点的累加误差
+func floatTrap(ctx context.Context, legacy *sql.DB, userID int64) (FloatTrapDemo, error) {
+	var d FloatTrapDemo
+
+	// 逐条取值（3 位小数原文）
+	rows, err := legacy.QueryContext(ctx, `
+SELECT CAST(money AS DECIMAL(20,3))
+  FROM t_user_expense_log WHERE userid = ? ORDER BY id`, userID)
+	if err != nil {
+		return d, err
+	}
+	var raws []string
+	for rows.Next() {
+		var s string
+		if err := rows.Scan(&s); err != nil {
+			rows.Close()
+			return d, err
+		}
+		raws = append(raws, s)
+	}
+	rows.Close()
+	if err := rows.Err(); err != nil {
+		return d, err
+	}
+	d.LegacyValues = raws
+
+	var exactThousandths int64
+	var floatSum float64
+	for _, raw := range raws {
+		th, err := thousandthsOf(raw)
+		if err != nil {
+			return d, err
+		}
+		exactThousandths += th
+		floatSum += float64Of(raw)
+
+		c, err := DecimalToCents(raw)
+		if err != nil {
+			return d, err
+		}
+		d.PerRowRounded = append(d.PerRowRounded, CentsToDecimal(c))
+		d.SumOfRoundedCents += c
+	}
+	d.SumRoundedStr = CentsToDecimal(d.SumOfRoundedCents)
+	d.ExactSumThousandths = exactThousandths
+	d.FloatSumRaw = floatSum
+	d.FloatAccumErr = floatSum - float64(exactThousandths)/1000
+
+	// 服务端口径：ROUND(Σ x, 2) —— 与 Σ ROUND(x,2) 可能差 1 分（这就是「先加后取整」的坑）
+	var roundOfSum string
+	if err := legacy.QueryRowContext(ctx,
+		`SELECT CAST(SUM(money) AS DECIMAL(12,2)) FROM t_user_expense_log WHERE userid = ?`,
+		userID).Scan(&roundOfSum); err != nil {
+		return d, err
+	}
+	d.RoundOfSumStr = roundOfSum
+	rc, err := DecimalToCents(roundOfSum)
+	if err != nil {
+		return d, err
+	}
+	d.DeltaVsPerRow = rc - d.SumOfRoundedCents
+	return d, nil
+}
+
+// SortedTableNames 返回核对表出现过的表名（去重后排序），供报告分节打印。
+func (r *VerifyReport) SortedTableNames() []string {
+	seen := map[string]bool{}
+	var out []string
+	for _, c := range r.Checks {
+		if !seen[c.Table] {
+			seen[c.Table] = true
+			out = append(out, c.Table)
+		}
+	}
+	sort.Strings(out)
+	return out
+}
+
+// FormatFloat 用 %.17g 展示 float64，暴露二进制尾差。
+func FormatFloat(v float64) string {
+	if math.IsNaN(v) {
+		return "NaN"
+	}
+	return strconv.FormatFloat(v, 'g', 17, 64)
+}
+
+// NowTruncSec 仅用于报告时间戳。
+func NowTruncSec() time.Time { return time.Now().Truncate(time.Second) }