github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-27 00:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(agents): database-backed storage for custom agent definitions (#4359)

Browse Source 
* 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.
This commit is contained in:
Aari 2026-07-23 08:03:21 +08:00 committed by GitHub
parent 4dd7cafef1
commit 0d4d0cb17d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
31 changed files with 1751 additions and 477 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
backend/app/gateway/deps.py
View File

@ -100,6 +100,38 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
)
def _validate_agent_storage(config: AppConfig) -> None:
"""Fail fast on an agent-storage backend the database cannot support.
``agent_storage.backend: db`` needs a durable, shared SQL database — a
``memory`` database is per-process, so agent definitions would silently
diverge across nodes (and there is no SQL URL to open). Mirrors deermem's
create_storage fail-fast and the multi-worker gate above.
Also warns when a multi-worker Postgres deployment leaves agent storage on
``file``: custom agents created on one node's local disk are invisible to
the others, exactly the divergence the db backend exists to fix.
"""
agent_storage = getattr(config, "agent_storage", None)
backend = getattr(agent_storage, "backend", "file")
db_backend = getattr(getattr(config, "database", None), "backend", None)
if backend == "db" and db_backend not in ("sqlite", "postgres"):
raise SystemExit(
f"agent_storage.backend='db' requires database.backend to be 'sqlite' or 'postgres', "
f"but database.backend is '{db_backend}'. A 'memory' database is per-process and cannot "
"share agent definitions across nodes. Set database.backend, or use agent_storage.backend='file'."
)
try:
workers = int(os.environ.get("GATEWAY_WORKERS", "1"))
except (TypeError, ValueError):
workers = 1
if workers > 1 and db_backend == "postgres" and backend == "file":
logger.warning(
"GATEWAY_WORKERS=%s with database.backend='postgres' but agent_storage.backend='file': custom agents are stored per-node on local disk and are not visible across workers/nodes. Set agent_storage.backend='db' to share them.",
workers,
)
async def _drain_inflight_runs(run_manager: RunManager) -> None:
"""Drain in-flight runs before the checkpointer is torn down (issue #3373).
@ -270,6 +302,9 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen
# SQLite write-locks cannot support concurrent multi-process access.
# ------------------------------------------------------------------
_enforce_postgres_for_multi_worker(startup_config)
# Reject agent_storage.backend='db' on a non-durable database, and warn on
# node-divergent file storage under multi-worker Postgres.
_validate_agent_storage(startup_config)
async with AsyncExitStack() as stack:
config = startup_config

172
backend/app/gateway/github/registry.py
View File

@ -1,43 +1,37 @@
"""Build the GitHub webhook → agent registry.
Walks every user's custom-agent directory under ``{base_dir}/users/`` plus
the legacy shared layout at ``{base_dir}/agents/`` and indexes every agent
that declares a ``github:`` block by the ``(repo, event)`` pairs it
declares an interest in.
Indexes every custom agent that declares a ``github:`` block by the
``(repo, event)`` pairs it is interested in, across all owners. Agent discovery
and change detection go through the configured agent store
(:mod:`deerflow.persistence.agents`), so both the ``file`` backend (per-user
directories + the legacy shared layout) and the ``db`` backend (the shared
``agents`` table) are covered by the same code.
The dispatcher calls :func:`build_github_agent_registry` once per webhook
delivery. We avoid re-parsing every ``config.yaml`` on each call via a
small mtime-keyed cache: the directory listing + ``stat()`` per config
file is cheap (~µs), while ``yaml.safe_load`` is the dominant cost
(~hundreds of µs per file). The cache key is the sorted tuple of
``(user_id, agent_name, config.yaml mtime)`` triples; any mtime change,
addition, or deletion invalidates the cache transparently. Operators
who hand-edit ``config.yaml`` see the change on the next webhook.
delivery. We avoid re-loading every agent on each call via a small cache keyed
on the store's :meth:`~deerflow.persistence.agents.base.AgentStore.signature`
change token: the file backend derives it from ``config.yaml`` mtimes (any
edit, addition, or deletion invalidates the cache transparently); the db backend
derives it from ``max(updated_at)`` + the row count. Operators who hand-edit a
``config.yaml`` see the change on the next webhook.
Cache invalidation caveat: mtime granularity on macOS HFS+ / APFS is
~1 µs but on some filesystems (FAT, network shares with caching) it's
1 s. Two edits inside the same coarse-tick would look identical. For
the dispatch path that's fine — webhooks are rare relative to operator
edits, and the next non-coincident write reconciles. If we ever land
operator tooling that batches sub-second edits, we can extend the
signature with file size or a content hash.
Cache invalidation caveat (file backend): mtime granularity on macOS HFS+ /
APFS is ~1 µs but on some filesystems (FAT, network shares with caching) it's
1 s. Two edits inside the same coarse-tick would look identical. For the
dispatch path that's fine — webhooks are rare relative to operator edits, and
the next non-coincident write reconciles.
"""
from __future__ import annotations
import logging
import threading
from collections.abc import Hashable
from dataclasses import dataclass
from pathlib import Path
from app.gateway.github.triggers import _resolved_trigger
from deerflow.config.agents_config import (
AgentConfig,
GitHubTriggerConfig,
load_agent_config,
)
from deerflow.config.paths import get_paths
from deerflow.runtime.user_context import DEFAULT_USER_ID
from deerflow.config.agents_config import AgentConfig, GitHubTriggerConfig
from deerflow.persistence.agents import get_agent_store
logger = logging.getLogger(__name__)
@ -67,124 +61,29 @@ class GitHubAgentMatch:
trigger: GitHubTriggerConfig
# Cache: (signature, registry). ``signature`` is a tuple of
# ``(user_id, agent_name, mtime)`` triples. Identical signature → registry
# is still valid, skip the YAML parses.
_Signature = tuple[tuple[str, str, float], ...]
# Cache: (signature, registry). ``signature`` is the store's opaque change
# token — identical token → registry is still valid, skip the reload.
_Registry = dict[tuple[str, str], list[GitHubAgentMatch]]
_cache: tuple[_Signature, _Registry] | None = None
_cache: tuple[Hashable, _Registry] | None = None
# Threading lock (not asyncio): build_github_agent_registry is invoked
# from asyncio.to_thread in the dispatcher, so the lock is acquired on
# the worker thread. A plain Lock is the right primitive here.
_cache_lock = threading.Lock()
def _discover_user_ids() -> list[str]:
"""Return all user-id directories under ``base_dir/users/``.
Falls back to ``[DEFAULT_USER_ID]`` so the no-auth dev setup (which
keeps everything in ``users/default/``) is always covered even before
the directory has been created on disk.
"""
paths = get_paths()
users_dir: Path = paths.base_dir / "users"
if not users_dir.exists():
return [DEFAULT_USER_ID]
found: list[str] = []
for entry in sorted(users_dir.iterdir()):
if entry.is_dir() and (entry / "agents").exists():
found.append(entry.name)
if DEFAULT_USER_ID not in found:
found.append(DEFAULT_USER_ID)
return found
def _gather_agent_signature() -> tuple[_Signature, list[tuple[str, str]]]:
"""Return (signature, [(user_id, agent_name)]) for every agent on disk.
The signature lets us skip the YAML parse on warm hits; the
discovered list lets the rebuilder process exactly the agents that
the signature covers. Doing iterdir + stat is cheap (~µs each); the
full cost we avoid is the ``yaml.safe_load`` per config.
Includes the legacy shared layout at ``{base_dir}/agents/`` under the
:data:`DEFAULT_USER_ID` bucket so unmigrated installations still
receive webhook fan-out. Per-user entries shadow legacy entries with
the same name (matching :func:`list_custom_agents`' precedence), so
once an install runs ``migrate_user_isolation.py`` the legacy entry
is silently superseded rather than producing duplicate rows.
"""
paths = get_paths()
sig: list[tuple[str, str, float]] = []
discovered: list[tuple[str, str]] = []
seen: set[tuple[str, str]] = set()
for user_id in _discover_user_ids():
agent_root = paths.user_agents_dir(user_id)
if not agent_root.exists():
continue
for entry in sorted(agent_root.iterdir()):
config = entry / "config.yaml"
if not entry.is_dir() or not config.exists():
continue
try:
mtime = config.stat().st_mtime
except OSError:
# Vanished between iterdir and stat — racing operator
# edit. Drop from this round; next call picks it up.
continue
sig.append((user_id, entry.name, mtime))
discovered.append((user_id, entry.name))
seen.add((user_id, entry.name))
# Legacy shared layout: {base_dir}/agents/{name}/. CLAUDE.md commits
# to this as a read-only fallback for unmigrated installs, and
# load_agent_config() / list_custom_agents() honour it — the webhook
# path must too, or an unmigrated install with a ``github:`` block on
# a shared agent silently fans out to nothing. Legacy entries map
# onto the DEFAULT_USER_ID bucket because that is the user-id
# ``load_agent_config(name)`` resolves them under at run-time.
legacy_root = paths.agents_dir
if legacy_root.exists():
for entry in sorted(legacy_root.iterdir()):
config = entry / "config.yaml"
if not entry.is_dir() or not config.exists():
continue
# Per-user shadow: if users/default/agents/{name} already
# exists, skip the legacy entry so we don't index the same
# agent twice with conflicting trigger sets.
if (DEFAULT_USER_ID, entry.name) in seen:
continue
try:
mtime = config.stat().st_mtime
except OSError:
continue
sig.append((DEFAULT_USER_ID, entry.name, mtime))
discovered.append((DEFAULT_USER_ID, entry.name))
seen.add((DEFAULT_USER_ID, entry.name))
return tuple(sig), discovered
def _rebuild(discovered: list[tuple[str, str]]) -> _Registry:
"""Parse every agent's config.yaml and build the (repo, event) index.
def _build_index(agents: list[tuple[str, AgentConfig]]) -> _Registry:
"""Build the ``(repo, event)`` index from the store's agents.
Each ``(repo, event)`` slot stores :class:`GitHubAgentMatch` rows — the
user_id + AgentConfig + the trigger already resolved (binding override
merged with per-event field defaults). The dispatcher then only needs
to apply the trigger; it never re-walks ``bindings`` to find the right
one. Single-binding-per-repo is enforced by
:class:`GitHubAgentConfig`'s validator, so a duplicate-repo config
fails to load here (logged as a skip) instead of producing duplicate
rows in this index.
one. Single-binding-per-repo is enforced by :class:`GitHubAgentConfig`'s
validator (an agent that fails to load is already dropped by the store).
"""
index: _Registry = {}
for user_id, agent_name in discovered:
try:
cfg = load_agent_config(agent_name, user_id=user_id)
except Exception as exc: # noqa: BLE001 — one bad agent must not kill the scan
logger.warning("github_registry: skipping agent %s/%s: %s", user_id, agent_name, exc)
continue
if cfg is None or cfg.github is None:
for user_id, cfg in agents:
if cfg.github is None:
continue
for binding in cfg.github.bindings:
for event, override in binding.triggers.items():
@ -211,17 +110,18 @@ def build_github_agent_registry() -> _Registry:
dispatcher will never fan a webhook out to it.
Warm path (no agents added/removed/edited since the last call) costs
only the iterdir + stat pass — no YAML parsing. Cold path parses
every config.yaml and refreshes the cache. The result is shared
across callers (returned by reference) since :class:`GitHubAgentMatch`
is frozen and the registry is intended as read-only.
only the store's ``signature()`` — cheap on both backends. Cold path
reloads every agent and refreshes the cache. The result is shared across
callers (returned by reference) since :class:`GitHubAgentMatch` is frozen
and the registry is intended as read-only.
"""
global _cache
store = get_agent_store()
with _cache_lock:
signature, discovered = _gather_agent_signature()
signature = store.signature()
if _cache is not None and _cache[0] == signature:
return _cache[1]
registry = _rebuild(discovered)
registry = _build_index(store.list_all())
_cache = (signature, registry)
return registry

171
backend/app/gateway/routers/agents.py
View File

@ -3,10 +3,8 @@
import asyncio
import logging
import re
import shutil
from typing import Literal
import yaml
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
@ -21,6 +19,7 @@ from deerflow.config.agents_config import (
)
from deerflow.config.app_config import get_app_config
from deerflow.config.paths import get_paths
from deerflow.persistence.agents import AgentExistsError, get_agent_store
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
@ -212,9 +211,16 @@ async def list_agents() -> AgentsListResponse:
_require_agents_api_enabled()
user_id = get_effective_user_id()
try:
def _list() -> AgentsListResponse:
# Worker thread: the store read plus the per-agent SOUL read inside
# _agent_config_to_response are filesystem IO (file backend) or DB round
# trips (db backend) and must stay off the event loop.
agents = list_custom_agents(user_id=user_id)
return AgentsListResponse(agents=[_agent_config_to_response(a, include_soul=True, user_id=user_id) for a in agents])
try:
return await asyncio.to_thread(_list)
except Exception as e:
logger.error(f"Failed to list agents: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to list agents: {str(e)}")
@ -241,12 +247,11 @@ async def check_agent_name(name: str) -> dict:
_validate_agent_name(name)
normalized = _normalize_agent_name(name)
user_id = get_effective_user_id()
paths = get_paths()
# Treat the name as taken if either the per-user path or the legacy shared
# path holds an agent — picking a name that collides with an unmigrated
# legacy agent would shadow the legacy entry once migration runs.
available = not paths.user_agent_dir(user_id, normalized).exists() and not paths.agent_dir(normalized).exists()
return {"available": available, "name": normalized}
# Availability is defined by the active backend and stays consistent with
# create()'s conflict rule (file: per-user or legacy dir; db: a row). The
# exists() probe is filesystem IO / a DB round trip, so keep it off the loop.
exists = await asyncio.to_thread(get_agent_store().exists, normalized, user_id=user_id)
return {"available": not exists, "name": normalized}
@router.get(
@ -272,9 +277,13 @@ async def get_agent(name: str) -> AgentResponse:
name = _normalize_agent_name(name)
user_id = get_effective_user_id()
try:
def _get() -> AgentResponse:
# Worker thread: config read + SOUL read must stay off the event loop.
agent_cfg = load_agent_config(name, user_id=user_id)
return _agent_config_to_response(agent_cfg, include_soul=True, user_id=user_id)
try:
return await asyncio.to_thread(_get)
except FileNotFoundError:
raise HTTPException(status_code=404, detail=f"Agent '{name}' not found")
except Exception as e:
@ -306,62 +315,37 @@ async def create_agent_endpoint(request: AgentCreateRequest) -> AgentResponse:
_validate_model_exists(request.model)
normalized_name = _normalize_agent_name(request.name)
user_id = get_effective_user_id()
paths = get_paths()
def _create_agent() -> AgentResponse | None:
# Worker thread: base-dir resolution, existence checks, directory/file
# creation, read-back, and failure cleanup are all blocking filesystem
# IO that must stay off the event loop.
agent_dir = paths.user_agent_dir(user_id, normalized_name)
legacy_dir = paths.agent_dir(normalized_name)
# Config document — only the fields the caller set, matching the historical
# writer (an omitted field stays absent rather than being materialized).
config_data: dict = {"name": normalized_name}
if request.description:
config_data["description"] = request.description
if request.tool_groups is not None:
config_data["tool_groups"] = request.tool_groups
if request.skills is not None:
config_data["skills"] = request.skills
# model / model_settings / thinking_enabled / reasoning_effort (issue #4336).
_apply_model_behavior(config_data, request)
if legacy_dir.exists():
return None # signals 409 to the caller
store = get_agent_store()
try:
try:
agent_dir.mkdir(parents=True, exist_ok=False)
except FileExistsError:
return None # signals 409 to the caller
# Write config.yaml
config_data: dict = {"name": normalized_name}
if request.description:
config_data["description"] = request.description
if request.tool_groups is not None:
config_data["tool_groups"] = request.tool_groups
if request.skills is not None:
config_data["skills"] = request.skills
_apply_model_behavior(config_data, request)
config_file = agent_dir / "config.yaml"
with open(config_file, "w", encoding="utf-8") as f:
yaml.dump(config_data, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
# Write SOUL.md
soul_file = agent_dir / "SOUL.md"
soul_file.write_text(request.soul, encoding="utf-8")
logger.info(f"Created agent '{normalized_name}' at {agent_dir}")
agent_cfg = load_agent_config(normalized_name, user_id=user_id)
return _agent_config_to_response(agent_cfg, include_soul=True, user_id=user_id)
except Exception:
# Clean up partial state on failure before surfacing the error.
if agent_dir.exists():
shutil.rmtree(agent_dir)
raise
def _create_agent() -> AgentResponse:
# Worker thread: existence checks + persistence (file IO or a DB round
# trip) must stay off the event loop.
store.create(normalized_name, config_data, request.soul, user_id=user_id)
logger.info("Created agent '%s'", normalized_name)
agent_cfg = load_agent_config(normalized_name, user_id=user_id)
return _agent_config_to_response(agent_cfg, include_soul=True, user_id=user_id)
try:
response = await asyncio.to_thread(_create_agent)
return await asyncio.to_thread(_create_agent)
except AgentExistsError:
raise HTTPException(status_code=409, detail=f"Agent '{normalized_name}' already exists")
except Exception as e:
logger.error(f"Failed to create agent '{request.name}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to create agent: {str(e)}")
if response is None:
raise HTTPException(status_code=409, detail=f"Agent '{normalized_name}' already exists")
return response
@router.put(
"/agents/{name}",
@ -388,21 +372,26 @@ async def update_agent(name: str, request: AgentUpdateRequest) -> AgentResponse:
user_id = get_effective_user_id()
try:
agent_cfg = load_agent_config(name, user_id=user_id)
agent_cfg = await asyncio.to_thread(load_agent_config, name, user_id=user_id)
except FileNotFoundError:
raise HTTPException(status_code=404, detail=f"Agent '{name}' not found")
paths = get_paths()
agent_dir = paths.user_agent_dir(user_id, name)
legacy_dir = paths.agent_dir(name)
# Require config.yaml, not bare directory existence — a per-user agent
# directory can exist containing only memory.json (written the first
# time this user chats with a legacy shared agent, before this route
# is ever called). Bare .exists() would miss that case and let this
# fall through to a silent fork of a brand-new config.yaml/SOUL.md
# into the memory-only directory instead of blocking (mirrors
# resolve_agent_dir's guard, see #3390).
if not (agent_dir / "config.yaml").exists() and (legacy_dir / "config.yaml").exists():
def _is_legacy_only_layout() -> bool:
# Require config.yaml, not bare directory existence — a per-user agent
# directory can exist containing only memory.json (written the first
# time this user chats with a legacy shared agent, before this route
# is ever called). Bare .exists() would miss that case and let this
# fall through to a silent fork of a brand-new config.yaml/SOUL.md
# into the memory-only directory instead of blocking (mirrors
# resolve_agent_dir's guard, see #3390). The db backend has no legacy
# shared layout, so this file-only guard is a no-op there. The .exists()
# probes are filesystem IO, so they run off the event loop.
paths = get_paths()
agent_dir = paths.user_agent_dir(user_id, name)
legacy_dir = paths.agent_dir(name)
return not (agent_dir / "config.yaml").exists() and (legacy_dir / "config.yaml").exists()
if await asyncio.to_thread(_is_legacy_only_layout):
raise HTTPException(
status_code=409,
detail=(f"Agent '{name}' only exists in the legacy shared layout and is not scoped to a user. Run scripts/migrate_user_isolation.py to move legacy agents into the per-user layout before updating."),
@ -418,8 +407,9 @@ async def update_agent(name: str, request: AgentUpdateRequest) -> AgentResponse:
fields_set = request.model_fields_set
config_changed = bool(fields_set & ({"description", "tool_groups", "skills"} | set(_MODEL_BEHAVIOR_FIELDS)))
updated: dict | None = None
if config_changed:
updated: dict = {
updated = {
"name": agent_cfg.name,
"description": request.description if "description" in fields_set else agent_cfg.description,
}
@ -452,19 +442,20 @@ async def update_agent(name: str, request: AgentUpdateRequest) -> AgentResponse:
for key, value in preserve_non_managed_fields(agent_cfg).items():
updated.setdefault(key, value)
config_file = agent_dir / "config.yaml"
with open(config_file, "w", encoding="utf-8") as f:
yaml.dump(updated, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
# Update SOUL.md if provided
if request.soul is not None:
soul_path = agent_dir / "SOUL.md"
soul_path.write_text(request.soul, encoding="utf-8")
store = get_agent_store()
# Persist config (when changed) and/or soul (when provided) off the
# event loop. A no-change PATCH commits nothing and re-reads current state.
if updated is not None or request.soul is not None:
await asyncio.to_thread(store.update, name, updated, request.soul, user_id=user_id)
logger.info(f"Updated agent '{name}'")
refreshed_cfg = load_agent_config(name, user_id=user_id)
return _agent_config_to_response(refreshed_cfg, include_soul=True, user_id=user_id)
def _refresh() -> AgentResponse:
# Worker thread: re-read config + SOUL off the event loop.
refreshed_cfg = load_agent_config(name, user_id=user_id)
return _agent_config_to_response(refreshed_cfg, include_soul=True, user_id=user_id)
return await asyncio.to_thread(_refresh)
except HTTPException:
raise
@ -558,23 +549,11 @@ async def delete_agent(name: str) -> None:
_validate_agent_name(name)
name = _normalize_agent_name(name)
user_id = get_effective_user_id()
paths = get_paths()
def _remove_agent_dir() -> tuple[str, str]:
# Runs in a worker thread: resolving the base dir, probing the directory
# (`exists`), and removing it (`rmtree`) are all blocking filesystem IO
# that must stay off the event loop.
agent_dir = paths.user_agent_dir(user_id, name)
if not agent_dir.exists():
outcome = "legacy" if paths.agent_dir(name).exists() else "missing"
return outcome, str(agent_dir)
if not (agent_dir / "config.yaml").is_file():
return "not-custom-agent", str(agent_dir)
shutil.rmtree(agent_dir)
return "deleted", str(agent_dir)
store = get_agent_store()
try:
outcome, agent_dir = await asyncio.to_thread(_remove_agent_dir)
# Off the event loop: file rmtree or a DB delete plus memory cleanup.
outcome = await asyncio.to_thread(store.delete, name, user_id=user_id)
except Exception as e:
logger.error(f"Failed to delete agent '{name}': {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to delete agent: {str(e)}")
@ -592,4 +571,4 @@ async def delete_agent(name: str) -> None:
detail=(f"Directory for '{name}' contains memory data but is not a custom agent because config.yaml is missing; it was preserved."),
)
logger.info(f"Deleted agent '{name}' from {agent_dir}")
logger.info(f"Deleted agent '{name}'")

11
backend/app/gateway/routers/assistants_compat.py
View File

@ -9,6 +9,7 @@ initialization requirements (``assistants.search()`` and ``assistants.get()``).
from __future__ import annotations
import asyncio
import logging
from datetime import UTC, datetime
from typing import Any
@ -91,7 +92,9 @@ async def search_assistants(body: AssistantSearchRequest | None = None) -> list[
Returns all registered assistants (lead_agent + custom agents from config).
"""
assistants = _list_assistants()
# _list_assistants reads custom agents through the sync agent store (file IO
# or a DB round trip), so keep it off the event loop.
assistants = await asyncio.to_thread(_list_assistants)
if body and body.graph_id:
assistants = [a for a in assistants if a.graph_id == body.graph_id]
@ -106,7 +109,7 @@ async def search_assistants(body: AssistantSearchRequest | None = None) -> list[
@router.get("/{assistant_id}", response_model=AssistantResponse)
async def get_assistant_compat(assistant_id: str) -> AssistantResponse:
"""Get an assistant by ID."""
for a in _list_assistants():
for a in await asyncio.to_thread(_list_assistants):
if a.assistant_id == assistant_id:
return a
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")
@ -119,7 +122,7 @@ async def get_assistant_graph(assistant_id: str) -> dict:
Returns a minimal graph description. Full graph introspection is
not supported in the Gateway — this stub satisfies SDK validation.
"""
found = any(a.assistant_id == assistant_id for a in _list_assistants())
found = any(a.assistant_id == assistant_id for a in await asyncio.to_thread(_list_assistants))
if not found:
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")
@ -136,7 +139,7 @@ async def get_assistant_schemas(assistant_id: str) -> dict:
Returns empty schemas — full introspection not supported in Gateway.
"""
found = any(a.assistant_id == assistant_id for a in _list_assistants())
found = any(a.assistant_id == assistant_id for a in await asyncio.to_thread(_list_assistants))
if not found:
raise HTTPException(status_code=404, detail=f"Assistant {assistant_id} not found")

25
backend/docs/CONFIGURATION.md
View File

@ -254,6 +254,31 @@ Notes:
- Manual trigger uses the same scheduled-task resource and run lifecycle.
- Scheduled task definitions and task-run history are persisted in the application database.
### Agent Storage
Custom agent **definitions** (`config.yaml` + `SOUL.md`) are stored per-user on
local disk by default. This is separate from the `database` backend (which holds
run/thread/event data) and from agent memory.
```yaml
agent_storage:
backend: file # file (default) | db
```
- `backend: file` — the historical layout under `{base_dir}/users/{user_id}/agents/`. Single-node by construction: an agent created on one node is not visible to other nodes without a shared mount.
- `backend: db` — one row per agent in the shared SQL persistence layer (a new `agents` table), so every node in a multi-instance deployment sees the same agents. Requires `database.backend` to be `sqlite` or `postgres`; the Gateway **fails fast at startup** if it is `memory` (a per-process database cannot share definitions).
- `agent_storage` is restart-required (the backend is captured at Gateway lifespan startup).
- In a multi-worker Postgres deployment (`GATEWAY_WORKERS > 1`), leaving `agent_storage.backend: file` logs a startup warning — agents written to one node's local disk are invisible to the others, which is exactly the divergence the `db` backend fixes.
Migrating an existing install from `file` to `db`:
```bash
python backend/scripts/migrate_agents_to_db.py # copy on-disk agents into the db
python backend/scripts/migrate_agents_to_db.py --dry-run # preview without writing
```
The importer is idempotent (already-present agents are skipped) and leaves the source files untouched, so reverting `agent_storage.backend` to `file` is a clean rollback. Agent *memory* (`memory.json`) is unaffected by this switch.
### Tools
Configure specific tools available to the agent:

37
backend/packages/harness/deerflow/config/agent_storage_config.py Normal file
View File

@ -0,0 +1,37 @@
"""Custom-agent definition storage configuration.
Controls where custom agent *definitions* (``config.yaml`` + ``SOUL.md``) are
persisted. This is orthogonal to :class:`DatabaseConfig` (which governs the
run/thread/event persistence layer) and to the deermem memory store.
Backends:
- file: Per-user files under ``{base_dir}/users/{user_id}/agents/{name}/``
(today's layout, unchanged). Single-node by construction — an agent created
on one node is invisible to other nodes without a shared mount. This is the
default so single-node and zero-config development are unaffected.
- db: A row in the ``agents`` table of the existing SQL persistence layer,
shared by every node. Requires ``database.backend`` to be ``sqlite`` or
``postgres`` (validated at startup; see the gateway ``deps`` module).
Agent *memory* (``memory.json``) is a separate concern handled by the deermem
storage layer and is not affected by this switch.
"""
from __future__ import annotations
from typing import Literal
from pydantic import BaseModel, Field
class AgentStorageConfig(BaseModel):
backend: Literal["file", "db"] = Field(
default="file",
description=(
"Storage backend for custom agent definitions (config.yaml + SOUL.md). "
"'file' (default) keeps today's per-user on-disk layout — single-node only. "
"'db' stores each agent as a row in the shared SQL persistence layer so a "
"multi-instance deployment sees the same agents on every node; it requires "
"database.backend to be 'sqlite' or 'postgres'."
),
)

137
backend/packages/harness/deerflow/config/agents_config.py
View File

@ -10,9 +10,8 @@ per-user layout.
import logging
import re
from pathlib import Path
from typing import Any, Literal
from typing import Literal
import yaml
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from deerflow.config.paths import get_paths
@ -290,10 +289,12 @@ def resolve_agent_dir(name: str, *, user_id: str | None = None) -> Path:
def load_agent_config(name: str | None, *, user_id: str | None = None) -> AgentConfig | None:
"""Load the custom or default agent's config from its directory.
"""Load the custom or default agent's config.
Reads from the per-user layout first; falls back to the legacy shared layout
for installations that have not yet been migrated.
Dispatches to the configured agent store (``agent_storage.backend``): the
``file`` backend reads the per-user layout first and falls back to the legacy
shared layout; the ``db`` backend reads the shared ``agents`` table. Behaviour
and error semantics are unchanged from the historical file-only loader.
Args:
name: The agent name.
@ -304,45 +305,25 @@ def load_agent_config(name: str | None, *, user_id: str | None = None) -> AgentC
AgentConfig instance, or ``None`` if ``name`` is ``None``.
Raises:
FileNotFoundError: If the agent directory or config.yaml does not exist.
ValueError: If config.yaml cannot be parsed.
FileNotFoundError: If the agent does not exist.
ValueError: If the stored config cannot be parsed.
"""
if name is None:
return None
# Lazy import: the store package imports back from this module.
from deerflow.persistence.agents import get_agent_store
name = validate_agent_name(name)
agent_dir = resolve_agent_dir(name, user_id=user_id)
config_file = agent_dir / "config.yaml"
if not agent_dir.exists():
raise FileNotFoundError(f"Agent directory not found: {agent_dir}")
if not config_file.exists():
raise FileNotFoundError(f"Agent config not found: {config_file}")
try:
with open(config_file, encoding="utf-8") as f:
data: dict[str, Any] = yaml.safe_load(f) or {}
except yaml.YAMLError as e:
raise ValueError(f"Failed to parse agent config {config_file}: {e}") from e
# Ensure name is set from directory name if not in file
if "name" not in data:
data["name"] = name
# Strip unknown fields before passing to Pydantic (e.g. legacy prompt_file)
known_fields = set(AgentConfig.model_fields.keys())
data = {k: v for k, v in data.items() if k in known_fields}
return AgentConfig(**data)
return get_agent_store().get(name, user_id=user_id)
def load_agent_soul(agent_name: str | None, *, user_id: str | None = None) -> str | None:
"""Read the SOUL.md file for a custom agent, if it exists.
"""Read the SOUL.md content for an agent, if any.
SOUL.md defines the agent's personality, values, and behavioral guardrails.
It is injected into the lead agent's system prompt as additional context.
The default agent (``agent_name`` falsy) always reads ``{base_dir}/SOUL.md``
directly — it is not a custom-agent record — regardless of backend. A named
agent dispatches to the configured store.
Args:
agent_name: The name of the agent or None for the default agent.
@ -350,86 +331,34 @@ def load_agent_soul(agent_name: str | None, *, user_id: str | None = None) -> st
current request context.
Returns:
The SOUL.md content as a string, or None if the file does not exist.
The SOUL.md content as a string, or None if not set.
"""
if agent_name:
agent_dir = resolve_agent_dir(agent_name, user_id=user_id)
soul_path = agent_dir / SOUL_FILENAME
# Fallback: resolve_agent_dir requires config.yaml to be present
# (see #3390), but SOUL.md loading does not depend on config.yaml.
# If the resolved dir doesn't have config.yaml (meaning the resolver
# returned its default path because no agent dir qualified) and also
# lacks SOUL.md, check the per-user and legacy directories directly
# so that agents configured via DEER_FLOW_CONFIG_PATH (or any setup
# where the agent dir has SOUL.md but no config.yaml) can still load
# their soul (#4135). The config.yaml guard ensures this fallback
# only fires for dirs the resolver couldn't resolve, not for a
# properly-resolved per-user agent that simply lacks SOUL.md -
# preserving the "per-user entries fully shadow legacy entries"
# invariant (agents_config.py:3-7, list_custom_agents).
if not soul_path.exists() and not (agent_dir / "config.yaml").exists():
paths = get_paths()
effective_user = user_id or get_effective_user_id()
for candidate in (
paths.user_agent_dir(effective_user, agent_name),
paths.agent_dir(agent_name),
):
if (candidate / SOUL_FILENAME).exists():
soul_path = candidate / SOUL_FILENAME
break
else:
agent_dir = get_paths().base_dir
soul_path = agent_dir / SOUL_FILENAME
if not soul_path.exists():
return None
content = soul_path.read_text(encoding="utf-8").strip()
return content or None
if not agent_name:
soul_path = get_paths().base_dir / SOUL_FILENAME
if not soul_path.exists():
return None
content = soul_path.read_text(encoding="utf-8").strip()
return content or None
from deerflow.persistence.agents import get_agent_store
return get_agent_store().get_soul(agent_name, user_id=user_id)
def list_custom_agents(*, user_id: str | None = None) -> list[AgentConfig]:
"""Scan the agents directory and return all valid custom agents.
"""Return all valid custom agents for ``user_id``.
Returns the union of agents in the per-user layout and the legacy shared
layout, so that pre-migration installations remain visible until they are
migrated. Per-user entries shadow legacy entries with the same name.
Dispatches to the configured agent store. The ``file`` backend returns the
union of the per-user layout and the legacy shared layout (per-user entries
shadow legacy entries with the same name); the ``db`` backend returns the
user's rows. Sorted by name.
Args:
user_id: Owner whose agents to list. Defaults to the effective user
from the current request context.
Returns:
List of AgentConfig for each valid agent directory found.
List of AgentConfig for each valid agent found.
"""
paths = get_paths()
effective_user = user_id or get_effective_user_id()
from deerflow.persistence.agents import get_agent_store
seen: set[str] = set()
agents: list[AgentConfig] = []
user_root = paths.user_agents_dir(effective_user)
legacy_root = paths.agents_dir
for root in (user_root, legacy_root):
if not root.exists():
continue
for entry in sorted(root.iterdir()):
if not entry.is_dir():
continue
if entry.name in seen:
continue
config_file = entry / "config.yaml"
if not config_file.exists():
logger.debug(f"Skipping {entry.name}: no config.yaml")
continue
try:
agent_cfg = load_agent_config(entry.name, user_id=effective_user)
if agent_cfg is None:
continue
agents.append(agent_cfg)
seen.add(entry.name)
except Exception as e:
logger.warning(f"Skipping agent '{entry.name}': {e}")
agents.sort(key=lambda a: a.name)
return agents
return get_agent_store().list(user_id=user_id)

8
backend/packages/harness/deerflow/config/app_config.py
View File

@ -10,6 +10,7 @@ from dotenv import load_dotenv
from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, model_validator
from deerflow.config.acp_config import ACPAgentConfig, load_acp_config_from_dict
from deerflow.config.agent_storage_config import AgentStorageConfig
from deerflow.config.agents_api_config import AgentsApiConfig, load_agents_api_config_from_dict
from deerflow.config.auth_config import AuthAppConfig
from deerflow.config.authorization_config import AuthorizationConfig, load_authorization_config_from_dict
@ -260,6 +261,13 @@ class AppConfig(BaseModel):
field_doc="Run-event store backend (memory for dev, db for production queries, jsonl for lightweight single-node persistence).",
),
)
agent_storage: AgentStorageConfig = Field(
default_factory=AgentStorageConfig,
description=format_field_description(
"agent_storage",
field_doc="Custom agent definition storage backend ('file' for today's per-user on-disk layout, 'db' to share definitions across nodes via the SQL persistence layer).",
),
)
scheduler: SchedulerConfig = Field(
default_factory=SchedulerConfig,
description=format_field_description(

25
backend/packages/harness/deerflow/config/database_config.py
View File

@ -124,3 +124,28 @@ class DatabaseConfig(BaseModel):
url = url.replace("postgres://", "postgresql+asyncpg://", 1)
return url
raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")
@property
def app_sync_sqlalchemy_url(self) -> str:
"""SQLAlchemy *synchronous* URL for the application ORM data.
Used by the ``agent_storage.backend: db`` store, whose consumers (the
LangGraph graph factory, the setup/update tools) are synchronous and may
run on the event loop or in a separate process from the gateway, where an
async engine cannot be driven. Points at the same database file/server as
:meth:`app_sqlalchemy_url`; only the driver differs (both drivers —
stdlib sqlite3 and psycopg — ship with the app, so this adds no
dependency).
"""
if self.backend == "sqlite":
return f"sqlite:///{self.sqlite_path}"
if self.backend == "postgres":
url = self.postgres_url
if url.startswith("postgresql+asyncpg://"):
url = url.replace("postgresql+asyncpg://", "postgresql+psycopg://", 1)
elif url.startswith("postgresql://"):
url = url.replace("postgresql://", "postgresql+psycopg://", 1)
elif url.startswith("postgres://"):
url = url.replace("postgres://", "postgresql+psycopg://", 1)
return url
raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")

1
backend/packages/harness/deerflow/config/reload_boundary.py
View File

@ -46,6 +46,7 @@ STARTUP_ONLY_FIELDS: dict[str, str] = {
"database": ("init_engine_from_config() runs once during langgraph_runtime() startup; the SQLAlchemy engine holds the connection pool and is not rebuilt on config.yaml edits."),
"checkpointer": ("make_checkpointer() binds the persistent checkpointer once at startup, including SQLite WAL / busy_timeout settings."),
"run_events": ("make_run_event_store() picks the memory- vs SQL-backed implementation at startup and is frozen onto app.state.run_events_config to stay paired with the underlying event store."),
"agent_storage": ("langgraph_runtime() validates agent_storage.backend against database.backend once at startup, and the db backend's synchronous SQLAlchemy engine is process-cached on first use; switching backend needs a restart."),
"stream_bridge": ("make_stream_bridge() constructs the stream-bridge singleton once during startup."),
"sandbox": ("get_sandbox_provider() caches the provider singleton (``_default_sandbox_provider``); a different ``sandbox.use`` class path only takes effect on next process start."),
"log_level": (

94
backend/packages/harness/deerflow/persistence/agents/__init__.py Normal file
View File

@ -0,0 +1,94 @@
"""Custom agent definition persistence — abstract store + file/db backends.
The public entry point is :func:`get_agent_store`, which the free functions in
:mod:`deerflow.config.agents_config` dispatch to. ``file`` (default) preserves
today's on-disk layout; ``db`` shares definitions across nodes via the SQL
persistence layer.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from deerflow.persistence.agents.base import (
AgentDeleteOutcome,
AgentExistsError,
AgentStore,
parse_agent_config,
)
from deerflow.persistence.agents.model import AgentRow
if TYPE_CHECKING:
from deerflow.config.app_config import AppConfig
__all__ = [
"AgentDeleteOutcome",
"AgentExistsError",
"AgentRow",
"AgentStore",
"get_agent_store",
"make_agent_store",
"parse_agent_config",
]
_file_store_singleton: AgentStore | None = None
def make_agent_store(config: AppConfig) -> AgentStore:
"""Build (or reuse) the store selected by ``config.agent_storage.backend``.
``db`` requires ``database.backend`` to be ``sqlite`` or ``postgres``; a
``memory`` database has no durable URL and is rejected here (the gateway
also fails fast at startup, but this guard covers the graph-process path).
"""
if config.agent_storage.backend == "db":
db_backend = config.database.backend
if db_backend not in ("sqlite", "postgres"):
raise ValueError(
f"agent_storage.backend='db' requires database.backend to be 'sqlite' or 'postgres', "
f"but database.backend is '{db_backend}'. A 'memory' database is per-process and cannot "
"share agent definitions across nodes; set database.backend accordingly or use "
"agent_storage.backend='file'."
)
from deerflow.persistence.agents.sql import SqlAgentStore
return SqlAgentStore(config.database.app_sync_sqlalchemy_url)
return _file_store()
def get_agent_store() -> AgentStore:
"""Return the store for the current process's configuration.
Defaults to the file backend when no app config can be resolved — the free
functions in ``agents_config`` must keep working in lightweight contexts
(CLI, tests, tools) that never load a full ``config.yaml``. Only an
explicit ``agent_storage.backend: db`` diverges from the file default.
Cross-process invariant (the ``db`` backend's whole point): the per-run
agent build runs in the **graph subprocess**, a different process from the
gateway. Its cross-node guarantee holds only because ``get_app_config()``
resolves ``config.yaml`` there too and returns ``backend: db`` — so the read
path in the graph process sees the same shared table the gateway wrote. The
``except`` below is a genuine *no resolvable config* fallback (CLI/tests),
**not** a mask for a misconfigured graph process: if ``config.yaml`` is
reachable there (it is, same working tree), ``db`` is honoured, not silently
downgraded to node-local ``file``. Pinned by
``test_get_agent_store_resolves_db_backend_from_on_disk_config``.
"""
from deerflow.config.app_config import get_app_config
try:
config = get_app_config()
except Exception: # noqa: BLE001 — no resolvable config → file default
return _file_store()
return make_agent_store(config)
def _file_store() -> AgentStore:
global _file_store_singleton
if _file_store_singleton is None:
from deerflow.persistence.agents.file import FileAgentStore
_file_store_singleton = FileAgentStore()
return _file_store_singleton

131
backend/packages/harness/deerflow/persistence/agents/base.py Normal file
View File

@ -0,0 +1,131 @@
"""Abstract interface for custom agent definition storage.
Two implementations:
- :class:`FileAgentStore` — the historical per-user on-disk layout
(``config.yaml`` + ``SOUL.md``), still the default and behaviourally
unchanged.
- :class:`SqlAgentStore` — a row per agent in the shared SQL persistence
layer, so every node in a multi-instance deployment sees the same agents.
The store is deliberately **synchronous**. Its consumers — the LangGraph graph
factory (``make_lead_agent``), the ``setup_agent`` / ``update_agent`` tools, and
the GitHub agent registry — are synchronous and may run on the event loop or in
a separate process from the gateway, where an async engine cannot be driven.
Async HTTP routes call the store via ``asyncio.to_thread`` (the same pattern the
agents router already uses for filesystem work).
``user_id`` semantics (kept identical to the pre-refactor free functions in
:mod:`deerflow.config.agents_config`, which is what makes the file backend a
behaviour-neutral move): ``None`` resolves to the effective user from the
request context — ``"default"`` in no-auth mode — via
:func:`deerflow.runtime.user_context.get_effective_user_id`. This is filesystem
bucket semantics, distinct from the AUTO/None sentinel used by the async
``thread_meta`` repositories.
"""
from __future__ import annotations
import abc
from collections.abc import Hashable
from typing import Any, Literal
from deerflow.config.agents_config import AgentConfig
def parse_agent_config(data: dict[str, Any], name: str) -> AgentConfig:
"""Build an :class:`AgentConfig` from a raw config *document*, shared by both backends.
Sets ``name`` from the natural key when the document omits it and strips
unknown keys (e.g. a legacy ``prompt_file``) before validation — identical
to the pre-refactor ``load_agent_config``.
"""
data = dict(data)
if "name" not in data:
data["name"] = name
known_fields = set(AgentConfig.model_fields.keys())
data = {k: v for k, v in data.items() if k in known_fields}
return AgentConfig(**data)
# Delete outcome, mirroring the agents router's result:
# a row/dir was removed ("deleted"); only a legacy shared-layout entry exists,
# which the current write path never removes ("legacy"); nothing was there
# ("missing"); or a per-user directory exists holding memory/facts data but is
# not a custom agent (no config.yaml), so it is preserved rather than deleting a
# user's memory ("not-custom-agent", #4279).
AgentDeleteOutcome = Literal["deleted", "legacy", "missing", "not-custom-agent"]
class AgentExistsError(Exception):
"""Raised by :meth:`AgentStore.create` when ``(user_id, name)`` already exists."""
class AgentStore(abc.ABC):
@abc.abstractmethod
def get(self, name: str, *, user_id: str | None = None) -> AgentConfig:
"""Return the agent's config.
Raises :class:`FileNotFoundError` if the agent does not exist — the
historical contract that ``routers/agents.py`` and ``update_agent`` rely
on to surface a 404 / "does not exist" error.
"""
@abc.abstractmethod
def exists(self, name: str, *, user_id: str | None = None) -> bool:
"""Return whether ``name`` is already taken for ``user_id``.
Consistent with :meth:`create`'s conflict rule (so an "available" name
never then 409s): the file backend treats any per-user or legacy
directory as taken; the db backend checks for a row.
"""
@abc.abstractmethod
def get_soul(self, name: str, *, user_id: str | None = None) -> str | None:
"""Return the agent's ``SOUL.md`` content, or ``None`` if unset/empty."""
@abc.abstractmethod
def list(self, *, user_id: str | None = None) -> list[AgentConfig]:
"""Return every custom agent owned by ``user_id``, sorted by name."""
@abc.abstractmethod
def list_all(self) -> list[tuple[str, AgentConfig]]:
"""Return ``(user_id, config)`` for every agent across all owners.
Used by the GitHub registry, which scans all users' agents for repo
bindings. Ordering is deterministic (by ``user_id`` then name).
"""
@abc.abstractmethod
def create(self, name: str, config: dict, soul: str, *, user_id: str | None = None) -> None:
"""Persist a new agent from the config *document* each write surface builds.
``config`` is the raw dict the caller assembled (the same one previously
written to ``config.yaml``); passing the document rather than a
re-serialized :class:`AgentConfig` keeps the on-disk bytes and the
"only present keys are written" behaviour identical to the pre-refactor
writers. Raises :class:`AgentExistsError` on an existing ``(user_id, name)``.
"""
@abc.abstractmethod
def update(self, name: str, config: dict | None, soul: str | None, *, user_id: str | None = None) -> None:
"""Write an agent's config and/or soul (upsert).
``config`` / ``soul`` are each independently optional: ``None`` means
"leave that part unchanged" — the agents router updates config only when
a config field changed, and soul only when one was supplied. Creates the
record if it does not exist (the ``setup_agent`` / first-write path).
"""
@abc.abstractmethod
def delete(self, name: str, *, user_id: str | None = None) -> AgentDeleteOutcome:
"""Delete an agent and its co-located memory, returning the outcome."""
@abc.abstractmethod
def signature(self) -> Hashable:
"""Return an opaque change token for cache invalidation.
Equal tokens mean "nothing changed since last read". The GitHub registry
keys its cache off this instead of ``stat()`` so it works for both
backends (mtime triples for ``file``; ``max(updated_at)`` + row count for
``db``).
"""

259
backend/packages/harness/deerflow/persistence/agents/file.py Normal file
View File

@ -0,0 +1,259 @@
"""Filesystem-backed agent store — today's per-user layout, behaviour-preserving.
The read methods are the pre-refactor bodies of ``load_agent_config`` /
``load_agent_soul`` / ``list_custom_agents`` (so the free functions in
:mod:`deerflow.config.agents_config` dispatch here without changing behaviour).
Writes use a staged temp-file + atomic ``os.replace`` commit — the crash-safety
the ``update_agent`` tool already had, applied uniformly to create/update.
Path/user resolution is done through the :mod:`deerflow.config.agents_config`
module object (``_ac.get_paths`` / ``_ac.get_effective_user_id``) rather than
direct imports, so it honours the same monkeypatch seams the existing agent
tests target.
"""
from __future__ import annotations
import logging
import shutil
import tempfile
from collections.abc import Hashable
from pathlib import Path
from typing import Any
import yaml
from deerflow.config import agents_config as _ac
from deerflow.config.agents_config import (
SOUL_FILENAME,
AgentConfig,
resolve_agent_dir,
validate_agent_name,
)
from deerflow.persistence.agents.base import (
AgentDeleteOutcome,
AgentExistsError,
AgentStore,
parse_agent_config,
)
from deerflow.runtime.user_context import DEFAULT_USER_ID
logger = logging.getLogger(__name__)
class FileAgentStore(AgentStore):
def get(self, name: str, *, user_id: str | None = None) -> AgentConfig:
name = validate_agent_name(name)
agent_dir = resolve_agent_dir(name, user_id=user_id)
config_file = agent_dir / "config.yaml"
if not agent_dir.exists():
raise FileNotFoundError(f"Agent directory not found: {agent_dir}")
if not config_file.exists():
raise FileNotFoundError(f"Agent config not found: {config_file}")
try:
with open(config_file, encoding="utf-8") as f:
data: dict[str, Any] = yaml.safe_load(f) or {}
except yaml.YAMLError as e:
raise ValueError(f"Failed to parse agent config {config_file}: {e}") from e
return parse_agent_config(data, name)
def exists(self, name: str, *, user_id: str | None = None) -> bool:
name = validate_agent_name(name)
paths = _ac.get_paths()
effective_user = user_id or _ac.get_effective_user_id()
return paths.user_agent_dir(effective_user, name).exists() or paths.agent_dir(name).exists()
def get_soul(self, name: str, *, user_id: str | None = None) -> str | None:
agent_dir = resolve_agent_dir(name, user_id=user_id)
soul_path = agent_dir / SOUL_FILENAME
# resolve_agent_dir requires config.yaml; SOUL.md loading does not, so
# when the resolver fell back to its default path (no qualifying dir),
# check the per-user and legacy dirs directly (#4135). The config.yaml
# guard keeps this fallback from firing for a properly-resolved per-user
# agent that merely lacks SOUL.md (preserves per-user-shadows-legacy).
if not soul_path.exists() and not (agent_dir / "config.yaml").exists():
paths = _ac.get_paths()
effective_user = user_id or _ac.get_effective_user_id()
for candidate in (
paths.user_agent_dir(effective_user, name),
paths.agent_dir(name),
):
if (candidate / SOUL_FILENAME).exists():
soul_path = candidate / SOUL_FILENAME
break
if not soul_path.exists():
return None
content = soul_path.read_text(encoding="utf-8").strip()
return content or None
def list(self, *, user_id: str | None = None) -> list[AgentConfig]:
paths = _ac.get_paths()
effective_user = user_id or _ac.get_effective_user_id()
seen: set[str] = set()
agents: list[AgentConfig] = []
for root in (paths.user_agents_dir(effective_user), paths.agents_dir):
if not root.exists():
continue
for entry in sorted(root.iterdir()):
if not entry.is_dir() or entry.name in seen:
continue
if not (entry / "config.yaml").exists():
logger.debug("Skipping %s: no config.yaml", entry.name)
continue
try:
agents.append(self.get(entry.name, user_id=effective_user))
seen.add(entry.name)
except Exception as e: # noqa: BLE001 — one bad agent must not hide the rest
logger.warning("Skipping agent '%s': %s", entry.name, e)
agents.sort(key=lambda a: a.name)
return agents
def list_all(self) -> list[tuple[str, AgentConfig]]:
result: list[tuple[str, AgentConfig]] = []
for user_id, name in self._discover():
try:
result.append((user_id, self.get(name, user_id=user_id)))
except Exception as e: # noqa: BLE001
logger.warning("list_all: skipping agent %s/%s: %s", user_id, name, e)
return result
def create(self, name: str, config: dict, soul: str, *, user_id: str | None = None) -> None:
name = validate_agent_name(name)
paths = _ac.get_paths()
effective_user = user_id or _ac.get_effective_user_id()
agent_dir = paths.user_agent_dir(effective_user, name)
# Refuse if a per-user directory OR a legacy shared directory already
# owns the name — the agents router's 409 semantics (a legacy agent must
# not be shadowed; a per-user dir may be memory-only but still blocks).
if agent_dir.exists() or paths.agent_dir(name).exists():
raise AgentExistsError(f"Agent '{name}' already exists for user '{effective_user}'")
try:
agent_dir.mkdir(parents=True, exist_ok=False)
except FileExistsError as e:
# A concurrent create passed the existence check above and reached
# mkdir first. Surface the router's 409 (via AgentExistsError) rather
# than a generic 500 — mirrors SqlAgentStore's IntegrityError path.
raise AgentExistsError(f"Agent '{name}' already exists for user '{effective_user}'") from e
try:
self._write(agent_dir, config, soul)
except Exception:
# The directory was newly created for this call; a failed write
# must not leave an empty/partial agent dir behind.
shutil.rmtree(agent_dir, ignore_errors=True)
raise
def update(self, name: str, config: dict | None, soul: str | None, *, user_id: str | None = None) -> None:
name = validate_agent_name(name)
effective_user = user_id or _ac.get_effective_user_id()
agent_dir = _ac.get_paths().user_agent_dir(effective_user, name)
pre_existing = agent_dir.exists()
agent_dir.mkdir(parents=True, exist_ok=True)
try:
self._write(agent_dir, config, soul)
except Exception:
# Only clean up a directory this call created — never delete a
# pre-existing agent on a failed write.
if not pre_existing:
shutil.rmtree(agent_dir, ignore_errors=True)
raise
def delete(self, name: str, *, user_id: str | None = None) -> AgentDeleteOutcome:
name = validate_agent_name(name)
paths = _ac.get_paths()
effective_user = user_id or _ac.get_effective_user_id()
agent_dir = paths.user_agent_dir(effective_user, name)
if not agent_dir.exists():
# A legacy shared-layout agent is intentionally left in place (the
# write path never targets it); report it distinctly.
return "legacy" if paths.agent_dir(name).exists() else "missing"
if not (agent_dir / "config.yaml").is_file():
# The directory holds memory/facts data but is not a custom agent
# (no config.yaml) — preserve it rather than deleting a user's memory
# (#4279). rmtree below would otherwise take the whole tree.
return "not-custom-agent"
# rmtree removes config.yaml, SOUL.md and the co-located memory.json in
# one shot — the historical behaviour.
shutil.rmtree(agent_dir)
return "deleted"
def signature(self) -> Hashable:
sig: list[tuple[str, str, float]] = []
for user_id, name in self._discover():
config = resolve_agent_dir(name, user_id=user_id) / "config.yaml"
try:
sig.append((user_id, name, config.stat().st_mtime))
except OSError:
continue
return tuple(sig)
# -- internals --
def _discover(self) -> list[tuple[str, str]]:
"""Enumerate ``(user_id, name)`` across per-user and legacy layouts.
A legacy shared-layout agent is attributed to ``DEFAULT_USER_ID`` and is
shadowed only by a ``users/default/`` agent of the same name — not by an
agent another user happens to own — matching the GitHub registry's
historical discovery (``load_agent_config(name)`` resolves a legacy agent
under ``DEFAULT_USER_ID``).
"""
paths = _ac.get_paths()
discovered: list[tuple[str, str]] = []
seen: set[tuple[str, str]] = set()
users_root = paths.base_dir / "users"
if users_root.exists():
for user_entry in sorted(users_root.iterdir()):
if not user_entry.is_dir():
continue
agents_root = paths.user_agents_dir(user_entry.name)
if not agents_root.exists():
continue
for entry in sorted(agents_root.iterdir()):
if entry.is_dir() and (entry / "config.yaml").exists():
key = (user_entry.name, entry.name)
discovered.append(key)
seen.add(key)
legacy_root = paths.agents_dir
if legacy_root.exists():
for entry in sorted(legacy_root.iterdir()):
if entry.is_dir() and (entry / "config.yaml").exists() and (DEFAULT_USER_ID, entry.name) not in seen:
discovered.append((DEFAULT_USER_ID, entry.name))
return discovered
@staticmethod
def _write(agent_dir: Path, config: dict | None, soul: str | None) -> None:
"""Write config.yaml and/or SOUL.md, each via an atomic ``os.replace``.
Each part is written only when supplied (``config``/``soul`` non-None),
staged to a temp file then committed with ``os.replace``, so neither file
is ever observed half-written. The two commits are sequential, **not** a
single transaction: a crash between them can leave a freshly-replaced
config.yaml beside a stale SOUL.md (single-node, sub-millisecond window).
The ``db`` backend commits both fields in one transaction; if cross-file
atomicity ever matters here, restore ``update_agent``'s partial-write
reporting.
"""
pending: list[tuple[Path, Path]] = []
staged: list[Path] = []
try:
if config is not None:
config_text = yaml.dump(config, default_flow_style=False, allow_unicode=True, sort_keys=False)
config_tmp = _stage_temp(agent_dir / "config.yaml", config_text)
staged.append(config_tmp)
pending.append((config_tmp, agent_dir / "config.yaml"))
if soul is not None:
soul_tmp = _stage_temp(agent_dir / SOUL_FILENAME, soul)
staged.append(soul_tmp)
pending.append((soul_tmp, agent_dir / SOUL_FILENAME))
for tmp, target in pending:
tmp.replace(target)
staged.remove(tmp)
finally:
for tmp in staged:
tmp.unlink(missing_ok=True)
def _stage_temp(target: Path, text: str) -> Path:
with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=target.parent, suffix=".tmp", delete=False) as f:
f.write(text)
return Path(f.name)

42
backend/packages/harness/deerflow/persistence/agents/model.py Normal file
View File

@ -0,0 +1,42 @@
"""ORM model for custom agent definitions.
One row per ``(user_id, name)`` custom agent. ``config`` holds the full
:class:`~deerflow.config.agents_config.AgentConfig` document *minus* ``name``
(which is the natural key, carried by the ``name`` column). Storing the config
as a single JSON document — rather than a column per field — is deliberate: the
codebase already declares, via ``preserve_non_managed_fields``, that any field
added to ``AgentConfig`` in the future must round-trip through writers that do
not know it. A document column honours that invariant with zero schema churn
(a new ``AgentConfig`` field needs no migration here); the only queries are by
``(user_id, name)`` and list-by-user, which are exactly the indexed columns.
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import JSON, DateTime, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
class AgentRow(Base):
__tablename__ = "agents"
__table_args__ = (UniqueConstraint("user_id", "name", name="uq_agents_user_name"),)
# Surrogate primary key (uuid4 hex). The natural key is (user_id, name),
# enforced by the UNIQUE constraint above; a surrogate PK keeps the row
# identity stable if an agent is ever renamed in a future revision.
id: Mapped[str] = mapped_column(String(64), primary_key=True)
user_id: Mapped[str] = mapped_column(String(64), index=True)
# Stored lowercase, matching the on-disk layout (Paths.user_agent_dir lowercases).
name: Mapped[str] = mapped_column(String(128))
config: Mapped[dict] = mapped_column(JSON, default=dict)
soul: Mapped[str] = mapped_column(Text, default="")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
default=lambda: datetime.now(UTC),
onupdate=lambda: datetime.now(UTC),
)

221
backend/packages/harness/deerflow/persistence/agents/sql.py Normal file
View File

@ -0,0 +1,221 @@
"""SQL-backed agent store (synchronous).
Serves the ``agent_storage.backend: db`` path. It is intentionally synchronous
and uses its own small engine (see :mod:`deerflow.persistence.agents.base` for
why the store is sync). The engine points at the same database the async
persistence layer manages — the ``agents`` table is created by that layer's
Alembic bootstrap (migration ``0006``); this store only reads and writes rows.
Both the sqlite (stdlib) and postgres (psycopg) sync drivers already ship with
the app, so this adds no dependency.
"""
from __future__ import annotations
import logging
import shutil
import threading
import uuid
from collections.abc import Hashable
from datetime import UTC, datetime
from sqlalchemy import Engine, create_engine, delete, event, func, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, sessionmaker
from deerflow.config.agents_config import AgentConfig
from deerflow.config.paths import get_paths
from deerflow.persistence.agents.base import (
AgentDeleteOutcome,
AgentExistsError,
AgentStore,
parse_agent_config,
)
from deerflow.persistence.agents.model import AgentRow
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.utils.time import coerce_iso
logger = logging.getLogger(__name__)
# Cache sync engines by URL: the store is constructed on demand in multiple
# places (gateway routes, the graph factory) and each process should reuse one
# engine/pool rather than opening a connection per call. The lock keeps two
# threads first-touching the same URL from building — and registering connect
# listeners on — duplicate engines.
_engines: dict[str, Engine] = {}
_engines_lock = threading.Lock()
def _build_engine(url: str) -> Engine:
connect_args = {"check_same_thread": False} if url.startswith("sqlite") else {}
engine = create_engine(url, future=True, pool_pre_ping=True, connect_args=connect_args)
if url.startswith("sqlite"):
# Mirror the async engine's per-connection PRAGMAs (persistence/engine.py).
# journal_mode=WAL is persistent on the DB file (the async bootstrap sets
# it), but synchronous and busy_timeout are per-connection: without this
# these sync connections run synchronous=FULL and pysqlite's default 5s
# busy_timeout rather than the async engine's NORMAL + 30s. Match them so
# both engines behave identically against the shared DB and a concurrent
# writer waits up to 30s instead of failing early on lock contention.
@event.listens_for(engine, "connect")
def _enable_sqlite_pragmas(dbapi_conn, _record): # noqa: ARG001 — SQLAlchemy contract
cursor = dbapi_conn.cursor()
try:
cursor.execute("PRAGMA journal_mode=WAL;")
cursor.execute("PRAGMA synchronous=NORMAL;")
cursor.execute("PRAGMA foreign_keys=ON;")
cursor.execute("PRAGMA busy_timeout=30000;")
finally:
cursor.close()
return engine
def _get_sessionmaker(url: str) -> sessionmaker[Session]:
engine = _engines.get(url)
if engine is None:
with _engines_lock:
engine = _engines.get(url)
if engine is None:
engine = _build_engine(url)
_engines[url] = engine
return sessionmaker(engine, expire_on_commit=False)
def _config_document(config: dict) -> dict:
"""Strip the natural key from the stored document (``name`` is its own column)."""
return {k: v for k, v in config.items() if k != "name"}
class SqlAgentStore(AgentStore):
def __init__(self, url: str) -> None:
self._Session = _get_sessionmaker(url)
def _row(self, session: Session, name: str, user_id: str) -> AgentRow | None:
stmt = select(AgentRow).where(AgentRow.user_id == user_id, AgentRow.name == name.lower())
return session.execute(stmt).scalar_one_or_none()
def get(self, name: str, *, user_id: str | None = None) -> AgentConfig:
effective_user = user_id or get_effective_user_id()
with self._Session() as session:
row = self._row(session, name, effective_user)
if row is None:
raise FileNotFoundError(f"Agent config not found: {name} (user {effective_user})")
return parse_agent_config(row.config or {}, row.name)
def exists(self, name: str, *, user_id: str | None = None) -> bool:
effective_user = user_id or get_effective_user_id()
with self._Session() as session:
return self._row(session, name, effective_user) is not None
def get_soul(self, name: str, *, user_id: str | None = None) -> str | None:
effective_user = user_id or get_effective_user_id()
with self._Session() as session:
row = self._row(session, name, effective_user)
if row is None:
return None
return row.soul or None
def list(self, *, user_id: str | None = None) -> list[AgentConfig]:
effective_user = user_id or get_effective_user_id()
stmt = select(AgentRow).where(AgentRow.user_id == effective_user).order_by(AgentRow.name.asc())
with self._Session() as session:
rows = list(session.execute(stmt).scalars())
return [parse_agent_config(r.config or {}, r.name) for r in rows]
def list_all(self) -> list[tuple[str, AgentConfig]]:
stmt = select(AgentRow).order_by(AgentRow.user_id.asc(), AgentRow.name.asc())
with self._Session() as session:
rows = list(session.execute(stmt).scalars())
return [(r.user_id, parse_agent_config(r.config or {}, r.name)) for r in rows]
def create(self, name: str, config: dict, soul: str, *, user_id: str | None = None) -> None:
effective_user = user_id or get_effective_user_id()
now = datetime.now(UTC)
row = AgentRow(
id=uuid.uuid4().hex,
user_id=effective_user,
name=name.lower(),
config=_config_document(config),
soul=soul or "",
created_at=now,
updated_at=now,
)
try:
with self._Session() as session:
session.add(row)
session.commit()
except IntegrityError as e:
# UNIQUE(user_id, name) turns the check-then-write race into a clean conflict.
raise AgentExistsError(f"Agent '{name}' already exists for user '{effective_user}'") from e
def update(self, name: str, config: dict | None, soul: str | None, *, user_id: str | None = None) -> None:
effective_user = user_id or get_effective_user_id()
with self._Session() as session:
row = self._row(session, name, effective_user)
if row is not None:
self._apply_update(row, config, soul)
session.commit()
return
# Upsert: setup_agent and any first-time write land here. Two
# concurrent first-time updates (e.g. two setup_agent handshakes) can
# both see row is None and both insert; UNIQUE(user_id, name) rejects
# the loser. Re-fetch the winner's row and apply the update to it
# rather than letting a raw IntegrityError surface as a 500 — a true
# upsert, symmetric with create()'s conflict handling.
row = AgentRow(
id=uuid.uuid4().hex,
user_id=effective_user,
name=name.lower(),
config=_config_document(config or {}),
soul=soul or "",
)
session.add(row)
try:
session.commit()
except IntegrityError:
session.rollback()
existing = self._row(session, name, effective_user)
if existing is None:
raise
self._apply_update(existing, config, soul)
session.commit()
@staticmethod
def _apply_update(row: AgentRow, config: dict | None, soul: str | None) -> None:
if config is not None:
row.config = _config_document(config)
if soul is not None:
row.soul = soul
def delete(self, name: str, *, user_id: str | None = None) -> AgentDeleteOutcome:
effective_user = user_id or get_effective_user_id()
with self._Session() as session:
result = session.execute(delete(AgentRow).where(AgentRow.user_id == effective_user, AgentRow.name == name.lower()))
session.commit()
row_deleted = result.rowcount > 0
agent_dir = get_paths().user_agent_dir(effective_user, name)
if row_deleted:
# The agent existed as a row; remove any co-located on-disk memory
# (deermem file backend) so it is not orphaned. Mirrors the file
# backend's rmtree, which bundles config + soul + memory.
if agent_dir.exists():
shutil.rmtree(agent_dir)
return "deleted"
# No agent row. A bare on-disk directory here holds only memory/facts
# data (in db mode the config lives in the row, not on disk), so preserve
# it rather than deleting a user's memory (#4279) — do not rmtree it.
if agent_dir.exists():
return "not-custom-agent"
return "missing"
def signature(self) -> Hashable:
# MAX(updated_at) is not covered by an index (only user_id and the
# (user_id, name) unique constraint are), so this is a small full scan.
# It runs only on the webhook registry's cache-freshness check against a
# tiny agents table, so an index is not warranted; revisit if agents ever
# grow into the thousands with frequent webhook deliveries.
with self._Session() as session:
max_updated, count = session.execute(select(func.max(AgentRow.updated_at), func.count(AgentRow.id))).one()
token = coerce_iso(max_updated) if isinstance(max_updated, datetime) else str(max_updated)
return (token, int(count))

50
backend/packages/harness/deerflow/persistence/migrations/versions/0006_agents.py Normal file
View File

@ -0,0 +1,50 @@
"""agents.
Revision ID: 0006_agents
Revises: 0005_run_stop_reason
Create Date: 2026-07-22
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "0006_agents"
down_revision: str | Sequence[str] | None = "0005_run_stop_reason"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
bind = op.get_bind()
inspector = sa.inspect(bind)
if inspector.has_table("agents"):
# Idempotent: a DB whose full-metadata create_all already provisioned
# the table (e.g. legacy test seeds) must not have it re-created here.
return
op.create_table(
"agents",
sa.Column("id", sa.String(length=64), nullable=False),
sa.Column("user_id", sa.String(length=64), nullable=False),
sa.Column("name", sa.String(length=128), nullable=False),
sa.Column("config", sa.JSON(), nullable=False),
# No server_default: matches the ORM's Python-side ``default=""`` (the
# store always supplies soul on insert), keeping create_all and the
# migration byte-identical (test_create_all_and_alembic_upgrade...).
sa.Column("soul", sa.Text(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("user_id", "name", name="uq_agents_user_name"),
)
with op.batch_alter_table("agents", schema=None) as batch_op:
batch_op.create_index("ix_agents_user_id", ["user_id"], unique=False)
def downgrade() -> None:
with op.batch_alter_table("agents", schema=None) as batch_op:
batch_op.drop_index("ix_agents_user_id")
op.drop_table("agents")

2
backend/packages/harness/deerflow/persistence/models/__init__.py
View File

@ -14,6 +14,7 @@ its storage implementation lives in ``deerflow.runtime.events.store.db`` and
there is no matching entity directory.
"""
from deerflow.persistence.agents.model import AgentRow
from deerflow.persistence.channel_connections.model import (
ChannelConnectionRow,
ChannelConversationRow,
@ -29,6 +30,7 @@ from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.user.model import UserRow
__all__ = [
"AgentRow",
"ChannelConnectionRow",
"ChannelConversationRow",
"ChannelCredentialRow",

44
backend/packages/harness/deerflow/tools/builtins/setup_agent_tool.py
View File

@ -1,12 +1,12 @@
import logging
import yaml
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langgraph.types import Command
from deerflow.config.agents_config import validate_agent_name
from deerflow.config.agents_config import SOUL_FILENAME, validate_agent_name
from deerflow.config.paths import get_paths
from deerflow.persistence.agents import get_agent_store
from deerflow.runtime.user_context import resolve_runtime_user_id
from deerflow.tools.types import Runtime
@ -48,39 +48,30 @@ def setup_agent(
)
agent_name: str | None = runtime.context.get("agent_name") if runtime.context else None
agent_dir = None
is_new_dir = False
try:
agent_name = validate_agent_name(agent_name)
paths = get_paths()
if agent_name:
# Custom agents are persisted under the current user's bucket so
# different users do not see each other's agents.
# Custom agents are persisted under the current user's bucket (via
# the configured store — file or db) so different users, and
# different nodes, resolve the same agent. setup is idempotent, so
# this is an upsert.
user_id = resolve_runtime_user_id(runtime)
agent_dir = paths.user_agent_dir(user_id, agent_name)
else:
# Default agent (no agent_name): SOUL.md lives at the global base dir.
agent_dir = paths.base_dir
is_new_dir = not agent_dir.exists()
agent_dir.mkdir(parents=True, exist_ok=True)
if agent_name:
# If agent_name is provided, we are creating a custom agent in the agents/ directory
config_data: dict = {"name": agent_name}
if description:
config_data["description"] = description
if skills is not None:
config_data["skills"] = skills
get_agent_store().update(agent_name, config_data, soul, user_id=user_id)
else:
# Default agent (no agent_name): SOUL.md lives at the global base
# dir. It is not a custom-agent record, so it stays file-based
# regardless of the agent-storage backend.
paths = get_paths()
paths.base_dir.mkdir(parents=True, exist_ok=True)
(paths.base_dir / SOUL_FILENAME).write_text(soul, encoding="utf-8")
config_file = agent_dir / "config.yaml"
with open(config_file, "w", encoding="utf-8") as f:
yaml.dump(config_data, f, default_flow_style=False, allow_unicode=True)
soul_file = agent_dir / "SOUL.md"
soul_file.write_text(soul, encoding="utf-8")
logger.info(f"[agent_creator] Created agent '{agent_name}' at {agent_dir}")
logger.info(f"[agent_creator] Created agent '{agent_name}'")
return Command(
update={
"created_agent_name": agent_name,
@ -89,10 +80,5 @@ def setup_agent(
)
except Exception as e:
import shutil
if agent_name and is_new_dir and agent_dir is not None and agent_dir.exists():
# Cleanup the custom agent directory only if it was newly created during this call
shutil.rmtree(agent_dir)
logger.error(f"[agent_creator] Failed to create agent '{agent_name}': {e}", exc_info=True)
return Command(update={"messages": [ToolMessage(content=f"Error: {e}", tool_call_id=runtime.tool_call_id)]})

120
backend/packages/harness/deerflow/tools/builtins/update_agent_tool.py
View File

@ -5,21 +5,30 @@ Bound to the lead agent only when ``runtime.context['agent_name']`` is set
this tool, and the bootstrap flow continues to use ``setup_agent`` for the
initial creation handshake.
The tool writes back to ``{base_dir}/users/{user_id}/agents/{agent_name}/{config.yaml,SOUL.md}``
so an agent created by one user is never visible to (or mutable by) another.
Writes are staged into temp files first; both files are renamed into place only
after both temp files are successfully written, so a partial failure cannot leave
config.yaml updated while SOUL.md still holds stale content.
The tool writes back through the configured agent store (file: per-user
``config.yaml``/``SOUL.md``; db: the shared ``agents`` table) so an agent created
by one user is never visible to (or mutable by) another.
Cross-field write atomicity depends on the backend: the ``db`` store commits
config and soul in a single transaction, so a partial failure never leaves one
updated and the other stale. The ``file`` store stages both to temp files and
commits them with two sequential ``os.replace`` calls (see
``FileAgentStore._write``): each file is all-or-nothing, but a crash *between*
the two replaces can leave a freshly-written config.yaml beside a stale SOUL.md
(single-node, sub-millisecond window). The pre-store tool reported that partial
window explicitly ("Partial update for agent 'X': ..."); routing through the
store drops the *reporting* (a mid-replace crash now surfaces as the generic
"Failed to update agent"). That is an intentional tradeoff — the stage-then-
replace *safety* is preserved (no corruption, no leftover temp files), only the
diagnostic is gone. If cross-file atomicity ever matters on ``file``, restore
that reporting in ``FileAgentStore._write``.
"""
from __future__ import annotations
import logging
import tempfile
from pathlib import Path
from typing import Annotated, Any
import yaml
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
from langgraph.types import Command
@ -28,6 +37,7 @@ from pydantic import BaseModel, BeforeValidator
from deerflow.config.agents_config import load_agent_config, preserve_non_managed_fields, validate_agent_name
from deerflow.config.app_config import get_app_config
from deerflow.config.paths import get_paths
from deerflow.persistence.agents import get_agent_store
from deerflow.runtime.user_context import resolve_runtime_user_id
from deerflow.tools.types import Runtime
@ -50,40 +60,6 @@ _MODEL_BEHAVIOR_FIELDS: tuple[str, ...] = (
)
def _stage_temp(path: Path, text: str) -> Path:
"""Write ``text`` into a sibling temp file and return its path.
The caller is responsible for ``Path.replace``-ing the temp into the target
once every staged file is ready, or for unlinking it on failure.
"""
path.parent.mkdir(parents=True, exist_ok=True)
fd = tempfile.NamedTemporaryFile(
mode="w",
dir=path.parent,
suffix=".tmp",
delete=False,
encoding="utf-8",
)
try:
fd.write(text)
fd.flush()
fd.close()
return Path(fd.name)
except BaseException:
fd.close()
Path(fd.name).unlink(missing_ok=True)
raise
def _cleanup_temps(temps: list[Path]) -> None:
"""Best-effort removal of staged temp files."""
for tmp in temps:
try:
tmp.unlink(missing_ok=True)
except OSError:
logger.debug("Failed to clean up temp file %s", tmp, exc_info=True)
def _is_nullish_string(value: object) -> bool:
return isinstance(value, str) and value.strip().lower() in _NULLISH_STRINGS
@ -261,58 +237,20 @@ def update_agent(
config_data.setdefault(key, value)
config_changed = bool({"description", "model", "tool_groups", "skills"} & set(updated_fields))
if soul is not None:
updated_fields.append("soul")
# Stage every file we intend to rewrite into a temp sibling. Only after
# *all* temp files exist do we rename them into place — so a failure on
# SOUL.md cannot leave config.yaml already replaced.
pending: list[tuple[Path, Path]] = []
staged_temps: list[Path] = []
try:
agent_dir.mkdir(parents=True, exist_ok=True)
if config_changed:
yaml_text = yaml.dump(config_data, default_flow_style=False, allow_unicode=True, sort_keys=False)
config_target = agent_dir / "config.yaml"
config_tmp = _stage_temp(config_target, yaml_text)
staged_temps.append(config_tmp)
pending.append((config_tmp, config_target))
if soul is not None:
soul_target = agent_dir / "SOUL.md"
soul_tmp = _stage_temp(soul_target, soul)
staged_temps.append(soul_tmp)
pending.append((soul_tmp, soul_target))
updated_fields.append("soul")
# Commit phase. ``Path.replace`` is atomic per file on POSIX/NTFS and
# the staging step above means any earlier failure has already been
# reported. The remaining failure mode is a crash *between* two
# ``replace`` calls, which is reported via the partial-write error
# branch below so the caller knows which files are now on disk.
committed: list[Path] = []
# Persist config (when a managed field changed) and/or soul through the
# store. The db backend commits both in one transaction; the file backend
# commits each atomically but sequentially (see this module's docstring).
# Nothing to write if the provided values all matched the existing config
# and no soul was supplied.
if config_changed or soul is not None:
try:
for tmp, target in pending:
tmp.replace(target)
committed.append(target)
get_agent_store().update(agent_name, config_data if config_changed else None, soul, user_id=user_id)
except Exception as e:
_cleanup_temps([t for t, _ in pending if t not in committed])
if committed:
logger.error(
"[update_agent] Partial write for agent '%s' (user=%s): committed=%s, failed during rename: %s",
agent_name,
user_id,
[p.name for p in committed],
e,
exc_info=True,
)
return _err(f"Partial update for agent '{agent_name}': {[p.name for p in committed]} were updated, but the rest failed ({e}). Re-run update_agent to retry the remaining fields.")
raise
except Exception as e:
_cleanup_temps(staged_temps)
logger.error("[update_agent] Failed to update agent '%s' (user=%s): %s", agent_name, user_id, e, exc_info=True)
return _err(f"Failed to update agent '{agent_name}': {e}")
logger.error("[update_agent] Failed to update agent '%s' (user=%s): %s", agent_name, user_id, e, exc_info=True)
return _err(f"Failed to update agent '{agent_name}': {e}")
if not updated_fields:
return Command(update={"messages": [ToolMessage(content=f"No changes applied to agent '{agent_name}'. The provided values matched the existing config.", tool_call_id=tool_call_id)]})

93
backend/scripts/migrate_agents_to_db.py Normal file
View File

@ -0,0 +1,93 @@
#!/usr/bin/env python
"""One-shot importer: copy file-backed custom agents into the ``db`` agent store.
For operators switching ``agent_storage.backend`` from ``file`` to ``db``. Reads
every agent from the on-disk layout (both the per-user
``{base_dir}/users/{user_id}/agents/`` and the legacy shared
``{base_dir}/agents/``, with the same shadowing rule the file store uses) and
writes each as a row in the shared ``agents`` table.
Design (mirrors ``scripts/migrate_user_isolation.py``):
- Explicit, operator-run. Nothing auto-imports on boot.
- Idempotent: an agent already present in the db is skipped, so re-running is safe.
- Non-destructive: the on-disk files are left untouched, so unsetting
``agent_storage.backend`` (back to ``file``) is a clean rollback.
Usage::
python scripts/migrate_agents_to_db.py [--dry-run]
Requires ``database.backend`` to be ``sqlite`` or ``postgres`` in config.yaml.
"""
from __future__ import annotations
import argparse
import asyncio
import logging
import sys
from deerflow.config.app_config import get_app_config
from deerflow.persistence.agents.base import AgentExistsError
from deerflow.persistence.agents.file import FileAgentStore
from deerflow.persistence.agents.sql import SqlAgentStore
logger = logging.getLogger("migrate_agents_to_db")
def main() -> int:
parser = argparse.ArgumentParser(description="Import file-backed custom agents into the db agent store.")
parser.add_argument("--dry-run", action="store_true", help="List what would be imported without writing to the database.")
args = parser.parse_args()
logging.basicConfig(level=logging.INFO, format="%(message)s")
config = get_app_config()
db_backend = getattr(config.database, "backend", None)
if db_backend not in ("sqlite", "postgres"):
logger.error(
"database.backend is %r; this importer needs 'sqlite' or 'postgres'. Set it in config.yaml (the same database the gateway uses).",
db_backend,
)
return 1
source = FileAgentStore()
agents = source.list_all()
if not agents:
logger.info("No file-backed agents found — nothing to import.")
return 0
if args.dry_run:
for user_id, cfg in agents:
logger.info("[dry-run] would import %s/%s", user_id, cfg.name)
logger.info("[dry-run] %d agent(s) would be imported. Source files are left in place.", len(agents))
return 0
# Ensure the schema exists (creates the ``agents`` table via the same
# Alembic bootstrap the gateway runs) before the sync store writes rows.
from deerflow.persistence.engine import init_engine_from_config
asyncio.run(init_engine_from_config(config.database))
dest = SqlAgentStore(config.database.app_sync_sqlalchemy_url)
imported = 0
skipped = 0
for user_id, cfg in agents:
soul = source.get_soul(cfg.name, user_id=user_id) or ""
# exclude_unset keeps the stored document as sparse as the source file
# (only the keys the operator actually wrote), matching the file layout.
document = cfg.model_dump(exclude_unset=True)
try:
dest.create(cfg.name, document, soul, user_id=user_id)
imported += 1
logger.info("imported %s/%s", user_id, cfg.name)
except AgentExistsError:
skipped += 1
logger.info("skip %s/%s: already present in db", user_id, cfg.name)
logger.info("Done: %d imported, %d already present. Source files left in place (rollback: revert agent_storage.backend to 'file').", imported, skipped)
return 0
if __name__ == "__main__":
sys.exit(main())

32
backend/tests/blocking_io/test_agents_router.py
View File

@ -21,7 +21,14 @@ from pathlib import Path
import pytest
from app.gateway.routers.agents import AgentCreateRequest, create_agent_endpoint, delete_agent
from app.gateway.routers.agents import (
AgentCreateRequest,
check_agent_name,
create_agent_endpoint,
delete_agent,
get_agent,
list_agents,
)
from deerflow.config.agents_api_config import load_agents_api_config_from_dict
from deerflow.config.paths import get_paths
from deerflow.runtime.user_context import get_effective_user_id
@ -62,3 +69,26 @@ async def test_delete_agent_does_not_block_event_loop(tmp_path: Path, monkeypatc
assert not await asyncio.to_thread(agent_dir.exists)
finally:
load_agents_api_config_from_dict({})
async def test_read_endpoints_do_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
# list/get/check read through the sync agent store; on the db backend each is
# a DB round trip. They must offload via asyncio.to_thread, or the strict
# Blockbuster gate raises BlockingError here (finding: reads on the loop).
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
monkeypatch.setattr("deerflow.config.paths._paths", None)
load_agents_api_config_from_dict({"enabled": True})
try:
await create_agent_endpoint(AgentCreateRequest(name="loop-read-agent", soul="You are a test agent."))
listed = await list_agents()
assert any(a.name == "loop-read-agent" for a in listed.agents)
got = await get_agent("loop-read-agent")
assert got.name == "loop-read-agent"
check = await check_agent_name("loop-read-agent")
assert check["available"] is False
assert (await check_agent_name("never-created-agent"))["available"] is True
finally:
load_agents_api_config_from_dict({})

231
backend/tests/test_agent_storage_backend.py Normal file
View File

@ -0,0 +1,231 @@
"""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()

232
backend/tests/test_agent_store_sql.py Normal file
View File

@ -0,0 +1,232 @@
"""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()

2
backend/tests/test_migration_0004_run_ownership_dedupe.py
View File

@ -156,7 +156,7 @@ async def test_migration_dedupes_duplicate_active_rows_before_unique_index(tmp_p
with sqlite3.connect(db_path) as raw:
version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
assert version_row[0] == "0005_run_stop_reason"
assert version_row[0] == "0006_agents"
# Sanity: the invariant the index enforces is now true — at most one
# active row per thread.

2
backend/tests/test_persistence_bootstrap.py
View File

@ -48,7 +48,7 @@ from deerflow.persistence.migrations._helpers import _normalize_default
asyncio_test = pytest.mark.asyncio
HEAD = "0005_run_stop_reason"
HEAD = "0006_agents"
BASELINE = "0001_baseline"

2
backend/tests/test_persistence_bootstrap_concurrency.py
View File

@ -28,7 +28,7 @@ from deerflow.persistence.bootstrap import bootstrap_schema
pytestmark = pytest.mark.asyncio
HEAD = "0005_run_stop_reason"
HEAD = "0006_agents"
def _url(tmp_path: Path) -> str:

4
backend/tests/test_persistence_bootstrap_regression.py
View File

@ -76,7 +76,7 @@ async def test_legacy_database_recovers_token_usage_column(tmp_path: Path) -> No
cols = {row[1] for row in raw.execute("PRAGMA table_info(runs)").fetchall()}
assert "token_usage_by_model" in cols
version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
assert version_row[0] == "0005_run_stop_reason"
assert version_row[0] == "0006_agents"
# And the read path that originally 500'd must now succeed.
sf = get_session_factory()
@ -116,6 +116,6 @@ async def test_legacy_database_with_manual_alter_still_bootstraps(tmp_path: Path
# No duplicate column -- list, not set, to catch dupes.
assert cols.count("token_usage_by_model") == 1
version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone()
assert version_row[0] == "0005_run_stop_reason"
assert version_row[0] == "0006_agents"
finally:
await close_engine()

8
backend/tests/test_setup_agent_e2e_user_isolation.py
View File

@ -316,7 +316,7 @@ async def test_real_graph_real_setup_agent_writes_to_authenticated_user_dir(tmp_
# Patch get_paths only (the file-system rooting); everything else is real
with patch(
"deerflow.tools.builtins.setup_agent_tool.get_paths",
"deerflow.config.agents_config.get_paths",
return_value=_make_paths_mock(tmp_path),
):
# Drive the real graph. This goes through real ToolNode + real Runtime merge.
@ -365,7 +365,7 @@ async def test_inject_failure_falls_back_to_default_proving_test_is_load_bearing
graph = _build_real_bootstrap_graph("does-not-matter")
with patch(
"deerflow.tools.builtins.setup_agent_tool.get_paths",
"deerflow.config.agents_config.get_paths",
return_value=_make_paths_mock(tmp_path),
):
await graph.ainvoke(
@ -434,7 +434,7 @@ async def test_subgraph_invocation_preserves_user_id_in_runtime(tmp_path: Path):
config.setdefault("configurable", {})["__pregel_runtime"] = runtime
with patch(
"deerflow.tools.builtins.setup_agent_tool.get_paths",
"deerflow.config.agents_config.get_paths",
return_value=_make_paths_mock(tmp_path),
):
# Direct sub-graph invoke (mimics what a subagent invocation looks like
@ -498,7 +498,7 @@ def test_sync_tool_dispatch_through_thread_pool_uses_runtime_context(tmp_path: P
config.setdefault("configurable", {})["__pregel_runtime"] = runtime
with patch(
"deerflow.tools.builtins.setup_agent_tool.get_paths",
"deerflow.config.agents_config.get_paths",
return_value=_make_paths_mock(tmp_path),
):
# Use SYNC invoke to hit the ContextThreadPoolExecutor path

12
backend/tests/test_setup_agent_tool.py
View File

@ -35,7 +35,7 @@ def _make_paths_mock(tmp_path: Path):
def _call_setup_agent(tmp_path: Path, soul: str, description: str, agent_name: str = "test-agent"):
"""Call the underlying setup_agent function directly, bypassing langchain tool wrapper."""
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)):
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)), patch("deerflow.config.agents_config.get_paths", return_value=_make_paths_mock(tmp_path)):
return setup_agent.func(
soul=soul,
description=description,
@ -89,9 +89,9 @@ class TestSetupAgentNoDataLoss:
old_soul = agent_dir / "SOUL.md"
old_soul.write_text("original soul content", encoding="utf-8")
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)):
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)), patch("deerflow.config.agents_config.get_paths", return_value=_make_paths_mock(tmp_path)):
# Force soul_file.write_text to raise after directory already exists
with patch.object(Path, "write_text", side_effect=OSError("disk full")):
with patch("yaml.dump", side_effect=OSError("disk full")):
setup_agent.func(
soul="new soul",
description="desc",
@ -109,7 +109,7 @@ class TestSetupAgentNoDataLoss:
agent_dir = tmp_path / "users" / "test-user-autouse" / "agents" / "test-agent"
assert not agent_dir.exists()
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)):
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)), patch("deerflow.config.agents_config.get_paths", return_value=_make_paths_mock(tmp_path)):
with patch("yaml.dump", side_effect=OSError("write error")):
setup_agent.func(
soul="new soul",
@ -137,7 +137,7 @@ class TestSetupAgentNoDataLoss:
tool_call_id="tool-3",
)
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)):
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)), patch("deerflow.config.agents_config.get_paths", return_value=_make_paths_mock(tmp_path)):
setup_agent.func(
soul="# My Agent",
description="A test agent",
@ -185,7 +185,7 @@ class TestSetupAgentEmptySoulGuard:
global_soul = tmp_path / "SOUL.md"
global_soul.write_text("original global soul", encoding="utf-8")
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)):
with patch("deerflow.tools.builtins.setup_agent_tool.get_paths", return_value=_make_paths_mock(tmp_path)), patch("deerflow.config.agents_config.get_paths", return_value=_make_paths_mock(tmp_path)):
setup_agent.func(
soul="",
description="desc",

2
backend/tests/test_update_agent_tool.py
View File

@ -451,7 +451,7 @@ def test_update_agent_soul_failure_does_not_replace_config(tmp_path, patched_pat
raise OSError("disk full while staging SOUL.md")
return real_named_temp_file(*args, **kwargs)
with patch("deerflow.tools.builtins.update_agent_tool.tempfile.NamedTemporaryFile", side_effect=_explode_on_soul):
with patch("deerflow.persistence.agents.file.tempfile.NamedTemporaryFile", side_effect=_explode_on_soul):
result = update_agent.func(runtime=_runtime(), description="new-desc", soul="new soul")
cfg = yaml.safe_load((agent_dir / "config.yaml").read_text())

23
config.example.yaml
View File

@ -1820,6 +1820,29 @@ run_events:
max_trace_content: 10240
track_token_usage: true
# ============================================================================
# Agent Storage Configuration
# ============================================================================
# Where custom agent DEFINITIONS (config.yaml + SOUL.md) are stored. This is
# separate from `database` (run/thread/event data) and from agent memory.
# Restart-required (the backend is captured at Gateway lifespan startup).
#
# backend: file -- Per-user files under {base_dir}/users/{uid}/agents/ (default).
# Single-node only: an agent created on one node is invisible
# to other nodes without a shared mount.
# backend: db -- A row per agent in the shared SQL persistence layer, so every
# node sees the same agents. Requires database.backend to be
# 'sqlite' or 'postgres' (rejected at startup on 'memory').
#
# Switching an existing install to 'db'? Import the on-disk agents once with:
# python backend/scripts/migrate_agents_to_db.py # --dry-run to preview
# The source files are left in place, so reverting to 'file' is a clean rollback.
#
# agent_storage:
# backend: file
agent_storage:
backend: file
# ============================================================================
# Scheduled Tasks Configuration
# ============================================================================