fix: filter position-granted agent favorites

This commit is contained in:
董宏凯 committed 2026-07-01 14:51:14 +08:00
1 parent f3e5a3bb35
commit fa04ab8b5a
4 files changed
+496 -85

No files matched your search

@@ -6,20 +6,21 @@ from datetime import UTC, datetime
from typing import Any
from uuid import uuid4
from sqlalchemy import and_, case, delete, func, or_, select
from sqlalchemy import and_, case, delete, func, literal, not_, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.config.agent_name_desensitize import source_terms_for_needle
from deerflow.persistence.agents.base import AgentStore
from deerflow.persistence.agents.model import (
AgentExtrasRow,
AgentFavoriteRow,
AgentPinRow,
AgentRow,
AgentSkillRow,
)
from deerflow.persistence.tags.model import TagAssignmentRow
from deerflow.persistence.agents.model import (
AgentExtrasRow,
AgentFavoriteRow,
AgentPinRow,
AgentRow,
AgentSkillRow,
)
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
from deerflow.persistence.tags.model import TagAssignmentRow
def _row_to_dict(row: AgentRow) -> dict[str, Any]:
@@ -32,9 +33,33 @@ def _row_to_dict(row: AgentRow) -> dict[str, Any]:
return data
class AgentRepository(AgentStore):
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
class AgentRepository(AgentStore):
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
@staticmethod
def _business_chain_agent_exists(agent_id_col):
return (
select(RoundtableChainRow.id)
.where(
or_(
RoundtableChainRow.seats.like(literal('%"agent_id": "') + agent_id_col + literal('"%')),
RoundtableChainRow.seats.like(literal('%"agent_id":"') + agent_id_col + literal('"%')),
)
)
.exists()
)
@classmethod
def _valid_favorite_filter(cls):
return or_(
AgentFavoriteRow.origin == "user",
and_(
AgentFavoriteRow.origin == "position",
AgentRow.user_id.is_not(None),
not_(cls._business_chain_agent_exists(AgentFavoriteRow.agent_id)),
),
)
@staticmethod
def _visible_filter(user_id: str):
@@ -52,15 +77,23 @@ class AgentRepository(AgentStore):
``published=false``, so this clause cannot lean on ``published``;
- favorited — the "我的" listing keeps favorited agents even after
the owner unpublishes them, so reads keep honoring the favorite.
"""
fav_subq = select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id)
return or_(
AgentRow.user_id.is_(None),
AgentRow.user_id == user_id,
AgentRow.published.is_(True),
AgentRow.is_private.is_(True),
AgentRow.id.in_(fav_subq),
)
"""
valid_favorite_exists = (
select(AgentFavoriteRow.id)
.where(
AgentFavoriteRow.user_id == user_id,
AgentFavoriteRow.agent_id == AgentRow.id,
AgentRepository._valid_favorite_filter(),
)
.exists()
)
return or_(
AgentRow.user_id.is_(None),
AgentRow.user_id == user_id,
AgentRow.published.is_(True),
AgentRow.is_private.is_(True),
valid_favorite_exists,
)
@staticmethod
def _listing_filter(user_id: str):
@@ -268,11 +301,20 @@ class AgentRepository(AgentStore):
or_(AgentRow.user_id.is_(None), AgentRow.published.is_(True)),
AgentRow.is_private.is_(False),
)
if scope == "mine":
# owned OR favorited. A correlated IN-subquery (not a JOIN) keeps a
# single row per agent even when it is both owned and favorited.
fav_subq = select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id)
return or_(AgentRow.user_id == user_id, AgentRow.id.in_(fav_subq))
if scope == "mine":
# owned OR valid favorites. Built-in agents and business-chain
# seats only enter "mine" when explicitly favorited by the user;
# the original position-granted behavior stays for ordinary agents.
valid_favorite_exists = (
select(AgentFavoriteRow.id)
.where(
AgentFavoriteRow.user_id == user_id,
AgentFavoriteRow.agent_id == AgentRow.id,
self._valid_favorite_filter(),
)
.exists()
)
return or_(AgentRow.user_id == user_id, valid_favorite_exists)
if scope == "private":
return AgentRow.is_private.is_(True)
if scope == "builtin":
@@ -437,18 +479,22 @@ class AgentRepository(AgentStore):
# --- Favorites ("收藏") ---
async def add_favorite(self, user_id: str, agent_id: str) -> None:
async with self._sf() as session:
existing = (
await session.execute(
select(AgentFavoriteRow.id)
.where(AgentFavoriteRow.user_id == user_id, AgentFavoriteRow.agent_id == agent_id)
.limit(1)
)
).scalar_one_or_none()
if existing is not None:
return
session.add(AgentFavoriteRow(id=uuid4().hex, user_id=user_id, agent_id=agent_id))
async def add_favorite(self, user_id: str, agent_id: str) -> None:
async with self._sf() as session:
existing = (
await session.execute(
select(AgentFavoriteRow)
.where(AgentFavoriteRow.user_id == user_id, AgentFavoriteRow.agent_id == agent_id)
.limit(1)
)
).scalar_one_or_none()
if existing is not None:
if existing.origin != "user" or existing.position_id is not None:
existing.origin = "user"
existing.position_id = None
await session.commit()
return
session.add(AgentFavoriteRow(id=uuid4().hex, user_id=user_id, agent_id=agent_id))
try:
await session.commit()
except IntegrityError:
@@ -467,12 +513,17 @@ class AgentRepository(AgentStore):
await session.commit()
return (result.rowcount or 0) > 0
async def list_favorite_ids(self, user_id: str) -> set[str]:
async with self._sf() as session:
result = await session.execute(
select(AgentFavoriteRow.agent_id).where(AgentFavoriteRow.user_id == user_id)
)
return {row[0] for row in result.all()}
async def list_favorite_ids(self, user_id: str) -> set[str]:
async with self._sf() as session:
result = await session.execute(
select(AgentFavoriteRow.agent_id)
.join(AgentRow, AgentRow.id == AgentFavoriteRow.agent_id)
.where(
AgentFavoriteRow.user_id == user_id,
self._valid_favorite_filter(),
)
)
return {row[0] for row in result.all()}
async def delete_favorites_for_agent(self, agent_id: str) -> None:
async with self._sf() as session:
@@ -0,0 +1,90 @@
"""Stop position-granted favorites from collecting built-ins/chain seats.
Built-in agents and agents used as business-chain seats should only enter
"mine" when a user explicitly clicks the favorite action. This migration
removes historical automatic ``origin='position'`` rows for those agents while
preserving user-created favorites (``origin='user'``) and ordinary
position-granted agent favorites.
Revision ID: 20260701_01
Revises: 20260625_04
Create Date: 2026-07-01
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260701_01"
down_revision = "20260625_04"
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 upgrade() -> None:
if not (
_has_table("agent_favorites")
and _has_column("agent_favorites", "origin")
and _has_column("agent_favorites", "agent_id")
):
return
favorites = sa.table(
"agent_favorites",
sa.column("agent_id", sa.String(64)),
sa.column("origin", sa.String(16)),
)
cleanup_conditions = []
if _has_table("agents") and _has_column("agents", "user_id"):
agents = sa.table(
"agents",
sa.column("id", sa.String(64)),
sa.column("user_id", sa.String(64)),
)
cleanup_conditions.append(
favorites.c.agent_id.in_(sa.select(agents.c.id).where(agents.c.user_id.is_(None)))
)
if _has_table("roundtable_chains") and _has_column("roundtable_chains", "seats"):
chains = sa.table(
"roundtable_chains",
sa.column("seats", sa.Text()),
)
cleanup_conditions.append(
sa.select(sa.literal(1))
.select_from(chains)
.where(
sa.or_(
chains.c.seats.like(sa.literal('%"agent_id": "') + favorites.c.agent_id + sa.literal('"%')),
chains.c.seats.like(sa.literal('%"agent_id":"') + favorites.c.agent_id + sa.literal('"%')),
)
)
.exists()
)
if not cleanup_conditions:
return
op.get_bind().execute(
favorites.delete().where(
favorites.c.origin == "position",
sa.or_(*cleanup_conditions),
)
)
def downgrade() -> None:
# Data cleanup only; removed auto-grants are not reconstructable.
pass
@@ -12,7 +12,8 @@ Resolution rule (per resource type):
* For a resource type **with** bound tags → the union of resources carrying any
of those tags.
* For a resource type **without** bound tags → the *default set*:
- ``agent`` → all built-in agents (``user_id IS NULL``)
- ``agent`` -> built-ins first, then filtered out of
position-granted favorites
- ``scheduled_task`` → empty (no built-in tasks)
- ``skill`` → empty (the 广场 still shows every visible skill;
a position only *adds* its tagged skills to the member's "我的技能")
@@ -24,9 +25,12 @@ Resolution rule (per resource type):
Materialization (``sync_user``) only ever touches rows it created itself
(``origin='position'``), so a user's own favorites / subscriptions are never
clobbered. ``skill`` is materialized into ``skill_favorites`` exactly like
``agent`` → ``agent_favorites``: the tagged skills are *added* to the user's
"我的技能" while the browse-all 广场 stays complete (it is **not** restricted).
clobbered. ``agent`` keeps the original position-granted behavior for ordinary
tagged agents, but filters out built-ins and business-chain seats; those only
enter "mine" after an explicit user favorite. ``skill`` is still materialized
into ``skill_favorites`` so tagged skills are added to the user's "mine"
collection while the browse-all square stays complete (it is **not**
restricted).
Only recommended-question visibility is computed live by :meth:`resolve_visible`
and is *not* materialized. (The admin-configured prompts shown to users ARE the
recommended questions — there is no separate prompt-template store.)
@@ -34,6 +38,7 @@ recommended questions — there is no separate prompt-template store.)
from __future__ import annotations
import json
import logging
import uuid
from datetime import UTC, datetime
@@ -43,13 +48,15 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
from deerflow.persistence.positions.base import PositionStore
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskSubscriptionRow
from deerflow.persistence.skills.model import SkillFavoriteRow
from deerflow.persistence.tags.base import TagStore
logger = logging.getLogger(__name__)
# Materialized into "我的" tables by sync_user.
# Reconciled by sync_user. ``agent`` preserves ordinary position-granted rows,
# but filters built-ins and business-chain seats.
_MATERIALIZED_TYPES = ("agent", "skill", "scheduled_task")
# Resolved live at query time (visibility filters), never materialized.
_VISIBILITY_TYPES = ("recommended_question",)
@@ -93,8 +100,12 @@ class PositionSyncService:
bindings = await self._positions.get_tag_bindings(pid) if pid else {}
tag_ids = bindings.get(resource_type) or []
if tag_ids:
return set(await self._tags.list_targets_for_tags(resource_type, tag_ids))
return await self._default_set(resource_type, session)
target = set(await self._tags.list_targets_for_tags(resource_type, tag_ids))
else:
target = await self._default_set(resource_type, session)
if resource_type == "agent":
return await self._filter_materializable_agents(target, session)
return target
async def _default_set(self, resource_type: str, session: AsyncSession) -> set[str]:
if resource_type == "agent":
@@ -103,6 +114,37 @@ class PositionSyncService:
# scheduled_task (and any other materialized type) defaults to empty.
return set()
async def _filter_materializable_agents(self, agent_ids: set[str], session: AsyncSession) -> set[str]:
if not agent_ids:
return set()
rows = (
await session.execute(
select(AgentRow.id).where(
AgentRow.id.in_(agent_ids),
AgentRow.user_id.is_not(None),
)
)
).all()
candidate_ids = {r[0] for r in rows}
return candidate_ids - await self._business_chain_agent_ids(session)
async def _business_chain_agent_ids(self, session: AsyncSession) -> set[str]:
rows = (await session.execute(select(RoundtableChainRow.seats))).scalars()
agent_ids: set[str] = set()
for raw in rows:
if not raw:
continue
try:
seats = json.loads(raw) if isinstance(raw, str) else raw
except Exception:
continue
if not isinstance(seats, list):
continue
for seat in seats:
if isinstance(seat, dict) and isinstance(seat.get("agent_id"), str):
agent_ids.add(seat["agent_id"])
return agent_ids
# --------------------------------------------------------- materialize
async def sync_user(self, user_id: str) -> None:
@@ -1,33 +1,131 @@
from __future__ import annotations
import pytest
import yaml
from app.gateway.routers.agents import (
_backfill_listing_cache,
_load_config_extras,
_write_agent_config,
)
from app.gateway.services import build_run_config
from deerflow.config import paths as paths_module
from deerflow.config.agents_config import AgentConfig
from deerflow.persistence.agents import MemoryAgentStore
from __future__ import annotations
import json
from uuid import uuid4
import pytest
import pytest_asyncio
import yaml
from sqlalchemy import select
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from app.gateway.routers.agents import (
_backfill_listing_cache,
_load_config_extras,
_write_agent_config,
)
from app.gateway.services import build_run_config
from deerflow.config import paths as paths_module
from deerflow.config.agents_config import AgentConfig
from deerflow.persistence.agents import AgentRepository, MemoryAgentStore
from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
from deerflow.persistence.base import Base
from deerflow.persistence.positions.memory import MemoryPositionStore
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
from deerflow.persistence.tags.memory import MemoryTagStore
from deerflow.services.position_sync import PositionSyncService
@pytest.fixture
def isolated_paths(tmp_path, monkeypatch):
"""Redirect get_paths() at a fresh temp directory for the duration of one test."""
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
monkeypatch.setattr(paths_module, "_paths", None)
yield tmp_path
monkeypatch.setattr(paths_module, "_paths", None)
def isolated_paths(tmp_path, monkeypatch):
"""Redirect get_paths() at a fresh temp directory for the duration of one test."""
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
monkeypatch.setattr(paths_module, "_paths", None)
yield tmp_path
monkeypatch.setattr(paths_module, "_paths", None)
@pytest_asyncio.fixture
async def sql_session_factory(tmp_path):
engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'agent-tests.db'}")
async with engine.begin() as conn:
# Importing PositionSyncService above registers the agent / skill /
# scheduled-task rows it materializes into.
await conn.run_sync(Base.metadata.create_all)
sf = async_sessionmaker(engine, expire_on_commit=False)
try:
yield sf
finally:
await engine.dispose()
async def _add_sql_agent(sf, *, agent_id: str, user_id: str | None) -> None:
async with sf() as session:
session.add(
AgentRow(
id=agent_id,
name=agent_id,
description="",
user_id=user_id,
published=True,
is_private=False,
square_id="",
)
)
await session.commit()
async def _add_sql_favorite(sf, *, user_id: str, agent_id: str, origin: str) -> None:
async with sf() as session:
session.add(
AgentFavoriteRow(
id=uuid4().hex,
user_id=user_id,
agent_id=agent_id,
origin=origin,
position_id="default" if origin == "position" else None,
)
)
await session.commit()
async def _add_sql_chain(sf, *, chain_id: str, agent_ids: list[str]) -> None:
async with sf() as session:
session.add(
RoundtableChainRow(
id=chain_id,
user_id="chain-owner",
title=chain_id,
description="",
seats=json.dumps(
[{"agent_id": agent_id, "name": agent_id, "description": ""} for agent_id in agent_ids],
ensure_ascii=False,
),
)
)
await session.commit()
async def _sql_favorite_rows(sf, user_id: str) -> set[tuple[str, str]]:
async with sf() as session:
rows = (
await session.execute(
select(AgentFavoriteRow.agent_id, AgentFavoriteRow.origin).where(
AgentFavoriteRow.user_id == user_id
)
)
).all()
return {(agent_id, origin) for agent_id, origin in rows}
async def _default_positions() -> MemoryPositionStore:
positions = MemoryPositionStore()
await positions.create_position(
{
"id": "default",
"name": "Default",
"description": "",
"is_default": True,
}
)
return positions
@pytest.mark.asyncio
async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility() -> None:
store = MemoryAgentStore()
chinese_name = "\u4e2d\u6587\u52a9\u624b"
async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility() -> None:
store = MemoryAgentStore()
chinese_name = "\u4e2d\u6587\u52a9\u624b"
await store.ensure_builtin({"id": "builtin-agent", "name": chinese_name, "description": "built in"})
await store.create({"id": "private-a", "name": chinese_name, "description": "private", "user_id": "user-a"})
await store.create({"id": "published-b", "name": chinese_name, "description": "published", "user_id": "user-b", "published": True})
@@ -35,14 +133,144 @@ async def test_agent_store_allows_duplicate_chinese_names_and_filters_visibility
user_a_ids = {agent["id"] for agent in await store.list_visible("user-a")}
assert user_a_ids == {"builtin-agent", "private-a", "published-b"}
user_c_ids = {agent["id"] for agent in await store.list_visible("user-c")}
assert user_c_ids == {"builtin-agent", "published-b"}
@pytest.mark.asyncio
async def test_agent_store_only_owner_can_update_or_delete() -> None:
store = MemoryAgentStore()
user_c_ids = {agent["id"] for agent in await store.list_visible("user-c")}
assert user_c_ids == {"builtin-agent", "published-b"}
@pytest.mark.asyncio
async def test_position_sync_default_does_not_collect_builtin_agents(sql_session_factory) -> None:
await _add_sql_agent(sql_session_factory, agent_id="builtin-auto", user_id=None)
await _add_sql_agent(sql_session_factory, agent_id="builtin-manual", user_id=None)
await _add_sql_favorite(
sql_session_factory,
user_id="user-a",
agent_id="builtin-auto",
origin="position",
)
await _add_sql_favorite(
sql_session_factory,
user_id="user-a",
agent_id="builtin-manual",
origin="user",
)
sync = PositionSyncService(
sql_session_factory,
await _default_positions(),
MemoryTagStore(),
)
await sync.sync_user("user-a")
assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
("builtin-manual", "user")
}
@pytest.mark.asyncio
async def test_position_sync_tagged_agents_are_not_auto_favorited(sql_session_factory) -> None:
await _add_sql_agent(sql_session_factory, agent_id="builtin-tagged", user_id=None)
await _add_sql_agent(sql_session_factory, agent_id="custom-tagged", user_id="owner")
await _add_sql_agent(sql_session_factory, agent_id="chain-tagged", user_id="owner")
await _add_sql_chain(sql_session_factory, chain_id="chain-1", agent_ids=["chain-tagged"])
positions = await _default_positions()
tags = MemoryTagStore()
await tags.create_tag({"id": "agent-tag", "name": "Agent tag", "tag_type": "agent"})
await tags.set_target_tags("agent", "builtin-tagged", ["agent-tag"])
await tags.set_target_tags("agent", "custom-tagged", ["agent-tag"])
await tags.set_target_tags("agent", "chain-tagged", ["agent-tag"])
await positions.set_tag_bindings("default", {"agent": ["agent-tag"]})
sync = PositionSyncService(sql_session_factory, positions, tags)
await sync.sync_user("user-a")
assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
("custom-tagged", "position")
}
@pytest.mark.asyncio
async def test_mine_scope_ignores_auto_agent_favorites_but_keeps_manual_builtin(sql_session_factory) -> None:
await _add_sql_agent(sql_session_factory, agent_id="builtin-auto", user_id=None)
await _add_sql_agent(sql_session_factory, agent_id="builtin-manual", user_id=None)
await _add_sql_agent(sql_session_factory, agent_id="custom-auto", user_id="owner")
await _add_sql_agent(sql_session_factory, agent_id="chain-auto", user_id="owner")
await _add_sql_chain(sql_session_factory, chain_id="chain-1", agent_ids=["chain-auto"])
await _add_sql_favorite(
sql_session_factory,
user_id="user-a",
agent_id="builtin-auto",
origin="position",
)
await _add_sql_favorite(
sql_session_factory,
user_id="user-a",
agent_id="builtin-manual",
origin="user",
)
await _add_sql_favorite(
sql_session_factory,
user_id="user-a",
agent_id="custom-auto",
origin="position",
)
await _add_sql_favorite(
sql_session_factory,
user_id="user-a",
agent_id="chain-auto",
origin="position",
)
store = AgentRepository(sql_session_factory)
rows, total = await store.list_paginated(
user_id="user-a",
scope="mine",
search=None,
tag_ids=None,
page=1,
page_size=20,
)
assert total == 2
assert {row["id"] for row in rows} == {"builtin-manual", "custom-auto"}
assert await store.list_favorite_ids("user-a") == {"builtin-manual", "custom-auto"}
@pytest.mark.asyncio
async def test_manual_favorite_upgrades_existing_auto_builtin_favorite(sql_session_factory) -> None:
await _add_sql_agent(sql_session_factory, agent_id="builtin-agent", user_id=None)
await _add_sql_favorite(
sql_session_factory,
user_id="user-a",
agent_id="builtin-agent",
origin="position",
)
store = AgentRepository(sql_session_factory)
assert await store.list_favorite_ids("user-a") == set()
await store.add_favorite("user-a", "builtin-agent")
assert await _sql_favorite_rows(sql_session_factory, "user-a") == {
("builtin-agent", "user")
}
assert await store.list_favorite_ids("user-a") == {"builtin-agent"}
rows, total = await store.list_paginated(
user_id="user-a",
scope="mine",
search=None,
tag_ids=None,
page=1,
page_size=20,
)
assert total == 1
assert [row["id"] for row in rows] == ["builtin-agent"]
@pytest.mark.asyncio
async def test_agent_store_only_owner_can_update_or_delete() -> None:
store = MemoryAgentStore()
await store.ensure_builtin({"id": "builtin-agent", "name": "Built in"})
await store.create({"id": "private-a", "name": "Owned", "user_id": "user-a"})