mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-25 22:16:19 +00:00
fix(persistence): apply postgres schema to sync agent stores (#5678)
Co-authored-by: NEEDI <298523066+sherxlg-gif@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
9ae1e585bb
commit
0627de2bcc
@ -309,6 +309,14 @@ class DatabaseConfig(BaseModel):
|
|||||||
return f"sqlite:///{self.sqlite_path}"
|
return f"sqlite:///{self.sqlite_path}"
|
||||||
if self.backend == "postgres":
|
if self.backend == "postgres":
|
||||||
url = self.postgres_url
|
url = self.postgres_url
|
||||||
|
if self.postgres_schema:
|
||||||
|
# The synchronous agent stores use psycopg via SQLAlchemy,
|
||||||
|
# so they need the same search_path as the async ORM engine
|
||||||
|
# and LangGraph stores. Keep the option merge in the shared
|
||||||
|
# PostgreSQL helper so existing libpq options are preserved.
|
||||||
|
from deerflow.persistence.postgres_schema import dsn_with_search_path
|
||||||
|
|
||||||
|
url = dsn_with_search_path(url, self.postgres_schema)
|
||||||
if url.startswith("postgresql+asyncpg://"):
|
if url.startswith("postgresql+asyncpg://"):
|
||||||
url = url.replace("postgresql+asyncpg://", "postgresql+psycopg://", 1)
|
url = url.replace("postgresql+asyncpg://", "postgresql+psycopg://", 1)
|
||||||
elif url.startswith("postgresql://"):
|
elif url.startswith("postgresql://"):
|
||||||
|
|||||||
@ -1,3 +1,5 @@
|
|||||||
# Persistence lifecycle
|
# Persistence lifecycle
|
||||||
|
|
||||||
Postgres bootstrap owns its session-scoped advisory lock until `pg_advisory_unlock` completes. Drain that unlock across host cancellation before leaving the SQLAlchemy connection context; repeated cancellation must not return a pooled session while it still holds the bootstrap mutex. Ordinary database errors remain best-effort and are logged.
|
Postgres bootstrap owns its session-scoped advisory lock until `pg_advisory_unlock` completes. Drain that unlock across host cancellation before leaving the SQLAlchemy connection context; repeated cancellation must not return a pooled session while it still holds the bootstrap mutex. Ordinary database errors remain best-effort and are logged.
|
||||||
|
|
||||||
|
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.
|
||||||
|
|||||||
@ -7,9 +7,9 @@ uses expect different mechanisms:
|
|||||||
- **asyncpg** (app ORM engine): only honours ``server_settings`` passed
|
- **asyncpg** (app ORM engine): only honours ``server_settings`` passed
|
||||||
via SQLAlchemy ``connect_args``. It does not understand libpq's
|
via SQLAlchemy ``connect_args``. It does not understand libpq's
|
||||||
``options=-c ...`` syntax.
|
``options=-c ...`` syntax.
|
||||||
- **psycopg** (LangGraph checkpointer/store): uses the libpq
|
- **psycopg** (LangGraph checkpointer/store and synchronous agent stores): uses
|
||||||
``options=-c search_path=...`` connection parameter, either as a pool
|
the libpq ``options=-c search_path=...`` connection parameter, either as a
|
||||||
kwarg or encoded into the DSN query string.
|
pool kwarg or encoded into the DSN query string.
|
||||||
|
|
||||||
Schema names are validated upstream by
|
Schema names are validated upstream by
|
||||||
:class:`deerflow.config.database_config.DatabaseConfig` to be plain
|
:class:`deerflow.config.database_config.DatabaseConfig` to be plain
|
||||||
|
|||||||
@ -115,6 +115,21 @@ class TestDatabaseConfig:
|
|||||||
assert "deerflow" not in url.replace("/db", "")
|
assert "deerflow" not in url.replace("/db", "")
|
||||||
assert url.startswith("postgresql+asyncpg://")
|
assert url.startswith("postgresql+asyncpg://")
|
||||||
|
|
||||||
|
def test_sync_postgres_url_uses_configured_schema(self):
|
||||||
|
c = DatabaseConfig(backend="postgres", postgres_url="postgresql://u:p@h:5432/db", postgres_schema="deerflow")
|
||||||
|
url = c.app_sync_sqlalchemy_url
|
||||||
|
assert url.startswith("postgresql+psycopg://")
|
||||||
|
assert "options=-c%20search_path%3Ddeerflow" in url
|
||||||
|
|
||||||
|
def test_sync_postgres_url_preserves_existing_libpq_options(self):
|
||||||
|
c = DatabaseConfig(
|
||||||
|
backend="postgres",
|
||||||
|
postgres_url="postgresql://u:p@h:5432/db?options=-c%20statement_timeout%3D5000",
|
||||||
|
postgres_schema="deerflow",
|
||||||
|
)
|
||||||
|
url = c.app_sync_sqlalchemy_url
|
||||||
|
assert "options=-c%20statement_timeout%3D5000%20-c%20search_path%3Ddeerflow" in url
|
||||||
|
|
||||||
|
|
||||||
# -- MemoryRunStore --
|
# -- MemoryRunStore --
|
||||||
|
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user