feat(skills): 自我进化 - 软归档/恢复 + Agent 归属 + 状态分层 UI

后端:
- 新增 PUT /api/skills/custom/{name}/state 与 /pinned 软归档/恢复 + 钉选 API
- skill_manage 工具: agent 创建写 DB 归属; 删除清理 .usage.json + DB row + cascade 引用; 加 ownership 预校验
- curator 全部软归档(文件保留在 custom/), 加 reason 必填(中文), 修 user_context 跨线程传播 / 协程泄漏 / archive 文件不再移动
- LLM 评审接 ToolErrorHandlingMiddleware, 工具异常自动重试不中止本轮; 替换 create_react_agent 为 create_agent
- skill_list 按当前 agent config.yaml 的 skills 字段过滤(custom agent), lead agent 仍可见全部
- 自我进化提示词对齐 Hermes 风格: 主动询问保存, reason 强制中文且面向用户
- 控制台日志增强: 自动归档/标过时/LLM 合并/删减建议 都按条目输出原因

前端:
- 技能管理页 Tab 改「全部 / 我的」, 正确包含他人发布的 custom 技能(修复 isCustom 互斥 bug)
- 新增「过时 N / 归档 N」右侧抽屉(恢复/查看/立即归档 ⋯ 菜单)
- 新增徽章: 钉选 / 来源(Agent 创建 / 用户上传) / 使用次数展示
- 内置技能徽章修复(原 isBuiltin && isCustom 永远 false)
- 自我进化日志 run_started sentinel 在前端展示为中文分隔条
- 移除冗余 enabled 开关(归档已覆盖此能力, 减少术语)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
xuhdandClaude Opus 4.7 committed 2026-05-15 00:00:52 +08:00
1 parent f8bc74b27f
commit 62d525dae7
16 files changed
+1270 -97

No files matched your search

