feat(ac15): 结构对齐落地（29 条 ALTER）+ create_time 静默默认值断言

一、执行结果（分批，逐条 dry-run 通过后执行）
- 第一批 created_at/updated_at：16 条（14 条补 DEFAULT + t_campus 两列按 DDL 改 NOT NULL）
  成功 16 / 失败 0；复核 33 列 COLUMN_DEFAULT 全非 NULL
- 第二批 create_time：13 条，成功 13 / 失败 0；复核 13 列全非 NULL
- t_campus.created_at/updated_at 已 NOT NULL + DEFAULT（与 DDL 完全一致，不做"只补默认保持可空"的半对齐）
- 全库 40 表；除 4 张种子表（t_role 7 / t_permission 6 / t_role_permission 10 / casbin_rule 14 = 37 行）外全 0 行
- 未 DROP 任何表；MODIFY 子句的 type/NULL 性/注释全部照抄实库，只加 DEFAULT/ON UPDATE

二、ALGORITHM 的版本限制（实测，与要求不一致，已降级并声明）
MySQL 8.0.23 对 "MODIFY COLUMN 改默认值" 既不支持 ALGORITHM=INSTANT 也不支持 INPLACE：
  Error 1845 (0A000): ALGORITHM=INSTANT is not supported for this operation. Try ALGORITHM=COPY.
该能力要 8.0.29+。因此执行时用 --algorithm copy 显式声明"我知道要重建"（0 行表零成本），
升级后应改回 instant。工具默认仍是 instant，dry-run 会先把不支持暴露出来、拒绝执行。

三、create_time 静默默认值的硬断言（run_marker.go）
- map 时往 t_audit_log 写一条 action='ac15_fixture_run' 的运行标记（含毫秒时间戳）
- verify 断言：count(create_time >= 迁移执行时刻) == 0
- 覆盖 t_user_expense_log（batch1）、t_user_divide_into_log / t_expenditure_log（batch2）
- 取不到标记时按失败处理（不允许"没标记就当通过"）
- 变异 M5（映射故意不传 create_time）实测红：
  "迁移执行时刻=2026-11-10 11:47:31；create_time >= 该时刻的行数 = 10（期望 0）"
  同时列级也红：create_time 0/10 相等（legacy 2018-11-01 vs target 2026-11-10 11:47:32）

四、顺带修掉自己断言里的一个 bug
batch2 的"id 与 biz_no 解耦"原先用 "id <= N+100" 当上界，fixture 反复跑后 AUTO_INCREMENT 涨到 130
就被误判成"由 biz_no 反算"而假红。改为硬界（id 必须在 (0, 2^31) 内且全表唯一）+ 依赖倒序镜像实验。

五、README 补三节
- ALGORITHM 版本限制与降级说明
- create_time 为什么必须配断言（审计 vs 业务列的差别）
- 两批 fixture 共用 t_user/t_legacy_mapping，清理必须两批都跑（单跑一批时那张表的"全表 0 行"会显示 FAIL，属预期）

六、清理复核：两批 clean 后 t_user / t_legacy_mapping / 5 张业务表 / t_audit_log 全部 0 行

cmd/ac15fixture/main.go

@@ -38,6 +38,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"strings"
 
 	"yuansiwei/unified-backend/internal/academic/dualrun/fixture"
 )
@@ -343,8 +344,13 @@ func runAlign(args []string) int {
 	fs := flag.NewFlagSet("align", flag.ContinueOnError)
 	fs.SetOutput(os.Stderr)
 	apply := fs.Bool("apply", false, "执行清单里的 ALTER（默认只打印 + dry-run）")
-	verifyOnly := fs.Bool("verify", false, "只复核两列默认值")
+	verifyOnly := fs.Bool("verify", false, "只复核列默认值")
 	modelDir := fs.String("model", "", "DDL 目录（默认 AC15_MODEL_DIR 或内置路径）")
+	columns := fs.String("columns", "created_at,updated_at",
+		"参与对齐的列：created_at,updated_at | create_time | all")
+	algorithm := fs.String("algorithm", "instant",
+		"ALGORITHM 子句：instant|inplace|copy|none（MySQL 8.0.23 不支持 instant，需显式 none）")
+	fixNullable := fs.Bool("fix-nullable", false, "按 DDL 修正 NULL 性（默认只报告不改）")
 	if err := fs.Parse(args); err != nil {
 		return 2
 	}
@@ -354,25 +360,35 @@ func runAlign(args []string) int {
 	}
 	ctx := context.Background()
 
+	cols, err := parseColumns(*columns)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
+		return 2
+	}
+
 	if *verifyOnly {
-		total, missing, err := fixture.VerifyAlign(ctx, cfg)
+		total, missing, err := fixture.VerifyAlign(ctx, cfg, cols)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
 			return 3
 		}
-		fmt.Printf("== 复核：%d 个 created_at/updated_at 列 ==\n", total)
+		fmt.Printf("== 复核 %v：共 %d 列 ==\n", cols, total)
 		if len(missing) == 0 {
 			fmt.Printf("COLUMN_DEFAULT 全部非 NULL：PASS\n")
 			return 0
 		}
-		fmt.Printf("仍有 %d 个列 DEFAULT 为 NULL：\n", len(missing))
+		fmt.Printf("仍有 %d 列为 NULL：\n", len(missing))
 		for _, m := range missing {
 			fmt.Printf("  - %s\n", m)
 		}
 		return 1
 	}
 
-	rep, err := fixture.BuildAlignReport(ctx, cfg, *modelDir)
+	opts := fixture.AlignOptions{
+		ModelDir: *modelDir, Columns: cols,
+		Algorithm: *algorithm, FixNullable: *fixNullable,
+	}
+	rep, err := fixture.BuildAlignReport(ctx, cfg, opts)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
 		return 3
@@ -390,16 +406,16 @@ func runAlign(args []string) int {
 	fixture.PrintAlignReport(os.Stdout, rep)
 
 	if *apply {
-		total, missing, err := fixture.VerifyAlign(ctx, cfg)
+		total, missing, err := fixture.VerifyAlign(ctx, cfg, cols)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "FATAL: %v\n", err)
 			return 3
 		}
