wsk/deerflow-code
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'ji' of http://192.168.0.200/knowledge-base-bj/deerflow-server into hwg

Browse Source
This commit is contained in:
wghe committed 2026-05-21 18:35:56 +08:00
2 parents b97548f8d6 f1d58983c2
commit 6ebbf3aa32
21 files changed
+1317 -119
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

offline-backend-20260512/backend/CLAUDE.md
+3 -1
View File
@@ -233,7 +233,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
| **Skills** (`/api/skills`) | `GET /` - list skills (optional `search` fuzzy-name + `tag_id` filters; each item carries `tags` + owner-authored `detail`); `GET /{name}` - details; `PUT /{name}` - update enabled; `PUT /{name}/detail` - set the long-form `detail` text (custom skill: owner or admin; built-in/public: admin only — materializes a `skills` row on demand); `GET /{name}/content` - raw SKILL.md source, restricted to an admin or the publisher (custom-skill owner); other users get 403 and see only description + detail; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`) |
| **Skills** (`/api/skills`) | `GET /` - list skills (optional `search` fuzzy-name + repeatable `tag_ids` multi-select filter, OR-matched; each item carries `tags` + owner-authored `detail`); `GET /{name}` - details; `PUT /{name}` - update enabled; `PUT /{name}/detail` - set the long-form `detail` text (custom skill: owner or admin; built-in/public: admin only — materializes a `skills` row on demand); `GET /{name}/content` - raw SKILL.md source, restricted to an admin or the publisher (custom-skill owner); other users get 403 and see only description + detail; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`) |
| **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data |
| **Agents** (`/api/agents`) | `GET /` - list built-in, owned, and published agents (optional `search` + repeatable `tag_ids` multi-select filter, OR-matched; each item carries `tags`); `POST /` - create user-owned agent; `GET/PUT/DELETE /{id}` - read/update/delete by stable agent id; `GET /check` - validate non-empty display name (duplicates allowed) |
| **Agents** (`/api/agents`) | `GET /` - list built-in, owned, and published agents (optional `search` + repeatable `tag_ids` multi-select filter, OR-matched; each item carries `tags` and `featured_order`); `POST /` - create user-owned agent; `GET/PUT/DELETE /{id}` - read/update/delete by stable agent id; `GET /check` - validate non-empty display name (duplicates allowed); `GET /selector` - homepage agent picker (single admin-curated list, visibility-filtered, sorted by `featured_order` asc); `PUT /{id}/featured` - **admin** sets/clears the homepage `featured_order` (must be `published=true` to feature; passing `order: null` clears unconditionally); `PUT /featured/order` - **admin** bulk-reorders the homepage list (body `{agent_ids: [...]}` → indices become 0..N-1) |
| **Tags** (`/api/tags`) | `GET /` - list tags (optional `tag_type` filter; open to all); `POST /` - create tag (admin); `PUT /{id}` / `DELETE /{id}` - update/delete tag (admin); `GET /assignments/{target_type}/{target_id}` - tags on a resource; `PUT /assignments/{target_type}/{target_id}` - replace a resource's tags (admin). Tags are typed `agent`/`skill`/`scheduled_task` and may only tag resources of their own type. |
| **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); `GET /list` - list; `DELETE /{filename}` - delete |
| **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; unexpected failures are logged server-side and return a generic 500 detail |
@@ -260,6 +260,8 @@ Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` →
Custom agent metadata lives in the `agents` ORM table (`deerflow.persistence.agents`). `agents.id` is the stable runtime id and the filesystem directory name under `.deer-flow/agents/{id}/`; `agents.name` is display-only, can contain Chinese characters, and is not unique. `user_id IS NULL` marks built-in agents, user-owned agents are visible only to their owner unless `published = true`, and update/delete operations require ownership. The runtime accepts `agent_id` in context/configurable, and non-default `assistant_id` values are treated as agent ids after visibility checks.
**Homepage selector (agent picker on the new-chat page)** — single admin-curated list backed by `agents.featured_order` (nullable BIGINT). NULL = not featured. Only `published = true` agents can be featured (router enforces on writes); any published agent is eligible regardless of ownership. Cleared with `order: null`. The list rendered to end users is `WHERE featured_order IS NOT NULL` + visibility-filter, sorted by `featured_order ASC`. Admins reorder by sending the desired ids to `PUT /api/agents/featured/order` (assigns indices 0..N-1). BIGINT is used rather than INT so the management UI can seed initial values from `Date.now()` (overflows INT on MySQL).
### Portable Column Types (`deerflow/persistence/types.py`)
Two `TypeDecorator`s keep ORM columns correct across the `sqlite` / `postgres` / `mysql` backends. **All ORM timestamp/JSON columns must use these, not the raw SQLAlchemy types.**
offline-backend-20260512/backend/app/gateway/routers/agents.py
+183 -2
View File
@@ -45,6 +45,11 @@ class AgentResponse(BaseModel):
tags: list[dict] = Field(default_factory=list, description="Tags assigned to this agent")
featured_order: int | None = Field(
default=None,
description="Admin-curated homepage rank. Null = not featured. Lower sorts first.",
)
class AgentsListResponse(BaseModel):
"""Response model for listing all visible agents."""
@@ -76,6 +81,28 @@ class AgentUpdateRequest(BaseModel):
tool_groups: list[str] | None = Field(default=None, description="Updated tool group whitelist (config.yaml)")
class AgentSelectorResponse(BaseModel):
"""Response for the homepage agent selector dropdown.
The list is fully curated by admins (``featured_order`` on each row).
Non-admin users get a read-only view of the same list.
"""
agents: list[AgentResponse] = Field(
default_factory=list,
description="Admin-curated agents, sorted by featured_order ascending.",
)
class AgentFeaturedRequest(BaseModel):
"""Request body for setting/clearing an agent's homepage featured rank."""
order: int | None = Field(
default=None,
description="Featured rank. Lower sorts first. Pass null to remove the agent from the homepage selector.",
)
def _require_agents_api_enabled() -> None:
"""Reject access unless the custom-agent management API is explicitly enabled."""
if not get_agents_api_config().enabled:
@@ -179,7 +206,45 @@ async def _sync_legacy_agents(store: AgentStore) -> None:
logger.exception("Failed to sync legacy folder-backed agents")
def _record_to_response(record: dict[str, Any], *, include_soul: bool = False) -> AgentResponse:
def _ensure_agent_dir_from_record(record: dict[str, Any]) -> None:
"""Lazily restore an agent's on-disk artifacts from its DB record.
The DB row is authoritative for the agent's existence. If the matching
``{base_dir}/agents/{id}/`` directory is missing (data dir moved between
machines, filesystem wiped, etc.) we recreate it with a minimal
``config.yaml`` + empty ``SOUL.md`` synthesized from the row. This restores
a usable agent without requiring the admin to manually delete + recreate.
"""
agent_id = record.get("id")
if not agent_id:
return
try:
agent_dir = get_paths().agent_dir(agent_id)
if agent_dir.exists():
return
logger.warning(
"Agent dir missing for '%s' — rebuilding minimal config.yaml + SOUL.md from DB",
agent_id,
)
_write_agent_config(
agent_id,
name=record.get("name") or agent_id,
description=record.get("description") or "",
)
_write_agent_soul(agent_id, "")
except Exception:
# Self-heal is best-effort. ``load_agent_config`` already tolerates a
# missing dir, so the worst case is the agent runs with default
# settings until someone re-edits it.
logger.exception("Failed to self-heal agent dir for '%s'", agent_id)
def _record_to_response(
record: dict[str, Any],
*,
include_soul: bool = False,
) -> AgentResponse:
_ensure_agent_dir_from_record(record)
soul: str | None = None
if include_soul:
soul = load_agent_soul(record["id"]) or ""
@@ -199,6 +264,7 @@ def _record_to_response(record: dict[str, Any], *, include_soul: bool = False) -
model=extras["model"],
tool_groups=extras["tool_groups"],
skills=extras["skills"],
featured_order=record.get("featured_order"),
)
@@ -284,6 +350,32 @@ async def check_agent_name(name: str) -> dict[str, Any]:
return {"available": True, "name": normalized}
@router.get(
"/agents/selector",
response_model=AgentSelectorResponse,
summary="Agent Homepage Selector",
description=(
"Return the admin-curated agent list shown on the new-chat homepage. "
"Sorted by ``featured_order`` ascending. Visibility-filtered, so an "
"agent the caller can't see (author un-published) is silently omitted."
),
)
async def get_agent_selector(request: Request) -> AgentSelectorResponse:
_require_agents_api_enabled()
try:
store = get_agent_store(request)
await _sync_legacy_agents(store)
user_id = _current_user_id(request)
records = await store.list_featured(user_id)
return AgentSelectorResponse(agents=[_record_to_response(r) for r in records])
except HTTPException:
raise
except Exception as e:
logger.error("Failed to load agent selector: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to load agent selector: {str(e)}") from e
@router.get(
"/agents/{agent_id}",
response_model=AgentResponse,
@@ -297,7 +389,8 @@ async def get_agent(agent_id: str, request: Request) -> AgentResponse:
try:
store = get_agent_store(request)
await _sync_legacy_agents(store)
record = await store.get_visible(agent_id, _current_user_id(request))
user_id = _current_user_id(request)
record = await store.get_visible(agent_id, user_id)
if record is None:
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
return _record_to_response(record, include_soul=True)
@@ -514,3 +607,91 @@ async def delete_agent(agent_id: str, request: Request) -> None:
except Exception as e:
logger.error("Failed to delete agent '%s': %s", agent_id, e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to delete agent: {str(e)}") from e
@router.put(
"/agents/{agent_id}/featured",
response_model=AgentResponse,
summary="Set Agent Featured Rank (admin)",
description=(
"Admin-only. Set or clear the homepage-selector featured rank for any agent. "
"Requires the agent to be ``published=true`` when a non-null rank is supplied; "
"passing ``order: null`` removes the agent from the selector and is always allowed."
),
)
async def set_agent_featured(
agent_id: str,
body: AgentFeaturedRequest,
request: Request,
) -> AgentResponse:
_require_agents_api_enabled()
agent_id = _validate_storage_id(agent_id)
if not _is_admin_user(request):
raise HTTPException(status_code=403, detail="Only system admins can set the homepage featured rank")
try:
store = get_agent_store(request)
await _sync_legacy_agents(store)
record = await store.get_any(agent_id)
if record is None:
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
if body.order is not None and not record.get("published"):
raise HTTPException(status_code=409, detail="Only published agents can be featured on the homepage")
updated = await store.set_featured(agent_id, body.order)
if updated is None:
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
logger.info("Set agent '%s' featured_order=%s by admin", agent_id, body.order)
return _record_to_response(updated, include_soul=True)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to set featured for agent '%s': %s", agent_id, e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to set featured: {str(e)}") from e
class AgentFeaturedReorderRequest(BaseModel):
"""Request body for admin to bulk-reorder the homepage featured list."""
agent_ids: list[str] = Field(
default_factory=list,
description=(
"Featured agent ids in the desired top→bottom display order. Ids not "
"currently featured are silently skipped, so the client can send the "
"current visible order without first reconciling with the server."
),
)
@router.put(
"/agents/featured/order",
status_code=204,
summary="Reorder Featured Agents (admin)",
description=(
"Admin-only. Bulk-assign ``featured_order`` to each id in the order received "
"(``0`` is topmost). After this, the homepage selector renders agents in "
"exactly the supplied order. Ids not already featured are ignored."
),
)
async def reorder_featured_agents(
body: AgentFeaturedReorderRequest,
request: Request,
) -> None:
_require_agents_api_enabled()
if not _is_admin_user(request):
raise HTTPException(status_code=403, detail="Only system admins can reorder the homepage selector")
try:
store = get_agent_store(request)
for index, agent_id in enumerate(body.agent_ids):
# ``set_featured`` is a no-op for unknown ids — it returns None and
# we move on without raising, matching the docstring.
await store.set_featured(agent_id, index)
logger.info("Admin reordered featured agents (%d ids)", len(body.agent_ids))
except HTTPException:
raise
except Exception as e:
logger.error("Failed to reorder featured agents: %s", e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to reorder featured agents: {str(e)}") from e
offline-backend-20260512/backend/packages/harness/deerflow/config/agents_config.py
+22 -5
View File
@@ -57,11 +57,18 @@ def load_agent_config(agent_id: str | None) -> AgentConfig | None:
agent_id: The agent storage/runtime id.
Returns:
AgentConfig instance.
AgentConfig instance, or ``None`` when the on-disk artifacts are missing.
Raises:
FileNotFoundError: If the agent directory or config.yaml does not exist.
ValueError: If config.yaml cannot be parsed.
ValueError: If config.yaml exists but cannot be parsed.
Tolerance:
A missing agent directory or missing config.yaml is **not** an error —
the runtime simply falls back to default Lead Agent behavior for that
id. This keeps the chat session usable when the DB still has an agent
row but its files have been deleted (e.g. data dir moved between
machines, or filesystem wiped). The router layer is expected to
lazily restore the directory next time the agent is fetched via API.
"""
if agent_id is None:
@@ -72,10 +79,20 @@ def load_agent_config(agent_id: str | None) -> AgentConfig | None:
config_file = agent_dir / "config.yaml"
if not agent_dir.exists():
raise FileNotFoundError(f"Agent directory not found: {agent_dir}")
logger.warning(
"Agent directory missing for id '%s' (%s); falling back to default config",
agent_id,
agent_dir,
)
return None
if not config_file.exists() and not (agent_dir / SOUL_FILENAME).exists():
raise FileNotFoundError(f"Agent config not found: {config_file}")
logger.warning(
"Agent has no config.yaml or %s at %s; falling back to default config",
SOUL_FILENAME,
agent_dir,
)
return None
data: dict[str, Any] = {}
if config_file.exists():
offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/base.py
+20
View File
@@ -61,3 +61,23 @@ class AgentStore(ABC):
async def count_owned(self, user_id: str) -> int:
"""Return the number of agents owned by ``user_id`` (user_id IS NOT NULL)."""
raise NotImplementedError
@abstractmethod
async def set_featured(self, agent_id: str, order: int | None) -> dict[str, Any] | None:
"""Set or clear the homepage-selector featured rank for an agent.
Returns the updated row, or ``None`` if the agent does not exist.
Caller is responsible for authorization (admin) and for verifying
that ``published`` is true when ``order`` is not ``None``.
"""
raise NotImplementedError
@abstractmethod
async def list_featured(self, user_id: str) -> list[dict[str, Any]]:
"""Return featured + still-visible-to-``user_id`` agents, ordered by ``featured_order`` asc."""
raise NotImplementedError
@abstractmethod
async def list_by_ids(self, agent_ids: list[str], user_id: str) -> list[dict[str, Any]]:
"""Return rows for ``agent_ids`` visible to ``user_id``, preserving the input order."""
raise NotImplementedError
offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/memory.py
+21
View File
@@ -55,6 +55,7 @@ class MemoryAgentStore(AgentStore):
"description": data.get("description") or "",
"user_id": data.get("user_id"),
"published": bool(data.get("published", False)),
"featured_order": None,
"created_at": now,
"updated_at": now,
}
@@ -70,6 +71,7 @@ class MemoryAgentStore(AgentStore):
"description": data.get("description") or "",
"user_id": None,
"published": bool(data.get("published", True)),
"featured_order": None,
"created_at": _now(),
"updated_at": _now(),
}
@@ -117,3 +119,22 @@ class MemoryAgentStore(AgentStore):
async def count_owned(self, user_id: str) -> int:
return sum(1 for a in self._agents.values() if a.get("user_id") == user_id)
async def set_featured(self, agent_id: str, order: int | None) -> dict[str, Any] | None:
row = self._agents.get(agent_id)
if row is None:
return None
row["featured_order"] = order
row["updated_at"] = _now()
return _copy(row)
async def list_featured(self, user_id: str) -> list[dict[str, Any]]:
candidates = [a for a in self._agents.values() if a.get("featured_order") is not None and self._is_visible(a, user_id)]
candidates.sort(key=lambda a: (a.get("featured_order"), a.get("created_at") or ""))
return [_copy(a) for a in candidates]
async def list_by_ids(self, agent_ids: list[str], user_id: str) -> list[dict[str, Any]]:
order_map = {aid: idx for idx, aid in enumerate(agent_ids)}
rows = [self._agents[aid] for aid in agent_ids if aid in self._agents and self._is_visible(self._agents[aid], user_id)]
rows.sort(key=lambda a: order_map.get(a["id"], 0))
return [_copy(a) for a in rows]
offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/model.py
+7 -1
View File
@@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Boolean, String
from sqlalchemy import BigInteger, Boolean, String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
@@ -21,5 +21,11 @@ class AgentRow(Base):
description: Mapped[str] = mapped_column(String(1024), nullable=False, default="")
user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
published: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, index=True)
# NULL = not featured on the homepage selector. Non-NULL = admin-curated
# featured rank (lower sorts first). Only ``published=True`` agents are
# eligible; the router enforces that invariant on writes. BIGINT so the
# frontend can seed initial ranks from ``Date.now()`` (ms-resolution
# timestamps overflow a 4-byte INT on MySQL).
featured_order: Mapped[int | None] = mapped_column(BigInteger, nullable=True, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
offline-backend-20260512/backend/packages/harness/deerflow/persistence/agents/sql.py
+32
View File
@@ -142,3 +142,35 @@ class AgentRepository(AgentStore):
stmt = select(func.count()).select_from(AgentRow).where(AgentRow.user_id == user_id)
async with self._sf() as session:
return await session.scalar(stmt) or 0
async def set_featured(self, agent_id: str, order: int | None) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(AgentRow, agent_id)
if row is None:
return None
row.featured_order = order
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
return _row_to_dict(row)
async def list_featured(self, user_id: str) -> list[dict[str, Any]]:
stmt = (
select(AgentRow)
.where(AgentRow.featured_order.is_not(None), self._visible_filter(user_id))
.order_by(AgentRow.featured_order.asc(), AgentRow.created_at.asc())
)
async with self._sf() as session:
result = await session.execute(stmt)
return [_row_to_dict(row) for row in result.scalars()]
async def list_by_ids(self, agent_ids: list[str], user_id: str) -> list[dict[str, Any]]:
if not agent_ids:
return []
order_map = {aid: idx for idx, aid in enumerate(agent_ids)}
stmt = select(AgentRow).where(AgentRow.id.in_(agent_ids), self._visible_filter(user_id))
async with self._sf() as session:
result = await session.execute(stmt)
rows = list(result.scalars())
rows.sort(key=lambda r: order_map.get(r.id, len(order_map)))
return [_row_to_dict(r) for r in rows]
offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py
+137 -78
View File
@@ -10,13 +10,14 @@ None and fall back to in-memory implementations.
from __future__ import annotations
import json
import logging
import sys
from importlib.util import find_spec
from pathlib import Path
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
import asyncio
import json
import logging
import sys
from importlib.util import find_spec
from pathlib import Path
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
def _json_serializer(obj: object) -> str:
@@ -26,37 +27,89 @@ def _json_serializer(obj: object) -> str:
logger = logging.getLogger(__name__)
_engine: AsyncEngine | None = None
_session_factory: async_sessionmaker[AsyncSession] | None = None
def _add_runtime_vendor_path() -> None:
"""Allow offline deployments to place optional DB drivers in DEER_FLOW_HOME.
The offline Docker image keeps the Python environment inside the image, while
``/app/backend/.deer-flow`` is a persistent bind mount. The MySQL switch
script installs ``asyncmy`` into ``{DEER_FLOW_HOME}/python-vendor`` so the
unchanged startup script can still import it after container recreation.
"""
try:
from deerflow.config.runtime_paths import runtime_home
vendor_dir = runtime_home() / "python-vendor"
except Exception:
vendor_dir = Path(".deer-flow/python-vendor").resolve()
vendor = str(vendor_dir)
if vendor_dir.exists() and vendor not in sys.path:
sys.path.insert(0, vendor)
def _mysql_driver_name(url: str) -> str:
if url.startswith("mysql+aiomysql://"):
return "aiomysql"
return "asyncmy"
async def _auto_create_postgres_db(url: str) -> None:
_engine: AsyncEngine | None = None
_session_factory: async_sessionmaker[AsyncSession] | None = None
def _add_runtime_vendor_path() -> None:
"""Allow offline deployments to place optional DB drivers in DEER_FLOW_HOME.
The offline Docker image keeps the Python environment inside the image, while
``/app/backend/.deer-flow`` is a persistent bind mount. The MySQL switch
script installs ``asyncmy`` into ``{DEER_FLOW_HOME}/python-vendor`` so the
unchanged startup script can still import it after container recreation.
"""
try:
from deerflow.config.runtime_paths import runtime_home
vendor_dir = runtime_home() / "python-vendor"
except Exception:
vendor_dir = Path(".deer-flow/python-vendor").resolve()
vendor = str(vendor_dir)
if vendor_dir.exists() and vendor not in sys.path:
sys.path.insert(0, vendor)
def _mysql_driver_name(url: str) -> str:
if url.startswith("mysql+aiomysql://"):
return "aiomysql"
return "asyncmy"
async def _run_alembic_upgrade(url: str) -> None:
"""Apply pending Alembic migrations against ``url`` (``alembic upgrade head``).
Runs after ``Base.metadata.create_all`` so brand-new deployments still get
a complete schema even without Alembic, while existing deployments pick up
in-place schema fixes (column widenings, new indexes, etc.) at the next
restart — no manual ``alembic upgrade`` step required.
Notes:
* Each historical migration in ``migrations/versions/`` already guards
itself with ``_has_table`` / ``_has_column``, so re-running them against
a database that was originally built via ``create_all`` is a sequence of
no-ops up to the first migration that actually has work to do.
* The Alembic ``env.py`` calls ``asyncio.run`` internally, which cannot run
inside an already-running loop — so we hop into a worker thread.
* Failures here are logged but do not abort startup. The app remains
operable on the ``create_all`` schema; the operator just won't get
in-place fixes until they investigate.
"""
ini_path = Path(__file__).parent / "migrations" / "alembic.ini"
if not ini_path.exists():
logger.warning("Alembic config not found at %s; skipping auto-migrate", ini_path)
return
# Alembic's ``Config`` is backed by ConfigParser, which performs ``%``
# interpolation on every option value. URL-encoded credentials such as
# ``kland%4088124932`` (the ``@`` in ``kland@88124932``) trip that up with
# ``ValueError: invalid interpolation syntax ... at position N``. Doubling
# the percent signs neutralises the interpolation; ConfigParser un-doubles
# them back to the original value when read.
escaped_url = url.replace("%", "%%")
def _do_upgrade() -> None:
# Import inside the worker so a missing Alembic install can't break
# the engine import chain.
from alembic import command
from alembic.config import Config
cfg = Config(str(ini_path))
cfg.set_main_option("sqlalchemy.url", escaped_url)
# The ``script_location`` in alembic.ini is ``%(here)s`` which points
# at the directory of alembic.ini itself — already correct for us.
command.upgrade(cfg, "head")
try:
await asyncio.to_thread(_do_upgrade)
logger.info("Alembic migrations applied (or already up-to-date)")
except Exception:
logger.exception("Auto-applying Alembic migrations failed; continuing with create_all schema")
async def _auto_create_postgres_db(url: str) -> None:
"""Connect to the ``postgres`` maintenance DB and CREATE DATABASE.
The target database name is extracted from *url*. The connection is
@@ -94,8 +147,8 @@ async def init_engine(
"""Create the async engine and session factory, then auto-create tables.
Args:
backend: "memory", "sqlite", "postgres", or "mysql".
url: SQLAlchemy async URL (for sqlite/postgres/mysql).
backend: "memory", "sqlite", "postgres", or "mysql".
url: SQLAlchemy async URL (for sqlite/postgres/mysql).
echo: Echo SQL to log.
pool_size: Postgres connection pool size.
sqlite_dir: Directory to create for SQLite (ensured to exist).
@@ -106,20 +159,20 @@ async def init_engine(
logger.info("Persistence backend=memory -- ORM engine not initialized")
return
if backend == "postgres":
try:
import asyncpg # noqa: F401
except ImportError:
raise ImportError("database.backend is set to 'postgres' but asyncpg is not installed.\nInstall it with:\n uv sync --extra postgres\nOr switch to backend: sqlite in config.yaml for single-node deployment.") from None
if backend == "mysql":
_add_runtime_vendor_path()
driver = _mysql_driver_name(url)
if find_spec(driver) is None:
raise ImportError(
"database.backend is set to 'mysql' but the MySQL async driver is not installed.\n"
"Install asyncmy, or run switch_database_to_mysql.sh so it can install asyncmy into "
f"{Path('.deer-flow/python-vendor')} from the offline wheelhouse."
)
if backend == "postgres":
try:
import asyncpg # noqa: F401
except ImportError:
raise ImportError("database.backend is set to 'postgres' but asyncpg is not installed.\nInstall it with:\n uv sync --extra postgres\nOr switch to backend: sqlite in config.yaml for single-node deployment.") from None
if backend == "mysql":
_add_runtime_vendor_path()
driver = _mysql_driver_name(url)
if find_spec(driver) is None:
raise ImportError(
"database.backend is set to 'mysql' but the MySQL async driver is not installed.\n"
"Install asyncmy, or run switch_database_to_mysql.sh so it can install asyncmy into "
f"{Path('.deer-flow/python-vendor')} from the offline wheelhouse."
)
if backend == "sqlite":
import os
@@ -150,29 +203,31 @@ async def init_engine(
cursor.execute("PRAGMA foreign_keys=ON;")
finally:
cursor.close()
elif backend == "postgres":
_engine = create_async_engine(
url,
echo=echo,
pool_size=pool_size,
pool_pre_ping=True,
json_serializer=_json_serializer,
)
elif backend == "mysql":
_engine = create_async_engine(
url,
echo=echo,
pool_size=pool_size,
pool_pre_ping=True,
pool_recycle=1800,
json_serializer=_json_serializer,
)
else:
raise ValueError(f"Unknown persistence backend: {backend!r}")
elif backend == "postgres":
_engine = create_async_engine(
url,
echo=echo,
pool_size=pool_size,
pool_pre_ping=True,
json_serializer=_json_serializer,
)
elif backend == "mysql":
_engine = create_async_engine(
url,
echo=echo,
pool_size=pool_size,
pool_pre_ping=True,
pool_recycle=1800,
json_serializer=_json_serializer,
)
else:
raise ValueError(f"Unknown persistence backend: {backend!r}")
_session_factory = async_sessionmaker(_engine, expire_on_commit=False)
# Auto-create tables (dev convenience). Production should use Alembic.
# Auto-create tables for fresh deployments, then let Alembic apply any
# in-place schema fixes on top (see ``_run_alembic_upgrade`` below). Both
# are idempotent so repeating them across restarts is harmless.
from deerflow.persistence.base import Base
# Import all models so Base.metadata discovers them.
@@ -200,6 +255,10 @@ async def init_engine(
else:
raise
# Apply in-place schema fixes (column widenings, etc.) that ``create_all``
# cannot do on its own. Failures don't abort startup — see the helper.
await _run_alembic_upgrade(url)
logger.info("Persistence engine initialized: backend=%s", backend)
@@ -209,10 +268,10 @@ async def init_engine_from_config(config) -> None:
await init_engine("memory")
return
await init_engine(
backend=config.backend,
url=config.app_sqlalchemy_url,
echo=config.echo_sql,
pool_size=config.pool_size,
backend=config.backend,
url=config.app_sqlalchemy_url,
echo=config.echo_sql,
pool_size=config.pool_size,
sqlite_dir=config.sqlite_dir if config.backend == "sqlite" else "",
)
offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260520_01_agent_pins_and_featured.py
+83
View File
@@ -0,0 +1,83 @@
"""Homepage agent selector: ``user_agent_pin`` table + ``agents.featured_order``.
Adds two pieces backing the new chat-homepage agent picker:
- ``agents.featured_order`` (nullable INT) — admin-curated featured rank.
NULL = not shown on the selector. Non-NULL rows sort ascending.
- ``user_agent_pin`` — per-user pinned agents (user's "favorites"). Composite
PK ``(user_id, agent_id)`` so a user pins each agent at most once. Foreign
key to ``agents.id`` with ``ON DELETE CASCADE`` so deleting an agent also
removes every user's pin of it.
Revision ID: 20260520_01
Revises: 20260519_02
Create Date: 2026-05-20
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260520_01"
down_revision = "20260519_02"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
inspector = inspect(op.get_bind())
return table_name in inspector.get_table_names()
def _has_column(table_name: str, column_name: str) -> bool:
if not _has_table(table_name):
return False
inspector = inspect(op.get_bind())
return any(col["name"] == column_name for col in inspector.get_columns(table_name))
def _has_index(table_name: str, index_name: str) -> bool:
if not _has_table(table_name):
return False
inspector = inspect(op.get_bind())
return any(idx["name"] == index_name for idx in inspector.get_indexes(table_name))
def upgrade() -> None:
# 1. agents.featured_order — ALTER existing table. Skipped on brand-new DBs
# where create_all already materialised the column.
if _has_table("agents") and not _has_column("agents", "featured_order"):
op.add_column("agents", sa.Column("featured_order", sa.Integer(), nullable=True))
if _has_table("agents") and not _has_index("agents", "ix_agents_featured_order"):
op.create_index("ix_agents_featured_order", "agents", ["featured_order"])
# 2. user_agent_pin — per-user pinned agents.
if not _has_table("user_agent_pin"):
op.create_table(
"user_agent_pin",
sa.Column("user_id", sa.String(64), nullable=False),
sa.Column(
"agent_id",
sa.String(64),
sa.ForeignKey("agents.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.PrimaryKeyConstraint("user_id", "agent_id", name="pk_user_agent_pin"),
)
op.create_index("ix_user_agent_pin_sort_order", "user_agent_pin", ["sort_order"])
def downgrade() -> None:
if _has_index("user_agent_pin", "ix_user_agent_pin_sort_order"):
op.drop_index("ix_user_agent_pin_sort_order", table_name="user_agent_pin")
if _has_table("user_agent_pin"):
op.drop_table("user_agent_pin")
if _has_index("agents", "ix_agents_featured_order"):
op.drop_index("ix_agents_featured_order", table_name="agents")
if _has_column("agents", "featured_order"):
op.drop_column("agents", "featured_order")
offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260520_02_featured_order_bigint.py
+66
View File
@@ -0,0 +1,66 @@
"""Widen ``agents.featured_order`` from INT to BIGINT.
The previous revision (``20260520_01``) created ``featured_order`` as plain
``INTEGER``. The frontend seeds the initial rank with ``Date.now()`` so admins
get "newest featured sinks to the bottom" with zero UI — but ms-resolution
timestamps (~1.78e12) overflow a 4-byte signed INT on MySQL, surfacing as
sqlalchemy.exc.DataError: (1264, "Out of range value for column 'featured_order'")
SQLite and PostgreSQL ``INTEGER`` is already 64-bit / arbitrary-precision and
needs no change, but the migration runs everywhere for schema parity.
Revision ID: 20260520_02
Revises: 20260520_01
Create Date: 2026-05-20
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260520_02"
down_revision = "20260520_01"
branch_labels = None
depends_on = None
def _has_column(table_name: str, column_name: str) -> bool:
inspector = inspect(op.get_bind())
if table_name not in inspector.get_table_names():
return False
return any(col["name"] == column_name for col in inspector.get_columns(table_name))
def upgrade() -> None:
if not _has_column("agents", "featured_order"):
return
bind = op.get_bind()
# SQLite has only INTEGER (dynamic precision) — alter_column would rebuild
# the table for nothing. PostgreSQL INTEGER is 32-bit too, so alter it.
if bind.dialect.name == "sqlite":
return
op.alter_column(
"agents",
"featured_order",
type_=sa.BigInteger(),
existing_nullable=True,
)
def downgrade() -> None:
if not _has_column("agents", "featured_order"):
return
bind = op.get_bind()
if bind.dialect.name == "sqlite":
return
# Any rows with values > 2^31-1 would be truncated. Downgrade is dev-only.
op.alter_column(
"agents",
"featured_order",
type_=sa.Integer(),
existing_nullable=True,
)
offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260520_03_drop_user_agent_pin.py
+61
View File
@@ -0,0 +1,61 @@
"""Drop ``user_agent_pin`` (homepage selector simplified to admin-only).
The original design split the homepage agent picker into two buckets:
admin-curated "featured" + per-user "pinned". Product feedback showed the
per-user side added complexity without clear value — admins now own the
single list shown to everyone. ``agents.featured_order`` is the only piece
that survives; this revision removes the now-unused pin table.
Revision ID: 20260520_03
Revises: 20260520_02
Create Date: 2026-05-20
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260520_03"
down_revision = "20260520_02"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
inspector = inspect(op.get_bind())
return table_name in inspector.get_table_names()
def _has_index(table_name: str, index_name: str) -> bool:
if not _has_table(table_name):
return False
inspector = inspect(op.get_bind())
return any(idx["name"] == index_name for idx in inspector.get_indexes(table_name))
def upgrade() -> None:
if _has_index("user_agent_pin", "ix_user_agent_pin_sort_order"):
op.drop_index("ix_user_agent_pin_sort_order", table_name="user_agent_pin")
if _has_table("user_agent_pin"):
op.drop_table("user_agent_pin")
def downgrade() -> None:
if not _has_table("user_agent_pin"):
op.create_table(
"user_agent_pin",
sa.Column("user_id", sa.String(64), nullable=False),
sa.Column(
"agent_id",
sa.String(64),
sa.ForeignKey("agents.id", ondelete="CASCADE"),
nullable=False,
),
sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.PrimaryKeyConstraint("user_id", "agent_id", name="pk_user_agent_pin"),
)
op.create_index("ix_user_agent_pin_sort_order", "user_agent_pin", ["sort_order"])