copy-to-postgres 在真实开发数据上核对不过的几处（复核 R-a1、R-a2、R-a3、R-a8、R-a13）

- R-a1 老库 NOT NULL 列为空：第一版补 schedules.trigger 时没带缺省值，存量定时任务这一列是 NULL，导入 PostgreSQL 撞 NOT NULL，连带触发记录、通知撞外键。迁移表里这一列改成带 DEFAULT '{}'；新增 _NOT_NULL_BACKFILLS，启动时按模型的缺省值把已经补过、留着 NULL 的填上（SQLite 这边也填，两种库存的一样；只更新这一列，不碰 updated_at）；导入时 _Fixer 遇到 NOT NULL、有缺省值的列的 NULL 也填缺省值，计入「转换」。迁移表里的 NOT NULL 列都过了一遍，只有这一列没带缺省值
- R-a2 孤儿行：导入前在快照上按模型外键扫（settle_orphans）：外键是 SET NULL、列可空的置空，计入转换；CASCADE / 不可空的从快照里拿掉、不导入，反复扫到没有为止，报告里单列「孤儿行（SQLite 当时没强制外键）」的表和条数，不算核对失败
- R-a3 停服务后 agentlab.db-wal 还在：lifespan 收尾 dispose 应用库引擎（只加了这几行），SQLite 最后一个连接关掉时并回、删掉 -wal，PostgreSQL 的池子也正常关
- R-a8 自增主键的序列取「目标表 MAX、快照 sqlite_sequence 的计数器、序列现值」三者最大，删过的号不复用（召回按 source_changes.seq 当游标）
- R-a13 插不进的行的原因不再带「<class 'asyncpg.exceptions.…'>:」前缀（errors.first_line 两种写法都去）；设了 AGENTLAB_DATABASE_URL 却没装 [postgres] 时，应用库引擎和断点库都直接说要装什么
- 测试：DROP COLUMN 再按第一版迁移补回造老表、关着外键写孤儿行，迁移填值、快照上置空和拿掉、导入 PostgreSQL 核对通过且重跑一样；NULL 落回模型缺省值；端到端里删掉最后一条变化，导过去新变化从 3 编起；起停一次服务后数据目录里没有 -wal / -shm（去掉 dispose 这条测试就挂）
- PostgreSQL 引擎上另挂了去掉文本参数里 NUL 的钩子（db/postgres.strip_nul，见下一个提交）

backend/app/core/errors.py

@@ -36,8 +36,8 @@ _UNREACHABLE = ("connection refused", "nodename nor servname", "name or service
                 "all connection attempts failed", "network is unreachable")
 
 #: SQLAlchemy 包过的驱动异常开头是「(sqlite3.OperationalError) 」；asyncpg（PostgreSQL）的后面还跟着
-#: 「<class 'asyncpg.exceptions.UniqueViolationError'>: 」，一起去掉
-_WRAPPED = re.compile(r"^\([\w.]+\)\s*(?:<class '[\w.]+'>:\s*)?")
+#: 「<class 'asyncpg.exceptions.UniqueViolationError'>: 」，一起去掉（直接拿 e.orig 的也只剩后一段，同样去掉）
+_WRAPPED = re.compile(r"^(?:\([\w.]+\)\s*)?(?:<class '[\w.]+'>:\s*)?")
 # 内部符号：`a.b.<locals>._run()`、`<function f at 0x…>`、模块路径。对使用者没有意义
 _LOCALS = re.compile(r"[\w.]*<locals>\.?[\w.]*(\(\))?")
 _REPR = re.compile(r"<(?:function|bound method|coroutine|object) [^>]*>")

backend/app/db/base.py

