From 0e997eb968d86c7a21f0dcf683ebc9ac9f10c0fa Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 19 Oct 2026 12:06:31 +0800 Subject: [PATCH] =?UTF-8?q?copy-to-postgres=20=E5=9C=A8=E7=9C=9F=E5=AE=9E?= =?UTF-8?q?=E5=BC=80=E5=8F=91=E6=95=B0=E6=8D=AE=E4=B8=8A=E6=A0=B8=E5=AF=B9?= =?UTF-8?q?=E4=B8=8D=E8=BF=87=E7=9A=84=E5=87=A0=E5=A4=84=EF=BC=88=E5=A4=8D?= =?UTF-8?q?=E6=A0=B8=20R-a1=E3=80=81R-a2=E3=80=81R-a3=E3=80=81R-a8?= =?UTF-8?q?=E3=80=81R-a13=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 插不进的行的原因不再带「:」前缀（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 | 4 +- backend/app/db/base.py | 54 +++++++++- backend/app/db/to_postgres.py | 106 ++++++++++++++++-- backend/app/engine/checkpoints.py | 3 + backend/app/main.py | 5 + backend/tests/test_copy_to_postgres.py | 142 ++++++++++++++++++++++++- 6 files changed, 299 insertions(+), 15 deletions(-) diff --git a/backend/app/core/errors.py b/backend/app/core/errors.py index a64cf211..01355678 100644 --- a/backend/app/core/errors.py +++ b/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）的后面还跟着 -#: 「: 」，一起去掉 -_WRAPPED = re.compile(r"^\([\w.]+\)\s*(?::\s*)?") +#: 「: 」，一起去掉（直接拿 e.orig 的也只剩后一段，同样去掉） +_WRAPPED = re.compile(r"^(?:\([\w.]+\)\s*)?(?::\s*)?") # 内部符号：`a.b.._run()`、``、模块路径。对使用者没有意义 _LOCALS = re.compile(r"[\w.]*\.?[\w.]*(\(\))?") _REPR = re.compile(r"<(?:function|bound method|coroutine|object) [^>]*>") diff --git a/backend/app/db/base.py b/backend/app/db/base.py index 7e741218..1486058c 100644 --- a/backend/app/db/base.py +++ b/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 上拿的会话级咨询锁的号：几个进程同时启动时，建表和迁移一个一个来（同一个号别处别用） diff --git a/backend/app/db/to_postgres.py b/backend/app/db/to_postgres.py index 4dec6a25..bb6093cc 100644 --- a/backend/app/db/to_postgres.py +++ b/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}") diff --git a/backend/app/engine/checkpoints.py b/backend/app/engine/checkpoints.py index 73ab7b7e..919ec46c 100644 --- a/backend/app/engine/checkpoints.py +++ b/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 diff --git a/backend/app/main.py b/backend/app/main.py index c96663a3..00c22b1d 100644 --- a/backend/app/main.py +++ b/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("已停止") diff --git a/backend/tests/test_copy_to_postgres.py b/backend/tests/test_copy_to_postgres.py index b1391432..23a41350 100644 --- a/backend/tests/test_copy_to_postgres.py +++ b/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