diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py
index decac902c..e11c75a53 100644
--- a/backend/app/gateway/deps.py
+++ b/backend/app/gateway/deps.py
@@ -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
diff --git a/backend/app/gateway/github/registry.py b/backend/app/gateway/github/registry.py
index 0ac128a7b..9ae68cc6a 100644
--- a/backend/app/gateway/github/registry.py
+++ b/backend/app/gateway/github/registry.py
@@ -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
 
diff --git a/backend/app/gateway/routers/agents.py b/backend/app/gateway/routers/agents.py
index 4c87d51be..e3c32dfb1 100644
--- a/backend/app/gateway/routers/agents.py
+++ b/backend/app/gateway/routers/agents.py
@@ -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}'")
diff --git a/backend/app/gateway/routers/assistants_compat.py b/backend/app/gateway/routers/assistants_compat.py
index 83708747c..b0479e81f 100644
--- a/backend/app/gateway/routers/assistants_compat.py
+++ b/backend/app/gateway/routers/assistants_compat.py
@@ -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")
 
diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md
index 07706bfa9..bf58c8282 100644
--- a/backend/docs/CONFIGURATION.md
+++ b/backend/docs/CONFIGURATION.md
@@ -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:
diff --git a/backend/packages/harness/deerflow/config/agent_storage_config.py b/backend/packages/harness/deerflow/config/agent_storage_config.py
new file mode 100644
index 000000000..bb26c2d15
--- /dev/null
+++ b/backend/packages/harness/deerflow/config/agent_storage_config.py
@@ -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'."
+        ),
+    )
diff --git a/backend/packages/harness/deerflow/config/agents_config.py b/backend/packages/harness/deerflow/config/agents_config.py
index 25663c03d..7b5e17c2a 100644
--- a/backend/packages/harness/deerflow/config/agents_config.py
+++ b/backend/packages/harness/deerflow/config/agents_config.py
@@ -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)
diff --git a/backend/packages/harness/deerflow/config/app_config.py b/backend/packages/harness/deerflow/config/app_config.py
index 66ee48b2e..78930ef77 100644
--- a/backend/packages/harness/deerflow/config/app_config.py
+++ b/backend/packages/harness/deerflow/config/app_config.py
@@ -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(
diff --git a/backend/packages/harness/deerflow/config/database_config.py b/backend/packages/harness/deerflow/config/database_config.py
index 4b970ac9e..cb578fff5 100644
--- a/backend/packages/harness/deerflow/config/database_config.py
+++ b/backend/packages/harness/deerflow/config/database_config.py
@@ -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}")
diff --git a/backend/packages/harness/deerflow/config/reload_boundary.py b/backend/packages/harness/deerflow/config/reload_boundary.py
index 96131fb00..31a6a99db 100644
--- a/backend/packages/harness/deerflow/config/reload_boundary.py
+++ b/backend/packages/harness/deerflow/config/reload_boundary.py
@@ -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": (
diff --git a/backend/packages/harness/deerflow/persistence/agents/__init__.py b/backend/packages/harness/deerflow/persistence/agents/__init__.py
new file mode 100644
index 000000000..4ab1f1797
--- /dev/null
+++ b/backend/packages/harness/deerflow/persistence/agents/__init__.py
@@ -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
diff --git a/backend/packages/harness/deerflow/persistence/agents/base.py b/backend/packages/harness/deerflow/persistence/agents/base.py
new file mode 100644
index 000000000..725dc765f
--- /dev/null
+++ b/backend/packages/harness/deerflow/persistence/agents/base.py
@@ -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``).
+        """
diff --git a/backend/packages/harness/deerflow/persistence/agents/file.py b/backend/packages/harness/deerflow/persistence/agents/file.py
new file mode 100644
index 000000000..0d8e7283d
--- /dev/null
+++ b/backend/packages/harness/deerflow/persistence/agents/file.py
@@ -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)
diff --git a/backend/packages/harness/deerflow/persistence/agents/model.py b/backend/packages/harness/deerflow/persistence/agents/model.py
new file mode 100644
index 000000000..58426b9d4
--- /dev/null
+++ b/backend/packages/harness/deerflow/persistence/agents/model.py
@@ -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),
+    )
diff --git a/backend/packages/harness/deerflow/persistence/agents/sql.py b/backend/packages/harness/deerflow/persistence/agents/sql.py
new file mode 100644
index 000000000..50ad21a97
--- /dev/null
+++ b/backend/packages/harness/deerflow/persistence/agents/sql.py
@@ -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))
diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0006_agents.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0006_agents.py
new file mode 100644
index 000000000..1759709e2
--- /dev/null
+++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0006_agents.py
@@ -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")
diff --git a/backend/packages/harness/deerflow/persistence/models/__init__.py b/backend/packages/harness/deerflow/persistence/models/__init__.py
index 986ad5db1..fa8e14cb6 100644
--- a/backend/packages/harness/deerflow/persistence/models/__init__.py
+++ b/backend/packages/harness/deerflow/persistence/models/__init__.py
@@ -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",
diff --git a/backend/packages/harness/deerflow/tools/builtins/setup_agent_tool.py b/backend/packages/harness/deerflow/tools/builtins/setup_agent_tool.py
index ea8b43f9b..c30823914 100644
--- a/backend/packages/harness/deerflow/tools/builtins/setup_agent_tool.py
+++ b/backend/packages/harness/deerflow/tools/builtins/setup_agent_tool.py
@@ -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)]})
diff --git a/backend/packages/harness/deerflow/tools/builtins/update_agent_tool.py b/backend/packages/harness/deerflow/tools/builtins/update_agent_tool.py
index 7e4910f70..a080b2a78 100644
--- a/backend/packages/harness/deerflow/tools/builtins/update_agent_tool.py
+++ b/backend/packages/harness/deerflow/tools/builtins/update_agent_tool.py
@@ -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)]})
diff --git a/backend/scripts/migrate_agents_to_db.py b/backend/scripts/migrate_agents_to_db.py
new file mode 100644
index 000000000..d7fabcf42
--- /dev/null
+++ b/backend/scripts/migrate_agents_to_db.py
@@ -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())
diff --git a/backend/tests/blocking_io/test_agents_router.py b/backend/tests/blocking_io/test_agents_router.py
index 1f7787397..a05a0afbd 100644
--- a/backend/tests/blocking_io/test_agents_router.py
+++ b/backend/tests/blocking_io/test_agents_router.py
@@ -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({})
diff --git a/backend/tests/test_agent_storage_backend.py b/backend/tests/test_agent_storage_backend.py
new file mode 100644
index 000000000..2f1e94794
--- /dev/null
+++ b/backend/tests/test_agent_storage_backend.py
@@ -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()
diff --git a/backend/tests/test_agent_store_sql.py b/backend/tests/test_agent_store_sql.py
new file mode 100644
index 000000000..f416073e3
--- /dev/null
+++ b/backend/tests/test_agent_store_sql.py
@@ -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()
diff --git a/backend/tests/test_migration_0004_run_ownership_dedupe.py b/backend/tests/test_migration_0004_run_ownership_dedupe.py
index 03d948050..28bbd611a 100644
--- a/backend/tests/test_migration_0004_run_ownership_dedupe.py
+++ b/backend/tests/test_migration_0004_run_ownership_dedupe.py
@@ -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.
diff --git a/backend/tests/test_persistence_bootstrap.py b/backend/tests/test_persistence_bootstrap.py
index 11cfe926c..ca9d84245 100644
--- a/backend/tests/test_persistence_bootstrap.py
+++ b/backend/tests/test_persistence_bootstrap.py
@@ -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"
 
 
diff --git a/backend/tests/test_persistence_bootstrap_concurrency.py b/backend/tests/test_persistence_bootstrap_concurrency.py
index 045c8a9bb..1848da0f5 100644
--- a/backend/tests/test_persistence_bootstrap_concurrency.py
+++ b/backend/tests/test_persistence_bootstrap_concurrency.py
@@ -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:
diff --git a/backend/tests/test_persistence_bootstrap_regression.py b/backend/tests/test_persistence_bootstrap_regression.py
index 6981fb27e..745fb4773 100644
--- a/backend/tests/test_persistence_bootstrap_regression.py
+++ b/backend/tests/test_persistence_bootstrap_regression.py
@@ -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()
diff --git a/backend/tests/test_setup_agent_e2e_user_isolation.py b/backend/tests/test_setup_agent_e2e_user_isolation.py
index 95c8e5f31..08f780e99 100644
--- a/backend/tests/test_setup_agent_e2e_user_isolation.py
+++ b/backend/tests/test_setup_agent_e2e_user_isolation.py
@@ -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
diff --git a/backend/tests/test_setup_agent_tool.py b/backend/tests/test_setup_agent_tool.py
index 4fd58de18..dc2f03f67 100644
--- a/backend/tests/test_setup_agent_tool.py
+++ b/backend/tests/test_setup_agent_tool.py
@@ -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",
diff --git a/backend/tests/test_update_agent_tool.py b/backend/tests/test_update_agent_tool.py
index 2e2e14e0f..6f271468f 100644
--- a/backend/tests/test_update_agent_tool.py
+++ b/backend/tests/test_update_agent_tool.py
@@ -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())
diff --git a/config.example.yaml b/config.example.yaml
index 2b7a54de2..4754262d9 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -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
 # ============================================================================