fix(persistence): auto-create the Postgres database only when it is missing (#7410)

* fix(persistence): auto-create the Postgres database only when it is missing

init_engine retried bootstrap through CREATE DATABASE whenever the error
text contained "does not exist". Postgres words missing relations,
columns, functions and roles the same way, so a bootstrap that failed
inside an existing database (for example a migration reading a dropped
column) went on to CREATE DATABASE. Startup then failed with
DuplicateDatabaseError, or with "permission denied to create database"
for a role without CREATEDB, and the original error appeared nowhere in
the traceback.

Recognize a missing database by SQLSTATE 3D000 (invalid_catalog_name)
instead, following __cause__ for a wrapped copy, and re-raise every
other bootstrap error unchanged. A live PostgreSQL test covers both the
missing-database path and an existing database whose migration fails.

* docs(changelog): reference #7410 for the Postgres auto-create fix

* fix(persistence): walk orig, __cause__ and __context__ for the missing-database SQLSTATE

Match _is_unique_violation's chain walk, with its cycle guard, so a wrapper
that keeps the driver error only on `orig` or `__context__` still triggers
auto-create. Drop the live test's English message assertion; the SQLSTATE
check already proves the original error surfaced.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

CHANGELOG.md

@@ -1397,6 +1397,14 @@ As of 2026-10-21, the milestone includes **829 merged pull requests**.
 
 ### Fixed
 
+- **persistence:** A PostgreSQL bootstrap failure inside an existing database
+  now reports its own error. The Gateway treated any error mentioning "does not
+  exist", such as a missing column or relation during a migration, as a missing
+  database and ran `CREATE DATABASE`. Startup then failed with `database ...
+  already exists` or `permission denied to create database`, and the original
+  error was missing from the traceback. Only a missing database (SQLSTATE
+  `3D000`) is auto-created now. ([#7410])
+
 - **mcp:** Ordinary users can now install DingTalk, WeCom and HubSpot from
   **Capability Center → Plugins**. Saving one of these bundled
   connections as a non-administrator, including editing or enabling it later,
@@ -13060,3 +13068,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
 [#7362]: https://github.com/bytedance/deer-flow/pull/7362
 [#7364]: https://github.com/bytedance/deer-flow/pull/7364
 [#7400]: https://github.com/bytedance/deer-flow/pull/7400
+[#7410]: https://github.com/bytedance/deer-flow/pull/7410

CHANGELOG_zh.md

@@ -1112,6 +1112,12 @@
 
 ### 修复
 
+- **持久化：** 在已存在的数据库中 PostgreSQL 启动初始化失败时，现在会报告原始错误。
+  此前 Gateway 会把任何包含 "does not exist" 的错误（例如迁移时缺少列或表）当作数据库
+  不存在并执行 `CREATE DATABASE`，启动随之以 `database ... already exists` 或
+  `permission denied to create database` 失败，原始错误也不会出现在 traceback 中。
+  现在只有数据库确实不存在（SQLSTATE `3D000`）时才会自动创建。([#7410])
+
 - **MCP：** 普通用户现在可以在“能力中心 → 插件”中安装钉钉、企业微信和 HubSpot。
   此前非管理员保存这些内置连接（包括之后编辑或启用）会返回
   `500 Internal Server Error`，因为个人连接检查读取了请求模型未声明的 `cwd` 字段。
@@ -10425,3 +10431,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
 [#7362]: https://github.com/bytedance/deer-flow/pull/7362
 [#7364]: https://github.com/bytedance/deer-flow/pull/7364
 [#7400]: https://github.com/bytedance/deer-flow/pull/7400
+[#7410]: https://github.com/bytedance/deer-flow/pull/7410

backend/packages/harness/deerflow/persistence/AGENTS.md

@@ -4,6 +4,8 @@ Postgres bootstrap owns its session-scoped advisory lock until `pg_advisory_unlo
 
 Acquire that lock by polling `pg_try_advisory_lock`, never with a blocking `pg_advisory_lock`: the app engine's asyncpg `command_timeout` bounds every statement, so a blocking acquire fails a second instance's startup whenever a peer's migration outlasts it.
 
+`init_engine` auto-creates a Postgres database only on SQLSTATE `3D000`. Missing relations, columns and roles also "do not exist"; a `CREATE DATABASE` retry would mask them.
+
 When `database.postgres_schema` is configured, both async ORM connections and the synchronous SQLAlchemy connections used by DB-backed custom agents and managed subagents must use the same `search_path`; preserve this invariant when adding another persistence entry point.
 
 Alembic stamp/upgrade workers started inside `bootstrap_schema()` remain owned by the bootstrap critical section until the worker finishes. Drain those `asyncio.to_thread()` calls across host cancellation before releasing the in-process SQLite bootstrap lock or PostgreSQL advisory lock; otherwise another bootstrap can overlap a still-running migration worker.

backend/packages/harness/deerflow/persistence/engine.py

@@ -56,6 +56,34 @@ logger = logging.getLogger(__name__)
 _engine: AsyncEngine | None = None
 _session_factory: async_sessionmaker[AsyncSession] | None = None
 
+# SQLSTATE invalid_catalog_name: the database named in the URL does not exist.
+_MISSING_DATABASE_SQLSTATE = "3D000"
+
+
+def _is_missing_database_error(exc: BaseException) -> bool:
+    """Return whether *exc* means the target PostgreSQL database is absent.
+
+    Postgres words missing relations, columns and roles as "does not exist"
+    too, so the message cannot tell them apart; only the SQLSTATE can.
+    asyncpg raises this one unwrapped from connect(); follow ``orig``,
+    ``__cause__`` and ``__context__`` (as ``_is_unique_violation`` in
+    ``runtime/runs/manager.py`` does) so a wrapped copy is recognized as well.
+    """
+    pending: list[BaseException] = [exc]
+    seen: set[int] = set()
+    while pending:
+        current = pending.pop()
+        if id(current) in seen:
+            continue
+        seen.add(id(current))
+        if getattr(current, "sqlstate", None) == _MISSING_DATABASE_SQLSTATE:
+            return True
+        for attr in ("orig", "__cause__", "__context__"):
+            inner = getattr(current, attr, None)
+            if isinstance(inner, BaseException):
+                pending.append(inner)
+    return False
+
 
 async def _auto_create_postgres_db(url: str) -> None:
     """Connect to the ``postgres`` maintenance DB and CREATE DATABASE.
@@ -218,8 +246,11 @@ async def init_engine(
         await _ensure_postgres_schema()
         await bootstrap_schema(_engine, backend=backend, postgres_schema=postgres_schema)
     except Exception as exc:
-        if backend == "postgres" and "does not exist" in str(exc):
+        if backend == "postgres" and _is_missing_database_error(exc):
             # Database not yet created -- attempt to auto-create it, then retry.
+            # Any other bootstrap failure is re-raised untouched: a
+            # CREATE DATABASE attempt would replace it with "already exists"
+            # or "permission denied to create database".
             await _auto_create_postgres_db(url)
             # Rebuild engine against the now-existing database. The rebuilt
             # engine MUST keep the same connect_args so the retried bootstrap

backend/tests/test_persistence_engine_auto_create_postgres.py

@@ -0,0 +1,94 @@
+"""Live PostgreSQL checks for init_engine's missing-database auto-create.
+
+Opt in with DEERFLOW_TEST_POSTGRES_URL or TEST_POSTGRES_URI (set by the
+backend unit-test CI job). The role must be allowed to create databases.
+"""
+
+from __future__ import annotations
+
+import os
+import uuid
+from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import text
+from sqlalchemy.engine.url import make_url
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from deerflow.persistence import engine as engine_mod
+
+
+def _asyncpg_url(url: str | None) -> str | None:
+    if not url:
+        return url
+    parts = urlsplit(url)
+    scheme = "postgresql+asyncpg" if parts.scheme in {"postgres", "postgresql"} else parts.scheme
+    query = urlencode([(key, value) for key, value in parse_qsl(parts.query, keep_blank_values=True) if key not in {"sslmode", "channel_binding"}])
+    return urlunsplit(parts._replace(scheme=scheme, query=query))
+
+
+POSTGRES_URL = _asyncpg_url(os.getenv("DEERFLOW_TEST_POSTGRES_URL") or os.getenv("TEST_POSTGRES_URI"))
+
+pytestmark = [
+    pytest.mark.asyncio,
+    pytest.mark.skipif(not POSTGRES_URL, reason="set DEERFLOW_TEST_POSTGRES_URL or TEST_POSTGRES_URI to run live PostgreSQL tests"),
+]
+
+
+async def _execute_on_server(*statements: str, database: str | None = None) -> None:
+    url = make_url(POSTGRES_URL or "")
+    if database is not None:
+        url = url.set(database=database)
+    engine = create_async_engine(url, isolation_level="AUTOCOMMIT")
+    try:
+        async with engine.connect() as conn:
+            for statement in statements:
+                await conn.execute(text(statement))
+    finally:
+        await engine.dispose()
+
+
+@pytest_asyncio.fixture
+async def missing_database():
+    name = f"deerflow_autocreate_{uuid.uuid4().hex[:12]}"
+    url = make_url(POSTGRES_URL or "").set(database=name).render_as_string(hide_password=False)
+    try:
+        yield name, url
+    finally:
+        await engine_mod.close_engine()
+        await _execute_on_server(f'DROP DATABASE IF EXISTS "{name}" WITH (FORCE)')
+
+
+async def test_missing_database_is_created_and_bootstrapped(missing_database) -> None:
+    name, url = missing_database
+
+    await engine_mod.init_engine("postgres", url=url)
+
+    async with engine_mod.get_engine().connect() as conn:
+        assert (await conn.execute(text("SELECT current_database()"))).scalar_one() == name
+        assert (await conn.execute(text("SELECT version_num FROM alembic_version"))).scalar_one()
+
+
+async def test_bootstrap_failure_in_an_existing_database_is_not_masked(missing_database) -> None:
+    name, url = missing_database
+    await engine_mod.init_engine("postgres", url=url)
+    await engine_mod.close_engine()
+    # Leave an unguarded migration statement to fail: rewind to 0012 and drop
+    # the column 0013's backfill UPDATE reads.
+    await _execute_on_server(
+        "UPDATE alembic_version SET version_num = '0012_mcp_task_results'",
+        "ALTER TABLE mcp_tasks DROP COLUMN updated_at CASCADE",
+        database=name,
+    )
+
+    with pytest.raises(Exception) as raised:
+        await engine_mod.init_engine("postgres", url=url)
+
+    # Before the fix, this tried CREATE DATABASE and raised DuplicateDatabaseError.
+    sqlstates = []
+    error: BaseException | None = raised.value
+    while error is not None:
+        sqlstates.append(getattr(error, "sqlstate", None))
+        error = error.__cause__
+    assert "42703" in sqlstates, raised.value

backend/tests/test_persistence_engine_postgres_config.py

@@ -16,6 +16,14 @@ from deerflow.config.database_config import DatabaseConfig
 from deerflow.persistence import engine as engine_mod
 
 
+class _PostgresError(Exception):
+    """Driver error carrying a SQLSTATE, as asyncpg's exceptions do."""
+
+    def __init__(self, sqlstate: str, message: str) -> None:
+        super().__init__(message)
+        self.sqlstate = sqlstate
+
+
 def test_postgres_engine_kwargs_include_connection_hardening() -> None:
     kwargs = engine_mod._postgres_engine_kwargs(echo=False, pool_size=5)
 
@@ -216,7 +224,7 @@ async def test_init_engine_postgres_retry_uses_hardened_kwargs(monkeypatch) -> N
     initial_engine.dispose = AsyncMock()
     retry_engine = MagicMock()
     retry_engine.dispose = AsyncMock()
-    bootstrap_schema = AsyncMock(side_effect=[Exception("database does not exist"), None])
+    bootstrap_schema = AsyncMock(side_effect=[_PostgresError("3D000", 'database "deerflow" does not exist'), None])
     auto_create = AsyncMock()
 
     monkeypatch.setitem(sys.modules, "asyncpg", ModuleType("asyncpg"))
@@ -241,6 +249,100 @@ async def test_init_engine_postgres_retry_uses_hardened_kwargs(monkeypatch) -> N
             await engine_mod.close_engine()
 
 
+def _wrapped_by_orig(missing: BaseException) -> BaseException:
+    from sqlalchemy.exc import OperationalError
+
+    return OperationalError("connect", {}, missing)
+
+
+def _wrapped_by_cause(missing: BaseException) -> BaseException:
+    wrapped = RuntimeError("bootstrap failed")
+    wrapped.__cause__ = missing
+    return wrapped
+
+
+def _wrapped_by_context(missing: BaseException) -> BaseException:
+    wrapped = RuntimeError("bootstrap failed")
+    wrapped.__context__ = missing
+    return wrapped
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("wrap", [_wrapped_by_orig, _wrapped_by_cause, _wrapped_by_context], ids=["orig", "cause", "context"])
+async def test_init_engine_postgres_retries_a_wrapped_missing_database_error(monkeypatch, wrap) -> None:
+    url = "postgresql+asyncpg://user:password@localhost/deerflow"
+    engine = MagicMock()
+    engine.dispose = AsyncMock()
+    wrapped = wrap(_PostgresError("3D000", 'database "deerflow" does not exist'))
+    bootstrap_schema = AsyncMock(side_effect=[wrapped, None])
+    auto_create = AsyncMock()
+
+    monkeypatch.setitem(sys.modules, "asyncpg", ModuleType("asyncpg"))
+    with (
+        patch.object(engine_mod, "create_async_engine", return_value=engine),
+        patch.object(engine_mod, "async_sessionmaker", return_value=MagicMock()),
+        patch.object(engine_mod, "_auto_create_postgres_db", new=auto_create),
+        patch("deerflow.persistence.bootstrap.bootstrap_schema", new=bootstrap_schema),
+    ):
+        try:
+            await engine_mod.init_engine(backend="postgres", url=url)
+
+            auto_create.assert_awaited_once_with(url)
+            assert bootstrap_schema.await_count == 2
+        finally:
+            await engine_mod.close_engine()
+
+
+def _cyclic_chain() -> BaseException:
+    # The classifier must terminate on a chain that loops back on itself.
+    error = _PostgresError("42703", 'column "updated_at" does not exist')
+    wrapper = RuntimeError("bootstrap failed")
+    wrapper.__cause__ = error
+    error.__context__ = wrapper
+    return error
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "error",
+    [
+        pytest.param(_PostgresError("42703", 'column "updated_at" does not exist'), id="undefined-column"),
+        pytest.param(_PostgresError("42P01", 'relation "runs" does not exist'), id="undefined-table"),
+        pytest.param(_PostgresError("42883", "function gen_random_uuid() does not exist"), id="undefined-function"),
+        pytest.param(_PostgresError("28000", 'role "deerflow" does not exist'), id="undefined-role"),
+        pytest.param(Exception('database "deerflow" does not exist'), id="no-sqlstate"),
+        pytest.param(_cyclic_chain(), id="cyclic-chain"),
+    ],
+)
+async def test_init_engine_postgres_reraises_other_does_not_exist_errors(monkeypatch, error) -> None:
+    # Postgres words all of these "does not exist". Only a missing database
+    # may trigger CREATE DATABASE; anything else must surface unchanged
+    # instead of as "already exists" or "permission denied to create database".
+    url = "postgresql+asyncpg://user:password@localhost/deerflow"
+    engine = MagicMock()
+    engine.dispose = AsyncMock()
+    bootstrap_schema = AsyncMock(side_effect=error)
+    auto_create = AsyncMock()
+
+    monkeypatch.setitem(sys.modules, "asyncpg", ModuleType("asyncpg"))
+    with (
+        patch.object(engine_mod, "create_async_engine", return_value=engine) as create_engine,
+        patch.object(engine_mod, "async_sessionmaker", return_value=MagicMock()),
+        patch.object(engine_mod, "_auto_create_postgres_db", new=auto_create),
+        patch("deerflow.persistence.bootstrap.bootstrap_schema", new=bootstrap_schema),
+    ):
+        try:
+            with pytest.raises(type(error)) as raised:
+                await engine_mod.init_engine(backend="postgres", url=url)
+
+            assert raised.value is error
+            auto_create.assert_not_awaited()
+            create_engine.assert_called_once()
+            bootstrap_schema.assert_awaited_once()
+        finally:
+            await engine_mod.close_engine()
+
+
 @pytest.mark.asyncio
 async def test_init_engine_sqlite_omits_postgres_kwargs_and_keeps_wal_listener(tmp_path) -> None:
     url = f"sqlite+aiosqlite:///{tmp_path / 'deerflow.db'}"

backend/tests/test_persistence_retry_dispose_cancellation.py

@@ -31,8 +31,11 @@ async def test_postgres_auto_create_retry_drains_provisional_engine_dispose_acro
     monkeypatch.setattr(engine_mod, "create_async_engine", lambda *args, **kwargs: engine)
     monkeypatch.setattr(engine_mod, "async_sessionmaker", lambda *args, **kwargs: object())
 
+    class MissingDatabaseError(Exception):
+        sqlstate = "3D000"
+
     async def fail_bootstrap(*args, **kwargs) -> None:
-        raise RuntimeError("database does not exist")
+        raise MissingDatabaseError('database "deerflow" does not exist')
 
     async def fake_auto_create(_url: str) -> None:
         return None