deer-flow/backend/tests/test_agent_store_sql.py
Aari 0d4d0cb17d
feat(agents): database-backed storage for custom agent definitions (#4359)
* 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.
2026-07-23 08:03:21 +08:00

233 lines
9.4 KiB
Python

"""SqlAgentStore — the db agent-storage backend.
Exercises the synchronous SQL store against a real sqlite file: CRUD, the
config-document round trip (name excluded from the JSON, restored on read),
per-user isolation, the UNIQUE(user_id, name) conflict, and the change-token
signature the GitHub registry keys its cache off.
"""
from __future__ import annotations
from unittest import mock
import pytest
from sqlalchemy import create_engine
from deerflow.config.agents_config import AgentConfig
from deerflow.persistence.agents.base import AgentExistsError
from deerflow.persistence.agents.model import AgentRow
from deerflow.persistence.agents.sql import SqlAgentStore
from deerflow.persistence.base import Base
@pytest.fixture()
def store(tmp_path):
# Unique file per test → the store's per-URL engine cache never collides.
url = f"sqlite:///{tmp_path}/agents.db"
create_engine(url) # touch the file
engine = create_engine(url)
Base.metadata.create_all(engine, tables=[AgentRow.__table__])
engine.dispose()
return SqlAgentStore(url)
def test_create_and_get_round_trips_config_and_soul(store):
store.create("reviewer", {"name": "reviewer", "description": "reviews code", "model": "gpt-x"}, "You review.", user_id="u1")
cfg = store.get("reviewer", user_id="u1")
assert isinstance(cfg, AgentConfig)
assert cfg.name == "reviewer"
assert cfg.description == "reviews code"
assert cfg.model == "gpt-x"
assert store.get_soul("reviewer", user_id="u1") == "You review."
def test_name_is_stored_lowercase_and_excluded_from_document(store):
store.create("Mixed", {"name": "Mixed", "description": "d"}, "s", user_id="u1")
# Stored lowercase (matches the on-disk layout), and the JSON document does
# not duplicate the name column.
with store._Session() as session:
row = session.query(AgentRow).one()
assert row.name == "mixed"
assert "name" not in row.config
assert store.get("mixed", user_id="u1").name == "mixed"
def test_get_missing_raises_file_not_found(store):
# The historical contract routers/tools rely on for a 404 / "does not exist".
with pytest.raises(FileNotFoundError):
store.get("nope", user_id="u1")
assert store.get_soul("nope", user_id="u1") is None
assert store.exists("nope", user_id="u1") is False
def test_create_duplicate_raises_agent_exists(store):
store.create("dup", {"name": "dup"}, "s", user_id="u1")
with pytest.raises(AgentExistsError):
store.create("dup", {"name": "dup"}, "s2", user_id="u1")
def test_update_is_upsert_and_partial(store):
# Upsert: update on a missing agent creates it.
store.update("a", {"name": "a", "description": "first"}, "soul1", user_id="u1")
assert store.get("a", user_id="u1").description == "first"
# config=None leaves config untouched, only soul changes.
store.update("a", None, "soul2", user_id="u1")
assert store.get("a", user_id="u1").description == "first"
assert store.get_soul("a", user_id="u1") == "soul2"
# soul=None leaves soul untouched, only config changes.
store.update("a", {"name": "a", "description": "second"}, None, user_id="u1")
assert store.get("a", user_id="u1").description == "second"
assert store.get_soul("a", user_id="u1") == "soul2"
def test_update_insert_race_recovers_via_upsert(store):
"""update()'s insert-on-missing branch is a true upsert under a write race.
Two concurrent first-time writes (e.g. two ``setup_agent`` handshakes) both
see ``row is None`` and both insert; ``UNIQUE(user_id, name)`` rejects the
loser. The loser must re-fetch the winner's row and apply its update rather
than surfacing the raw ``IntegrityError`` as a 500. Simulated deterministically
by making the first ``_row`` probe (the check-then-insert window) miss the
already-committed winner, forcing the ``IntegrityError`` recovery path.
"""
store.create("a", {"name": "a", "description": "winner"}, "winner-soul", user_id="u1")
real_row = store._row # bound method captured before patching
seen = {"n": 0}
def racing_row(session, name, user_id):
seen["n"] += 1
# First call is the pre-insert probe: pretend the winner isn't there yet.
return None if seen["n"] == 1 else real_row(session, name, user_id)
with mock.patch.object(store, "_row", side_effect=racing_row):
# Must not raise: the loser recovers into a true upsert (last write wins).
store.update("a", {"name": "a", "description": "loser"}, "loser-soul", user_id="u1")
assert store.get("a", user_id="u1").description == "loser"
assert store.get_soul("a", user_id="u1") == "loser-soul"
def test_list_and_list_all(store):
store.create("b-agent", {"name": "b-agent"}, "s", user_id="u1")
store.create("a-agent", {"name": "a-agent"}, "s", user_id="u1")
store.create("c-agent", {"name": "c-agent"}, "s", user_id="u2")
u1 = store.list(user_id="u1")
assert [c.name for c in u1] == ["a-agent", "b-agent"] # sorted by name
assert store.list(user_id="u2") == store.list(user_id="u2")
assert [c.name for c in store.list(user_id="u2")] == ["c-agent"]
all_agents = store.list_all()
assert sorted((uid, cfg.name) for uid, cfg in all_agents) == [
("u1", "a-agent"),
("u1", "b-agent"),
("u2", "c-agent"),
]
def test_user_isolation(store):
store.create("shared-name", {"name": "shared-name", "description": "u1's"}, "s1", user_id="u1")
# Same name under a different user is a distinct agent (no conflict).
store.create("shared-name", {"name": "shared-name", "description": "u2's"}, "s2", user_id="u2")
assert store.get("shared-name", user_id="u1").description == "u1's"
assert store.get("shared-name", user_id="u2").description == "u2's"
with pytest.raises(FileNotFoundError):
store.get("shared-name", user_id="u3")
def test_delete_reports_outcome(store):
store.create("gone", {"name": "gone"}, "s", user_id="u1")
assert store.delete("gone", user_id="u1") == "deleted"
assert store.delete("gone", user_id="u1") == "missing"
with pytest.raises(FileNotFoundError):
store.get("gone", user_id="u1")
def test_signature_changes_on_mutation(store):
empty = store.signature()
store.create("x", {"name": "x"}, "s", user_id="u1")
after_create = store.signature()
assert after_create != empty
# A no-op read does not change the token.
assert store.signature() == after_create
store.update("x", {"name": "x", "description": "changed"}, None, user_id="u1")
assert store.signature() != after_create
def test_sync_engine_mirrors_async_pragmas(tmp_path):
# The db backend's sync engine must set the same per-connection SQLite PRAGMAs
# the async engine does (persistence/engine.py), not leave synchronous=FULL
# and pysqlite's default 5s busy_timeout. WAL is persistent on the file;
# synchronous / busy_timeout are per-connection and must be re-applied here.
from sqlalchemy import text
from deerflow.persistence.agents.sql import _get_sessionmaker
url = f"sqlite:///{tmp_path}/pragma.db"
Session = _get_sessionmaker(url)
with Session() as session:
busy_timeout = session.execute(text("PRAGMA busy_timeout")).scalar()
synchronous = session.execute(text("PRAGMA synchronous")).scalar()
journal_mode = session.execute(text("PRAGMA journal_mode")).scalar()
assert busy_timeout == 30000
assert synchronous == 1 # NORMAL
assert str(journal_mode).lower() == "wal"
def test_engine_cache_is_reused_per_url(tmp_path):
# Two stores on the same URL share one cached engine (the lock-guarded
# double-checked cache), so we never build duplicate engines/pools.
from deerflow.persistence.agents.sql import _get_sessionmaker
url = f"sqlite:///{tmp_path}/reuse.db"
first = _get_sessionmaker(url)
second = _get_sessionmaker(url)
assert first.kw["bind"] is second.kw["bind"]
def test_delete_preserves_memory_only_dir_when_no_row(store, tmp_path, monkeypatch):
# #4279 invariant carried into the db backend: with no agent row, a bare
# on-disk directory holds only memory/facts data (config lives in the row in
# db mode), so delete must preserve it and report "not-custom-agent" instead
# of rmtree-ing a user's memory.
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
from deerflow.config import paths as paths_module
monkeypatch.setattr(paths_module, "_paths", None)
from deerflow.config.paths import get_paths
facts_dir = get_paths().user_agent_dir("u1", "ghost") / "facts"
facts_dir.mkdir(parents=True)
fact = facts_dir / "fact_keep.md"
fact.write_text("memory data", encoding="utf-8")
assert store.delete("ghost", user_id="u1") == "not-custom-agent"
assert fact.read_text(encoding="utf-8") == "memory data"
def test_delete_removes_memory_dir_when_row_exists(store, tmp_path, monkeypatch):
# The complement: when the agent row exists, its co-located on-disk memory is
# cleaned along with the row.
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
from deerflow.config import paths as paths_module
monkeypatch.setattr(paths_module, "_paths", None)
from deerflow.config.paths import get_paths
store.create("real", {"name": "real"}, "s", user_id="u1")
mem_dir = get_paths().user_agent_dir("u1", "real")
mem_dir.mkdir(parents=True, exist_ok=True)
(mem_dir / "memory.json").write_text("{}", encoding="utf-8")
assert store.delete("real", user_id="u1") == "deleted"
assert not mem_dir.exists()