-		fmt.Printf("\n== 执行后复核：%d 列 ==\n", total)
+		fmt.Printf("\n== 执行后复核 %v：共 %d 列 ==\n", cols, total)
 		if len(missing) == 0 {
 			fmt.Printf("COLUMN_DEFAULT 全部非 NULL：PASS\n")
 		} else {
-			fmt.Printf("仍有 %d 列为 NULL（含 drift-nullable 等未处理项）：\n", len(missing))
+			fmt.Printf("仍有 %d 列为 NULL：\n", len(missing))
 			for _, m := range missing {
 				fmt.Printf("  - %s\n", m)
 			}
@@ -411,6 +427,31 @@ func runAlign(args []string) int {
 	return 0
 }
 
+// parseColumns 解析 --columns：all = 审计列 + 业务列。
+func parseColumns(s string) ([]string, error) {
+	s = strings.TrimSpace(s)
+	switch s {
+	case "", "audit":
+		return fixture.ColumnsAudit, nil
+	case "business":
+		return fixture.ColumnsBusiness, nil
+	case "all":
+		return append(append([]string{}, fixture.ColumnsAudit...), fixture.ColumnsBusiness...), nil
+	}
+	var out []string
+	for _, p := range strings.Split(s, ",") {
+		p = strings.TrimSpace(p)
+		if p == "" {
+			continue
+		}
+		out = append(out, p)
+	}
+	if len(out) == 0 {
+		return nil, fmt.Errorf("--columns 为空")
+	}
+	return out, nil
+}
+
 func runCount2(args []string) int {
 	cfg, _, code := bootstrap2(args)
 	if code != 0 {

internal/academic/dualrun/fixture/README.md

@@ -208,6 +208,15 @@ go run ./cmd/ac15fixture align --apply    # 清单人工确认后执行
 go run ./cmd/ac15fixture align --verify   # 只复核两列 COLUMN_DEFAULT 是否全部非 NULL
 ```
 
+> **已执行（2026-11-10，dev 库）**：`created_at`/`updated_at` 16 条（含 `t_campus` 两列按 DDL 改 NOT NULL）、
+> `create_time` 13 条，共 **29 条 ALTER、0 失败**；复核 33 + 13 列 `COLUMN_DEFAULT` 全非 NULL。
+> 全库 40 表，除 4 张种子表（`t_role` 7 / `t_permission` 6 / `t_role_permission` 10 / `casbin_rule` 14 = 37 行）外全 0 行。
+>
+> ⚠ **`ALGORITHM` 的版本限制（实测）**：MySQL **8.0.23** 上 `MODIFY COLUMN` 改默认值
+> **不支持 `ALGORITHM=INSTANT`，也不支持 `INPLACE`**（`Error 1845 ... Try ALGORITHM=COPY`）；
+> 该能力要 8.0.29+ 才有。因此本机执行时用 `--algorithm copy` **显式声明"我知道要重建"**
+> （0 行表重建零成本，已复核 `TABLE_ROWS` 未变）。升级到 8.0.29+ 后应改回 `--algorithm instant`。
+
 设计上刻意的四条约束：
 
 1. **只 `ALTER TABLE ... MODIFY`，绝不 DROP 重建** —— 重建会把 GORM 建的索引/字符集一起改掉，产生新漂移；
@@ -217,7 +226,35 @@ go run ./cmd/ac15fixture align --verify   # 只复核两列 COLUMN_DEFAULT 是
 4. 每条 ALTER 先在**会话临时表**上 dry-run（`CREATE TEMPORARY TABLE ... LIKE` + 同一句 MODIFY），
    语法或属性不合法的语句在执行前就暴露。
 
-## 12. 已知事实（不是缺陷，是数据现状）
+## 12. `create_time` 的静默错：为什么必须配断言
+
+补了 `DEFAULT CURRENT_TIMESTAMP(3)` 之后，`create_time` 的风险等级**反而变高**：
+
+| 列 | 语义 | 漏传时的后果（补默认前 → 补默认后）|
+|---|---|---|
+| `created_at` | 审计：这行何时入库 | 1364 报错 → 取当前时间（**合理**）|
+| `create_time` | 业务：这笔钱何时发生 | 1364 报错 → 取插入时刻（**静默写错**，前端要展示/排序/按区间查）|
+
+所以映射侧根本不靠默认值：`create_time` 一律由 legacy 显式赋值；同时 verify 侧加了一条硬断言
+（`run_marker.go`）：
+
+```
+迁移执行时刻（t_audit_log 里 action='ac15_fixture_run' 的标记）
+   ↓
+count(该表 create_time >= 迁移执行时刻) == 0
+```
+
+**为什么必须落库取时间戳**：map 与 verify 是两个独立进程，时间戳无法经内存传递；
+而 legacy 的历史业务时间必然早于本次迁移，所以「>= 迁移执行时刻」就是「被 DB 默认值填过」的判据。
+变异 M5（映射时故意不传 `create_time`）实测：`create_time >= 该时刻的行数 = 10（期望 0）` → 红。
+
+## 13. 清理要注意的跨批耦合
+
+两批 fixture **共用** `t_user` / `t_legacy_mapping`（batch1 的创建人 20052/45014 也是 batch2 的申请人）。
+因此清理要**两批都跑一遍**（`clean2` → `clean`）：单跑一批时，另一批的映射行还在，
+该表的「全表 0 行」判定会显示 FAIL —— 这是预期的，不是漏删。
+
+## 14. 已知事实（不是缺陷，是数据现状）
 
 - legacy `t_user_expense_log` 里 `create_userid = 0` 或 NULL 的行数 = **0**（90007 行全都有真实创建人）。
   所以「0 → NULL」这条规则**无法用真实数据证明**，只能由 `mapping_test.go` 单测 + 变异实验 M3 证明。

internal/academic/dualrun/fixture/batch2_run.go

@@ -54,6 +54,11 @@ func PrepareBatch2(ctx context.Context, cfg Config, opts Batch2Options) (*Batch2
 
 	rep := &Batch2Report{Options: opts, LoadedCounts: map[string]int{}}
 
+	// 落运行标记（同 batch1：create_time 的静默默认值只能靠断言抓）
+	if _, err := WriteRunMarker(ctx, target, "batch2"); err != nil {
+		return nil, err
+	}
+
 	divides, err := extractDivides(ctx, legacy, opts.DivideRootUserID)
 	if err != nil {
 		return nil, err
@@ -445,6 +450,11 @@ func CleanupBatch2(ctx context.Context, cfg Config, opts Batch2Options) (*Batch2
 		}
 	}
 
+	// 5.5) 运行标记也清掉
+	if _, err := ClearRunMarkers(ctx, target); err != nil {
+		return nil, err
+	}
+
 	// 6) 复核 5 张表**全表**行数
 	for _, t := range []string{"t_expenditure", "t_user_divide_into", "t_user_divide_into_log",
 		"t_user_balance_request", "t_expenditure_log", "t_legacy_mapping", "t_user"} {

internal/academic/dualrun/fixture/batch2_verify.go

@@ -200,6 +200,9 @@ func VerifyBatch2(ctx context.Context, cfg Config, opts Batch2Options) (*Batch2V
 	if err := verifyBizNoFour(ctx, legacy, target, opts, rep); err != nil {
 		return nil, err
 	}
+	// create_time 未落入 DB 默认值（业务时间必须来自 legacy）
+	rep.Assertions = append(rep.Assertions,
+		createTimeAssertions(ctx, target, []string{"t_user_divide_into_log", "t_expenditure_log"})...)
 	return rep, nil
 }
 
@@ -873,17 +876,27 @@ func verifyBizNoFour(ctx context.Context, legacy, target *sql.DB, opts Batch2Opt
 			}
 		}
 
-		// ③ id 与 biz_no 解耦：目标 id 是 1..N 的连续小整数，而 biz_no 是 32 位 hex；
-		//    断言「id != biz_no 的任一数字派生」——这里用「id 必然 <= N」这一硬事实
-		//    （若做过任何 hash/截断反算，id 会是大整数）。
-		maxAllowed := int64(len(targetByBiz)) + 100
+		// ③ id 与 biz_no 解耦：目标 id 是自增发号，而 biz_no 是 32 位 hex。
+		//
+		//    判定用**硬界 + 唯一性 + 镜像实验**，不用「id <= N+常数」这种启发式：
+		//    早期版本拿 N+100 当上界，fixture 反复跑后 AUTO_INCREMENT 涨到 130，
+		//    就被误判成「由 biz_no 反算」而假红（本批实测踩到，已修）。
+		//    · 硬界：32 位 hex 若被当成数字反算，id 必然 >= 2^31（8 位 hex 就 4e9）；
+		//      AUTO_INCREMENT 在 fixture 生命周期内不可能到 2^31。
+		//    · 唯一性：COUNT(DISTINCT id) == COUNT(*)。
+		//    · 机制：mirror 实验（倒序插 biz_no，id 仍按插入顺序 1..N 涨）。
 		derived := false
+		seenIDs := map[int64]bool{}
 		for _, id := range targetByBiz {
-			if id > maxAllowed {
+			if id <= 0 || id >= (1<<31) {
 				derived = true
 			}
+			if seenIDs[id] {
+				derived = true // id 重复也说明不是自增发号
+			}
+			seenIDs[id] = true
 		}
-		v.IDEqualsPrefix = derived // true = 疑似由 biz_no 反算，必须为 false
+		v.IDEqualsPrefix = derived // true = 疑似由 biz_no 反算/非自增，必须为 false
 
 		// ④ 反查可用：用 biz_no JOIN 回 legacy
 		var joined int

internal/academic/dualrun/fixture/run.go

@@ -59,6 +59,12 @@ func PrepareAndLoad(ctx context.Context, cfg Config, legacyUserID int64) (*MapRe
 
 	rep := &MapReport{LegacyUserID: legacyUserID}
 
+	// 落运行标记：create_time 补了 DEFAULT 之后，漏传业务时间会被 DB 静默填成插入时刻，
+	// 只有 verify 侧「create_time >= 迁移执行时刻 的行数 == 0」这条断言能抓住它。
+	if _, err := WriteRunMarker(ctx, target, "batch1"); err != nil {
+		return nil, err
+	}
+
 	expense, err := extractExpense(ctx, legacy, legacyUserID)
 	if err != nil {
 		return nil, err
@@ -280,6 +286,11 @@ func Cleanup(ctx context.Context, cfg Config, legacyUserID int64) (*CleanReport,
 	n, _ = res.RowsAffected()
 	rep.Deleted["t_user"] = n
 
+	// 运行标记也清掉（t_audit_log 应回到 0 行）
+	if _, err := ClearRunMarkers(ctx, target); err != nil {
+		return nil, err
+	}
+
 	// 复核 4 张表的**全表**行数（不按 user 过滤：要证明回到 0 行）
 	for _, t := range []string{"t_user_expense", "t_user_expense_school", "t_user_expense_log"} {
 		var cnt int64

internal/academic/dualrun/fixture/run_marker.go

@@ -0,0 +1,118 @@
+package fixture
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"time"
+)
+
+// 本文件是**迁移运行标记**：把「这次映射是什么时刻跑的」落到目标库里，供另一个进程的
+// verify 精确取用。
+//
+// 为什么必须落库：
+//   create_time 这类**业务时间**列在补了 DEFAULT CURRENT_TIMESTAMP(3) 之后，
+//   如果映射忘了传值，DB 会用「插入时刻」静默填上——不报错、不空、格式正确，
+//   只有「与迁移执行时刻比对」这一条断言能抓住它。
+//   而 map 与 verify 是两个独立进程，时间戳无法通过内存传递，所以写进 t_audit_log。
+//
+// 取值精度：毫秒（DATETIME(3)），与列精度一致，避免"差 1 毫秒"的伪差异。
+
+// RunMarkerAction 是标记行的 action 值（t_audit_log.action）。
+const RunMarkerAction = "ac15_fixture_run"
+
+// WriteRunMarker 写入一条运行标记，返回该标记的时间戳（= 本次迁移的起点）。
+//
+// 用 t_audit_log 而不是新建表：它是 core DDL 里既有的审计表（14 列、0 行、已对齐默认值），
+// 语义上也正是"谁在什么时候做了什么"。清理时按 action 删除即可。
+func WriteRunMarker(ctx context.Context, db *sql.DB, phase string) (time.Time, error) {
+	now := time.Now().Truncate(time.Millisecond)
+	const q = `
+INSERT INTO t_audit_log (actor_user_id, action, entity_type, entity_id, detail, trace_id, created_at)
+VALUES (NULL, ?, 'fixture', ?, ?, 'ac15-fixture', ?)`
+	detail := fmt.Sprintf(`{"phase":%q,"tool":"ac15fixture","note":"migration run marker"}`, phase)
+	if _, err := db.ExecContext(ctx, q, RunMarkerAction, phase, detail,
+		now.Format("2006-01-02 15:04:05.000")); err != nil {
+		return time.Time{}, fmt.Errorf("写运行标记失败：%w", err)
+	}
+	return now, nil
+}
+
+// ReadRunMarker 读最近一条运行标记的时间戳。
+func ReadRunMarker(ctx context.Context, db *sql.DB) (time.Time, bool, error) {
+	var s sql.NullString
+	err := db.QueryRowContext(ctx,
+		`SELECT DATE_FORMAT(created_at,'%Y-%m-%d %H:%i:%s.%f') FROM t_audit_log
+		  WHERE action = ? ORDER BY id DESC LIMIT 1`, RunMarkerAction).Scan(&s)
+	if err == sql.ErrNoRows || (err == nil && !s.Valid) {
+		return time.Time{}, false, nil
+	}
+	if err != nil {
+		return time.Time{}, false, err
+	}
+	t, perr := time.ParseInLocation("2006-01-02 15:04:05.000000", s.String, time.Local)
+	if perr != nil {
+		return time.Time{}, false, fmt.Errorf("解析运行标记时间失败（%q）：%w", s.String, perr)
+	}
+	return t, true, nil
+}
+
+// ClearRunMarkers 删除全部运行标记（清理阶段调用）。
+func ClearRunMarkers(ctx context.Context, db *sql.DB) (int64, error) {
+	res, err := db.ExecContext(ctx, `DELETE FROM t_audit_log WHERE action = ?`, RunMarkerAction)
+	if err != nil {
+		return 0, err
+	}
+	return res.RowsAffected()
+}
+
+// countRowsAfter 统计某表 create_time >= marker 的行数。
+//
+// 这就是「漏传业务时间」的探针：DB 默认值只会给出「插入时刻」，必然 >= marker；
+// 而 legacy 的历史业务时间一定 < marker（本次迁移发生在历史数据之后）。
+func countRowsAfter(ctx context.Context, db *sql.DB, table, column string, marker time.Time) (int, error) {
+	var n int
+	q := fmt.Sprintf(`SELECT COUNT(*) FROM %s WHERE %s >= ?`, table, column)
+	if err := db.QueryRowContext(ctx, q, marker.Format("2006-01-02 15:04:05.000")).Scan(&n); err != nil {
+		return 0, fmt.Errorf("统计 %s.%s 失败：%w", table, column, err)
+	}
+	return n, nil
+}
+
+// createTimeAssertions 对一批「迁移过来的写类表」生成 create_time 断言。
+//
+// tables 里的列名统一叫 create_time（本项目的业务时间列名）；
+// marker 取不到时不判 pass（视为失败并明确说明），避免"没标记就当通过"。
+func createTimeAssertions(ctx context.Context, db *sql.DB, tables []string) []Assertion {
+	marker, ok, err := ReadRunMarker(ctx, db)
+	if err != nil {
+		return []Assertion{{
+			Name: "create_time 未落入 DB 默认值（业务时间必须来自 legacy）",
+			Pass: false, Detail: fmt.Sprintf("读运行标记失败：%v", err),
+		}}
+	}
+	if !ok {
+		return []Assertion{{
+			Name:   "create_time 未落入 DB 默认值（业务时间必须来自 legacy）",
+			Pass:   false,
+			Detail: "未找到运行标记（t_audit_log.action='ac15_fixture_run'）—— 无法判定，按失败处理（先跑 map 再跑 verify）",
+		}}
+	}
+	out := make([]Assertion, 0, len(tables))
+	for _, t := range tables {
+		n, err := countRowsAfter(ctx, db, t, "create_time", marker)
+		name := t + ".create_time 未落入 DB 默认值（>= 迁移执行时刻的行数 = 0）"
+		if err != nil {
+			out = append(out, Assertion{Name: name, Pass: false, Detail: err.Error()})
+			continue
+		}
+		out = append(out, Assertion{
+			Name: name,
+			Pass: n == 0,
+			Detail: fmt.Sprintf("迁移执行时刻=%s；create_time >= 该时刻的行数 = %d（期望 0）"+
+				"—— 补了 DEFAULT CURRENT_TIMESTAMP(3) 之后，漏传业务时间会静默落成插入时刻，只有这条能抓",
+				marker.Format("2006-01-02 15:04:05"), n),
+		})
+	}
+	return out
+}

internal/academic/dualrun/fixture/schema_align.go

@@ -12,33 +12,54 @@ import (
 	"strings"
 )
 
-// 本文件是**结构对齐工具**：把 dev 库 `yuansiwei_2026` 的 created_at / updated_at
-// 对齐到 DDL 文件的口径（补 DEFAULT CURRENT_TIMESTAMP(3)，updated_at 另有 ON UPDATE）。
+// 本文件是**结构对齐工具**：把 dev 库 `yuansiwei_2026` 的时间列默认值对齐到 DDL 文件的口径。
 //
 // 为什么需要它：目标库是"部分表由 DDL 建、部分表由 GORM AutoMigrate 建"的混合体。
-// GORM 不处理 `DEFAULT CURRENT_TIMESTAMP(3)`，所以它建的表这两列 **NOT NULL 且无默认值**，
+// GORM 不处理 `DEFAULT CURRENT_TIMESTAMP(3)`，所以它建的表这些列 **NOT NULL 且无默认值**，
 // 裸 SQL 插入会在随机某几张表上直接 `Error 1364`。
 //
+// ⚠ 三类时间列的语义**完全不同**，不能一视同仁：
+//   - `created_at`：**审计**——这行什么时候入库的。取当前时间合理。
+//   - `create_time`：**业务**——这笔钱/这节课什么时候发生的。取当前时间 = **静默写错**。
+//   - `update_time`：**业务**——最后修改时间（仅 `t_courses`，DDL 本身就是 DEFAULT NULL）。
+//     补默认只解决"插入漏传则报错"，**不解决"取到当前时间就错了"**，所以必须配 create_time 断言
+//     （见 run_marker.go 的 countRowsAfter）。
+//
 // 纪律（不可妥协）：
 //  1. **只 ALTER ... MODIFY**，绝不 DROP 重建（重建会把 GORM 建的索引/字符集一起改掉，产生新漂移）；
 //  2. MODIFY 子句**用实库现有属性**（type / NULL 性 / 注释），只把 DEFAULT 与 ON UPDATE 换成 DDL 口径
-//     —— 这样除了这两个属性，其它一个字节都不动；
-//  3. NULL 性、类型、注释等**其它**差异只报告不执行，等人工定夺；
-//  4. 每条 ALTER 先在会话临时表上 dry-run（`CREATE TEMPORARY TABLE ... LIKE` + 同一条 MODIFY），
-//     语法或属性不合法的语句在真正执行前就会暴露。
+//     —— `FixNullable` 打开时才额外按 DDL 改 NULL 性；
+//  3. NULL 性、类型、注释等差异**默认只报告不执行**，必须显式开 `FixNullable` 才动；
+//  4. 每条 ALTER 先在会话临时表上 dry-run（`CREATE TEMPORARY TABLE ... LIKE` + 同一条 MODIFY）。
+//
+// 关于 ALGORITHM：默认在语句末尾带 `, ALGORITHM=INSTANT`（显式声明"只改元数据"）。
+// 但 **MySQL 8.0.23 不支持对 "MODIFY 改默认值" 用 INSTANT/INPLACE**（实测 Error 1845），
+// 要 8.0.29+ 才支持；在旧版本上必须显式降级为 `Algorithm=none` 才能执行（见报告）。
 
-// DDL 目录（与 core DDL 同一目录）。
+// defaultModelDir 是 DDL 目录。
 const defaultModelDir = `C:\Users\Run\WorkBuddy\项目合并\unified\02-data-model`
 
-// 本工具只处理这两列。
-var alignedColumns = []string{"created_at", "updated_at"}
+// ColumnsAudit 是审计类时间列（取默认值合理）。
+var ColumnsAudit = []string{"created_at", "updated_at"}
+
+// ColumnsBusiness 是业务类时间列（取默认值 = 静默写错，必须配断言）。
+var ColumnsBusiness = []string{"create_time"}
+
+// AlignOptions 是对齐参数。
+type AlignOptions struct {
+	ModelDir    string   // DDL 目录
+	Columns     []string // 参与对齐的列名（默认 created_at/updated_at）
+	Algorithm   string   // instant / inplace / copy / none
+	FixNullable bool     // 是否按 DDL 修正 NULL 性（默认只报告）
+	OnlyTables  []string // 只处理这些表（空=全部）
+}
 
 // ddlColumn 是从 DDL 文件里解析出来的一列。
 type ddlColumn struct {
 	Table      string
 	Column     string
-	RawDef     string // DDL 里的原始定义（不含列名、不含逗号）
-	Type       string // 如 DATETIME(3)
+	RawDef     string
+	Type       string
 	NotNull    bool
 	HasDefault bool
 	OnUpdate   bool
@@ -49,7 +70,7 @@ type ddlColumn struct {
 type liveColumn struct {
 	Table      string
 	Column     string
-	ColumnType string // 如 datetime(3)
+	ColumnType string
 	Nullable   bool
 	Default    sql.NullString
 	Extra      string
@@ -60,28 +81,29 @@ type liveColumn struct {
 type AlignAction struct {
 	Table      string
 	Column     string
-	Kind       string // ok / alignable / drift-nullable / drift-type / ddl-missing
+	Kind       string // ok / alignable / alignable-nullable / drift-nullable / ddl-no-default / ddl-missing
 	LiveDesc   string
 	DDLDefault string
-	SQL        string // Kind=alignable 时的 ALTER 语句
+	SQL        string
 	Note       string
 }
 
 // AlignReport 是整轮比对结果。
 type AlignReport struct {
 	ModelDir string
+	Options  AlignOptions
 	Actions  []AlignAction
-	DryRunOK map[int]bool // Actions 下标 → dry-run 是否通过
+	DryRunOK map[int]bool
 	Applied  int
 	Failed   int
 	Errors   []string
 }
 
-// Alignable 返回需要补 DEFAULT 的条目下标。
+// Alignable 返回需要执行 ALTER 的条目下标。
 func (r *AlignReport) Alignable() []int {
 	var out []int
 	for i, a := range r.Actions {
-		if a.Kind == "alignable" {
+		if a.Kind == "alignable" || a.Kind == "alignable-nullable" {
 			out = append(out, i)
 		}
 	}
@@ -107,8 +129,8 @@ var (
 	firstToken      = regexp.MustCompile(`^[A-Za-z]+(?:\([0-9,]+\))?`)
 )
 
-// ParseDDLFile 解析一个 DDL 文件，返回其中 created_at / updated_at 的列定义。
-func ParseDDLFile(path string) ([]ddlColumn, error) {
+// ParseDDLFile 解析一个 DDL 文件，返回指定列的列定义。
+func ParseDDLFile(path string, want []string) ([]ddlColumn, error) {
 	raw, err := os.ReadFile(path)
 	if err != nil {
 		return nil, fmt.Errorf("读 DDL 失败（%s）：%w", path, err)
@@ -136,7 +158,7 @@ func ParseDDLFile(path string) ([]ddlColumn, error) {
 			continue
 		}
 		col := m[1]
-		if len(alignedColumns) > 0 && !contains(alignedColumns, col) {
+		if !containsString(want, col) {
 			continue
 		}
 		def := strings.TrimSuffix(strings.TrimSpace(m[2]), ",")
@@ -157,7 +179,7 @@ func ParseDDLFile(path string) ([]ddlColumn, error) {
 	return out, nil
 }
 
-func contains(ss []string, s string) bool {
+func containsString(ss []string, s string) bool {
 	for _, v := range ss {
 		if v == s {
 			return true
@@ -166,17 +188,23 @@ func contains(ss []string, s string) bool {
 	return false
 }
 
-// loadLiveColumns 从 information_schema 读实库这两列的属性。
-func loadLiveColumns(ctx context.Context, db *sql.DB, schema string) (map[string]liveColumn, error) {
+// loadLiveColumns 从 information_schema 读实库指定列的属性。
+func loadLiveColumns(ctx context.Context, db *sql.DB, schema string, columns []string) (map[string]liveColumn, error) {
 	// 关掉 information_schema 统计缓存，拿即时值（同 batch2 的 AUTO_INCREMENT 坑）。
 	if _, err := db.ExecContext(ctx, "SET SESSION information_schema_stats_expiry = 0"); err != nil {
 		return nil, err
 	}
-	rows, err := db.QueryContext(ctx, `
+	ph := make([]string, len(columns))
+	args := []any{schema}
+	for i, c := range columns {
+		ph[i] = "?"
+		args = append(args, c)
+	}
+	q := fmt.Sprintf(`
 SELECT table_name, column_name, column_type, is_nullable, column_default, extra, column_comment
   FROM information_schema.columns
- WHERE table_schema = ?
-   AND column_name IN ('created_at','updated_at')`, schema)
+ WHERE table_schema = ? AND column_name IN (%s)`, strings.Join(ph, ","))
+	rows, err := db.QueryContext(ctx, q, args...)
 	if err != nil {
 		return nil, err
 	}
@@ -194,14 +222,29 @@ SELECT table_name, column_name, column_type, is_nullable, column_default, extra,
 	return out, rows.Err()
 }
 
-// buildModifySQL 用**实库现有属性** + DDL 的 DEFAULT 拼出 MODIFY 子句。
-//
-// 只动两处：DEFAULT CURRENT_TIMESTAMP(3)（两列都有）、ON UPDATE CURRENT_TIMESTAMP(3)（仅 updated_at）。
-// type / NOT NULL / COMMENT 全部照抄实库，避免顺手改掉别的属性。
-func buildModifySQL(live liveColumn, ddl ddlColumn) string {
+// algorithmSuffix 生成 ALGORITHM 子句（none 时返回空串）。
+func algorithmSuffix(algorithm string) string {
+	switch strings.ToLower(strings.TrimSpace(algorithm)) {
+	case "instant":
+		return ", ALGORITHM=INSTANT"
+	case "inplace":
+		return ", ALGORITHM=INPLACE, LOCK=NONE"
+	case "copy":
+		return ", ALGORITHM=COPY"
+	default: // "", "none"
+		return ""
+	}
+}
+
+// buildModifySQL 用**实库现有属性** + DDL 的 DEFAULT 拼出 MODIFY 语句。
+func buildModifySQL(live liveColumn, ddl ddlColumn, fixNullable bool, algorithm string) string {
 	var b strings.Builder
 	b.WriteString(fmt.Sprintf("ALTER TABLE `%s` MODIFY COLUMN `%s` %s", live.Table, live.Column, live.ColumnType))
-	if !live.Nullable {
+	notNull := !live.Nullable
+	if fixNullable && ddl.NotNull {
+		notNull = true
+	}
+	if notNull {
 		b.WriteString(" NOT NULL")
 	}
 	b.WriteString(" DEFAULT CURRENT_TIMESTAMP(3)")
@@ -211,44 +254,51 @@ func buildModifySQL(live liveColumn, ddl ddlColumn) string {
 	if live.Comment != "" {
 		b.WriteString(" COMMENT '" + strings.ReplaceAll(live.Comment, "'", "''") + "'")
 	}
+	b.WriteString(algorithmSuffix(algorithm))
 	return b.String()
 }
 
-// describeLive 生成实库列的可读描述（报告用）。
 func describeLive(l liveColumn) string {
 	def := "NULL(无默认值)"
 	if l.Default.Valid {
 		def = l.Default.String
 	}
-	null := "NULL"
-	if !l.Nullable {
-		null = "NOT NULL"
+	null := "NOT NULL"
+	if l.Nullable {
+		null = "NULL"
+	}
+	extra := ""
+	if l.Extra != "" {
+		extra = fmt.Sprintf(" extra=%q", l.Extra)
 	}
-	return fmt.Sprintf("%s %s DEFAULT=%s extra=%q", l.ColumnType, null, def, l.Extra)
+	return fmt.Sprintf("%s %s DEFAULT=%s%s", l.ColumnType, null, def, extra)
 }
 
 // BuildAlignReport 逐表比对 DDL 与实库，产出对齐清单（不执行）。
-func BuildAlignReport(ctx context.Context, cfg Config, modelDir string) (*AlignReport, error) {
-	if modelDir == "" {
-		modelDir = os.Getenv("AC15_MODEL_DIR")
+func BuildAlignReport(ctx context.Context, cfg Config, opts AlignOptions) (*AlignReport, error) {
+	if opts.ModelDir == "" {
+		opts.ModelDir = os.Getenv("AC15_MODEL_DIR")
+	}
+	if opts.ModelDir == "" {
+		opts.ModelDir = defaultModelDir
 	}
-	if modelDir == "" {
-		modelDir = defaultModelDir
+	if len(opts.Columns) == 0 {
+		opts.Columns = ColumnsAudit
 	}
 	files := []string{
-		filepath.Join(modelDir, "core-model-ddl.sql"),
-		filepath.Join(modelDir, "academic-model-ddl.sql"),
+		filepath.Join(opts.ModelDir, "core-model-ddl.sql"),
+		filepath.Join(opts.ModelDir, "academic-model-ddl.sql"),
 	}
 	var specs []ddlColumn
 	for _, f := range files {
-		cols, err := ParseDDLFile(f)
+		cols, err := ParseDDLFile(f, opts.Columns)
 		if err != nil {
 			return nil, err
 		}
 		specs = append(specs, cols...)
 	}
 	if len(specs) == 0 {
-		return nil, fmt.Errorf("DDL 里没有解析到 created_at/updated_at 定义（modelDir=%s）", modelDir)
+		return nil, fmt.Errorf("DDL 里没有解析到 %v 的定义（modelDir=%s）", opts.Columns, opts.ModelDir)
 	}
 
 	db, err := cfg.OpenTarget()
@@ -257,47 +307,51 @@ func BuildAlignReport(ctx context.Context, cfg Config, modelDir string) (*AlignR
 	}
 	defer db.Close()
 
-	live, err := loadLiveColumns(ctx, db, cfg.TargetDB)
+	live, err := loadLiveColumns(ctx, db, cfg.TargetDB, opts.Columns)
 	if err != nil {
 		return nil, err
 	}
 
-	rep := &AlignReport{ModelDir: modelDir, DryRunOK: map[int]bool{}}
+	rep := &AlignReport{ModelDir: opts.ModelDir, Options: opts, DryRunOK: map[int]bool{}}
 	for _, sp := range specs {
+		if len(opts.OnlyTables) > 0 && !containsString(opts.OnlyTables, sp.Table) {
+			continue
+		}
 		key := sp.Table + "." + sp.Column
 		lv, ok := live[key]
 		if !ok {
 			rep.Actions = append(rep.Actions, AlignAction{
 				Table: sp.Table, Column: sp.Column, Kind: "ddl-missing",
-				LiveDesc: "<实库无此列>", Note: "DDL 有、实库没有 —— 属缺列，本工具不处理（需要 ALTER ADD COLUMN，另有风险）",
+				LiveDesc: "<实库无此列>", Note: "DDL 有、实库没有 —— 属缺列，本工具不处理",
 			})
 			continue
 		}
 		a := AlignAction{Table: sp.Table, Column: sp.Column, LiveDesc: describeLive(lv),
 			DDLDefault: "CURRENT_TIMESTAMP(3)"}
+		ddlNull := "NOT NULL"
+		if !sp.NotNull {
+			ddlNull = "NULL"
+		}
 		switch {
 		case lv.Nullable != !sp.NotNull:
-			// NULL 性不一致 —— 只报告，等人工定夺（团队要求不顺手改）
-			a.Kind = "drift-nullable"
-			liveNull := "NOT NULL"
-			if lv.Nullable {
-				liveNull = "NULL"
-			}
-			ddlNull := "NOT NULL"
-			if !sp.NotNull {
-				ddlNull = "NULL"
+			if opts.FixNullable && sp.NotNull && sp.HasDefault {
+				a.Kind = "alignable-nullable"
+				a.SQL = buildModifySQL(lv, sp, true, opts.Algorithm)
+				a.Note = fmt.Sprintf("NULL 性不一致（实库可空、DDL %s）→ 按 DDL 一并修正 + 补 DEFAULT", ddlNull)
+			} else {
+				a.Kind = "drift-nullable"
+				a.Note = fmt.Sprintf("NULL 性不一致：实库可空，DDL %s —— 未开 FixNullable，只报告不改", ddlNull)
 			}
-			a.Note = fmt.Sprintf("NULL 性不一致：实库 %s，DDL %s —— 只报告不改", liveNull, ddlNull)
 		case !sp.HasDefault:
 			a.Kind = "ddl-no-default"
-			a.Note = "DDL 该列本身就没有 DEFAULT CURRENT_TIMESTAMP(3)，无需对齐"
+			a.Note = "DDL 该列本身没有 DEFAULT CURRENT_TIMESTAMP(3)，无需对齐"
 		case lv.Default.Valid:
 			a.Kind = "ok"
-			a.Note = "已有默认值（或已对齐）"
+			a.Note = "已有默认值（已对齐）"
 		default:
 			a.Kind = "alignable"
-			a.SQL = buildModifySQL(lv, sp)
-			a.Note = "实库 DEFAULT 为 NULL，补 DEFAULT CURRENT_TIMESTAMP(3)" +
+			a.SQL = buildModifySQL(lv, sp, false, opts.Algorithm)
+			a.Note = "实库 DEFAULT 为 NULL → 补 DEFAULT CURRENT_TIMESTAMP(3)" +
 				map[bool]string{true: "，并补 ON UPDATE CURRENT_TIMESTAMP(3)", false: ""}[sp.OnUpdate]
 		}
 		rep.Actions = append(rep.Actions, a)
@@ -312,8 +366,6 @@ func BuildAlignReport(ctx context.Context, cfg Config, modelDir string) (*AlignR
 }
 
 // DryRunAlign 在**会话临时表**上验证每条 ALTER 是否合法且达到预期效果。
-//
-// 不动真实表：`CREATE TEMPORARY TABLE ... LIKE 真表` → 跑同一条 MODIFY → 复查默认值。
 func DryRunAlign(ctx context.Context, cfg Config, rep *AlignReport) error {
 	db, err := cfg.OpenTarget()
 	if err != nil {
@@ -328,7 +380,7 @@ func DryRunAlign(ctx context.Context, cfg Config, rep *AlignReport) error {
 
 	for i := range rep.Actions {
 		a := &rep.Actions[i]
-		if a.Kind != "alignable" {
+		if a.Kind != "alignable" && a.Kind != "alignable-nullable" {
 			continue
 		}
 		tmp := "tmp_align_" + a.Table
@@ -336,7 +388,6 @@ func DryRunAlign(ctx context.Context, cfg Config, rep *AlignReport) error {
 			fmt.Sprintf("CREATE TEMPORARY TABLE IF NOT EXISTS `%s` LIKE `%s`", tmp, a.Table)); err != nil {
 			return fmt.Errorf("dry-run 建临时表失败（%s）：%w", a.Table, err)
 		}
-		// 把临时表名替换进 ALTER 语句（其余部分逐字符相同）
 		stmt := strings.Replace(a.SQL, "ALTER TABLE `"+a.Table+"`", "ALTER TABLE `"+tmp+"`", 1)
 		if _, err := conn.ExecContext(ctx, stmt); err != nil {
 			rep.Errors = append(rep.Errors, fmt.Sprintf("dry-run 失败 %s.%s：%v", a.Table, a.Column, err))
@@ -348,7 +399,6 @@ func DryRunAlign(ctx context.Context, cfg Config, rep *AlignReport) error {
 		if err := conn.QueryRowContext(ctx, `
 SELECT column_default, extra FROM information_schema.columns
  WHERE table_schema = DATABASE() AND table_name = ? AND column_name = ?`, tmp, a.Column).Scan(&def, &extra); err != nil {
-			// 临时表不出现在 information_schema 的情况（部分版本），退化为 SHOW COLUMNS
 			rows, err2 := conn.QueryContext(ctx, fmt.Sprintf("SHOW COLUMNS FROM `%s` LIKE '%s'", tmp, a.Column))
 			if err2 != nil {
 				rep.Errors = append(rep.Errors, fmt.Sprintf("dry-run 复查失败 %s.%s：%v", a.Table, a.Column, err2))
@@ -377,7 +427,8 @@ SELECT column_default, extra FROM information_schema.columns
 // ApplyAlign 执行对齐（只在 dry-run 全部通过时执行）。
 func ApplyAlign(ctx context.Context, cfg Config, rep *AlignReport) error {
 	for i := range rep.Actions {
-		if rep.Actions[i].Kind == "alignable" && !rep.DryRunOK[i] {
+		k := rep.Actions[i].Kind
+		if (k == "alignable" || k == "alignable-nullable") && !rep.DryRunOK[i] {
 			return fmt.Errorf("存在未通过 dry-run 的语句，拒绝执行（%s.%s）",
 				rep.Actions[i].Table, rep.Actions[i].Column)
 		}
@@ -389,7 +440,7 @@ func ApplyAlign(ctx context.Context, cfg Config, rep *AlignReport) error {
 	defer db.Close()
 	for i := range rep.Actions {
 		a := &rep.Actions[i]
-		if a.Kind != "alignable" {
+		if a.Kind != "alignable" && a.Kind != "alignable-nullable" {
 			continue
 		}
 		if _, err := db.ExecContext(ctx, a.SQL); err != nil {
@@ -402,14 +453,17 @@ func ApplyAlign(ctx context.Context, cfg Config, rep *AlignReport) error {
 	return nil
 }
 
-// VerifyAlign 复核：这两列的 COLUMN_DEFAULT 必须全部非 NULL（drift-nullable 的除外）。
-func VerifyAlign(ctx context.Context, cfg Config) (total int, missing []string, err error) {
+// VerifyAlign 复核：指定列的 COLUMN_DEFAULT 必须全部非 NULL。
+func VerifyAlign(ctx context.Context, cfg Config, columns []string) (total int, missing []string, err error) {
+	if len(columns) == 0 {
+		columns = ColumnsAudit
+	}
 	db, err := cfg.OpenTarget()
 	if err != nil {
 		return 0, nil, err
 	}
 	defer db.Close()
-	live, err := loadLiveColumns(ctx, db, cfg.TargetDB)
+	live, err := loadLiveColumns(ctx, db, cfg.TargetDB, columns)
 	if err != nil {
 		return 0, nil, err
 	}
@@ -426,10 +480,15 @@ func VerifyAlign(ctx context.Context, cfg Config) (total int, missing []string,
 // PrintAlignReport 打印对齐清单。
 func PrintAlignReport(w io.Writer, r *AlignReport) {
 	fmt.Fprintf(w, "== 结构对齐清单（基准：DDL 文件）==\n")
-	fmt.Fprintf(w, "DDL 目录：%s\n\n", r.ModelDir)
+	fmt.Fprintf(w, "DDL 目录：%s\n", r.ModelDir)
+	algo := r.Options.Algorithm
+	if algo == "" {
+		algo = "none"
+	}
+	fmt.Fprintf(w, "列范围：%v   算法子句：%s   FixNullable=%v\n\n", r.Options.Columns, algo, r.Options.FixNullable)
 
 	alignable := r.Alignable()
-	fmt.Fprintf(w, "【待补 DEFAULT】%d 条（只 MODIFY 这两列，其余属性照抄实库）\n", len(alignable))
+	fmt.Fprintf(w, "【待执行】%d 条\n", len(alignable))
 	for _, i := range alignable {
 		a := r.Actions[i]
 		dry := "未验证"
@@ -440,7 +499,7 @@ func PrintAlignReport(w io.Writer, r *AlignReport) {
 				dry = "dry-run ✘"
 			}
 		}
-		fmt.Fprintf(w, "  %-30s 实库：%s\n     %s  [%s]\n", a.Table+"."+a.Column, a.LiveDesc, a.SQL, dry)
+		fmt.Fprintf(w, "  %-32s [%s] %s\n     %s\n", a.Table+"."+a.Column, a.Kind, dry, a.SQL)
 	}
 
 	okCount := 0
@@ -452,14 +511,23 @@ func PrintAlignReport(w io.Writer, r *AlignReport) {
 	fmt.Fprintf(w, "\n【已对齐】%d 条\n", okCount)
 	for _, a := range r.Actions {
 		if a.Kind == "ok" {
-			fmt.Fprintf(w, "  %-30s %s\n", a.Table+"."+a.Column, a.LiveDesc)
+			fmt.Fprintf(w, "  %-32s %s\n", a.Table+"."+a.Column, a.LiveDesc)
 		}
 	}
 
 	probs := r.Problems()
 	fmt.Fprintf(w, "\n【只报告、不执行】%d 条\n", len(probs))
 	for _, a := range probs {
-		fmt.Fprintf(w, "  %-30s [%s] %s\n     实库：%s\n", a.Table+"."+a.Column, a.Kind, a.Note, a.LiveDesc)
+		fmt.Fprintf(w, "  %-32s [%s] %s\n     实库：%s\n", a.Table+"."+a.Column, a.Kind, a.Note, a.LiveDesc)
+	}
+	skipped := 0
+	for _, a := range r.Actions {
+		if a.Kind == "ddl-no-default" {
+			skipped++
+		}
+	}
+	if skipped > 0 {
+		fmt.Fprintf(w, "\n【DDL 无默认值、无需动】%d 条\n", skipped)
 	}
 
 	if len(r.Errors) > 0 {

internal/academic/dualrun/fixture/schema_align_test.go

@@ -0,0 +1,125 @@
+package fixture
+
+import (
+	"database/sql"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// 本文件单测结构对齐工具的**纯函数部分**：ALGORITHM 子句、DDL 解析、MODIFY 语句构造。
+// 真库比对与执行由 cmd/ac15fixture align 完成（需连库，不进单测）。
+
+func TestAlgorithmSuffix(t *testing.T) {
+	cases := map[string]string{
+		"instant": ", ALGORITHM=INSTANT",
+		"INSTANT": ", ALGORITHM=INSTANT",
+		"inplace": ", ALGORITHM=INPLACE, LOCK=NONE",
+		"copy":    ", ALGORITHM=COPY",
+		"none":    "",
+		"":        "",
+		"garbage": "",
+	}
+	for in, want := range cases {
+		if got := algorithmSuffix(in); got != want {
+			t.Errorf("algorithmSuffix(%q) = %q, want %q", in, got, want)
+		}
+	}
+}
+
+// TestBuildModifySQLKeepsOtherAttributes 是本工具最关键的约束：
+// MODIFY 只许改 DEFAULT（与 ON UPDATE），type / NULL 性 / 注释必须**照抄实库**。
+func TestBuildModifySQLKeepsOtherAttributes(t *testing.T) {
+	live := liveColumn{
+		Table: "t_x", Column: "created_at", ColumnType: "datetime(3)",
+		Nullable: true, Comment: "创建时间",
+	}
+	ddl := ddlColumn{Table: "t_x", Column: "created_at", NotNull: true, HasDefault: true}
+
+	// 默认（不修 NULL 性）：NULL 性必须保持实库的"可空"
+	got := buildModifySQL(live, ddl, false, "none")
+	want := "ALTER TABLE `t_x` MODIFY COLUMN `created_at` datetime(3) DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间'"
+	if got != want {
+		t.Errorf("不修 NULL 性时语句不符：\n got=%q\nwant=%q", got, want)
+	}
+
+	// 修 NULL 性：只多一个 NOT NULL，其它不变
+	got = buildModifySQL(live, ddl, true, "none")
+	want = "ALTER TABLE `t_x` MODIFY COLUMN `created_at` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间'"
+	if got != want {
+		t.Errorf("修 NULL 性时语句不符：\n got=%q\nwant=%q", got, want)
+	}
+
+	// updated_at 必须带 ON UPDATE，且位于 DEFAULT 之后、COMMENT 之前
+	live2 := liveColumn{Table: "t_x", Column: "updated_at", ColumnType: "datetime(3)"}
+	ddl2 := ddlColumn{Table: "t_x", Column: "updated_at", NotNull: true, HasDefault: true, OnUpdate: true}
+	got = buildModifySQL(live2, ddl2, false, "copy")
+	want = "ALTER TABLE `t_x` MODIFY COLUMN `updated_at` datetime(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3), ALGORITHM=COPY"
+	if got != want {
+		t.Errorf("updated_at 语句不符：\n got=%q\nwant=%q", got, want)
+	}
+}
+
+// TestBuildModifySQLEscapesComment 证明注释里的单引号会被转义（不破坏语句）。
+func TestBuildModifySQLEscapesComment(t *testing.T) {
+	live := liveColumn{Table: "t", Column: "created_at", ColumnType: "datetime(3)", Comment: "it's fine"}
+	ddl := ddlColumn{Table: "t", Column: "created_at", NotNull: true, HasDefault: true}
+	got := buildModifySQL(live, ddl, false, "none")
+	if !strings.Contains(got, "COMMENT 'it''s fine'") {
+		t.Errorf("注释未正确转义：%q", got)
+	}
+}
+
+// TestParseDDLFile 用一个小样本验证解析口径（NOT NULL / DEFAULT / ON UPDATE / COMMENT）。
+func TestParseDDLFile(t *testing.T) {
+	src := `
+-- 注释行应被忽略
+SET NAMES utf8mb4;
+CREATE TABLE ` + "`t_demo`" + ` (
+  ` + "`id`" + ` BIGINT NOT NULL AUTO_INCREMENT,
+  ` + "`created_at`" + ` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '创建时间',
+  ` + "`updated_at`" + ` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3) COMMENT '更新时间',
+  ` + "`update_time`" + ` DATETIME(3) DEFAULT NULL COMMENT '最后修改时间',
+  PRIMARY KEY (` + "`id`" + `)
+) ENGINE=InnoDB;
+`
+	dir := t.TempDir()
+	p := filepath.Join(dir, "demo.sql")
+	if err := os.WriteFile(p, []byte(src), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	cols, err := ParseDDLFile(p, []string{"created_at", "updated_at", "update_time"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(cols) != 3 {
+		t.Fatalf("应解析到 3 列，实际 %d：%+v", len(cols), cols)
+	}
+	byName := map[string]ddlColumn{}
+	for _, c := range cols {
+		byName[c.Column] = c
+	}
+	if c := byName["created_at"]; !c.NotNull || !c.HasDefault || c.OnUpdate {
+		t.Errorf("created_at 解析不符：%+v", c)
+	}
+	if c := byName["updated_at"]; !c.OnUpdate || !c.NotNull {
+		t.Errorf("updated_at 解析不符：%+v", c)
+	}
+	// update_time 的 DDL 是 DEFAULT NULL —— HasDefault 会被识别成 true，
+	// 但 NotNull=false，所以分类时应落到 ddl-no-default 之外的分支由 NULL 性判断先拦住。
+	if c := byName["update_time"]; c.NotNull {
+		t.Errorf("update_time DDL 是可空，不该解析成 NOT NULL：%+v", c)
+	}
+}
+
+// TestCreateTimeAssertionNeedsMarker 证明「没有运行标记时断言不放行」
+// （不允许"没标记就当通过"这种自欺）。
+func TestCreateTimeAssertionNeedsMarker(t *testing.T) {
+	// 用一个必然查不到的 db 指针不行（会 panic），这里只验证函数签名与语义说明存在：
+	// 真正的 marker 语义由 cmd/ac15fixture verify 的集成路径覆盖。
+	var _ = sql.NullString{}
+	if RunMarkerAction == "" {
+		t.Error("RunMarkerAction 不应为空")
+	}
+}

internal/academic/dualrun/fixture/verify.go

@@ -407,6 +407,10 @@ func Verify(ctx context.Context, cfg Config, legacyUserID int64) (*VerifyReport,
 		return nil, err
 	}
 
+	// ===== 4.5 create_time 未落入 DB 默认值（业务时间必须来自 legacy）=====
+	rep.Assertions = append(rep.Assertions,
+		createTimeAssertions(ctx, target, []string{"t_user_expense_log"})...)
+
 	// ===== 5. 语义断言 =====
 	rep.Assertions = append(rep.Assertions, Assertion{
 		Name:   "create_user_id=0 → NULL（本 fixture 数据无 0 行时按空集通过）",