@@ -197,6 +197,7 @@ def make_engine(url: str | None = None, *, application_name: str = APPLICATION_N
         return made
     from app.db import postgres
 
+    _require_postgres_driver("asyncpg")
     pool: dict[str, Any] = (
         {"poolclass": NullPool} if settings.db_pool_size <= 0 else {
             "pool_size": settings.db_pool_size, "max_overflow": max(settings.db_max_overflow, 0),
@@ -207,9 +208,21 @@ def make_engine(url: str | None = None, *, application_name: str = APPLICATION_N
         connect_args={"server_settings": {"application_name": application_name, "timezone": "UTC"}},
         **pool)
     postgres.install(made.sync_engine.dialect)
+    event.listen(made.sync_engine, "before_cursor_execute", postgres.strip_nul, retval=True)
     return made
 
 
+def _require_postgres_driver(*modules: str) -> None:
+    """设了 AGENTLAB_DATABASE_URL 却没装 [postgres]：说清楚要装什么，不丢一个光秃秃的 ModuleNotFoundError。"""
+    import importlib.util
+
+    missing = [m for m in modules if importlib.util.find_spec(m) is None]
+    if missing:
+        raise RuntimeError(
+            f"应用库设成了 PostgreSQL（AGENTLAB_DATABASE_URL），但没装它要的包（{'、'.join(missing)}）："
+            "pip install -e 'backend[postgres]' -c backend/requirements.lock")
+
+
 def describe_database() -> str:
     """启动日志里的一句：用的是哪种库、在哪（不写密码）。"""
     from app.db.urls import describe
@@ -356,7 +369,9 @@ _COLUMN_MIGRATIONS: list[tuple[str, str, str]] = [
     ("schedule_fires", "eval_run_id", "VARCHAR(32)"),
     # 编排完善（docs/design/orchestration-round/00-总方案.md §4）：定时任务的触发方式。存量行都是 cron
     ("schedules", "trigger_kind", "VARCHAR(20) DEFAULT 'cron'"),
-    ("schedules", "trigger", "JSON"),
+    # 第一版补这一列时没带缺省值：在那之前建过定时任务的库里这一列是 NULL（读时 or {} 兜着），模型是 NOT NULL。
+    # 现在带上缺省值；已经补过、留着 NULL 的由 _NOT_NULL_BACKFILLS 填上
+    ("schedules", "trigger", "JSON DEFAULT '{}'"),
     # 第 2 波：服务账号（users.kind）；剖析上传表格源时的快照
     ("users", "kind", "VARCHAR(20) DEFAULT 'person'"),
     ("catalog_profile_tasks", "snapshot_id", "VARCHAR(32)"),
@@ -381,6 +396,42 @@ _INDEX_MIGRATIONS: list[tuple[str, str, str | tuple[str, ...]]] = [
 ]
 
 
+#: 补列时没带缺省值、存量行留着 NULL 的 NOT NULL 列：每次启动按模型的缺省值填上（UPDATE … WHERE 列 IS NULL，填完再跑
+#: 什么都不改）。两种库存的才一样，导入 PostgreSQL 时这一列也不会撞 NOT NULL。以后新加的 NOT NULL 列在迁移表里一律带
+#: DEFAULT（test_db_portable 守着），就不用往这里加
+_NOT_NULL_BACKFILLS: list[tuple[str, str]] = [("schedules", "trigger")]
+
+
+def model_default(column: Any) -> tuple[bool, Any]:
+    """模型列的缺省值：(有没有, 值)。标量原样，无参的可调用（dict、list、utcnow 之类）调一次；要按执行上下文算的不算。"""
+    default = getattr(column, "default", None)
+    if default is None:
+        return False, None
+    if default.is_scalar:
+        return True, default.arg
+    if default.is_callable:
+        try:
+            return True, default.arg(None)
+        except Exception:  # noqa: BLE001 - 要读执行上下文的缺省值：这里给不出
+            return False, None
+    return False, None
+
+
+def _backfill_not_null(conn: Any, columns: dict[str, set[str]]) -> None:
+    from sqlalchemy import Table, update
+
+    for table_name, column_name in _NOT_NULL_BACKFILLS:
+        if column_name not in (columns.get(table_name) or set()):
+            continue
+        model = Base.metadata.tables[table_name].c[column_name]
+        known, value = model_default(model)
+        if not known:
+            continue
+        # 只带这一列的表：模型表上 updated_at 的 onupdate 会跟着改（这是补数据，不是改了这一行），老库也未必有那一列
+        bare = Table(table_name, MetaData(), Column(column_name, model.type))
+        conn.execute(update(bare).where(bare.c[column_name].is_(None)).values({column_name: value}))
+
+
 #: 迁移补建的唯一索引：(索引名, 表, 列, 部分索引的条件, 建之前的去重语句)。只在索引还不存在时先去重再建——存量库里可能
 #: 已经有并发留下的重复行，直接建会失败。去重的规矩写在各自的语句里：多出的钩子行删掉（只留最早的那一行，它的密钥照旧
 #: 能用），多出的助手对话归档（不删，历史还在）。条件和语句里的 {false} / {true} 按方言换成布尔的字面量（SQLite 是 0 / 1）
@@ -449,6 +500,7 @@ def _migrate(conn) -> None:  # pragma: no cover - 同步回调
         conn.execute(text(dedupe.format(**literals)))
         conn.execute(text(f"CREATE UNIQUE INDEX {name} ON {table} ({', '.join(wanted)})"
                           + (f" WHERE {where.format(**literals)}" if where else "")))
+    _backfill_not_null(conn, columns)
 
 
 #: init_db 在 PostgreSQL 上拿的会话级咨询锁的号：几个进程同时启动时，建表和迁移一个一个来（同一个号别处别用）

backend/app/db/to_postgres.py

@@ -23,7 +23,13 @@ pending writes 按条数核对，逐项打印；有来源的行目标上没有
 
 **值的转换。** 读写都按模型的列类型走：时间是不带时区的 UTC、布尔是真布尔、JSON 原文照搬（SQL 的 NULL 和 JSON 的
 null 分开，不混）。SQLite 不查类型、PostgreSQL 查：整数列里的整数值的浮点、数字字符串换成整数，文本列里的非文本换成
-字符串；文本里的 NUL 字符（PostgreSQL 不收）去掉，JSON 里的 NaN / Infinity 写成 null，各自计数打印。
+字符串；文本里的 NUL 字符（PostgreSQL 不收）去掉，JSON 里的 NaN / Infinity 写成 null；模型里 NOT NULL、有缺省值的列，
+老库补列时没带缺省值留下的 NULL 填成缺省值（启动时的迁移在 SQLite 上也这么填）；各自计数打印。
+
+**孤儿行。** 外键指向已经不在的行（SQLite 当时没强制外键留下的）：外键是 SET NULL 的置空、计入转换；别的从快照里拿掉、
+不导入，按表列出条数，不算核对失败（settle_orphans）。
+
+**自增主键。** 导入后序列接在已用过的最大号后面：目标表的 MAX、SQLite 的 AUTOINCREMENT 计数器（删过的号不复用）取大的。
 """
 
 from __future__ import annotations
@@ -53,8 +59,10 @@ DEFAULT_BATCH = 1000
 @dataclass
 class TableResult:
     name: str
-    #: 来源的行数
+    #: 来源的行数（不含孤儿行）
     source: int = 0
+    #: 外键指向已不存在的行、又不能置空的（外键是 CASCADE 或不可空）：SQLite 当时没强制外键留下的，没导入
+    orphans: int = 0
     #: 来源的这些主键，目标上导入前已经有的（--dry-run 时就是「不用补的」）、导入后有的
     present_before: int = 0
     present_after: int = 0
@@ -77,6 +85,8 @@ class Report:
     checkpoints_written: int = 0
     threads: int = 0
     fixed: dict[str, int] = field(default_factory=dict)
+    #: 没导入的孤儿行，按表（见 settle_orphans）
+    orphans: dict[str, int] = field(default_factory=dict)
     dry_run: bool = False
 
     @property
@@ -139,6 +149,49 @@ def snapshot(db: Path, into: Path) -> Path:
     return out
 
 
+def settle_orphans(path: Path, *, fixed: dict[str, int]) -> dict[str, int]:
+    """快照上按模型的外键找孤儿行：外键列有值、指向的行却不在（SQLite 当时没强制外键：老版本、或者那时外键还没加上）。
+    PostgreSQL 一律拒收，重跑也补不上，所以导入前在快照上（来源不动）先处理：
+
+    - 外键是 ON DELETE SET NULL、列可空的：置空，计入「转换」——父行被删时本来就该这样；
+    - 别的（CASCADE、不可空）：从快照里拿掉、不导入，按表计数交回——父行被删时本来就该跟着删。拿掉一行可能让下一层又成了
+      孤儿，反复扫到没有为止。
+
+    这些不算核对失败：报告里照实列出哪张表、几行。"""
+    from sqlalchemy import and_, delete, exists, update
+
+    from app.db import models  # noqa: F401
+    from app.db.base import Base
+
+    orphans: dict[str, int] = {}
+    engine = create_sync_engine(f"sqlite:///{path}")
+    try:
+        while True:
+            removed = 0
+            with engine.begin() as conn:
+                for table in Base.metadata.sorted_tables:
+                    for fk in table.foreign_key_constraints:
+                        parent = fk.referred_table.alias("parent")
+                        pairs = [(el.parent, parent.c[el.column.name]) for el in fk.elements]
+                        lost = and_(*(child.is_not(None) for child, _ in pairs),
+                                    ~exists(select(1).select_from(parent).where(
+                                        and_(*(p == child for child, p in pairs)))))
+                        n = int(conn.execute(select(func.count()).select_from(table).where(lost)).scalar() or 0)
+                        if not n:
+                            continue
+                        if (fk.ondelete or "").upper() == "SET NULL" and all(c.nullable for c, _ in pairs):
+                            conn.execute(update(table).where(lost).values({c.name: None for c, _ in pairs}))
+                            fixed["外键指向已不存在的行，置空"] = fixed.get("外键指向已不存在的行，置空", 0) + n
+                        else:
+                            conn.execute(delete(table).where(lost))
+                            orphans[table.name] = orphans.get(table.name, 0) + n
+                            removed += n
+            if not removed:
+                return orphans
+    finally:
+        engine.dispose()
+
+
 def upgrade_snapshot(path: Path) -> None:
     """在快照上跑一遍启动时的建表补列：老版本的库按当前的表结构读。"""
     from app.db import models  # noqa: F401
@@ -173,7 +226,7 @@ class _Fixer:
 
     def value(self, column: Any, value: Any) -> Any:
         if value is None:
-            return None
+            return self.missing(column)
         if _is_json(column):
             return self.json(value)
         kind = column.type
@@ -201,6 +254,19 @@ class _Fixer:
             value = value.replace("\x00", "")
         return value
 
+    def missing(self, column: Any) -> Any:
+        """SQL 的 NULL。模型列 NOT NULL、又有缺省值的（老版本补列时没带缺省值，存量行留着 NULL；SQLite 的 ADD COLUMN
+        不带 NOT NULL，照存）填成缺省值：PostgreSQL 的表照模型建，带 NOT NULL，不填就整行插不进。"""
+        from app.db.base import model_default, pg_json_dumps
+
+        if column.nullable or column.primary_key:
+            return None
+        known, default = model_default(column)
+        if not known:
+            return None
+        self._count("NOT NULL 列的空值填成缺省值")
+        return pg_json_dumps(default) if _is_json(column) else default
+
     def json(self, raw: Any) -> str:
         """JSON 原文 → 写进 PostgreSQL 的原文。平常的原样照搬（键的先后不变）；有 NaN / NUL 的才重写一遍。"""
         from app.db.base import pg_json_dumps
@@ -252,7 +318,7 @@ async def _write_batch(target: Any, statement: Any, rows: list[dict[str, Any]],
             from app.core.errors import first_line
 
             ident = ", ".join(f"{k}={row.get(k)!r}" for k in keys)
-            refused.append(f"{ident}：{first_line(e.orig or e)}")
+            refused.append(f"{ident}：{first_line(e)}")
 
 
 def _present(table: Any, keys: list[tuple[Any, ...]]) -> Any:
@@ -276,8 +342,12 @@ async def copy_tables(snapshot_db: Path, target: Any, *, batch: int, dry_run: bo
     source = create_sync_engine(f"sqlite:///{snapshot_db}")
     try:
         with source.connect() as src:
+            # AUTOINCREMENT 表的计数器（删过最后几行时比 MAX 大）：PostgreSQL 的序列要接在它后面，不复用删掉的号——
+            # 召回按 source_changes.seq 当游标往后拉，复用了就跳过新变化
+            counters = dict(src.execute(text("SELECT name, seq FROM sqlite_sequence")).all()) if src.execute(text(
+                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'sqlite_sequence'")).first() else {}
             for table in Base.metadata.sorted_tables:
-                result = TableResult(table.name)
+                result = TableResult(table.name, orphans=report.orphans.get(table.name, 0))
                 report.tables.append(result)
                 result.source = int(src.execute(select(func.count()).select_from(table)).scalar() or 0)
                 async with target.connect() as conn:
@@ -306,7 +376,7 @@ async def copy_tables(snapshot_db: Path, target: Any, *, batch: int, dry_run: bo
                 if dry_run:
                     result.present_after, result.target_after = result.present_before, result.target_before
                     continue
-                await _reset_sequences(target, table)
+                await _reset_sequences(target, table, int(counters.get(table.name) or 0))
                 async with target.connect() as conn:
                     result.target_after = await _count(conn, table)
     finally:
@@ -319,15 +389,22 @@ def _has_table(sync: Any, name: str) -> bool:
     return inspect(sync).has_table(name)
 
 
-async def _reset_sequences(target: Any, table: Any) -> None:
-    """自增主键的序列接到已有的最大值后面：导入时写的是来源的主键值，序列自己没动过，不改的话下一条撞主键。"""
+async def _reset_sequences(target: Any, table: Any, floor: int = 0) -> None:
+    """自增主键的序列接到已用过的最大号后面：导入时写的是来源的主键值，序列自己没动过，不改的话下一条撞主键。
+    已用过的取三者最大：目标表里的 MAX、SQLite 的 AUTOINCREMENT 计数器（floor）、序列现在的值（重跑时不往回拨）。"""
     for column in table.primary_key.columns:
         if column is not table.autoincrement_column:
             continue
         async with target.begin() as conn:
-            await conn.execute(text(
-                f"SELECT setval(pg_get_serial_sequence(:t, :c), GREATEST(COALESCE(MAX({column.name}), 0), 1), "
-                f"MAX({column.name}) IS NOT NULL) FROM {table.name}"), {"t": table.name, "c": column.name})
+            name = (await conn.execute(text("SELECT pg_get_serial_sequence(:t, :c)"),
+                                       {"t": table.name, "c": column.name})).scalar()
+            if not name:
+                continue
+            top = int((await conn.execute(select(func.max(column)))).scalar() or 0)
+            last, called = (await conn.execute(text(f"SELECT last_value, is_called FROM {name}"))).one()
+            used = max(top, floor, int(last) if called else int(last) - 1)
+            await conn.execute(text("SELECT setval(:n, :v, :called)"),
+                               {"n": name, "v": max(used, 1), "called": used >= 1})
 
 
 async def services_on(target: Any) -> int:
@@ -425,6 +502,7 @@ async def copy_to_postgres(source: str | Path, target_url: str, *, checkpoint_ur
         with tempfile.TemporaryDirectory(prefix="agentlab-copy-", dir=work_dir) as tmp:
             snap = snapshot(db, Path(tmp))
             upgrade_snapshot(snap)
+            report.orphans = settle_orphans(snap, fixed=report.fixed)
             if not dry_run:
                 await create_schema(target)
             await copy_tables(snap, target, batch=batch, dry_run=dry_run, report=report, progress=progress)
@@ -451,8 +529,14 @@ def render(report: Report) -> list[str]:
         else:
             verdict = "一致" + (f"（这次补了 {t.present_after - t.present_before} 行）"
                                if t.present_after > t.present_before else "（都已经有了）")
+        if t.orphans:
+            verdict += f"；另有 {t.orphans} 行孤儿行没导入"
         lines.append(f"{t.name:<28}{t.source:>10}{t.present_before:>10}{t.present_after:>10}{t.target_after:>10}"
                      f"  {verdict}")
+    if report.orphans:
+        total = sum(report.orphans.values())
+        lines.append(f"孤儿行（SQLite 当时没强制外键，外键指向的行已经不在）：{total} 行没导入，不算核对失败——"
+                     + "、".join(f"{name} {n} 行" for name, n in report.orphans.items()))
     for t in report.tables:
         for item in t.refused[:20]:
             lines.append(f"  {t.name} 没插进去：{item}")

backend/app/engine/checkpoints.py

@@ -32,6 +32,9 @@ def _pool_size() -> int:
 async def open_postgres_saver(conninfo: str, *, max_size: int | None = None,
                               application_name: str = "agentlab") -> AsyncIterator[BaseCheckpointSaver]:
     """PostgreSQL 的断点库：开连接池、建表（幂等），退出时关池子。application_name 同应用库（db/base.APPLICATION_NAME）。"""
+    from app.db.base import _require_postgres_driver
+
+    _require_postgres_driver("psycopg", "psycopg_pool", "langgraph.checkpoint.postgres")
     from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
     from psycopg.rows import dict_row
     from psycopg_pool import AsyncConnectionPool

backend/app/main.py

@@ -182,6 +182,11 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]:
         from app.sandbox.manager import sandbox_manager
 
         await sandbox_manager.close()
+        # 应用库的连接全关掉：SQLite 最后一个连接关掉时才把 -wal 并回、删掉（copy-to-postgres 按它认服务停没停），
+        # PostgreSQL 的连接池也正常关
+        from app.db.base import engine as db_engine
+
+        await db_engine.dispose()
         logger.info("已停止")
 
 

backend/tests/test_copy_to_postgres.py

@@ -14,6 +14,7 @@ import sys
 from pathlib import Path
 
 import pytest
+from sqlalchemy import text
 
 from app.core.config import settings
 from app.db import to_postgres
@@ -68,6 +69,11 @@ async def main():
         s.add(source)
         await s.flush()
         s.add(SourceChange(source_id=source.id, kind="column_added", fingerprint="f" * 64, table_name="订单"))
+        dropped = SourceChange(source_id=source.id, kind="column_removed", fingerprint="d" * 64, table_name="订单")
+        s.add(dropped)
+        await s.flush()
+        # 删掉最后一条：SQLite 的 AUTOINCREMENT 计数器停在 2，表里的 MAX 是 1。导过去之后新的变化要从 3 编起
+        await s.delete(dropped)
         long = Run(workflow_name="长" * 300, last_seq=5_000_000_000, origin={"b": 2, "a": 1})
         sql_null = Run(workflow_name="SQL 的空", origin={"x": 1})
         json_null = Run(workflow_name="JSON 的空", origin=None)
@@ -263,7 +269,7 @@ async def test_a_waiting_run_is_imported_and_finishes_on_postgresql(source_dir):
         seqs = done["seqs"]
         assert seqs == sorted(set(seqs)) and seqs[-1] > made["last_seq"], "接着导入前的序号往下编"
         assert done["artifact"] == {"区域": "虚构", "数": 1}
-        assert done["new_change_seq"] > 1, "自增主键的序列接在导入的最大值后面"
+        assert done["new_change_seq"] == 3, "序列接在 SQLite 的 AUTOINCREMENT 计数器后面，删过的号不复用（召回按它当游标）"
 
 
 @needs_pg
@@ -294,3 +300,137 @@ async def test_a_target_with_the_service_still_connected_is_refused(source_dir):
             with pytest.raises(CopyRefused, match="服务的连接"):
                 await copy_to_postgres(data, target, progress=lambda _: None)
         assert _sql(target, "SELECT count(*) FROM information_schema.tables WHERE table_name = 'runs'") == [(0,)]
+
+
+# --------------------------------------------------------------------------
+# 老库：补列时没带缺省值的 NOT NULL 列、SQLite 当时没强制外键留下的孤儿行
+# --------------------------------------------------------------------------
+
+
+def _legacy_source(folder: Path) -> Path:
+    """照模型建一个 SQLite 应用库，再做旧：定时任务的 trigger 列删掉、按第一版迁移（不带缺省值）补回来，存量行是 NULL；
+    关着外键写几行孤儿行（对话轮次指向已删的运行、事件指向已删的运行、轮次指向已删的对话）。"""
+    from datetime import datetime, timezone
+
+    from sqlalchemy import create_engine, insert
+
+    from app.db import models  # noqa: F401
+    from app.db.base import Base
+
+    folder.mkdir(parents=True, exist_ok=True)
+    path = folder / "agentlab.db"
+    engine = create_engine(f"sqlite:///{path}")
+    tables = Base.metadata.tables
+    try:
+        Base.metadata.create_all(engine)
+        with engine.begin() as conn:
+            conn.execute(insert(tables["schedules"]).values(id="s1", workflow_id="w1", cron="0 9 * * *",
+                                                             timezone="Asia/Shanghai"))
+            conn.execute(insert(tables["schedule_fires"]).values(
+                id="f1", schedule_id="s1", fire_at=datetime(2026, 1, 1, tzinfo=timezone.utc)))
+            conn.execute(insert(tables["conversations"]).values(id="c1", title="虚构的对话"))
+            conn.execute(insert(tables["conversation_turns"]).values(id="t1", conversation_id="c1", seq=0,
+                                                                     run_id="run-gone"))
+            conn.execute(insert(tables["conversation_turns"]).values(id="t2", conversation_id="conv-gone", seq=0))
+            conn.execute(insert(tables["run_events"]).values(id="e1", run_id="run-gone", seq=1, type="log", ts=0.0,
+                                                             data={}))
+        with engine.begin() as conn:
+            conn.execute(text("ALTER TABLE schedules DROP COLUMN \"trigger\""))
+            conn.execute(text("ALTER TABLE schedules ADD COLUMN \"trigger\" JSON"))
+    finally:
+        engine.dispose()
+    return path
+
+
+def test_old_sqlite_databases_get_null_not_null_columns_filled(tmp_path):
+    """SQLite 这边：启动时的迁移把老版本补列留下的 NULL 填成模型的缺省值，两种库存的一样。"""
+    from sqlalchemy import create_engine
+
+    from app.db.base import _migrate
+
+    path = _legacy_source(tmp_path / "old")
+    engine = create_engine(f"sqlite:///{path}")
+    try:
+        with engine.begin() as conn:
+            assert conn.execute(text("SELECT \"trigger\" FROM schedules")).scalar_one() is None
+            _migrate(conn)
+            assert conn.execute(text("SELECT \"trigger\" FROM schedules")).scalar_one() == "{}"
+    finally:
+        engine.dispose()
+
+
+def test_orphan_rows_are_nulled_or_left_out_on_the_snapshot(tmp_path):
+    path = _legacy_source(tmp_path / "old")
+    fixed: dict[str, int] = {}
+    orphans = to_postgres.settle_orphans(path, fixed=fixed)
+    assert fixed == {"外键指向已不存在的行，置空": 1}
+    assert orphans == {"conversation_turns": 1, "run_events": 1}
+    import sqlite3
+
+    db = sqlite3.connect(path)
+    try:
+        assert db.execute("SELECT id, run_id FROM conversation_turns").fetchall() == [("t1", None)]
+        assert db.execute("SELECT count(*) FROM run_events").fetchone() == (0,)
+    finally:
+        db.close()
+
+
+@needs_pg
+async def test_an_old_database_with_nulls_and_orphans_imports_cleanly(tmp_path):
+    """真实开发库上核对不过的那两种：定时任务 trigger 为 NULL（NOT NULL 列），对话轮次指向已删的运行。导入照样核对通过，
+    孤儿行照实列出、不算失败，重跑也一样。"""
+    from tests.appdb import fresh_postgres_database
+
+    folder = tmp_path / "old"
+    _legacy_source(folder)
+    with fresh_postgres_database("legacy") as target:
+        report = await copy_to_postgres(folder, target, progress=lambda _: None)
+        lines = to_postgres.render(report)
+        assert report.ok, "\n".join(lines)
+        assert report.orphans == {"conversation_turns": 1, "run_events": 1}
+        assert report.fixed.get("外键指向已不存在的行，置空") == 1
+        assert any("孤儿行" in line and "run_events 1 行" in line for line in lines)
+        assert _sql(target, "SELECT CAST(\"trigger\" AS TEXT) FROM schedules") == [("{}",)]
+        assert _sql(target, "SELECT count(*) FROM schedule_fires") == [(1,)]
+        assert _sql(target, "SELECT run_id FROM conversation_turns") == [(None,)]
+        again = await copy_to_postgres(folder, target, progress=lambda _: None)
+        assert again.ok and again.orphans == report.orphans
+
+
+def test_null_not_null_values_fall_back_to_the_model_default():
+    from app.db.base import Base
+
+    schedules = Base.metadata.tables["schedules"]
+    fixed: dict[str, int] = {}
+    fix = to_postgres._Fixer(fixed)
+    assert fix.value(schedules.c["trigger"], None) == "{}"          # JSON 列按原文绑定
+    assert fix.value(schedules.c["trigger_kind"], None) == "cron"
+    assert fix.value(schedules.c["target_id"], None) is None        # 可空的照旧是 NULL
+    assert fix.value(schedules.c["cron"], None) is None             # 没有缺省值的：照实交给库去拒
+    assert fixed == {"NOT NULL 列的空值填成缺省值": 2}
+
+
+# --------------------------------------------------------------------------
+# 停服务之后数据目录里不留 -wal（导入按它认服务停没停）
+# --------------------------------------------------------------------------
+
+START_STOP = r'''
+from fastapi.testclient import TestClient
+from app.main import app
+
+with TestClient(app) as client:
+    assert client.get("/api/health").status_code == 200
+'''
+
+
+def test_stopping_the_service_leaves_no_wal_behind(tmp_path):
+    data = tmp_path / "data"
+    data.mkdir()
+    env = {**os.environ, "AGENTLAB_DATA_DIR": str(data), "AGENTLAB_DATABASE_URL": "", "AGENTLAB_CHECKPOINT_URL": "",
+           "AGENTLAB_SCHEDULER": "0", "PYTHONPATH": str(BACKEND)}
+    done = subprocess.run([sys.executable, "-c", START_STOP], env=env, capture_output=True, text=True, timeout=180,
+                          cwd=str(data))
+    assert done.returncode == 0, done.stderr[-4000:]
+    assert (data / "agentlab.db").is_file() and (data / "checkpoints.db").is_file()
+    assert sorted(p.name for p in data.iterdir() if p.name.endswith(("-wal", "-shm"))) == []
+    assert to_postgres.source_in_use(data / "agentlab.db", data / "checkpoints.db") is None