mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-27 00:17:53 +00:00
* feat(agents): database-backed storage for custom agent definitions Add an agent_storage.backend switch (default file, behaviour-unchanged) with a db backend that stores each custom agent as a row in the shared SQL persistence layer, so a multi-instance deployment sees the same agents on every node (#4331, #4357). Introduces an AgentStore interface routing all read/write surfaces, an agents table + migration 0006, startup validation, and a file->db importer. Follows the thread_meta store / run_events backend-switch / 0003_scheduled_tasks migration patterns; no new dependency. * fix(agents): make db storage path production-ready (review round 1) Addresses review feedback on the db/sync agent-storage path: - sql.py: mirror the async engine's per-connection SQLite PRAGMAs on the sync engine (busy_timeout=30000, synchronous=NORMAL, foreign_keys=ON, WAL) so both engines behave identically against the shared DB; guard the engine cache with a lock (double-checked) so concurrent first-touch cannot build duplicate engines or register the connect listener twice. - routers/agents.py + routers/assistants_compat.py: offload the sync-store reads that ran on the event loop (list/get/check, update's pre-read + legacy guard + refresh, and assistants_compat's four list routes) via asyncio.to_thread — on db+postgres each was a network round trip stalling the loop. Writes were already offloaded. - file.py: translate the create() mkdir(exist_ok=False) race FileExistsError into AgentExistsError (router 409, matching SqlAgentStore's IntegrityError path); correct the _write docstring — per-file atomic replace, two commits sequential not transactional. Tests: sync-engine PRAGMA + engine-cache reuse assertions; file create-race -> AgentExistsError; strict Blockbuster anchor over the read endpoints so a regression back onto the loop fails CI. * fix(agents): address round-2 review on the db store path - update_agent tool: align the docstring/inline comment with FileAgentStore._write. Cross-field write atomicity is db-only; the file backend commits config then soul via two sequential os.replace (a crash between them can leave a fresh config.yaml beside a stale SOUL.md). The dropped partial-write *reporting* is an intentional tradeoff — the stage-then-replace safety is preserved (test_update_agent_soul_failure_does_not_replace_config still holds). - SqlAgentStore.update(): true upsert. Catch IntegrityError on the insert-on-missing branch, re-fetch and apply, so two concurrent first-time writes (e.g. two setup_agent handshakes) converge instead of surfacing a raw UNIQUE(user_id, name) violation as a 500. Symmetric with create(). - get_agent_store(): document the graph-subprocess config-resolution invariant (the except->file fallback is a genuine no-config path, not a mask for a misconfigured graph process) and pin it with two tests driving the real get_app_config() file resolution: db resolves from an on-disk config.yaml, file fallback when config is unresolvable. * test(agents): cover SqlAgentStore.update() write-race upsert recovery Mandatory-TDD test for the round-2 fix in 0680340a: two concurrent first-time update()s where the loser's insert hits UNIQUE(user_id, name). Deterministically forces the IntegrityError recovery path by making the first _row probe miss the committed winner, and asserts last-writer-wins instead of a surfaced 500.
232 lines
9.4 KiB
Python
232 lines
9.4 KiB
Python
"""Agent-storage backend selection, startup validation, and the db importer."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yaml
|
|
from sqlalchemy import create_engine
|
|
|
|
from app.gateway.deps import _validate_agent_storage
|
|
from deerflow.config.agent_storage_config import AgentStorageConfig
|
|
from deerflow.config.app_config import reset_app_config
|
|
from deerflow.config.database_config import DatabaseConfig
|
|
from deerflow.persistence.agents import get_agent_store, make_agent_store
|
|
from deerflow.persistence.agents.file import FileAgentStore
|
|
from deerflow.persistence.agents.model import AgentRow
|
|
from deerflow.persistence.agents.sql import SqlAgentStore
|
|
from deerflow.persistence.base import Base
|
|
|
|
|
|
def _cfg(agent_backend: str, db_backend: str, sqlite_dir: str = "/tmp/agent-store-test") -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
agent_storage=AgentStorageConfig(backend=agent_backend),
|
|
database=DatabaseConfig(backend=db_backend, sqlite_dir=sqlite_dir),
|
|
)
|
|
|
|
|
|
# -- make_agent_store selection --------------------------------------------
|
|
|
|
|
|
def test_file_is_the_default_backend():
|
|
assert isinstance(make_agent_store(_cfg("file", "memory")), FileAgentStore)
|
|
|
|
|
|
def test_db_backend_builds_sql_store(tmp_path):
|
|
assert isinstance(make_agent_store(_cfg("db", "sqlite", str(tmp_path))), SqlAgentStore)
|
|
|
|
|
|
def test_db_backend_on_memory_database_is_rejected():
|
|
with pytest.raises(ValueError, match="requires database.backend"):
|
|
make_agent_store(_cfg("db", "memory"))
|
|
|
|
|
|
# -- startup validation (deps) ---------------------------------------------
|
|
|
|
|
|
def test_validation_rejects_db_on_memory_database():
|
|
with pytest.raises(SystemExit):
|
|
_validate_agent_storage(_cfg("db", "memory"))
|
|
|
|
|
|
def test_validation_allows_file_and_db_on_sql(tmp_path):
|
|
_validate_agent_storage(_cfg("file", "memory")) # no raise
|
|
_validate_agent_storage(_cfg("db", "sqlite", str(tmp_path))) # no raise
|
|
|
|
|
|
def test_validation_warns_on_file_under_multiworker_postgres(monkeypatch, caplog):
|
|
monkeypatch.setenv("GATEWAY_WORKERS", "4")
|
|
cfg = SimpleNamespace(
|
|
agent_storage=AgentStorageConfig(backend="file"),
|
|
database=DatabaseConfig(backend="postgres", postgres_url="postgresql://u:p@h/db"),
|
|
)
|
|
with caplog.at_level("WARNING"):
|
|
_validate_agent_storage(cfg)
|
|
assert any("not visible across workers" in r.message for r in caplog.records)
|
|
|
|
|
|
# -- importer: file layout → db --------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def file_home(tmp_path, monkeypatch):
|
|
"""Root the file store at a temp DEER_FLOW_HOME with two seeded agents."""
|
|
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
|
|
from deerflow.config import paths as paths_module
|
|
|
|
monkeypatch.setattr(paths_module, "_paths", None)
|
|
fs = FileAgentStore()
|
|
fs.create("reviewer", {"name": "reviewer", "description": "reviews"}, "review soul", user_id="u1")
|
|
fs.create("planner", {"name": "planner", "description": "plans", "model": "m1"}, "plan soul", user_id="u2")
|
|
return tmp_path
|
|
|
|
|
|
def _patch_importer(monkeypatch, cfg):
|
|
import pathlib
|
|
|
|
import scripts.migrate_agents_to_db as importer
|
|
|
|
async def _noop_init(_):
|
|
return None
|
|
|
|
# Pre-create the schema (stand in for the alembic bootstrap the real run
|
|
# does, which also creates the sqlite directory).
|
|
pathlib.Path(cfg.database.sqlite_dir).mkdir(parents=True, exist_ok=True)
|
|
engine = create_engine(cfg.database.app_sync_sqlalchemy_url)
|
|
Base.metadata.create_all(engine, tables=[AgentRow.__table__])
|
|
engine.dispose()
|
|
|
|
monkeypatch.setattr(importer, "get_app_config", lambda: cfg)
|
|
monkeypatch.setattr("deerflow.persistence.engine.init_engine_from_config", _noop_init)
|
|
return importer
|
|
|
|
|
|
def test_importer_copies_all_agents_into_db(file_home, monkeypatch):
|
|
cfg = _cfg("db", "sqlite", str(file_home / "db"))
|
|
importer = _patch_importer(monkeypatch, cfg)
|
|
monkeypatch.setattr(sys, "argv", ["migrate_agents_to_db"])
|
|
|
|
assert importer.main() == 0
|
|
|
|
dest = SqlAgentStore(cfg.database.app_sync_sqlalchemy_url)
|
|
assert dest.get("reviewer", user_id="u1").description == "reviews"
|
|
assert dest.get_soul("reviewer", user_id="u1") == "review soul"
|
|
assert dest.get("planner", user_id="u2").model == "m1"
|
|
|
|
|
|
def test_importer_is_idempotent(file_home, monkeypatch):
|
|
cfg = _cfg("db", "sqlite", str(file_home / "db"))
|
|
importer = _patch_importer(monkeypatch, cfg)
|
|
monkeypatch.setattr(sys, "argv", ["migrate_agents_to_db"])
|
|
|
|
assert importer.main() == 0
|
|
# Second run must not raise on the already-present rows.
|
|
assert importer.main() == 0
|
|
dest = SqlAgentStore(cfg.database.app_sync_sqlalchemy_url)
|
|
assert len(dest.list_all()) == 2
|
|
|
|
|
|
def test_importer_dry_run_writes_nothing(file_home, monkeypatch):
|
|
cfg = _cfg("db", "sqlite", str(file_home / "db"))
|
|
importer = _patch_importer(monkeypatch, cfg)
|
|
monkeypatch.setattr(sys, "argv", ["migrate_agents_to_db", "--dry-run"])
|
|
|
|
assert importer.main() == 0
|
|
dest = SqlAgentStore(cfg.database.app_sync_sqlalchemy_url)
|
|
assert dest.list_all() == []
|
|
|
|
|
|
def test_read_free_functions_dispatch_to_db_backend(file_home, monkeypatch):
|
|
"""The headline invariant: under the db backend the standard read path (the
|
|
same free functions the per-run agent build calls) resolves from the shared
|
|
DB, not from node-local files — so on-disk agents are invisible and db agents
|
|
are visible everywhere."""
|
|
cfg = _cfg("db", "sqlite", str(file_home / "db"))
|
|
_patch_importer(monkeypatch, cfg) # creates the schema
|
|
monkeypatch.setattr("deerflow.config.app_config.get_app_config", lambda: cfg)
|
|
|
|
from deerflow.config.agents_config import list_custom_agents, load_agent_config, load_agent_soul
|
|
|
|
# The file store seeded 'reviewer'/'planner' on disk; the db is empty, so
|
|
# the free functions (now db-backed) do not see them.
|
|
assert list_custom_agents(user_id="u1") == []
|
|
with pytest.raises(FileNotFoundError):
|
|
load_agent_config("reviewer", user_id="u1")
|
|
|
|
# An agent written to the shared db is visible through the same free functions.
|
|
SqlAgentStore(cfg.database.app_sync_sqlalchemy_url).create("dbonly", {"name": "dbonly", "description": "shared"}, "db soul", user_id="u1")
|
|
assert [c.name for c in list_custom_agents(user_id="u1")] == ["dbonly"]
|
|
assert load_agent_config("dbonly", user_id="u1").description == "shared"
|
|
assert load_agent_soul("dbonly", user_id="u1") == "db soul"
|
|
|
|
|
|
def test_file_create_race_maps_file_exists_to_agent_exists(tmp_path, monkeypatch):
|
|
# TOCTOU: the existence guard passes (no agent yet), but a concurrent create
|
|
# wins the race so mkdir(exist_ok=False) raises FileExistsError. The store
|
|
# must translate that to AgentExistsError so the router returns 409, not a
|
|
# generic 500 — matching SqlAgentStore's IntegrityError path.
|
|
import pathlib
|
|
|
|
from deerflow.persistence.agents.base import AgentExistsError
|
|
|
|
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
|
|
from deerflow.config import paths as paths_module
|
|
|
|
monkeypatch.setattr(paths_module, "_paths", None)
|
|
|
|
def _racing_mkdir(self, *args, **kwargs):
|
|
raise FileExistsError(str(self))
|
|
|
|
monkeypatch.setattr(pathlib.Path, "mkdir", _racing_mkdir)
|
|
|
|
fs = FileAgentStore()
|
|
with pytest.raises(AgentExistsError):
|
|
fs.create("racy", {"name": "racy"}, "soul", user_id="u1")
|
|
|
|
|
|
# -- graph-subprocess config resolution (db backend's core cross-process invariant) --
|
|
|
|
|
|
def _write_min_config(path, extra: dict) -> None:
|
|
"""Minimal but valid config.yaml (sandbox + models are the only hard requirements)."""
|
|
doc = {
|
|
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
|
|
"models": [{"name": "m", "use": "langchain_openai:ChatOpenAI", "model": "gpt-test"}],
|
|
**extra,
|
|
}
|
|
path.write_text(yaml.safe_dump(doc), encoding="utf-8")
|
|
|
|
|
|
def test_get_agent_store_resolves_db_backend_from_on_disk_config(tmp_path, monkeypatch):
|
|
"""Pins the db backend's headline cross-process guarantee.
|
|
|
|
The per-run agent build runs in the graph subprocess, a different process
|
|
from the gateway; its db visibility holds only because ``get_agent_store()``
|
|
resolves ``agent_storage.backend: db`` from the real on-disk ``config.yaml``
|
|
there (not a monkeypatched stub) rather than silently falling back to
|
|
node-local ``file``. Existing coverage monkeypatches ``get_app_config``;
|
|
this drives the genuine file-resolution path a fresh process would take.
|
|
"""
|
|
cfg_path = tmp_path / "config.yaml"
|
|
_write_min_config(cfg_path, {"agent_storage": {"backend": "db"}, "database": {"backend": "sqlite", "sqlite_dir": str(tmp_path / "db")}})
|
|
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(cfg_path))
|
|
try:
|
|
reset_app_config() # force a fresh read from the on-disk file
|
|
assert isinstance(get_agent_store(), SqlAgentStore)
|
|
finally:
|
|
reset_app_config() # don't leak the custom config into other tests
|
|
|
|
|
|
def test_get_agent_store_falls_back_to_file_without_config(tmp_path, monkeypatch):
|
|
"""The ``except -> file`` fallback is for genuinely unresolvable config only
|
|
(CLI/tests); it must not fire when a config exists — that asymmetry is what
|
|
keeps a misconfigured graph process from silently downgrading db to file."""
|
|
monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(tmp_path / "does-not-exist.yaml"))
|
|
try:
|
|
reset_app_config()
|
|
assert isinstance(get_agent_store(), FileAgentStore)
|
|
finally:
|
|
reset_app_config()
|