feat: 技能收藏与常驻(always-on)支持

- 新增 skill_favorites / skill_always_on 迁移与持久化模型字段
- skills 路由、存储、前端设置页支持收藏与常驻
- 调整摘要中间件与 lead agent 提示词
- 同步岗位 position_sync 及相关配置清理

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed 2026-06-16 16:37:43 +08:00
1 parent 5ebbb82d66
commit 5182ea5d11
19 files changed
+472 -107

No files matched your search

@@ -2,6 +2,7 @@
import {
AlertCircle,
Anchor,
ArchiveIcon,
ArrowDownUpIcon,
BookmarkIcon,
@@ -77,6 +78,7 @@ import {
useGenerateSkillNameZh,
useInstallSkillUpload,
useReconcileCustomSkills,
useSetSkillAlwaysOn,
useSetSkillFeatured,
useSetSkillNameZh,
useSetSkillPinned,
@@ -84,7 +86,11 @@ import {
useSkills,
useValidateSkillUpload,
} from "@/core/skills/hooks";
import { usePublishSquares, useViewMode } from "@/core/system-settings/hooks";
import {
usePublishSquares,
useSkillCompressionSettings,
useViewMode,
} from "@/core/system-settings/hooks";
import type { Skill } from "@/core/skills/type";
import { formatDateTime } from "@/core/utils/datetime";
import { env } from "@/env";
@@ -181,8 +187,10 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
const isAdmin = storedAuth?.system_role === "admin";
// Name + tag filtering is resolved server-side via the /api/skills query.
// Non-admins get their 岗位 (position) visibility applied (scope=mine): a
// position that binds skill tags restricts which skills they can see.
// A 岗位 (position) no longer *restricts* the list — it grants skills into the
// member's 我的技能 (each such skill comes back with ``favorited: true``), so
// the 广场 always shows every visible skill and the 我的 tab includes the
// position-granted ones. ``scope`` is kept for API compatibility only.
const { skills, isLoading, error } = useSkills({
search,
tagIds,
@@ -217,7 +225,9 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
if (isBuiltin || !!skill.published) {
all.push(skill);
}
if (isOwnedByMe) {
// "我的" = skills I own OR skills granted into my collection by my 岗位
// (position) — mirroring how favorited agents land in 我的智能体.
if (isOwnedByMe || !!skill.favorited) {
mine.push(skill);
}
}
@@ -601,6 +611,8 @@ export function SkillRow({
const { data: cardPublishSquares } = usePublishSquares();
const setPinned = useSetSkillPinned();
const setFeatured = useSetSkillFeatured();
const setAlwaysOn = useSetSkillAlwaysOn();
const { data: compression } = useSkillCompressionSettings();
const deleteSkill = useDeleteCustomSkill();
const duplicateSkillMutation = useDuplicateSkill();
const [confirmDeleteOpen, setConfirmDeleteOpen] = useState(false);
@@ -620,6 +632,17 @@ export function SkillRow({
// Admin can edit/delete/unpublish anyone's skill; non-admin only their own.
const canManage = isCustom && (isOwned || isAdmin);
const canTogglePublish = canManage && !isBuiltin;
// 常驻(免压缩):内置技能仅管理员可设;自定义技能其所有者或管理员可设(与后端一致)。
const canSetAlwaysOn = isAdmin || isOwned;
async function handleToggleAlwaysOn(next: boolean) {
try {
await setAlwaysOn.mutateAsync({ skillName: skill.name, alwaysOn: next });
toast.success(next ? "已设为常驻(免压缩)" : "已取消常驻");
} catch (err) {
toast.error(err instanceof Error ? err.message : String(err));
}
}
async function handleTogglePublished(next: boolean) {
// Publishing: prompt for which square to publish into (when configured).
@@ -932,6 +955,27 @@ export function SkillRow({
/>
</Button>
)}
{compression?.enabled && canSetAlwaysOn && (
<Button
size="icon"
variant="ghost"
className={cn(
"h-8 w-8 shrink-0",
skill.always_on && "text-amber-600 hover:text-amber-700",
)}
disabled={setAlwaysOn.isPending}
onClick={() => void handleToggleAlwaysOn(!skill.always_on)}
title={
skill.always_on
? "取消常驻(将被压缩,改由 search_skills 检索)"
: "设为常驻(免压缩,始终全文注入提示词)"
}
>
<Anchor
className={cn("size-3.5", skill.always_on && "fill-current")}
/>
</Button>
)}
{canTogglePublish && (
<Button
size="icon"
+1 -1
View File
@@ -24,7 +24,7 @@ export const POSITION_RESOURCE_LABELS: Record<PositionResourceType, string> = {
export const POSITION_RESOURCE_HINT: Record<PositionResourceType, string> = {
agent: "不配置标签 = 默认加载全部内置智能体;配置后只加载所选标签下的智能体。",
skill: "不配置标签 = 可见全部技能;配置后只可见所选标签下的技能。",
skill: "不配置标签 = 不向「我的技能」添加任何技能;配置后将所选标签下的技能加入成员的「我的技能」(技能广场仍展示全部技能)。",
scheduled_task: "不配置标签 = 不订阅任何定时任务;配置后订阅所选标签下的定时任务。",
recommended_question: "不配置标签 = 展示全部推荐问题;配置后只展示所选标签下的推荐问题(聊天框上方的推荐问题)。",
};
+3
View File
@@ -61,6 +61,9 @@ export interface Skill {
/** Publish square (发布广场) the skill is filed under once published. "" = default square. */
square_id?: string;
builtin?: boolean;
// True when this skill is in the caller's 我的技能 collection — e.g. granted by
// their 岗位 (position). Owned skills are detected separately via owner_user_id.
favorited?: boolean;
// Long-form detail/introduction authored by the skill owner or an admin.
// Distinct from ``description`` (parsed from SKILL.md); supports Markdown.
detail?: string | null;
+32 -14
View File
@@ -360,16 +360,24 @@ shown to users ARE the **recommended questions** (`recommended_questions` table)
extended with `recommended_question`).
**Sync engine** (`deerflow/services/position_sync.py`, wired as
`app.state.position_sync`). `PositionSyncService.resolve_visible(user, type)`
returns `set | None` (None = unrestricted) for the **query-time** visibility
types (`skill`, `recommended_question`). `sync_user(user)` **materializes** the
other two into the existing "我的" tables, idempotently:
`app.state.position_sync`). `sync_user(user)` **materializes** three resource
types into the existing "我的" tables, idempotently:
- agents → `agent_favorites` rows with `origin='position'`
- **skills → `skill_favorites` rows with `origin='position'`** (table
`SkillFavoriteRow`, migration `20260616_02`, auto-created by `create_all`)
- scheduled tasks → `scheduled_task_subscriptions` rows with `origin='position'`
Skills behave **exactly like agents**: a position only *adds* its tagged skills
to the member's 我的技能 — the browse-all 广场 is never restricted. (This replaced
the earlier query-time visibility filter, which wrongly shrank the square to the
bound skill.) `PositionSyncService.resolve_visible(user, type)` returns
`set | None` (None = unrestricted) and is now used **only** for the one remaining
query-time type, `recommended_question`.
Sync only ever adds/removes `origin='position'` rows, so a user's own
favorites/subscriptions are never clobbered. Default sets: agents = built-ins
(`user_id IS NULL`); scheduled tasks = empty; skills/questions = unrestricted.
favorites/subscriptions are never clobbered. Default sets (no binding for a
type): agents = built-ins (`user_id IS NULL`); skills = empty (square stays
full, nothing added to 我的); scheduled tasks = empty; questions = unrestricted.
`origin`/`position_id` columns were added to `agent_favorites` and
`scheduled_task_subscriptions` — see migration `20260616_01` +
`scripts/sql/column_additions_mysql.sql`. The running app also auto-adds these
@@ -390,8 +398,11 @@ unassigned users inherit, else `null`); powers the top-bar 岗位 label
(`strategy-components/Header.tsx` user cluster, via `useMyPosition`). Related
changes: `GET /api/recommended-questions` is now
filtered by the caller's position (`?all=1` bypasses for admin management) and
admin can tag questions via `/api/tags/assignments/recommended_question/{id}`;
`GET /api/skills?scope=mine` applies skill visibility. Frontend: `core/positions`
admin can tag questions via `/api/tags/assignments/recommended_question/{id}`.
`GET /api/skills` no longer restricts by position; each skill carries a
`favorited` flag (true = in the caller's 我的技能, e.g. position-granted) so the
skills page's 我的 tab includes owned ∪ position-granted skills while the 广场
stays complete. Frontend: `core/positions`
+ admin `PositionManagementPage` (侧边栏「综合管理 → 岗位管理」 + 右下角设置菜单).
Recommended questions stay rendered **above** the chat input (their original
position), just filtered by the caller's position. Tests:
@@ -499,11 +510,18 @@ prompt body and surfaces the rest on demand via keyword retrieval.
default 5, `keep_index_in_prompt` default true). Read/written by
`GET`/`PUT /api/system-settings/skill-compression` (GET open to any user — the
skill detail page reads it; PUT admin-only, refreshes the prompt cache).
- **Per-skill 常驻 flag** — `extensions_config.json → skills.{name}.always_on`
(`SkillStateConfig.always_on`, alongside `enabled`). Surfaced on
`Skill.always_on` (populated in `load_skills`) and `SkillResponse.always_on`;
toggled via `PUT /api/skills/{name}/always-on`. `update_skill` /
`update_skill_display` preserve it when rewriting the config entry.
- **Per-skill 常驻 flag** — stored in the **DB**: `SkillRow.always_on`
(`Boolean`, migration `20260616_03`, idempotent SQL in
`column_additions_mysql.sql`), **not** in `extensions_config.json` (the
source-tree config file is read-only / unsafe to rewrite in some deployments,
and built-in skills carry no file entry). Threaded through `SkillStore`
create/ensure_legacy/update/update_admin + `list_always_on_names()` exactly
like `name_zh`/`detail`; surfaced on `SkillResponse.always_on` from the
ownership record in `_skill_to_response`. Toggled via
`PUT /api/skills/{name}/always-on` (custom skill → owner or admin; built-in /
public → admin only, materializes a row on demand via `ensure_legacy`). The
prompt builder stamps `Skill.always_on` from `list_always_on_names()` via a
sync DB bridge (`_apply_always_on` in `lead_agent/prompt.py`).
- **Prompt split** — `get_skills_prompt_section` (`lead_agent/prompt.py`) reads
the compression settings (skipped for restricted custom agents — they already
have a curated allowlist). When **off**, behaviour is unchanged (all enabled
@@ -688,7 +706,7 @@ Both endpoints are whitelisted in `auth_middleware._PUBLIC_EXACT_PATHS`. The sha
**`extensions_config.json`**:
- `mcpServers` - Map of server name ? config (enabled, type, command, args, env, url, headers, oauth, description)
- `skills` - Map of skill name ? state (enabled, always_on)
- `skills` - Map of skill name ? state (enabled). 常驻(免压缩)/always_on lives in the `skills` DB table, not here.
Both can be modified at runtime via Gateway API endpoints or `DeerFlowClient` methods.
@@ -169,6 +169,7 @@ class SkillResponse(BaseModel):
published: bool = Field(default=False, description="Whether the owner has made this skill visible to other users")
square_id: str = Field(default="", description="Publish square (发布广场) the skill is filed under once published. Empty string = the default square.")
builtin: bool = Field(default=False, description="True when owner_user_id is null (no human owner)")
favorited: bool = Field(default=False, description="True when this skill is in the caller's 我的技能 collection (e.g. granted by their 岗位/position). Owned skills are detected separately via owner_user_id.")
detail: str | None = Field(default=None, description="Human-authored long-form detail text (Markdown), editable by owner/admin")
display: dict[str, Any] | None = Field(default=None, description="Optional frontend display adapter")
display_sample: dict[str, Any] | None = Field(default=None, alias="displaySample", description="Persisted sample output used by the display adapter editor")
@@ -280,6 +281,7 @@ def _skill_to_response(
*,
display_override: dict[str, Any] | None | object = _UNSET,
display_sample: dict[str, Any] | None = None,
favorited: bool = False,
) -> SkillResponse:
"""Convert a Skill object to a SkillResponse.
@@ -317,6 +319,9 @@ def _skill_to_response(
detail: str | None = None
name_zh: str | None = None
square_id = ""
# 常驻(免压缩) is stored in the DB (SkillRow.always_on), surfaced via the
# ownership record — never read from the on-disk extensions_config.json.
always_on = False
if ownership is not None:
owner_user_id = ownership.get("owner_user_id")
published = bool(ownership.get("published", False))
@@ -324,6 +329,7 @@ def _skill_to_response(
detail = ownership.get("detail")
name_zh = ownership.get("name_zh")
square_id = str(ownership.get("square_id") or "")
always_on = bool(ownership.get("always_on", False))
display = None if display_override is _UNSET else display_override
if display_override is _UNSET:
@@ -341,7 +347,7 @@ def _skill_to_response(
license=skill.license,
category=skill.category,
enabled=skill.enabled,
always_on=skill.always_on,
always_on=always_on,
use_count=entry.get("use_count", 0),
view_count=entry.get("view_count", 0),
patch_count=entry.get("patch_count", 0),
@@ -355,6 +361,7 @@ def _skill_to_response(
published=published,
square_id=square_id,
builtin=builtin,
favorited=favorited,
detail=detail,
display=display,
displaySample=display_sample,
@@ -810,6 +817,15 @@ async def list_skills(
ownership = {record["name"]: record for record in ownership_records}
else:
ownership = await _build_ownership_index(skill_store, user_id)
# Skills the caller's 岗位 (position) granted into their "我的技能"
# collection. Admins browse everything, so the badge is moot for them.
favorite_names: set[str] = set()
if not admin:
try:
favorite_names = await skill_store.list_favorite_skill_names(user_id)
except Exception:
logger.debug("skill favorites lookup failed", exc_info=True)
favorite_names = set()
display_adapters = await _list_skill_display_adapters([skill.name for skill in all_skills])
responses: list[SkillResponse] = []
for skill in all_skills:
@@ -825,22 +841,30 @@ async def list_skills(
ownership.get(skill.name),
display_override=_display_override_from_adapter(adapter),
display_sample=adapter.get("sample") if adapter else None,
favorited=skill.name in favorite_names,
)
)
continue
record = ownership.get(skill.name)
if record is None:
if not admin:
if admin:
# Admin sees every custom skill on disk even without an
# ownership row (legacy / orphaned), so they can clean it up.
record = None
elif skill.name in favorite_names:
# A position granted this skill even though it isn't
# otherwise visible to the user (e.g. an unpublished custom
# skill). Surface it so it lands in their 我的技能.
record = await skill_store.get_any(skill.name)
else:
continue
# Admin sees every custom skill on disk even without an
# ownership row (legacy / orphaned), so they can clean it up.
record = None
responses.append(
_skill_to_response(
skill,
record,
display_override=_display_override_from_adapter(adapter),
display_sample=adapter.get("sample") if adapter else None,
favorited=skill.name in favorite_names,
)
)
responses = await _attach_and_filter_tags(tag_store, responses, search, tag_ids)
@@ -855,19 +879,12 @@ async def list_skills(
for r in responses
if (r.square_id or "") == square_id or (square_is_default and not (r.square_id or ""))
]
# 岗位 (position) visibility: when the caller asks for "我的技能",
# restrict to the skills their position grants. ``None`` from the sync
# service = unrestricted (no skill tag binding → all skills visible).
if scope == "mine" and not admin:
sync = getattr(request.app.state, "position_sync", None)
if sync is not None:
try:
visible = await sync.resolve_visible(user_id, "skill")
except Exception:
logger.debug("skill visibility resolve failed", exc_info=True)
visible = None
if visible is not None:
responses = [r for r in responses if r.name in visible]
# 岗位 (position) no longer *restricts* the skill list: a position grants
# skills by materializing them into the member's 我的技能 (the ``favorited``
# flag above + the ``skill_favorites`` table, synced by PositionSyncService),
# mirroring how agents are added to 我的智能体. The browse-all 广场 therefore
# always shows every visible skill — the ``scope`` param is kept for API
# compatibility but intentionally does not filter here.
responses = _sort_skill_responses(responses, sort)
return SkillsListResponse(skills=responses)
except HTTPException:
@@ -2194,7 +2211,6 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest, config: App
current_skill_config = extensions_config.skills.get(skill_name)
extensions_config.skills[skill_name] = SkillStateConfig(
enabled=request.enabled,
always_on=current_skill_config.always_on if current_skill_config else skill.always_on,
display=current_skill_config.display if current_skill_config else None,
)
@@ -2229,41 +2245,50 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest, config: App
summary="Update Skill 常驻(免压缩) Flag",
description="Mark a skill as 常驻(免压缩) so it is injected full-text into the system prompt even when skill compression is enabled.",
)
async def update_skill_always_on(skill_name: str, request: SkillAlwaysOnUpdateRequest, config: AppConfig = Depends(get_config)) -> SkillResponse:
async def update_skill_always_on(
skill_name: str,
request: SkillAlwaysOnUpdateRequest,
http_request: Request,
config: AppConfig = Depends(get_config),
skill_store: SkillStore = Depends(get_skill_store),
) -> SkillResponse:
"""Persist the 常驻(免压缩) flag in the **DB** (skills table), not in
extensions_config.json — the source-tree config file is read-only / unsafe to
rewrite in some deployments, and built-in skills carry no file entry anyway.
Mirrors the detail/name_zh endpoints: custom skill → owner or admin; built-in
/ public skill → admin only (materializes a row on demand)."""
try:
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
user_id = _current_user_id(http_request)
is_admin = _is_admin_user(http_request)
skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
skill = next((s for s in skills if s.name == skill_name), None)
if skill is None:
raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
config_path = ExtensionsConfig.resolve_config_path()
if config_path is None:
config_path = Path.cwd().parent / "extensions_config.json"
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
if skill.category == SkillCategory.CUSTOM:
_record, mode = await _get_editable_skill_or_raise(skill_store, skill_name, user_id, is_admin=is_admin)
if mode == "admin":
updated = await skill_store.update_admin(skill_name, {"always_on": request.always_on})
else:
updated = await skill_store.update(skill_name, user_id, {"always_on": request.always_on})
else:
# Built-in / public skill — admin only; no DB row by default.
if not is_admin:
raise HTTPException(status_code=403, detail="Only admin users can change a built-in skill's 常驻 flag")
if await skill_store.get_any(skill_name) is None:
await skill_store.ensure_legacy({"name": skill_name, "owner_user_id": None, "published": True})
updated = await skill_store.update_admin(skill_name, {"always_on": request.always_on})
extensions_config = get_extensions_config()
current_skill_config = extensions_config.skills.get(skill_name)
extensions_config.skills[skill_name] = SkillStateConfig(
enabled=current_skill_config.enabled if current_skill_config else skill.enabled,
always_on=request.always_on,
display=current_skill_config.display if current_skill_config else None,
)
if updated is None:
raise HTTPException(status_code=500, detail="Failed to persist skill 常驻 flag")
with open(config_path, "w", encoding="utf-8") as f:
json.dump(_dump_extensions_config(extensions_config), f, indent=2)
reload_extensions_config()
# always_on lives in the system prompt's skill injection — refresh the cache.
await refresh_skills_system_prompt_cache_async()
skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
updated_skill = next((s for s in skills if s.name == skill_name), None)
if updated_skill is None:
raise HTTPException(status_code=500, detail=f"Failed to reload skill '{skill_name}' after update")
logger.info(f"Skill '{skill_name}' always_on updated to {request.always_on}")
return _skill_to_response(updated_skill)
logger.info("Skill '%s' always_on updated to %s", skill_name, request.always_on)
return _skill_to_response(skill, updated)
except HTTPException:
raise
@@ -2300,7 +2325,6 @@ async def update_skill_display(
current_skill_config = extensions_config.skills.get(skill_name)
extensions_config.skills[skill_name] = SkillStateConfig(
enabled=current_skill_config.enabled if current_skill_config else skill.enabled,
always_on=current_skill_config.always_on if current_skill_config else skill.always_on,
display=request.display,
)
@@ -86,8 +86,58 @@ def _drop_archived_skills(skills: list[Skill]) -> list[Skill]:
return skills
def _run_coro_blocking(coro):
"""Run an async coroutine to completion from sync code.
Mirrors ``skill_tools._run_coro_blocking``: when already inside a running
event loop (e.g. prompt built during an async agent run), hand the coroutine
to a fresh loop in a worker thread; otherwise run it directly.
"""
try:
asyncio.get_running_loop()
in_loop = True
except RuntimeError:
in_loop = False
if in_loop:
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(asyncio.run, coro).result()
return asyncio.run(coro)
def _load_always_on_names() -> set[str]:
"""Names marked 常驻(免压缩), read from the **DB** (skills table).
Returns an empty set on any failure / no-DB deployment, so compression then
simply treats every skill as compressible (safe default).
"""
try:
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.skills import make_skill_store
store = make_skill_store(get_session_factory())
return set(_run_coro_blocking(store.list_always_on_names()))
except Exception:
logger.debug("Failed to load always_on skill names from DB", exc_info=True)
return set()
def _apply_always_on(skills: list[Skill]) -> list[Skill]:
"""Stamp ``Skill.always_on`` from the DB-backed 常驻 set (in place)."""
try:
always_on = _load_always_on_names()
if always_on:
for skill in skills:
skill.always_on = skill.name in always_on
except Exception:
logger.debug("Failed to apply always_on flags to skills", exc_info=True)
return skills
def _load_enabled_skills_sync() -> list[Skill]:
return _drop_archived_skills(list(get_or_new_skill_storage().load_skills(enabled_only=True)))
return _apply_always_on(_drop_archived_skills(list(get_or_new_skill_storage().load_skills(enabled_only=True))))
def _start_enabled_skills_refresh_thread() -> None:
@@ -188,7 +238,7 @@ def _get_enabled_skills_for_config(app_config: AppConfig | None = None) -> list[
"""
if app_config is None:
return _get_enabled_skills()
return _drop_archived_skills(list(get_or_new_skill_storage(app_config=app_config).load_skills(enabled_only=True)))
return _apply_always_on(_drop_archived_skills(list(get_or_new_skill_storage(app_config=app_config).load_skills(enabled_only=True))))
def _skill_mutability_label(category: SkillCategory | str) -> str:
@@ -314,16 +314,36 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
logger.debug("summarization: thread_id unresolved; cross-turn summary cache disabled for this call")
cached = self._get_cache_entry(thread_id) if thread_id else None
present_ids = {m.id for m in messages}
if cached is not None and cached.reusable and cached.summarized_ids <= present_ids:
cache_usable = cached is not None and cached.reusable and cached.summarized_ids <= present_ids
if cache_usable:
preserved = [m for m in messages if m.id not in cached.summarized_ids]
effective = [self._summary_message(cached.summary_text), *preserved]
if not self._should_summarize(effective, self.token_counter(effective)):
# Decide re-summarization on the *counted* size of the compressed
# context only — NOT ``_should_summarize`` — because that folds in the
# last AIMessage's reported ``usage_metadata`` (system prompt + tool
# schemas, fixed overhead compression can't shrink). Using it here
# would re-summarize on every single turn once that overhead alone
# approaches the trigger.
if not self._exceeds_trigger_by_count(effective):
# Refresh LRU recency and reuse without an LLM call.
self._touch_cache_entry(thread_id)
return ("reuse", effective)
# New messages have pushed the compressed context back over the
# threshold — fall through and re-summarize a larger prefix.
# Genuine (blocking) LLM summarize path. Log *why* we are paying for it so
# an "it compacts every turn" report is diagnosable straight from the
# backend log: cache=miss → thread_id/key problem; unusable → message-id
# instability; stale → preserved genuinely grew back over the trigger.
cache_state = "stale" if cache_usable else ("miss" if cached is None else "unusable")
logger.info(
"summarization: compacting thread=%s cache=%s counted_tokens=%d msgs=%d",
thread_id,
cache_state,
self.token_counter(messages),
len(messages),
)
cutoff_index = self._determine_cutoff_index(messages)
if cutoff_index <= 0:
return None
@@ -386,6 +406,25 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
while len(self._summary_cache) > _SUMMARY_CACHE_MAX_THREADS:
self._summary_cache.popitem(last=False)
def _exceeds_trigger_by_count(self, messages: list[AnyMessage]) -> bool:
"""Trigger check using only the *counted* size of ``messages``.
Deliberately ignores the stock middleware's reported-``usage_metadata``
heuristic: that number includes fixed system-prompt + tool-schema
overhead which compression can never shrink, so relying on it to decide
*re*-summarization makes a long, tool-heavy thread re-compact on every
turn. The first-ever decision to compress still uses the full
``_should_summarize`` (reported usage is a fair signal there); only the
cache-reuse-vs-recompact decision uses this stricter, content-only check.
"""
total = self.token_counter(messages)
for kind, value in getattr(self, "_trigger_conditions", []):
if kind == "messages" and len(messages) >= value:
return True
if kind == "tokens" and total >= value:
return True
return False
def _apply_stickiness_headroom(self, messages: list[AnyMessage], cutoff_index: int) -> int:
"""Advance ``cutoff_index`` (keep fewer recent messages) so the preserved
suffix lands below every trigger threshold with headroom.
@@ -910,7 +910,6 @@ class DeerFlowClient:
current_skill_config = extensions_config.skills.get(name)
extensions_config.skills[name] = SkillStateConfig(
enabled=enabled,
always_on=current_skill_config.always_on if current_skill_config else False,
display=current_skill_config.display if current_skill_config else None,
)
@@ -161,10 +161,6 @@ class SkillStateConfig(BaseModel):
"""Configuration for a single skill's state."""
enabled: bool = Field(default=True, description="Whether this skill is enabled")
always_on: bool = Field(
default=False,
description="常驻(免压缩):开启技能压缩后,本技能仍以全文形式注入系统提示词,而非降级为索引+search_skills 检索",
)
display: SkillDisplayConfig | None = Field(default=None, description="Optional frontend display adapter")
@@ -317,21 +313,6 @@ class ExtensionsConfig(BaseModel):
return skill_category in ("public", "custom")
return skill_config.enabled
def is_skill_always_on(self, skill_name: str) -> bool:
"""Check if a skill is marked 常驻(免压缩).
Args:
skill_name: Name of the skill
Returns:
True if the skill should bypass skill compression, False otherwise
(default for skills without an explicit config entry).
"""
skill_config = self.skills.get(skill_name)
if skill_config is None:
return False
return skill_config.always_on
_extensions_config: ExtensionsConfig | None = None
@@ -0,0 +1,54 @@
"""我的技能 (skill favorites) table.
New table ``skill_favorites`` — the per-user "我的技能" collection, mirroring
``agent_favorites``. A 岗位 (position) materializes its skill tag bindings into
this table (``origin='position'``) so granted skills land in the member's 我的
list while the browse-all 广场 stays complete.
Note: the running app also auto-creates this table via ``create_all`` on
startup (new tables are always created, even on existing DBs); this migration
keeps the alembic history consistent per the repo convention (see
backend/CLAUDE.md → Schema Migrations).
Revision ID: 20260616_02
Revises: 20260616_01
Create Date: 2026-06-16
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260616_02"
down_revision = "20260616_01"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
return table_name in inspect(op.get_bind()).get_table_names()
def upgrade() -> None:
if not _has_table("skill_favorites"):
op.create_table(
"skill_favorites",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("user_id", sa.String(64), nullable=False),
sa.Column("skill_name", sa.String(64), nullable=False),
sa.Column("origin", sa.String(16), nullable=False, server_default="user"),
sa.Column("position_id", sa.String(64), nullable=True),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.UniqueConstraint("user_id", "skill_name", name="uq_skill_favorites_user_skill"),
)
op.create_index("ix_skill_favorites_user_id", "skill_favorites", ["user_id"])
op.create_index("ix_skill_favorites_skill_name", "skill_favorites", ["skill_name"])
op.create_index("ix_skill_favorites_position_id", "skill_favorites", ["position_id"])
def downgrade() -> None:
if _has_table("skill_favorites"):
op.drop_table("skill_favorites")
@@ -0,0 +1,45 @@
"""Add always_on (常驻/免压缩) column to skills table.
Revision ID: 20260616_03
Revises: 20260616_02
Create Date: 2026-06-16
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260616_03"
down_revision = "20260616_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 upgrade() -> None:
# ``skills`` may not exist yet on a brand-new DB — create_all handles that
# case with the column already present, so only patch existing tables.
if _has_table("skills") and not _has_column("skills", "always_on"):
op.add_column(
"skills",
sa.Column("always_on", sa.Boolean(), nullable=False, server_default="0"),
)
def downgrade() -> None:
if _has_column("skills", "always_on"):
op.drop_column("skills", "always_on")
@@ -60,7 +60,7 @@ from deerflow.persistence.scheduled_tasks.model import (
ScheduledTaskSubscriptionRow,
SchedulerThreadRow,
)
from deerflow.persistence.skills.model import SkillDisplayAdapterRow, SkillRow
from deerflow.persistence.skills.model import SkillDisplayAdapterRow, SkillFavoriteRow, SkillRow
from deerflow.persistence.tags.model import TagAssignmentRow, TagRow
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.embed_sessions.model import EmbedSessionRow
@@ -110,6 +110,7 @@ __all__ = [
"SchedulerThreadRow",
"SkillRow",
"SkillDisplayAdapterRow",
"SkillFavoriteRow",
"TagAssignmentRow",
"TagRow",
"EmbedSessionRow",
@@ -85,3 +85,23 @@ class SkillStore(ABC):
DB round-trip per candidate skill.
"""
raise NotImplementedError
@abstractmethod
async def list_always_on_names(self) -> set[str]:
"""Return the set of skill names marked 常驻(免压缩) (``always_on = True``).
Batch query used by the prompt builder to decide which skills bypass
skill compression — one query instead of a per-skill round-trip.
"""
raise NotImplementedError
@abstractmethod
async def list_favorite_skill_names(self, user_id: str) -> set[str]:
"""Return the skill names in ``user_id``'s "我的技能" collection.
These are skills the user does not own but that landed in their list —
currently only via their 岗位 (position) sync (``origin='position'``).
Owned skills are NOT included here (the caller detects those via
``owner_user_id``).
"""
raise NotImplementedError
@@ -20,12 +20,15 @@ def _copy(record: dict[str, Any] | None) -> dict[str, Any] | None:
def _with_derived(record: dict[str, Any]) -> dict[str, Any]:
out = dict(record)
out["builtin"] = out.get("owner_user_id") is None
out.setdefault("always_on", False)
return out
class MemorySkillStore(SkillStore):
def __init__(self) -> None:
self._skills: dict[str, dict[str, Any]] = {}
# user_id -> {skill_name}
self._favorites: dict[str, set[str]] = {}
@staticmethod
def _is_visible(record: dict[str, Any], user_id: str) -> bool:
@@ -71,6 +74,7 @@ class MemorySkillStore(SkillStore):
"published": bool(data.get("published", False)),
"detail": data.get("detail"),
"name_zh": data.get("name_zh"),
"always_on": bool(data.get("always_on", False)),
"square_id": str(data.get("square_id") or ""),
"created_at": _now(),
"updated_at": _now(),
@@ -107,6 +111,8 @@ class MemorySkillStore(SkillStore):
record["detail"] = data["detail"]
if "name_zh" in data:
record["name_zh"] = data["name_zh"]
if "always_on" in data:
record["always_on"] = bool(data["always_on"])
if "square_id" in data:
record["square_id"] = str(data["square_id"] or "")
record["updated_at"] = _now()
@@ -124,6 +130,8 @@ class MemorySkillStore(SkillStore):
record["detail"] = data["detail"]
if "name_zh" in data:
record["name_zh"] = data["name_zh"]
if "always_on" in data:
record["always_on"] = bool(data["always_on"])
if "square_id" in data:
record["square_id"] = str(data["square_id"] or "")
record["updated_at"] = _now()
@@ -153,3 +161,13 @@ class MemorySkillStore(SkillStore):
async def list_published_names(self) -> set[str]:
return {name for name, record in self._skills.items() if record.get("published")}
async def list_always_on_names(self) -> set[str]:
return {name for name, record in self._skills.items() if record.get("always_on")}
async def list_favorite_skill_names(self, user_id: str) -> set[str]:
return set(self._favorites.get(user_id, set()))
# test helper — not part of the abstract API
def add_favorite(self, user_id: str, skill_name: str) -> None:
self._favorites.setdefault(user_id, set()).add(skill_name)
@@ -10,10 +10,11 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Boolean, DateTime, String, Text
from sqlalchemy import Boolean, DateTime, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
class SkillRow(Base):
@@ -33,6 +34,10 @@ class SkillRow(Base):
# the on-disk SKILL.md) so it can be set for built-in skills too, and so the
# shared source file is never rewritten.
name_zh: Mapped[str | None] = mapped_column(String(128), nullable=True, default=None)
# 常驻(免压缩):开启技能压缩后,标记此项的技能仍以全文形式注入系统提示词,而非
# 降级为索引 + search_skills 检索。存 DB(而非 extensions_config.json),内置技能
# 也能标记,且避免在某些部署里回写源码目录下的配置文件导致报错。
always_on: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="0", index=True)
# Publish square (发布广场) this skill is filed under once published. Pure
# organizational category, no access control. Empty string = unassigned →
# the listing layer resolves it to the default square. One square per skill.
@@ -41,6 +46,32 @@ class SkillRow(Base):
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
class SkillFavoriteRow(Base):
"""A user's "我的技能" membership for a skill they don't own.
Mirrors :class:`deerflow.persistence.agents.model.AgentFavoriteRow`: lets a
skill land in a user's "我的" list without the user owning it. Skills are
keyed by ``name`` (the on-disk folder / DB primary key), so this row
references ``skill_name`` rather than an opaque id. One row per
(user, skill); the unique constraint makes the grant idempotent.
"""
__tablename__ = "skill_favorites"
__table_args__ = (UniqueConstraint("user_id", "skill_name", name="uq_skill_favorites_user_skill"),)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
skill_name: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
# ``"user"`` = the user added it themselves (reserved for a future favorite
# button); ``"position"`` = auto-granted by the user's 岗位 (position) via
# the subscription sync engine. Sync only ever adds/removes ``"position"``
# rows, so a user's own picks are never clobbered.
origin: Mapped[str] = mapped_column(String(16), nullable=False, default="user", server_default="user")
# The position that granted this favorite (only set when ``origin='position'``).
position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
class SkillDisplayAdapterRow(Base):
__tablename__ = "skill_display_adapters"
@@ -9,7 +9,7 @@ from sqlalchemy import func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.skills.base import SkillStore
from deerflow.persistence.skills.model import SkillRow
from deerflow.persistence.skills.model import SkillFavoriteRow, SkillRow
def _row_to_dict(row: SkillRow) -> dict[str, Any]:
@@ -65,6 +65,7 @@ class SkillRepository(SkillStore):
published=bool(data.get("published", False)),
detail=data.get("detail"),
name_zh=data.get("name_zh"),
always_on=bool(data.get("always_on", False)),
square_id=str(data.get("square_id") or ""),
created_at=now,
updated_at=now,
@@ -110,6 +111,8 @@ class SkillRepository(SkillStore):
row.detail = data["detail"]
if "name_zh" in data:
row.name_zh = data["name_zh"]
if "always_on" in data:
row.always_on = bool(data["always_on"])
if "square_id" in data:
row.square_id = str(data["square_id"] or "")
row.updated_at = datetime.now(UTC)
@@ -130,6 +133,8 @@ class SkillRepository(SkillStore):
row.detail = data["detail"]
if "name_zh" in data:
row.name_zh = data["name_zh"]
if "always_on" in data:
row.always_on = bool(data["always_on"])
if "square_id" in data:
row.square_id = str(data["square_id"] or "")
row.updated_at = datetime.now(UTC)
@@ -171,3 +176,15 @@ class SkillRepository(SkillStore):
async with self._sf() as session:
result = await session.execute(stmt)
return {row[0] for row in result.all()}
async def list_always_on_names(self) -> set[str]:
stmt = select(SkillRow.name).where(SkillRow.always_on.is_(True))
async with self._sf() as session:
result = await session.execute(stmt)
return {row[0] for row in result.all()}
async def list_favorite_skill_names(self, user_id: str) -> set[str]:
stmt = select(SkillFavoriteRow.skill_name).where(SkillFavoriteRow.user_id == user_id)
async with self._sf() as session:
result = await session.execute(stmt)
return {row[0] for row in result.all()}
@@ -1,9 +1,9 @@
"""岗位 (position) → resource subscription **sync engine**.
A user's position decides which agents / scheduled tasks / custom prompts they
are subscribed to, and which skills / recommended questions they can see. This
service turns a position's *tag bindings* into concrete, materialized rows in
the existing "我的" tables (idempotent, re-runnable).
A user's position decides which agents / skills / scheduled tasks they are
subscribed to, and which recommended questions they can see. This service turns
a position's *tag bindings* into concrete, materialized rows in the existing
"我的" tables (idempotent, re-runnable).
Resolution rule (per resource type):
@@ -14,15 +14,19 @@ Resolution rule (per resource type):
* For a resource type **without** bound tags → the *default set*:
- ``agent`` → all built-in agents (``user_id IS NULL``)
- ``scheduled_task`` → empty (no built-in tasks)
- ``skill`` / ``recommended_question`` → *unrestricted* (handled at query
time, never materialized — see :meth:`resolve_visible`)
- ``skill`` → empty (the 广场 still shows every visible skill;
a position only *adds* its tagged skills to the member's "我的技能")
- ``recommended_question`` → *unrestricted* (handled at query time, never
materialized — see :meth:`resolve_visible`)
Materialization (``sync_user``) only ever touches rows it created itself
(``origin='position'``), so a user's own favorites / subscriptions are never
clobbered. The skill + recommended-question visibility filters are computed live
by :meth:`resolve_visible` and are *not* materialized. (The admin-configured
prompts shown to users ARE the recommended questions — there is no separate
prompt-template store.)
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).
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.)
"""
from __future__ import annotations
@@ -37,14 +41,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.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.
_MATERIALIZED_TYPES = ("agent", "scheduled_task")
_MATERIALIZED_TYPES = ("agent", "skill", "scheduled_task")
# Resolved live at query time (visibility filters), never materialized.
_VISIBILITY_TYPES = ("skill", "recommended_question")
_VISIBILITY_TYPES = ("recommended_question",)
class PositionSyncService:
@@ -64,9 +69,10 @@ class PositionSyncService:
async def resolve_visible(self, user_id: str, resource_type: str) -> set[str] | None:
"""Return the set of visible ``resource_type`` ids, or ``None`` = unrestricted.
Used by the skills list + recommended-questions filters. ``None`` means
"no position restriction → show everything"; a set means "restrict to
exactly these ids".
Used by the recommended-questions filter (the only remaining query-time
visibility type; skills are now *materialized*, not filtered). ``None``
means "no position restriction → show everything"; a set means "restrict
to exactly these ids".
"""
pid = await self._effective_position_id(user_id)
if not pid:
@@ -101,6 +107,7 @@ class PositionSyncService:
pid = await self._effective_position_id(user_id)
async with self._sf() as session:
await self._sync_agents(session, user_id, pid)
await self._sync_skills(session, user_id, pid)
await self._sync_tasks(session, user_id, pid)
await session.commit()
@@ -116,6 +123,18 @@ class PositionSyncService:
if agent_id not in target:
await session.delete(row)
async def _sync_skills(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
target = await self._resolve_materialized(user_id, "skill", session)
rows = list((await session.execute(select(SkillFavoriteRow).where(SkillFavoriteRow.user_id == user_id))).scalars())
owned_any = {r.skill_name for r in rows}
position_rows = {r.skill_name: r for r in rows if r.origin == "position"}
now = datetime.now(UTC)
for skill_name in target - owned_any:
session.add(SkillFavoriteRow(id=uuid.uuid4().hex, user_id=user_id, skill_name=skill_name, origin="position", position_id=position_id, created_at=now))
for skill_name, row in position_rows.items():
if skill_name not in target:
await session.delete(row)
async def _sync_tasks(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
target = await self._resolve_materialized(user_id, "scheduled_task", session)
rows = list((await session.execute(select(ScheduledTaskSubscriptionRow).where(ScheduledTaskSubscriptionRow.user_id == user_id))).scalars())
@@ -247,7 +247,6 @@ class SkillStorage(ABC):
extensions_config = ExtensionsConfig.from_file()
for skill in skills:
skill.enabled = extensions_config.is_skill_enabled(skill.name, skill.category)
skill.always_on = extensions_config.is_skill_always_on(skill.name)
except Exception as e:
logger.warning("Failed to load extensions config: %s", e)
@@ -65,6 +65,9 @@ CALL _add_column_if_missing('skills', 'detail', 'TEXT NULL DEFAULT NULL');
-- square_id: 技能所属的发布广场(发布广场功能),对应 migration 20260611_01。
-- ORM: SkillRow.square_id = String(64), nullable=False, default=''。
CALL _add_column_if_missing('skills', 'square_id', "VARCHAR(64) NOT NULL DEFAULT ''");
-- always_on: 常驻(免压缩) —— 开启技能压缩后仍全文注入系统提示词,对应 migration 20260616_03。
-- ORM: SkillRow.always_on = Boolean, nullable=False, default=False。
CALL _add_column_if_missing('skills', 'always_on', 'TINYINT(1) NOT NULL DEFAULT 0');
-- ---- agents ------------------------------------------------------------------
-- square_id: 智能体所属的发布广场(发布广场功能),对应 migration 20260611_01。
@@ -133,7 +136,7 @@ DEALLOCATE PREPARE stmt;
SELECT table_name, column_name
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND ((table_name = 'skills' AND column_name IN ('name_zh', 'detail', 'square_id'))
AND ((table_name = 'skills' AND column_name IN ('name_zh', 'detail', 'square_id', 'always_on'))
OR (table_name = 'agents' AND column_name IN ('square_id'))
OR (table_name = 'roundtable_drafts' AND column_name = 'task_id'))
ORDER BY table_name, column_name;