@@ -641,6 +641,96 @@ async def update_custom_skill_published(
raise HTTPException(status_code=500, detail=f"Failed to update published state: {str(e)}")
_ALLOWED_USER_STATES = {"active", "archived"}
class SkillStateRequest(BaseModel):
"""Request body for switching a custom skill's lifecycle state."""
state: str = Field(..., description="New state: 'active' (restore) or 'archived' (soft archive)")
class SkillPinRequest(BaseModel):
"""Request body for pinning / unpinning a custom skill."""
pinned: bool = Field(..., description="Whether to pin (true) so the curator never auto-archives this skill")
def _reload_skill_response(skill_name: str, ownership: dict[str, Any], config: AppConfig) -> SkillResponse:
"""Reload a single skill from disk and return its API response form."""
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 on disk")
return _skill_to_response(skill, ownership)
@router.put(
"/skills/custom/{skill_name}/state",
response_model=SkillResponse,
summary="Update Custom Skill Lifecycle State",
description="Restore (active) or soft-archive a custom skill. Only the owner or admins may invoke.",
)
async def update_custom_skill_state(
skill_name: str,
body: SkillStateRequest,
request: Request,
config: AppConfig = Depends(get_config),
skill_store: SkillStore = Depends(get_skill_store),
) -> SkillResponse:
try:
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
if body.state not in _ALLOWED_USER_STATES:
raise HTTPException(
status_code=400,
detail=f"state must be one of {sorted(_ALLOWED_USER_STATES)}",
)
user_id = _current_user_id(request)
is_admin = _is_admin_user(request)
record, _mode = await _get_editable_skill_or_raise(skill_store, skill_name, user_id, is_admin=is_admin)
from deerflow.skills.usage import set_state as _usage_set_state
_usage_set_state(skill_name, body.state)
await refresh_skills_system_prompt_cache_async()
return _reload_skill_response(skill_name, record, config)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to update state for %s: %s", skill_name, e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to update state: {e}")
@router.put(
"/skills/custom/{skill_name}/pinned",
response_model=SkillResponse,
summary="Pin or Unpin a Custom Skill",
description="Pin a skill so the curator never auto-archives it. Only the owner or admins may invoke.",
)
async def update_custom_skill_pinned(
skill_name: str,
body: SkillPinRequest,
request: Request,
config: AppConfig = Depends(get_config),
skill_store: SkillStore = Depends(get_skill_store),
) -> SkillResponse:
try:
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
user_id = _current_user_id(request)
is_admin = _is_admin_user(request)
record, _mode = await _get_editable_skill_or_raise(skill_store, skill_name, user_id, is_admin=is_admin)
from deerflow.skills.usage import set_pinned as _usage_set_pinned
_usage_set_pinned(skill_name, body.pinned)
return _reload_skill_response(skill_name, record, config)
except HTTPException:
raise
except Exception as e:
logger.error("Failed to update pinned for %s: %s", skill_name, e, exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to update pinned: {e}")
@router.get("/skills/custom/{skill_name}/history", response_model=CustomSkillHistoryResponse, summary="Get Custom Skill History")
async def get_custom_skill_history(skill_name: str, config: AppConfig = Depends(get_config)) -> CustomSkillHistoryResponse:
try:
@@ -191,14 +191,23 @@ def _build_skill_evolution_section(skill_evolution_enabled: bool) -> str:
return ""
return """
## Skill Self-Evolution
After completing a task, consider creating or updating a skill when:
- The task required 5+ tool calls to resolve
- You overcame non-obvious errors or pitfalls
- The user corrected your approach and the corrected version worked
- You discovered a non-trivial, recurring workflow
If you used a skill and encountered issues not covered by it, patch it immediately.
Prefer patch over edit. Before creating a new skill, confirm with the user first.
Skip simple one-off tasks.
You have access to `skill_manage` (plus `skill_list` / `skill_view`).
After completing a non-trivial task — 5+ tool calls to resolve, recovery from
a non-obvious error, a corrected approach that worked, or a recurring
workflow — **offer to save the approach as a reusable skill** so you can
replay it next time. Briefly propose a hyphen-case name and a one-line
description, then ask the user to confirm before calling
`skill_manage(action="create", ...)`. Capture the generalised steps, not the
user's specific data, paths, or credentials.
When using an existing skill and finding it outdated, incomplete, or wrong,
**patch it immediately** with `skill_manage(action="patch")` — don't wait to
be asked. Skills that aren't maintained become liabilities.
Skip the offer for one-off queries, casual chat, or short single-tool
answers — and never create a skill silently without user confirmation.
"""
@@ -9,19 +9,70 @@ memory.json and threads/ dirs).
from __future__ import annotations
import asyncio
import concurrent.futures
import json
import logging
import os
import re
import tempfile
import threading
from collections.abc import Coroutine
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any, TypeVar
logger = logging.getLogger(__name__)
_CURATOR_STATE_FILENAME = ".curator_state"
_T = TypeVar("_T")
def _run_async_blocking(coro: Coroutine[Any, Any, _T]) -> _T:
"""Run an async coroutine to completion from sync curator code.
Curator is invoked from a background thread, but ``agent.invoke()`` inside
``_run_llm_review`` (LangChain 1.x ``create_agent``) spins up its own
asyncio loop which can leave the calling thread in an "already has a
running loop" state. In that case raw ``asyncio.run(coro)`` raises
``RuntimeError`` *after* the coroutine object has been materialised but
before anyone awaits it — producing the ``coroutine was never awaited``
RuntimeWarning that pollutes our logs.
This helper detects a running loop in the caller's thread and offloads
to a dedicated worker thread; otherwise it falls back to ``asyncio.run``.
Either way it guarantees the coroutine is closed on the failure path,
so no RuntimeWarning ever escapes.
"""
try:
asyncio.get_running_loop()
in_loop = True
except RuntimeError:
in_loop = False
if in_loop:
try:
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(asyncio.run, coro).result()
except BaseException:
# ``asyncio.run`` ran in the worker (it consumes the coroutine on
# success or surfaces the inner exception); but if the executor
# itself failed to submit, close to be safe.
try:
coro.close()
except Exception:
pass
raise
try:
return asyncio.run(coro)
except BaseException:
try:
coro.close()
except Exception:
pass
raise
def _get_skills_root() -> Path:
from deerflow.skills.storage import get_or_new_skill_storage
@@ -102,7 +153,7 @@ def _list_user_owned_skill_names(user_id: str) -> set[str] | None:
async def _run() -> list[dict]:
return await store.list_visible(user_id)
records = asyncio.run(_run())
records = _run_async_blocking(_run())
return {r["name"] for r in records if r.get("owner_user_id") == user_id}
except Exception:
logger.debug("_list_user_owned_skill_names failed", exc_info=True)
@@ -173,7 +224,7 @@ def _is_skill_published(skill_name: str) -> bool:
async def _run() -> dict | None:
return await store.get_any(skill_name)
record = asyncio.run(_run())
record = _run_async_blocking(_run())
return bool(record and record.get("published", False))
except Exception:
logger.debug("_is_skill_published(%s) failed", skill_name, exc_info=True)
@@ -276,6 +327,8 @@ def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> di
"skipped_published": skipped_published,
}
from deerflow.skills import curator_log
for name, entry in eligible.items():
if entry.get("pinned"):
continue
@@ -295,9 +348,19 @@ def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> di
if days_idle >= archive_days and current_state != STATE_ARCHIVED:
set_state(name, STATE_ARCHIVED)
counters["archived"] += 1
curator_log.record(
"info",
f" 自动归档 [{name}]:已闲置 {days_idle} 天(阈值 {archive_days} 天)",
user_id=user_id,
)
elif days_idle >= stale_days and current_state == STATE_ACTIVE:
set_state(name, STATE_STALE)
counters["marked_stale"] += 1
curator_log.record(
"info",
f" 标记过时 [{name}]:已闲置 {days_idle} 天(阈值 {stale_days} 天)",
user_id=user_id,
)
return counters
@@ -351,9 +414,19 @@ _CURATOR_REVIEW_PROMPT = (
"Hard rules — do not violate:\n"
"1. DO NOT touch public (built-in) skills. The candidate list below "
"is already filtered to agent-created custom skills only.\n"
"2. DO NOT delete any skill. Archiving (moving the skill's directory "
"into the .archive/ folder under the skills root) is the maximum "
"destructive action. Archives are recoverable; deletion is not.\n"
"2. DO NOT delete any skill. Archiving (a soft state flip via the "
"skill_archive tool — the file stays in custom/<name>/ and the user "
"can restore it from the Web UI) is the maximum action you may take. "
"Archives are recoverable; deletion is not. "
"When you call skill_archive(name, reason) you MUST pass a `reason` "
"string written **in Chinese**, in plain language a human operator "
"would read — NOT English jargon or terse debug notes. Examples of "
"acceptable reasons: '已合并到 code-formatter,该技能同时覆盖 Python 与 JS', "
"'与 foo-bar 内容重复,保留 foo-bar', '重构后过时,被 api-v2-search 替代', "
"'一次性 POC 实验,无复用价值'. Bad reasons (do NOT use): 'merged', "
"'duplicate', 'obsolete', 'consolidated'. The reason is shown in the "
"Web UI run-history for user audit, so write it as you would explain "
"it to the user themselves.\n"
"3. DO NOT touch skills shown as pinned=yes. Skip them entirely.\n"
"4. DO NOT use usage counters as a reason to skip consolidation. The "
"counters are new and often mostly zero. Judge overlap on CONTENT, "
@@ -425,8 +498,8 @@ _CURATOR_REVIEW_PROMPT = (
" - name: <skill-name>\n"
" reason: <one short sentence — why archived with no merge target>\n"
"```\n\n"
"Every skill you moved to .archive/ MUST appear in exactly one of the "
"two lists. If you consolidated X into umbrella Y (patched Y, wrote "
"Every skill you archived via skill_archive MUST appear in exactly one "
"of the two lists. If you consolidated X into umbrella Y (patched Y, wrote "
"a references file to Y, or created Y with X's content absorbed), X "
"goes under `consolidations` with `into: Y`. If you archived X with "
"no absorption — truly stale, irrelevant, or obsolete — X goes under "
@@ -437,28 +510,48 @@ _CURATOR_REVIEW_PROMPT = (
def _make_skill_archive_tool(user_id: str):
"""Curator-only tool: move a skill directory to .archive/ (recoverable)."""
"""Curator-only tool: soft-archive a skill (sets state, files stay put)."""
from langchain.tools import tool as lc_tool
from deerflow.skills import curator_log
from deerflow.skills.storage import get_or_new_skill_storage
@lc_tool("skill_archive")
def _skill_archive(name: str) -> str:
"""Move an agent-created skill to the .archive/ directory (recoverable). Use this instead of delete.
def _skill_archive(name: str, reason: str) -> str:
"""Soft-archive an agent-created skill (sets state=archived; files stay in custom/).
The user can recover the skill from the Web UI's archive drawer.
This is intentionally non-destructive — the on-disk SKILL.md and any
supporting files are preserved so nothing is lost.
Args:
name: The skill name to archive.
reason: 必填。用**中文**写一句话,清晰说明为什么归档这个技能 —
直接给读它的运维人员看,不要写英文术语 / 调试信息。例如:
- "已合并到 code-formatter,该技能同时覆盖 Python 与 JS"
- "重构后过时,被 api-v2-search 替代"
- "与现有 foo-bar 技能重复,内容完全一致"
- "一次性 POC 实验,无复用价值"
- "命名过窄(含具体 issue 号),已并入通用伞技能 X"
此原因会展示在策展日志和 Web UI 运行历史中,供用户审计。
"""
from deerflow.skills.usage import STATE_ARCHIVED, get_entry, set_state
# Defensive: tolerate empty/None reason but log a clear placeholder
# rather than crashing the curator run.
reason_text = (reason or "").strip() or "未提供原因"
storage = get_or_new_skill_storage()
if not storage.custom_skill_exists(name):
curator_log.record("warning", f"skill_archive: '{name}' not found or already archived", user_id=user_id)
curator_log.record("warning", f"归档:未找到技能 [{name}],或它已被归档", user_id=user_id)
return f"Skill '{name}' not found or already archived."
if get_entry(name).get("pinned"):
curator_log.record("info", f"skill_archive: '{name}' is pinned — skipping", user_id=user_id)
entry = get_entry(name)
if entry.get("pinned"):
curator_log.record("info", f"归档:技能 [{name}] 已被钉选,跳过", user_id=user_id)
return f"Skill '{name}' is pinned — skipping."
if entry.get("state") == STATE_ARCHIVED:
curator_log.record("info", f"归档:技能 [{name}] 已是归档状态,跳过", user_id=user_id)
return f"Skill '{name}' is already archived."
# Safety net: even if the LLM hallucinated a name not in the
# candidate list, refuse to archive anything that is published or
@@ -466,7 +559,7 @@ def _make_skill_archive_tool(user_id: str):
if _is_skill_published(name):
curator_log.record(
"warning",
f"skill_archive: '{name}' is published — refused (curator must not touch published skills)",
f"归档:已拒绝处理 [{name}],该技能已发布(策展不允许操作已发布的技能)",
user_id=user_id,
)
return f"Refused: skill '{name}' is published; manual unpublish is required first."
@@ -474,24 +567,25 @@ def _make_skill_archive_tool(user_id: str):
if name in referenced:
curator_log.record(
"warning",
f"skill_archive: '{name}' is referenced by an agent's skills config — refused",
f"归档:已拒绝处理 [{name}],该技能正被某个 Agent 引用",
user_id=user_id,
)
return f"Refused: skill '{name}' is still referenced by one or more agents."
skill_dir = storage.get_custom_skill_dir(name)
archive_root = Path(storage.get_skills_root_path()) / "custom" / ".archive"
archive_root.mkdir(parents=True, exist_ok=True)
stamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S")
dest = archive_root / f"{name}-{stamp}"
# Soft archive: flip the lifecycle state. The skill directory stays in
# ``custom/<name>/`` untouched so the user can recover it from the
# Web UI's archive drawer (which lists skills by state, not by location).
try:
skill_dir.rename(dest)
set_state(name, STATE_ARCHIVED)
except Exception as e:
curator_log.record("error", f"skill_archive: archive failed for '{name}': {e}", user_id=user_id)
curator_log.record("error", f"归档失败:技能 [{name}] — {e}", user_id=user_id)
return f"Archive failed: {e}"
set_state(name, STATE_ARCHIVED)
curator_log.record("info", f"Archived '{name}' → custom/.archive/{dest.name}", user_id=user_id)
return f"Archived '{name}' → custom/.archive/{dest.name}."
curator_log.record(
"info",
f"已归档(软):[{name}] — 原因:{reason_text} — 文件保留在 custom/{name}/",
user_id=user_id,
)
return f"Archived '{name}' (reason: {reason_text}; files preserved in custom/{name}/)."
return _skill_archive
@@ -501,18 +595,25 @@ def _run_llm_review(candidate_list: str, *, user_id: str) -> str:
from deerflow.skills import curator_log
try:
from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from langgraph.prebuilt import create_react_agent
from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
from deerflow.models import create_chat_model
from deerflow.tools.builtins.skill_tools import skill_list_tool, skill_view_tool
from deerflow.tools.skill_manage_tool import skill_manage_tool
model = create_chat_model()
tools = [skill_list_tool, skill_view_tool, skill_manage_tool, _make_skill_archive_tool(user_id)]
agent = create_react_agent(model, tools)
# Reuse the lead-agent error middleware: when the LLM mis-calls a tool
# (wrong action / missing required args), the exception is converted
# into a ToolMessage(status="error") so the LLM can read the error and
# retry, instead of aborting the entire review pass.
# NOTE: ``create_react_agent`` (langgraph.prebuilt) does NOT accept a
# middleware kwarg — use LangChain 1.x's ``create_agent`` which does.
agent = create_agent(model=model, tools=tools, middleware=[ToolErrorHandlingMiddleware()])
curator_log.record("info", "LLM review: invoking React agent", user_id=user_id)
curator_log.record("info", "LLM 评审:开始调用 React Agent", user_id=user_id)
prompt = _CURATOR_REVIEW_PROMPT + f"\n\nSkill candidates for this pass:\n{candidate_list}"
result = agent.invoke({"messages": [HumanMessage(content=prompt)]})
@@ -521,20 +622,20 @@ def _run_llm_review(candidate_list: str, *, user_id: str) -> str:
if not getattr(msg, "tool_calls", None):
content = getattr(msg, "content", "")
if isinstance(content, str) and content.strip():
curator_log.record("info", "LLM review: agent returned final answer", user_id=user_id)
curator_log.record("info", "LLM 评审:Agent 已返回最终结论", user_id=user_id)
return content
if isinstance(content, list):
text = " ".join(
block.get("text", "") if isinstance(block, dict) else str(block)
for block in content
).strip()
curator_log.record("info", "LLM review: agent returned final answer", user_id=user_id)
curator_log.record("info", "LLM 评审:Agent 已返回最终结论", user_id=user_id)
return text
curator_log.record("warning", "LLM review: no final assistant message produced", user_id=user_id)
curator_log.record("warning", "LLM 评审:本次未产出最终回复", user_id=user_id)
return ""
except Exception as e:
logger.warning("Curator LLM review failed: %s", e)
curator_log.record("warning", f"LLM review failed: {e}", user_id=user_id)
curator_log.record("warning", f"LLM 评审失败:{e}", user_id=user_id)
return ""
@@ -563,16 +664,16 @@ def run_curator_review(user_id: str) -> dict:
now = datetime.now(UTC)
logger.info("Curator[%s]: starting run at %s", user_id, now.isoformat())
curator_log.record("info", f"Curator starting run at {now.isoformat()}", user_id=user_id)
curator_log.record("info", f"本次策展开始,启动时间 {now.isoformat()}", user_id=user_id)
transition_counters = apply_automatic_transitions(user_id, now)
logger.info("Curator[%s]: transitions=%s", user_id, transition_counters)
curator_log.record(
"info",
(
f"Age-based transitions: checked={transition_counters['checked']}, "
f"marked_stale={transition_counters['marked_stale']}, "
f"archived={transition_counters['archived']}"
f"按年龄自动流转完成:已检查 {transition_counters['checked']} 个,"
f"标记陈旧 {transition_counters['marked_stale']} 个,"
f"自动归档 {transition_counters['archived']} 个"
),
user_id=user_id,
)
@@ -580,7 +681,7 @@ def run_curator_review(user_id: str) -> dict:
candidate_list = _render_candidate_list(user_id)
candidate_line_count = 0 if not candidate_list.strip() else len([line for line in candidate_list.splitlines() if line.startswith("- ")])
curator_log.record(
"info", f"Candidates for LLM review: {candidate_line_count} skill(s)", user_id=user_id
"info", f"进入 LLM 评审的候选技能:共 {candidate_line_count} 个", user_id=user_id
)
llm_text = _run_llm_review(candidate_list, user_id=user_id)
@@ -588,36 +689,82 @@ def run_curator_review(user_id: str) -> dict:
curator_log.record(
"info",
(
f"LLM suggestions parsed: consolidations={len(suggestions['consolidations'])}, "
f"prunings={len(suggestions['prunings'])}"
f"LLM 建议已解析:建议合并 {len(suggestions['consolidations'])} 项,"
f"建议删减 {len(suggestions['prunings'])} 项"
),
user_id=user_id,
)
# Surface per-item reasoning so operators can audit *why* each action
# was taken, not just the count.
for item in suggestions.get("consolidations", []) or []:
if not isinstance(item, dict):
continue
name = item.get("from", "?")
into = item.get("into", "?")
reason = (item.get("reason") or "无说明").strip()
curator_log.record(
"info",
f" 合并建议 [{name}] → [{into}]:{reason}",
user_id=user_id,
)
for item in suggestions.get("prunings", []) or []:
if not isinstance(item, dict):
continue
name = item.get("name", "?")
reason = (item.get("reason") or "无说明").strip()
curator_log.record(
"info",
f" 删减建议 [{name}]:{reason}",
user_id=user_id,
)
state = _read_curator_state(user_id)
state["last_run_at"] = now.isoformat()
state["run_count"] = state.get("run_count", 0) + 1
state["last_run_summary"] = (
f"auto: {transition_counters['marked_stale']} marked stale, "
f"{transition_counters['archived']} archived; "
f"llm: {len(suggestions['consolidations'])} consolidations suggested"
f"自动流转:标记陈旧 {transition_counters['marked_stale']} 个、"
f"归档 {transition_counters['archived']} 个;"
f"LLM:建议合并 {len(suggestions['consolidations'])} 项"
)
state["last_consolidation_suggestions"] = suggestions
_write_curator_state(state, user_id)
logger.info("Curator[%s]: run complete — %s", user_id, state["last_run_summary"])
curator_log.record("info", f"Curator run complete — {state['last_run_summary']}", user_id=user_id)
curator_log.record("info", f"本次策展已完成 — {state['last_run_summary']}", user_id=user_id)
return {"transitions": transition_counters, "suggestions": suggestions}
class _CuratorUser:
"""Minimal ``CurrentUser`` protocol implementation for background runs.
Curator threads are spawned via ``threading.Thread`` (not an asyncio task),
so the ``deerflow.runtime.user_context`` ContextVar starts unset. Without
a stand-in, any tool the LLM review invokes (e.g. ``skill_manage(create)``)
would call ``get_effective_user_id()`` and silently fall back to the
``DEFAULT_USER_ID`` sentinel — causing newly-created skills to be owned
by ``"default"`` instead of the user the curator is running for.
"""
__slots__ = ("id",)
def __init__(self, user_id: str) -> None:
self.id = user_id
def _curator_thread_target(user_id: str) -> None:
from deerflow.runtime.user_context import reset_current_user, set_current_user
from deerflow.skills import curator_log
# Bind the per-thread user context for the entire curator run so every
# downstream tool / repository call resolves the correct user_id.
token = set_current_user(_CuratorUser(user_id))
try:
run_curator_review(user_id)
except Exception as exc:
logger.exception("Curator[%s] background run failed", user_id)
curator_log.record("error", f"Curator background run failed: {exc}", user_id=user_id)
curator_log.record("error", f"策展后台任务失败:{exc}", user_id=user_id)
finally:
reset_current_user(token)
async def maybe_run_curator() -> None:
@@ -5,11 +5,60 @@ from __future__ import annotations
import json
import logging
from langchain.tools import tool
from langchain.tools import ToolRuntime, tool
logger = logging.getLogger(__name__)
def _resolve_active_agent_id(runtime: ToolRuntime | None) -> str | None:
"""Extract the agent_id under which the current tool call is running.
Custom agents pass their id via either the ``context`` dict or the
``configurable`` section of the LangGraph runtime config. The default
lead agent runs without an explicit agent_id — callers should treat
a None return as "no per-agent skills filter applies".
"""
if runtime is None:
return None
try:
if getattr(runtime, "context", None):
ctx_id = runtime.context.get("agent_id")
if ctx_id:
return str(ctx_id)
cfg = getattr(runtime, "config", None)
if isinstance(cfg, dict):
cid = cfg.get("configurable", {}).get("agent_id")
if cid:
return str(cid)
except Exception:
return None
return None
def _load_agent_skill_whitelist(agent_id: str | None) -> set[str] | None:
"""Return the explicit skill name set declared in an agent's config.yaml.
Semantics:
- ``None`` returned → no agent_id, or config missing — caller should
fall back to "all skills visible".
- ``set()`` (empty) → agent has ``skills: []`` declared — visibility
is intentionally empty.
- Non-empty set → only these names are visible to this agent.
"""
if not agent_id:
return None
try:
from deerflow.config.agents_config import load_agent_config
cfg = load_agent_config(agent_id)
if cfg is None or cfg.skills is None:
return None
return set(cfg.skills)
except Exception:
logger.debug("Failed to load agent config for skill_list filter", exc_info=True)
return None
@tool("skill_view")
def skill_view_tool(name: str) -> str:
"""Read the full SKILL.md content of a skill by name.
@@ -48,21 +97,36 @@ def skill_view_tool(name: str) -> str:
@tool("skill_list")
def skill_list_tool() -> str:
"""List all available skills with metadata and usage stats. Does NOT read SKILL.md content.
def skill_list_tool(runtime: ToolRuntime) -> str:
"""List skills available to this agent, with metadata and usage stats.
Does NOT read SKILL.md content.
Returns a JSON array with fields: name, description, category, enabled, state,
use_count, view_count, patch_count, last_activity_at, pinned, source.
For a **custom agent**, the result is filtered to skills explicitly
declared in the agent's ``.deer-flow/agents/{agent_id}/config.yaml``
under the ``skills:`` field — the agent only sees what it was wired to
use.
For the **default lead agent** (no agent_id in runtime), there is no
per-agent whitelist and every visible skill is returned.
Returns a JSON array with fields: name, description, category, enabled,
state, use_count, view_count, patch_count, last_activity_at, pinned,
source.
"""
from deerflow.skills.storage import get_or_new_skill_storage
from deerflow.skills.usage import all_entries, latest_activity_at
agent_id = _resolve_active_agent_id(runtime)
allowed = _load_agent_skill_whitelist(agent_id)
try:
storage = get_or_new_skill_storage()
skills = storage.load_skills(enabled_only=False)
usage = all_entries()
result = []
for skill in skills:
if allowed is not None and skill.name not in allowed:
continue
entry = usage.get(skill.name, {})
result.append({
"name": skill.name,
@@ -13,6 +13,8 @@ from langgraph.typing import ContextT
from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
from deerflow.agents.thread_state import ThreadState
from deerflow.mcp.tools import _make_sync_tool_wrapper
from deerflow.persistence.skills import SkillStore
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.skills.security_scanner import scan_skill_content
from deerflow.skills.storage import get_or_new_skill_storage
from deerflow.skills.storage.skill_storage import SkillStorage
@@ -64,6 +66,85 @@ async def _to_thread(func, /, *args, **kwargs):
return await asyncio.to_thread(func, *args, **kwargs)
async def _ensure_skill_db_registered(store: SkillStore, name: str, user_id: str) -> None:
"""Idempotently record an agent-created skill's ownership in the DB.
Mirrors the ``/api/skills/install-upload`` path so agent-created skills
show up in the Web UI listing (which filters custom skills by DB
ownership). ``ensure_legacy`` is idempotent — it never overwrites an
existing owner — so re-running the create flow on an already-owned skill
is a no-op. Failures are swallowed so DB outages can never break the
agent's create flow itself.
"""
try:
await store.ensure_legacy(
{"name": name, "owner_user_id": user_id, "published": False}
)
except Exception:
logger.debug("Failed to register skill ownership for %r", name, exc_info=True)
def _default_skill_store() -> SkillStore | None:
"""Build a ``SkillStore`` backed by the global session factory.
Returns ``None`` if no DB session factory is configured (e.g. ephemeral
test environments), letting the caller skip the registration step
silently.
"""
try:
from deerflow.persistence.engine import get_session_factory
from deerflow.persistence.skills import make_skill_store
factory = get_session_factory()
if factory is None:
return None
return make_skill_store(factory)
except Exception:
logger.debug("Failed to build default skill store", exc_info=True)
return None
def _default_agent_store() -> Any | None:
"""Build an ``AgentStore`` backed by the global session factory."""
try:
from deerflow.persistence.agents import make_agent_store
from deerflow.persistence.engine import get_session_factory
factory = get_session_factory()
if factory is None:
return None
return make_agent_store(factory)
except Exception:
logger.debug("Failed to build default agent store", exc_info=True)
return None
async def _assert_agent_may_delete(store: SkillStore, name: str, user_id: str) -> dict[str, Any]:
"""Pre-flight ownership check before agent-delete touches the on-disk file.
LocalSkillStorage doesn't know about users — it deletes anything that
matches by name. Without this guard, a malicious or hallucinating agent
could destroy another user's skill just by knowing the name. We refuse
unless the DB confirms ``user_id`` owns the skill outright.
Returns the DB record for use in the subsequent cascade step. Raises
``ValueError`` (mapped to a tool error visible to the LLM) on refusal.
"""
record = await store.get_any(name)
if record is None:
# No DB row — could be a legacy/orphan skill. Refuse rather than
# delete blindly; admin can clean up via the Web UI.
raise ValueError(
f"Refused: skill '{name}' has no ownership record. Ask a human to delete it via the Web UI."
)
owner = record.get("owner_user_id")
if owner is None:
raise ValueError(f"Refused: skill '{name}' is built-in and cannot be deleted by an agent.")
if owner != user_id:
raise ValueError(f"Refused: skill '{name}' is owned by another user.")
return record
async def _skill_manage_impl(
runtime: ToolRuntime[ContextT, ThreadState],
action: str,
@@ -124,6 +205,13 @@ async def _skill_manage_impl(
await _to_thread(mark_agent_created, name)
except Exception:
pass
# Register the new skill in the DB so the web UI listing actually
# surfaces it (custom skills without a DB row are filtered out by
# ``GET /api/skills``). Resolve the user_id *here* on the main
# async stack — contextvars don't propagate into ``to_thread``.
store = _default_skill_store()
if store is not None:
await _ensure_skill_db_registered(store, name, get_effective_user_id())
await refresh_skills_system_prompt_cache_async()
return f"Created custom skill '{name}'."
@@ -169,6 +257,15 @@ async def _skill_manage_impl(
return f"Patched custom skill '{name}' ({replacement_count} replacement(s) applied, {occurrences} match(es) found)."
if action == "delete":
# Ownership pre-check: LocalSkillStorage deletes purely by name,
# so without this an agent could blow away another user's skill
# just by knowing the name. Raise BEFORE touching the filesystem.
user_id = get_effective_user_id()
store = _default_skill_store()
record: dict[str, Any] | None = None
if store is not None:
record = await _assert_agent_may_delete(store, name, user_id)
await _to_thread(
skill_storage.delete_custom_skill,
name,
@@ -181,6 +278,36 @@ async def _skill_manage_impl(
scanner={"decision": "allow", "reason": "Deletion requested."},
),
)
# Mirror what ``DELETE /api/skills/custom/{name}`` does: drop the
# ``.usage.json`` entry and the DB ownership row, then cascade
# the removal to every agent that referenced this skill so
# ``config.yaml`` doesn't end up with a dangling reference.
# Failures are best-effort — the on-disk delete already happened
# and is authoritative.
try:
from deerflow.skills.usage import forget
await _to_thread(forget, name)
except Exception:
logger.debug("Failed to forget usage entry for %r", name, exc_info=True)
if store is not None:
try:
await store.delete(name, user_id)
except Exception:
logger.debug("Failed to delete DB row for %r", name, exc_info=True)
agent_store = _default_agent_store()
if agent_store is not None:
try:
from deerflow.skills.cascade import cascade_skill_unpublished_or_deleted
await cascade_skill_unpublished_or_deleted(
name,
actor_user_id=user_id,
operation="delete",
skill_owner_user_id=(record.get("owner_user_id") if record else user_id),
agent_store=agent_store,
)
except Exception:
logger.debug("Cascade failed for deleted skill %r", name, exc_info=True)
await refresh_skills_system_prompt_cache_async()
return f"Deleted custom skill '{name}'."