diff --git a/deploy/dmig.py b/deploy/dmig.py
index 01fe693..be04980 100755
--- a/deploy/dmig.py
+++ b/deploy/dmig.py
@@ -101,16 +101,24 @@ def ensure_ledger(conf):
 
 
 def ledger_state(conf):
-    out, err = mysql(conf, "SELECT batch, MAX(step) FROM %s WHERE status='applied' "
+    """已处理批次：{batch: (max_step, has_failed)}。applied+skipped 都算已处理，
+    含 failed 记录的批次不算完成。"""
+    out, err = mysql(conf, "SELECT batch, MAX(step), SUM(status='failed') FROM %s "
+                           "WHERE status IN ('applied','skipped','failed') "
                            "GROUP BY batch" % LEDGER_TABLE)
     applied = {}
     for ln in (out or "").splitlines():
         p = ln.split("\t")
-        if len(p) == 2:
-            applied[p[0]] = int(p[1])
+        if len(p) == 3:
+            applied[p[0]] = (int(p[1]), int(p[2]) > 0)
     return applied
 
 
+def _esc(s):
+    """转义 SQL 单引号（detail/backup_file 入台账前必用）。"""
+    return str(s).replace("\\", "\\\\").replace("'", "''")
+
+
 def load_migrations():
     migs = {}
     if not os.path.isdir(MIG_DIR):
@@ -183,16 +191,38 @@ def cmd_status(conf):
     print("=== 迁移台账 ===")
     for mid in sorted(migs):
         m = migs[mid]
-        st = "applied(步骤%d)" % applied[mid] if mid in applied else "待执行"
-        print("  %s  %-10s  %s" % (mid, st, m.get("title", "")[:50]))
-    print("\n待执行批次数:", len([m for m in migs if m not in applied]))
+        if mid in applied:
+            mx, failed = applied[mid]
+            st = "失败(步骤≤%d)" % mx if failed else "已处理(步骤≤%d)" % mx
+        else:
+            st = "待执行"
+        print("  %s  %-16s  %s" % (mid, st, m.get("title", "")[:50]))
+    pending = [mid for mid in migs
+               if mid not in applied or applied[mid][1]]
+    print("\n待执行批次数:", len(pending), ("(%s)" % ",".join(pending)) if pending else "")
+
+
+def _pending(migs, applied):
+    """未完成的批次：未执行过，或含 failed 记录（修复后可重试）。"""
+    return [b for b in sorted(migs) if b not in applied or applied[b][1]]
+
+
+def _ledger_write(conf, mid, step, status, detail, backup_file=""):
+    _, err = mysql(conf,
+                   "INSERT INTO %s (id, batch, step, status, backup_file, detail) "
+                   "VALUES ('%s', '%s', %d, '%s', '%s', '%s') "
+                   "ON DUPLICATE KEY UPDATE status='%s', detail='%s'"
+                   % (LEDGER_TABLE, _new_id(), mid, step, status,
+                      _esc(backup_file[:200]), _esc(detail[:200]), status, _esc(detail[:200])))
+    if err:
+        print("   [WARN] 台账写入失败(不影响库变更): %s" % err[:150])
 
 
 def cmd_plan(conf, batches):
     ensure_ledger(conf)
     applied = ledger_state(conf)
     migs = load_migrations()
-    targets = [b for b in (batches or sorted(migs)) if b not in applied]
+    targets = [b for b in (batches or _pending(migs, applied)) if b not in applied or applied[b][1]]
     if not targets:
         print("无待执行批次")
         return
@@ -215,7 +245,7 @@ def cmd_apply(conf, batches):
     ensure_ledger(conf)
     applied = ledger_state(conf)
     migs = load_migrations()
-    targets = [b for b in (batches or sorted(migs)) if b not in applied]
+    targets = [b for b in (batches or _pending(migs, applied)) if b not in applied or applied[b][1]]
     if not targets:
         print("无待执行批次")
         return
@@ -237,24 +267,17 @@ def cmd_apply(conf, batches):
         for i, step in enumerate(m.get("up", []), 1):
             sqls, skip = render_step(conf, step, "up")
             if skip:
-                mysql(conf, "INSERT INTO %s (id, batch, step, status, detail) "
-                            "VALUES ('%s', '%s', %d, 'skipped', '%s') "
-                            "ON DUPLICATE KEY UPDATE status='skipped'"
-                      % (LEDGER_TABLE, _new_id(), mid, i, skip[:200]))
+                _ledger_write(conf, mid, i, "skipped", skip)
                 print("   步骤%d 跳过(%s)" % (i, skip))
                 continue
             for s in sqls:
                 out, err = mysql(conf, s)
                 if err:
-                    mysql(conf, "INSERT INTO %s (id, batch, step, status, detail) "
-                                "VALUES ('%s', '%s', %d, 'failed', '%s')"
-                          % (LEDGER_TABLE, _new_id(), mid, i, err[:200]))
+                    _ledger_write(conf, mid, i, "failed", err)
                     die("步骤%d 失败: %s\n（已完成步骤可用 `dmig.py rollback %s` 回退）"
                         % (i, err, mid))
                 print("   步骤%d OK: %s" % (i, s[:90]))
-            mysql(conf, "INSERT INTO %s (id, batch, step, status, backup_file, detail) "
-                        "VALUES ('%s', '%s', %d, 'applied', '%s', '%s')"
-                  % (LEDGER_TABLE, _new_id(), mid, i, bk[:200], str(step)[:200]))
+            _ledger_write(conf, mid, i, "applied", str(step), bk)
         print("   ✅ %s 完成" % mid)
     print("\n全部完成。验证: ./py3/bin/python deploy/dmig.py status && ddiff.py --checklist")