mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-05 20:38:41 +00:00
0d4d0cb17d
* feat(agents): database-backed storage for custom agent definitions Add an agent_storage.backend switch (default file, behaviour-unchanged) with a db backend that stores each custom agent as a row in the shared SQL persistence layer, so a multi-instance deployment sees the same agents on every node (#4331, #4357). Introduces an AgentStore interface routing all read/write surfaces, an agents table + migration 0006, startup validation, and a file->db importer. Follows the thread_meta store / run_events backend-switch / 0003_scheduled_tasks migration patterns; no new dependency. * fix(agents): make db storage path production-ready (review round 1) Addresses review feedback on the db/sync agent-storage path: - sql.py: mirror the async engine's per-connection SQLite PRAGMAs on the sync engine (busy_timeout=30000, synchronous=NORMAL, foreign_keys=ON, WAL) so both engines behave identically against the shared DB; guard the engine cache with a lock (double-checked) so concurrent first-touch cannot build duplicate engines or register the connect listener twice. - routers/agents.py + routers/assistants_compat.py: offload the sync-store reads that ran on the event loop (list/get/check, update's pre-read + legacy guard + refresh, and assistants_compat's four list routes) via asyncio.to_thread — on db+postgres each was a network round trip stalling the loop. Writes were already offloaded. - file.py: translate the create() mkdir(exist_ok=False) race FileExistsError into AgentExistsError (router 409, matching SqlAgentStore's IntegrityError path); correct the _write docstring — per-file atomic replace, two commits sequential not transactional. Tests: sync-engine PRAGMA + engine-cache reuse assertions; file create-race -> AgentExistsError; strict Blockbuster anchor over the read endpoints so a regression back onto the loop fails CI. * fix(agents): address round-2 review on the db store path - update_agent tool: align the docstring/inline comment with FileAgentStore._write. Cross-field write atomicity is db-only; the file backend commits config then soul via two sequential os.replace (a crash between them can leave a fresh config.yaml beside a stale SOUL.md). The dropped partial-write *reporting* is an intentional tradeoff — the stage-then-replace safety is preserved (test_update_agent_soul_failure_does_not_replace_config still holds). - SqlAgentStore.update(): true upsert. Catch IntegrityError on the insert-on-missing branch, re-fetch and apply, so two concurrent first-time writes (e.g. two setup_agent handshakes) converge instead of surfacing a raw UNIQUE(user_id, name) violation as a 500. Symmetric with create(). - get_agent_store(): document the graph-subprocess config-resolution invariant (the except->file fallback is a genuine no-config path, not a mask for a misconfigured graph process) and pin it with two tests driving the real get_app_config() file resolution: db resolves from an on-disk config.yaml, file fallback when config is unresolvable. * test(agents): cover SqlAgentStore.update() write-race upsert recovery Mandatory-TDD test for the round-2 fix in 0680340a: two concurrent first-time update()s where the loser's insert hits UNIQUE(user_id, name). Deterministically forces the IntegrityError recovery path by making the first _row probe miss the committed winner, and asserts last-writer-wins instead of a surfaced 500.
51 lines
1.8 KiB
Python
51 lines
1.8 KiB
Python
"""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")
|