From 268eb55f6c297c13935906b0c10033a0db1cbaa0 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Wed, 14 Oct 2026 18:56:25 +0900 Subject: [PATCH] fix(persistence): compare safe_add_column drift as the database spells it (#6865) * fix(persistence): compare safe_add_column drift as the database spells it safe_add_column's drift check compared the model's generic type and default spellings (String -> STRING, BigInteger -> BIGINTEGER, server_default="0" -> "DefaultClause('0', ...)", sa.false()) with the reflected dialect spellings (VARCHAR, BIGINT, Postgres TIMESTAMP, '0'). Only JSON/JSONB ever matched, so every column a revision found already present was reported as drift: - any database created before 0023 logged a runs.change_seq warning on upgrade, because 0025 re-runs 0023's safe_add_column; - a schema whose head stamp was lost (the empty branch commits create_all before stamping) logged 29 warnings on SQLite and 36 on Postgres, burying the real TEXT-vs-JSON drift the check exists to surface. Render both sides through the migration connection's dialect before comparing: types via TypeEngine.compile (parameters stripped, the Postgres time-zone suffix kept), defaults by unwrapping DefaultClause, compiling SQL expressions and stripping one layer of string quoting. Types the dialect cannot render (SQLite NullType) stay "missing info". * docs(changelog): reference #6865 for the drift comparator fix * fix(persistence): keep an empty-string default distinct from no default Stripping one layer of quotes turned a reflected DEFAULT '' into "", and `return text or None` then collapsed it to None. A desired server_default="" (passed as a bare "") also normalized to None, so _check_column_drift stopped warning when a column had no default but the model wanted '' (or the reverse). Render a plain-string DefaultClause argument as the quoted literal the DDL emits, and return the unquoted value even when it is empty. None now means only "no default". --- CHANGELOG.md | 10 + CHANGELOG_zh.md | 8 + .../deerflow/persistence/migrations/AGENTS.md | 2 +- .../persistence/migrations/_helpers.py | 121 ++++++++--- backend/tests/test_persistence_bootstrap.py | 190 ++++++++++++++++++ 5 files changed, 298 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9cf6f5198e..0a4a1a32ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5917,6 +5917,15 @@ This release closes that milestone with **621 merged pull requests**. 80 such requests failed. The same race in IM conversation binding, such as a first message arriving together with `/new`, no longer fails the message. The losing write now updates the row the other one created. ([#6852]) +- **persistence:** Startup schema upgrades no longer log a false + `safe_add_column ... drifts from the model definition` warning for columns + that already match the model. The check compared the model's generic type + and default names (`STRING`, `BIGINTEGER`, `DefaultClause('0')`) with the + database's own (`VARCHAR`, `BIGINT`, `'0'`), so every Postgres or SQLite + database created before revision 0023 logged one on upgrade, and a schema + whose head stamp was lost logged 29 (SQLite) or 36 (Postgres), burying the + real type mismatch the warning exists to catch. Both sides are now compared + as the connected database spells them. ([#6865]) ### Security @@ -11701,3 +11710,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6843]: https://github.com/bytedance/deer-flow/pull/6843 [#6844]: https://github.com/bytedance/deer-flow/pull/6844 [#6852]: https://github.com/bytedance/deer-flow/pull/6852 +[#6865]: https://github.com/bytedance/deer-flow/pull/6865 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index a761d0ba2e..93fc0b591c 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -4494,6 +4494,13 @@ `(thread, run, user)` 唯一约束；在 Postgres 上，80 个这样的请求中有 26 个失败。 IM 会话绑定中的同类竞争（例如首条消息与 `/new` 同时到达）也不再导致消息处理失败。 落后的写入现在会更新另一方创建的记录。([#6852]) +- **持久化：** 启动时的 schema 升级不再为已与模型一致的列记录虚假的 + `safe_add_column ... drifts from the model definition` 警告。该检查拿模型的通用类型名 + 和默认值写法（`STRING`、`BIGINTEGER`、`DefaultClause('0')`）去比较数据库自己的写法 + （`VARCHAR`、`BIGINT`、`'0'`），因此 revision 0023 之前创建的 Postgres 或 SQLite + 数据库升级时都会记录一条，丢失 head stamp 的 schema 会记录 29 条（SQLite）或 + 36 条（Postgres），淹没了这个警告本应发现的真实类型不一致。现在两侧都按所连接 + 数据库的写法进行比较。([#6865]) ### 安全 @@ -9337,3 +9344,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6843]: https://github.com/bytedance/deer-flow/pull/6843 [#6844]: https://github.com/bytedance/deer-flow/pull/6844 [#6852]: https://github.com/bytedance/deer-flow/pull/6852 +[#6865]: https://github.com/bytedance/deer-flow/pull/6865 diff --git a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md index 731307470d..c3f0326019 100644 --- a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md @@ -15,7 +15,7 @@ DeerFlow's application tables (`runs`, `threads_meta`, `feedback`, `users`, `run | `0019_thread_incarnations` missing local tables/columns | refuse startup; offline recovery required | | unknown revision, empty version table, or multiple version rows | fail closed and refuse to start | -The legacy branch handles pre-alembic databases that already have at least one DeerFlow-owned table. `create_all` runs first because stamping at `0001_baseline` makes alembic skip the baseline's own `create_table` DDL on the subsequent upgrade — so any baseline table introduced into `Base.metadata` after the user's DB was first provisioned (e.g. the `channel_*` tables from PR #1930 for users upgrading across multiple releases) would otherwise never be created, and the first request hitting that table would 500 with `no such table`. The backfill is **restricted to `_BASELINE_TABLE_NAMES`** so it does not also create tables that future revisions introduce — those revisions' own `op.create_table` would otherwise fail with `relation already exists`. A guard test pins `_BASELINE_TABLE_NAMES` against `0001_baseline.upgrade()`'s actual output, so editing 0001 to add or remove a table forces a matching update to the constant. Column-level shape (pre-#3658 vs post-#3658 vs manual-ALTER for `token_usage_by_model`) is answered by each `versions/*.py` revision via the idempotent helpers in `migrations/_helpers.py` (`safe_add_column` / `safe_drop_column`) which no-op when the change is already present and `logger.warning` on shape drift. **Adding a new ORM column / table only requires a new revision file — no edit to `bootstrap.py` is needed** *unless* the new revision adds a new baseline table (rare; only happens when a new model is part of the baseline rather than introduced by its own revision). +The legacy branch handles pre-alembic databases that already have at least one DeerFlow-owned table. `create_all` runs first because stamping at `0001_baseline` makes alembic skip the baseline's own `create_table` DDL on the subsequent upgrade — so any baseline table introduced into `Base.metadata` after the user's DB was first provisioned (e.g. the `channel_*` tables from PR #1930 for users upgrading across multiple releases) would otherwise never be created, and the first request hitting that table would 500 with `no such table`. The backfill is **restricted to `_BASELINE_TABLE_NAMES`** so it does not also create tables that future revisions introduce — those revisions' own `op.create_table` would otherwise fail with `relation already exists`. A guard test pins `_BASELINE_TABLE_NAMES` against `0001_baseline.upgrade()`'s actual output, so editing 0001 to add or remove a table forces a matching update to the constant. Column-level shape (pre-#3658 vs post-#3658 vs manual-ALTER for `token_usage_by_model`) is answered by each `versions/*.py` revision via the idempotent helpers in `migrations/_helpers.py` (`safe_add_column` / `safe_drop_column`) which no-op when the change is already present and `logger.warning` on shape drift (both sides rendered through the connection's dialect, so model-shaped columns stay quiet). **Adding a new ORM column / table only requires a new revision file — no edit to `bootstrap.py` is needed** *unless* the new revision adds a new baseline table (rare; only happens when a new model is part of the baseline rather than introduced by its own revision). The empty-DB path keeps using `create_all` because `Base.metadata` is the only authoritative schema source — `create_all` renders both SQLite (JSON, type affinity) and Postgres (JSONB, partial indexes) correctly without anyone having to keep a hand-written baseline in lockstep. `0001_baseline.upgrade()` is therefore almost never executed in practice; it exists as a stamp target + chain root. diff --git a/backend/packages/harness/deerflow/persistence/migrations/_helpers.py b/backend/packages/harness/deerflow/persistence/migrations/_helpers.py index 9428a8a8ea..323e049d11 100644 --- a/backend/packages/harness/deerflow/persistence/migrations/_helpers.py +++ b/backend/packages/harness/deerflow/persistence/migrations/_helpers.py @@ -27,11 +27,15 @@ JSON`` without ``NOT NULL DEFAULT '{}'``, or the wrong-type variant would produce on a fresh DB. To surface that silent drift, ``safe_add_column`` compares the existing column's ``nullable`` / ``server_default`` / ``type`` against the desired ``sa.Column`` and emits ``logger.warning`` on mismatch. -Type comparison goes through ``_type_equivalent``, which treats known -dialect-synonym pairs (e.g. ``JSON`` vs ``JSONB``) as equivalent to avoid -false positives while still catching wholesale type mismatches like -``TEXT`` vs ``JSON``. We do not auto-repair -- a warning is enough for -operators to notice and decide. +Both sides are rendered through the migration connection's dialect before +they are compared: the model says ``sa.String`` / ``sa.BigInteger`` / +``server_default="0"`` while reflection reports ``VARCHAR`` / ``BIGINT`` / +``'0'``, so comparing the raw Python objects would warn on every column a +revision finds already present. Type comparison goes through +``_type_equivalent``, which treats known dialect-synonym pairs (e.g. ``JSON`` +vs ``JSONB``) as equivalent to avoid false positives while still catching +wholesale type mismatches like ``TEXT`` vs ``JSON``. We do not auto-repair -- +a warning is enough for operators to notice and decide. """ from __future__ import annotations @@ -48,21 +52,32 @@ def _inspector() -> sa.Inspector: return sa.inspect(op.get_bind()) -def _normalize_default(value: object) -> str | None: +def _normalize_default(value: object, dialect: sa.engine.Dialect | None = None) -> str | None: """Normalize a server-default value for cross-source comparison. The desired value comes from ``sa.Column.server_default`` (a - ``DefaultClause`` / ``TextClause`` literal, ``None``, or a Python literal); - the reflected value comes from ``Inspector.get_columns()['default']`` as a - dialect-rendered string. Strip outer parens / whitespace / Postgres-style - type casts so textually-equivalent forms compare equal across dialects. + ``DefaultClause`` wrapping a string, ``TextClause`` or SQL expression such + as ``sa.false()``; ``None``; or a Python literal); the reflected value + comes from ``Inspector.get_columns()['default']`` as a dialect-rendered + string. Expressions are compiled with *dialect* (``sa.false()`` is ``0`` + on SQLite and ``false`` on Postgres), then outer parens / whitespace / + Postgres-style type casts / one layer of single quotes are stripped so + ``server_default="0"`` (DDL ``DEFAULT '0'``), SQLite's reflected ``'0'`` + and Postgres's reflected ``0`` or ``'0'::bigint`` all compare equal. + Returns ``None`` only when there is no default: a quoted empty literal + (``server_default=""``, reflected ``''``) stays ``""``. """ if value is None: return None + if isinstance(value, sa.schema.DefaultClause): + value = value.arg + if isinstance(value, str): + # DDL renders a plain-string default as a quoted literal. + value = "'" + value.replace("'", "''") + "'" if isinstance(value, sa.sql.elements.TextClause): text = value.text - elif isinstance(value, sa.schema.DefaultClause) and isinstance(value.arg, sa.sql.elements.TextClause): - text = value.arg.text + elif isinstance(value, sa.sql.elements.ClauseElement): + text = str(value.compile(dialect=dialect, compile_kwargs={"literal_binds": True})) else: text = str(value) text = text.strip() @@ -72,23 +87,62 @@ def _normalize_default(value: object) -> str | None: # Strip Postgres-style type casts like ``'{}'::jsonb``. if "::" in text: text = text.split("::", 1)[0].strip() + # Strip one layer of SQL string quoting: a plain-string ``server_default`` + # renders as a quoted literal, and dialects disagree on whether a numeric + # column reflects it back quoted. The unquoted value is returned even + # when empty: ``DEFAULT ''`` is a default, unlike no default at all. + if len(text) >= 2 and text.startswith("'") and text.endswith("'"): + return text[1:-1].replace("''", "'") return text or None -def _normalize_type(value: object) -> str: +def _strip_type_parameters(rendered: str) -> str: + """Drop every parenthesized group, keeping words around it. + + ``TIMESTAMP(3) WITH TIME ZONE`` keeps its ``WITH TIME ZONE`` suffix, and + nested reprs such as ``JSONB(astext_type=Text())`` lose the whole group. + """ + out: list[str] = [] + depth = 0 + for char in rendered: + if char == "(": + depth += 1 + elif char == ")": + depth = max(depth - 1, 0) + elif depth == 0: + out.append(char) + return " ".join("".join(out).split()) + + +def _normalize_type(value: object, dialect: sa.engine.Dialect | None = None) -> str: """Normalize a SQLAlchemy ``TypeEngine`` (or reflected type) for comparison. - Returns the upper-cased type-class name with any parameters stripped - (e.g. ``JSON()`` → ``"JSON"``, ``VARCHAR(255)`` → ``"VARCHAR"``). Length - parameters are dropped on purpose: drift warnings target wholesale type + A ``TypeEngine`` is rendered as the DDL *dialect* would emit for it, so + the model's generic ``String(128)`` / ``BigInteger()`` / + ``DateTime(timezone=True)`` and the reflected ``VARCHAR(128)`` / + ``BIGINT`` / Postgres ``TIMESTAMP(timezone=True)`` land on the same name. + Strings are taken as already-rendered type names. Returns the upper-cased + name with any parameters stripped (e.g. ``VARCHAR(255)`` → ``"VARCHAR"``, + Postgres ``TIMESTAMP WITH TIME ZONE`` stays whole). Length parameters are + dropped on purpose: drift warnings target wholesale type misconfigurations (the JSON-vs-TEXT review case), not dialect-rendered - size defaults. An empty string signals "missing info" -- callers should - not equality-check empty strings. + size defaults. An empty string signals "missing info" (no type, or one + the dialect cannot render, such as SQLite's ``NullType`` for an + unrecognized declared type) -- callers should not equality-check empty + strings. """ if value is None: return "" - s = value if isinstance(value, str) else repr(value) - return s.upper().split("(", 1)[0].strip() + if isinstance(value, str): + rendered = value + elif isinstance(value, sa.types.TypeEngine): + try: + rendered = str(value.compile(dialect=dialect)) + except sa.exc.CompileError: + return "" + else: + rendered = repr(value) + return _strip_type_parameters(rendered.upper()) # Known dialect-synonym pairs that must NOT fire a type-drift warning. @@ -104,14 +158,15 @@ def _normalize_type(value: object) -> str: _EQUIVALENT_TYPE_FAMILIES: tuple[frozenset[str], ...] = (frozenset({"JSON", "JSONB"}),) -def _type_equivalent(actual: object, desired: object) -> bool: +def _type_equivalent(actual: object, desired: object, dialect: sa.engine.Dialect | None = None) -> bool: """True if *actual* and *desired* are the same type or a known equivalent. - Returns True when either side is missing reflected info so missing-data - cases never false-positive into a noisy warning. + Both sides are rendered with *dialect* (see ``_normalize_type``). Returns + True when either side is missing reflected info so missing-data cases + never false-positive into a noisy warning. """ - a = _normalize_type(actual) - d = _normalize_type(desired) + a = _normalize_type(actual, dialect) + d = _normalize_type(desired, dialect) if not a or not d: return True if a == d: @@ -120,7 +175,7 @@ def _type_equivalent(actual: object, desired: object) -> bool: return any(pair <= fam for fam in _EQUIVALENT_TYPE_FAMILIES) -def _check_column_drift(table: str, desired: sa.Column, actual: dict) -> None: +def _check_column_drift(table: str, desired: sa.Column, actual: dict, dialect: sa.engine.Dialect | None = None) -> None: """Warn if an existing column's attributes diverge from the desired model. Equality is checked on ``nullable`` and ``server_default`` directly, and @@ -128,7 +183,9 @@ def _check_column_drift(table: str, desired: sa.Column, actual: dict) -> None: pairs like ``JSON`` vs ``JSONB`` as equivalent). The reflected and desired type reprs are also echoed in the warning payload regardless of whether type was the failing dimension, so an operator triaging the log - line sees the type context at a glance. + line sees the type context at a glance. *dialect* is the migration + connection's, so both sides are rendered the way that database spells + them. """ diffs: list[str] = [] @@ -137,13 +194,13 @@ def _check_column_drift(table: str, desired: sa.Column, actual: dict) -> None: if desired_nullable != actual_nullable: diffs.append(f"nullable actual={actual_nullable} desired={desired_nullable}") - desired_default = _normalize_default(desired.server_default) - actual_default = _normalize_default(actual.get("default")) + desired_default = _normalize_default(desired.server_default, dialect) + actual_default = _normalize_default(actual.get("default"), dialect) if desired_default != actual_default: diffs.append(f"server_default actual={actual_default!r} desired={desired_default!r}") - if not _type_equivalent(actual.get("type"), desired.type): - diffs.append(f"type actual={_normalize_type(actual.get('type'))!r} desired={_normalize_type(desired.type)!r}") + if not _type_equivalent(actual.get("type"), desired.type, dialect): + diffs.append(f"type actual={_normalize_type(actual.get('type'), dialect)!r} desired={_normalize_type(desired.type, dialect)!r}") if diffs: logger.warning( @@ -171,7 +228,7 @@ def safe_add_column(table: str, column: sa.Column) -> None: return existing = {c["name"]: c for c in insp.get_columns(table)} if column.name in existing: - _check_column_drift(table, column, existing[column.name]) + _check_column_drift(table, column, existing[column.name], insp.dialect) return with op.batch_alter_table(table) as batch: batch.add_column(column) diff --git a/backend/tests/test_persistence_bootstrap.py b/backend/tests/test_persistence_bootstrap.py index c12b083e9c..7299de3ad3 100644 --- a/backend/tests/test_persistence_bootstrap.py +++ b/backend/tests/test_persistence_bootstrap.py @@ -22,11 +22,14 @@ table-level backfill works. from __future__ import annotations import asyncio +import os +import uuid from pathlib import Path import pytest import sqlalchemy as sa from sqlalchemy.ext.asyncio import create_async_engine +from support.postgres import asyncpg_test_url # Pre-import models so Base.metadata is populated before bootstrap reads it. import deerflow.persistence.models # noqa: F401 @@ -42,6 +45,7 @@ from deerflow.persistence.bootstrap import ( bootstrap_schema, ) from deerflow.persistence.migrations._helpers import _normalize_default +from deerflow.persistence.postgres_schema import build_asyncpg_connect_args # Mark only async tests via the decorator below; module-level pytestmark would # spuriously warn for the sync ``TestDecideState`` cases. @@ -308,6 +312,8 @@ async def test_legacy_with_manual_workaround_column_warns_on_drift( # JSON ≈ JSON, so the equivalence check must NOT produce a "type" diff # entry here -- that would be a false positive on the matching-type case. assert "type actual=" not in msg, f"unexpected type drift on matching JSON column: {msg!r}" + helper_warnings = [r.getMessage() for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers"] + assert helper_warnings == [msg], f"expected only the token_usage_by_model drift, got: {helper_warnings!r}" finally: await engine.dispose() @@ -348,10 +354,99 @@ async def test_legacy_with_wrong_type_workaround_warns_on_type_drift( # Nullability + server_default match the model -- no other diffs. assert "nullable" not in msg, f"unexpected nullability drift on matching column: {msg!r}" assert "server_default" not in msg, f"unexpected server_default drift on matching column: {msg!r}" + # The seed is the full model schema minus this one column, so every + # other column a revision finds already present matches the model and + # the real drift must not be buried among false positives. + helper_warnings = [r.getMessage() for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers"] + assert helper_warnings == [msg], f"expected only the token_usage_by_model drift, got: {helper_warnings!r}" finally: await engine.dispose() +# --------------------------------------------------------------------------- +# No false positives: a column a revision finds already present in the model's +# own shape must stay quiet on both dialects. The model spells types and +# defaults generically (``sa.String``, ``sa.BigInteger``, ``sa.DateTime``, +# ``server_default="0"``, ``sa.false()``) while reflection reports the +# dialect's spelling (``VARCHAR``, ``BIGINT``, Postgres ``TIMESTAMP``, ``'0'``), +# so the comparator must render the model side through the dialect first. +# --------------------------------------------------------------------------- + + +async def _drift_engine(tmp_path: Path, backend: str): + if backend == "sqlite": + return create_async_engine(_url(tmp_path)), "" + uri = os.environ.get("TEST_POSTGRES_URI") + if not uri: + pytest.skip("requires TEST_POSTGRES_URI (real Postgres drift reflection)") + schema = f"drift_{uuid.uuid4().hex}" + engine = create_async_engine(asyncpg_test_url(uri), connect_args=build_asyncpg_connect_args(schema)) + async with engine.begin() as conn: + await conn.execute(sa.text(f'CREATE SCHEMA "{schema}"')) + return engine, schema + + +def _seed_at_revision(revision: str, *, drop_version_row: bool): + async def seed(engine, schema: str) -> None: + await asyncio.to_thread(_upgrade, _get_alembic_config(engine, postgres_schema=schema), revision) + if drop_version_row: + async with engine.begin() as conn: + await conn.execute(sa.text("DROP TABLE alembic_version")) + + return seed + + +async def _seed_model_schema_without_version_row(engine, schema: str) -> None: + # The empty branch commits ``create_all`` before it stamps head; a process + # that dies in between leaves exactly this, and the next start takes the + # legacy branch over columns every revision finds already present. + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + +async def _seed_model_schema_with_text_token_usage(engine, schema: str) -> None: + await _seed_model_schema_without_version_row(engine, schema) + async with engine.begin() as conn: + await conn.execute(sa.text("ALTER TABLE runs DROP COLUMN token_usage_by_model")) + await conn.execute(sa.text("ALTER TABLE runs ADD COLUMN token_usage_by_model TEXT NOT NULL DEFAULT '{}'")) + + +@asyncio_test +@pytest.mark.parametrize("backend", ["sqlite", "postgres"]) +@pytest.mark.parametrize( + ("seed", "expected_drift"), + [ + pytest.param(_seed_at_revision(BASELINE, drop_version_row=True), [], id="pre_alembic_baseline"), + # 0025 re-runs 0023's ``safe_add_column`` for ``runs.change_seq``. + pytest.param(_seed_at_revision("0022_scheduled_occurrence_seq", drop_version_row=False), [], id="upgrade_across_change_seq_repair"), + pytest.param(_seed_model_schema_without_version_row, [], id="model_schema_without_version_row"), + pytest.param(_seed_model_schema_with_text_token_usage, ["runs.token_usage_by_model"], id="real_drift_only"), + ], +) +async def test_bootstrap_warns_only_on_columns_that_really_drift( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + backend: str, + seed, + expected_drift: list[str], +) -> None: + engine, schema = await _drift_engine(tmp_path, backend) + try: + await seed(engine, schema) + + with caplog.at_level("WARNING", logger="deerflow.persistence.migrations._helpers"): + await bootstrap_schema(engine, backend=backend, postgres_schema=schema) + + assert await _alembic_version(engine) == HEAD + warned = [r.getMessage() for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers"] + assert [m.removeprefix("safe_add_column: ").split(" ", 1)[0] for m in warned] == expected_drift, f"unexpected drift warnings: {warned!r}" + finally: + if schema: + async with engine.begin() as conn: + await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + await engine.dispose() + + # --------------------------------------------------------------------------- # _type_equivalent unit tests: pin the JSON/JSONB equivalence so Postgres # dialect synonyms stay quiet, and pin the TEXT/JSON divergence so the @@ -397,6 +492,101 @@ def test_type_equivalent_returns_true_on_missing_info() -> None: assert _type_equivalent("", "JSON") is True +def test_type_equivalent_renders_model_types_through_the_dialect() -> None: + """The model's generic types meet reflection's dialect types on equal terms.""" + from sqlalchemy.dialects import postgresql, sqlite + + from deerflow.persistence.migrations._helpers import _type_equivalent + + sqlite_dialect = sqlite.dialect() + postgres_dialect = postgresql.dialect() + # Reflected (actual) on the left, model (desired) on the right. + assert _type_equivalent(sa.VARCHAR(length=128), sa.String(length=128), sqlite_dialect) is True + assert _type_equivalent(sa.BIGINT(), sa.BigInteger(), sqlite_dialect) is True + assert _type_equivalent(sqlite.DATETIME(), sa.DateTime(timezone=True), sqlite_dialect) is True + assert _type_equivalent(sa.VARCHAR(length=64), sa.String(length=64), postgres_dialect) is True + assert _type_equivalent(sa.BIGINT(), sa.BigInteger(), postgres_dialect) is True + assert _type_equivalent(postgresql.TIMESTAMP(timezone=True), sa.DateTime(timezone=True), postgres_dialect) is True + assert _type_equivalent(postgresql.JSONB(), sa.JSON(), postgres_dialect) is True + # Precision is a parameter like length; the time-zone suffix is not. + assert _type_equivalent(postgresql.TIMESTAMP(timezone=True, precision=3), sa.DateTime(timezone=True), postgres_dialect) is True + assert _type_equivalent(postgresql.TIMESTAMP(timezone=False, precision=3), sa.DateTime(timezone=True), postgres_dialect) is False + # Rendering both sides must not hide a real mismatch. + assert _type_equivalent(sa.TEXT(), sa.JSON(), sqlite_dialect) is False + assert _type_equivalent(sa.TEXT(), sa.JSON(), postgres_dialect) is False + assert _type_equivalent(postgresql.TIMESTAMP(timezone=False), sa.DateTime(timezone=True), postgres_dialect) is False + assert _type_equivalent(sa.INTEGER(), sa.BigInteger(), postgres_dialect) is False + # SQLite reflects an unrecognized declared type as NullType, which no + # dialect can render: missing info, not drift. + assert _type_equivalent(sa.types.NullType(), sa.String(length=64), sqlite_dialect) is True + + +def test_normalize_default_renders_model_defaults_like_reflection() -> None: + """``server_default="0"`` / ``sa.false()`` compare equal to how each dialect reflects them.""" + from sqlalchemy.dialects import postgresql, sqlite + + sqlite_dialect = sqlite.dialect() + postgres_dialect = postgresql.dialect() + string_zero = sa.Column("c", sa.Integer(), server_default="0").server_default + false_default = sa.Column("c", sa.Boolean(), server_default=sa.false()).server_default + + # SQLite reflects ``DEFAULT '0'`` quoted; Postgres reflects it bare on an + # integer column and as ``'0'::bigint`` on a bigint one. + assert _normalize_default(string_zero, sqlite_dialect) == _normalize_default("'0'", sqlite_dialect) == "0" + assert _normalize_default(string_zero, postgres_dialect) == _normalize_default("0", postgres_dialect) == "0" + assert _normalize_default("'0'::bigint", postgres_dialect) == "0" + assert _normalize_default(false_default, sqlite_dialect) == _normalize_default("0", sqlite_dialect) + assert _normalize_default(false_default, postgres_dialect) == _normalize_default("false", postgres_dialect) + assert _normalize_default(sa.text("'{}'")) == _normalize_default("'{}'::jsonb") == "{}" + # A different default still differs. + assert _normalize_default(string_zero, sqlite_dialect) != _normalize_default("'1'", sqlite_dialect) + assert _normalize_default(false_default, postgres_dialect) != _normalize_default("true", postgres_dialect) + + # An explicit empty-string default is a default, not the absence of one. + empty_string = sa.Column("c", sa.Text(), server_default="").server_default + assert _normalize_default(empty_string, sqlite_dialect) == _normalize_default("''", sqlite_dialect) == "" + assert _normalize_default("''::text", postgres_dialect) == "" + assert _normalize_default(None) is None + + +@asyncio_test +@pytest.mark.parametrize("backend", ["sqlite", "postgres"]) +async def test_empty_string_default_stays_distinct_from_no_default( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, + backend: str, +) -> None: + """``DEFAULT ''`` and no default drift from each other, through real reflection.""" + from deerflow.persistence.migrations._helpers import _check_column_drift + + engine, schema = await _drift_engine(tmp_path, backend) + try: + async with engine.begin() as conn: + await conn.execute(sa.text("CREATE TABLE drift_defaults (empty_default TEXT DEFAULT '', no_default TEXT)")) + + def check(sync_conn) -> None: + inspector = sa.inspect(sync_conn) + columns = {column["name"]: column for column in inspector.get_columns("drift_defaults")} + _check_column_drift("drift_defaults", sa.Column("no_default", sa.Text(), server_default=""), columns["no_default"], inspector.dialect) + _check_column_drift("drift_defaults", sa.Column("empty_default", sa.Text()), columns["empty_default"], inspector.dialect) + # Matching empty-string defaults stay quiet. + _check_column_drift("drift_defaults", sa.Column("empty_default", sa.Text(), server_default=""), columns["empty_default"], inspector.dialect) + + with caplog.at_level("WARNING", logger="deerflow.persistence.migrations._helpers"): + async with engine.connect() as conn: + await conn.run_sync(check) + + warned = [r.getMessage() for r in caplog.records if r.levelname == "WARNING" and r.name == "deerflow.persistence.migrations._helpers"] + assert len(warned) == 2, f"expected exactly the two mismatched defaults, got: {warned!r}" + assert "drift_defaults.no_default" in warned[0] and "server_default actual=None desired=''" in warned[0], warned[0] + assert "drift_defaults.empty_default" in warned[1] and "server_default actual='' desired=None" in warned[1], warned[1] + finally: + if schema: + async with engine.begin() as conn: + await conn.execute(sa.text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')) + await engine.dispose() + + # --------------------------------------------------------------------------- # Branch 4: versioned DB # ---------------------------------------------------------------------------