fix(skills): curator 自我进化周期调度 + 并发/配置/安全修复

调度:
- app.py 新增常驻 curator 调度器,每小时检查一次"需不需要进化",
  替代原来"仅启动跑一次";多 worker 下用跨平台文件锁做 leader 选举,
  只有一个进程跑调度;启动延迟 5 分钟避开启动高峰

curator 逻辑修复:
- C1: should_run_now / apply_automatic_transitions / get_curator_status
  调用 _get_curator_config 时传 user_id,使 per-user 配置在运行时生效
- C3: usage.py 的 .usage.json 读-改-写加进程内锁,bump_* 改回调式,
  避免 curator 多线程并发丢状态更新
- C4a: curator 受限版 skill_manage 在工具层屏蔽 delete,
  堵住绕过 skill_archive 安全网删技能的口子
- C4b: 进化记录同时保存 LLM 自述(suggestions)与真实工具调用(actual_actions)
- 无候选技能时跳过 LLM 评审,省掉无意义的模型调用

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

offline-backend-20260512/backend/app/gateway/app.py

@@ -1,6 +1,7 @@
 import asyncio
 import logging
 import os
+import sys
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
@@ -268,6 +269,93 @@ async def _run_db_migrations() -> None:
         logger.exception("Database migration failed — schema may be out of date")
 
 
+# Skill curator 调度参数。
+# - 检查间隔:每隔这么久检查一次"需不需要进化";真正是否执行由 should_run_now() 判定。
+# - 启动延迟:启动后先等这么久再首次检查,避开启动高峰。
+_CURATOR_CHECK_INTERVAL_SECONDS = 3600  # 1 小时
+_CURATOR_STARTUP_DELAY_SECONDS = 300  # 5 分钟
+
+
+def _acquire_singleton_lock(lock_path):
+    """跨平台非阻塞独占文件锁,用于多 worker 下的 leader 选举。
+
+    成功 -> 返回持有锁的文件句柄(调用方须保持其存活,直到进程退出/主动释放);
+    抢不到锁(已有其它 worker 持有)或打开失败 -> 返回 None。
+    进程异常退出时 OS 会自动释放该锁。
+    """
+    try:
+        lock_path.parent.mkdir(parents=True, exist_ok=True)
+        handle = open(lock_path, "a+")
+    except OSError:
+        logger.exception("Curator scheduler: 打开锁文件失败")
+        return None
+    try:
+        if sys.platform == "win32":
+            import msvcrt
+
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except OSError:
+        handle.close()
+        return None
+    return handle
+
+
+def _release_singleton_lock(handle) -> None:
+    """释放并关闭 _acquire_singleton_lock 拿到的锁文件句柄。"""
+    try:
+        if sys.platform == "win32":
+            import msvcrt
+
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+        else:
+            import fcntl
+
+            fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+    except OSError:
+        pass
+    finally:
+        try:
+            handle.close()
+        except OSError:
+            pass
+
+
+async def _curator_scheduler_loop() -> None:
+    """常驻后台任务:每隔 _CURATOR_CHECK_INTERVAL_SECONDS 检查一次需不需要进化。
+
+    多 worker 部署下用文件锁做 leader 选举 —— 只有抢到锁的那个 worker 进程真正
+    运行调度,其余 worker 直接退出,避免 curator 被重复触发、重复调用 LLM。
+    被 cancel 时 CancelledError 会穿过 ``except Exception``(它是 BaseException
+    子类,不会被捕获),协程随之干净退出并在 finally 里释放锁。
+    """
+    from deerflow.config.runtime_paths import runtime_home
+    from deerflow.skills.curator import maybe_run_curator
+
+    lock_handle = _acquire_singleton_lock(runtime_home() / ".curator_scheduler.lock")
+    if lock_handle is None:
+        logger.info("Curator scheduler: 另一个 worker 已持有调度锁,本 worker 跳过")
+        return
+
+    logger.info("Curator scheduler: 取得调度锁,本 worker 负责 curator 调度")
+    try:
+        # 启动后延迟一会儿再首次检查,避开启动高峰
+        await asyncio.sleep(_CURATOR_STARTUP_DELAY_SECONDS)
+        while True:
+            try:
+                await maybe_run_curator()
+            except Exception:
+                logger.exception("Curator scheduler tick failed (non-fatal)")
+            await asyncio.sleep(_CURATOR_CHECK_INTERVAL_SECONDS)
+    finally:
+        _release_singleton_lock(lock_handle)
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan handler."""
@@ -302,12 +390,12 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             logger.exception("Skill ownership migration failed (non-fatal)")
 
-        # Kick off skill curator in background if the run interval has elapsed
+        # 启动 skill curator 调度器:常驻后台任务,每小时检查一次需不需要进化
+        # (per-user interval_hours 仍是真正决定要不要跑进化的闸)。
         try:
-            from deerflow.skills.curator import maybe_run_curator
+            app.state.curator_scheduler_task = asyncio.create_task(_curator_scheduler_loop())
-            asyncio.create_task(maybe_run_curator())
         except Exception:
-            logger.exception("Skill curator startup task failed (non-fatal)")
+            logger.exception("Skill curator scheduler failed to start (non-fatal)")
 
         # Start IM channel service if any channels are configured
         try:
@@ -335,6 +423,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
 
         yield
 
+        # 停止 curator 调度器
+        curator_task = getattr(app.state, "curator_scheduler_task", None)
+        if curator_task is not None:
+            curator_task.cancel()
+            try:
+                await curator_task
+            except asyncio.CancelledError:
+                pass
+            except Exception:
+                logger.exception("Curator scheduler shutdown error")
+
         try:
             scheduler_service = getattr(app.state, "scheduled_task_service", None)
             if scheduler_service is not None:

offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py

@@ -264,7 +264,7 @@ def _eligible_for_curation(
 def should_run_now(user_id: str) -> bool:
     """Return True if the curator interval has elapsed for ``user_id`` and it
     is not paused."""
-    curator_cfg = _get_curator_config()
+    curator_cfg = _get_curator_config(user_id)
     if curator_cfg is None or not getattr(curator_cfg, "enabled", True):
         return False
     state = _read_curator_state(user_id)
@@ -297,7 +297,7 @@ def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> di
     if now is None:
         now = datetime.now(UTC)
 
-    curator_cfg = _get_curator_config()
+    curator_cfg = _get_curator_config(user_id)
     stale_days = getattr(curator_cfg, "stale_after_days", 30) if curator_cfg else 30
     archive_days = getattr(curator_cfg, "archive_after_days", 90) if curator_cfg else 90
 
@@ -590,8 +590,72 @@ def _make_skill_archive_tool(user_id: str):
     return _skill_archive
 
 
-def _run_llm_review(candidate_list: str, *, user_id: str) -> str:
+def _make_curator_skill_manage_tool():
-    """Run the curator consolidation pass as a React agent with skill tools bound."""
+    """curator 受限版 ``skill_manage``:在工具层屏蔽 delete 动作。
+
+    curator 做 umbrella 合并需要 create / patch / edit / write_file,但绝不能删整个
+    技能 —— prompt 硬规则也写明"归档(skill_archive)是允许的最大操作"。通用
+    skill_manage 的 delete 走 ``_assert_agent_may_delete``,**绕过了** skill_archive
+    对"已发布 / 被 agent 引用"技能的拒绝保护。只靠 prompt 文字约束不可靠,这里在
+    工具层直接拒掉 delete,curator 的 LLM 无论如何都删不掉技能。
+    """
+    from langchain.tools import tool as lc_tool
+
+    from deerflow.tools.skill_manage_tool import _skill_manage_impl
+
+    @lc_tool("skill_manage")
+    def _curator_skill_manage(
+        action: str,
+        name: str,
+        content: str | None = None,
+        path: str | None = None,
+        find: str | None = None,
+        replace: str | None = None,
+        expected_count: int | None = None,
+    ) -> str:
+        """管理 skills/custom/ 下的自定义技能(curator 受限版)。
+
+        支持 action:create / patch / edit / write_file / remove_file / list / view。
+        **不支持 delete** —— curator 不得删除技能;如需下架请改用 skill_archive。
+
+        Args:
+            action: 操作类型(见上;delete 会被拒绝)。
+            name: 连字符命名的技能名。
+            content: create / edit / write_file 的新文件内容。
+            path: write_file / remove_file 的支持文件路径。
+            find: patch 要替换的原文本。
+            replace: patch 的替换文本。
+            expected_count: patch 的预期替换次数(可选)。
+        """
+        if action == "delete":
+            return (
+                "Refused: curator 不允许删除技能。归档是允许的最大操作 —— "
+                "请改用 skill_archive(name, reason) 软下架(文件保留,用户可在 Web UI 恢复)。"
+            )
+        # curator 不绑定到具体会话线程,runtime 传 None(_get_thread_id 兼容 None)。
+        return _run_async_blocking(
+            _skill_manage_impl(
+                runtime=None,
+                action=action,
+                name=name,
+                content=content,
+                path=path,
+                find=find,
+                replace=replace,
+                expected_count=expected_count,
+            )
+        )
+
+    return _curator_skill_manage
+
+
+def _run_llm_review(candidate_list: str, *, user_id: str) -> tuple[str, list[dict]]:
+    """Run the curator consolidation pass as a React agent with skill tools bound.
+
+    返回 ``(llm_text, tool_actions)``:
+    - ``llm_text``    —— LLM 最终回复文字(它"自述"做了什么);
+    - ``tool_actions`` —— 从消息流提取的**真实**工具调用记录(它"实际"做了什么)。
+    """
     from deerflow.skills import curator_log
 
     try:
@@ -601,10 +665,10 @@ def _run_llm_review(candidate_list: str, *, user_id: str) -> str:
         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)]
+        # 用 curator 受限版 skill_manage(屏蔽 delete),不用通用版 —— 见 _make_curator_skill_manage_tool。
+        tools = [skill_list_tool, skill_view_tool, _make_curator_skill_manage_tool(), _make_skill_archive_tool(user_id)]
         # 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
@@ -618,25 +682,26 @@ def _run_llm_review(candidate_list: str, *, user_id: str) -> str:
         result = agent.invoke({"messages": [HumanMessage(content=prompt)]})
 
         messages = result.get("messages", [])
+        tool_actions = _extract_tool_actions(messages)
         for msg in reversed(messages):
             if not getattr(msg, "tool_calls", None):
                 content = getattr(msg, "content", "")
                 if isinstance(content, str) and content.strip():
                     curator_log.record("info", "LLM 评审:Agent 已返回最终结论", user_id=user_id)
-                    return content
+                    return content, tool_actions
                 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 评审:Agent 已返回最终结论", user_id=user_id)
-                    return text
+                    return text, tool_actions
         curator_log.record("warning", "LLM 评审:本次未产出最终回复", user_id=user_id)
-        return ""
+        return "", tool_actions
     except Exception as e:
         logger.warning("Curator LLM review failed: %s", e)
         curator_log.record("warning", f"LLM 评审失败:{e}", user_id=user_id)
-        return ""
+        return "", []
 
 
 def _parse_structured_summary(text: str) -> dict:
@@ -653,6 +718,44 @@ def _parse_structured_summary(text: str) -> dict:
         return {"consolidations": [], "prunings": []}
 
 
+def _extract_tool_actions(messages: list) -> list[dict]:
+    """从 agent 消息流里提取**真实发生的**工具调用 + 结果。
+
+    与 _parse_structured_summary(解析 LLM 自述的 YAML)互补:这里记录的是 curator
+    实际调了哪些工具、入参是什么、成功还是失败 —— 用于审计"嘴上说的"与"实际做的"
+    是否一致。每条形如 ``{tool, args, status, result}``。
+    """
+    # 先按 tool_call_id 收集每次调用的工具名与入参(来自 AIMessage.tool_calls)
+    calls: dict[str, dict] = {}
+    for msg in messages:
+        for tc in getattr(msg, "tool_calls", None) or []:
+            tc_id = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None)
+            if not tc_id:
+                continue
+            calls[tc_id] = {
+                "tool": tc.get("name") if isinstance(tc, dict) else getattr(tc, "name", None),
+                "args": (tc.get("args") if isinstance(tc, dict) else getattr(tc, "args", None)) or {},
+            }
+    # 再按 ToolMessage(带 tool_call_id)收集真实结果
+    actions: list[dict] = []
+    for msg in messages:
+        tc_id = getattr(msg, "tool_call_id", None)
+        if not tc_id:
+            continue
+        call = calls.get(tc_id, {})
+        raw = getattr(msg, "content", "")
+        result = raw if isinstance(raw, str) else str(raw)
+        actions.append(
+            {
+                "tool": call.get("tool") or getattr(msg, "name", None),
+                "args": call.get("args", {}),
+                "status": getattr(msg, "status", None) or "success",
+                "result": result[:500],
+            }
+        )
+    return actions
+
+
 def run_curator_review(user_id: str) -> dict:
     """Main curator run for ``user_id``: transitions → LLM consolidation → state.
 
@@ -684,16 +787,23 @@ def run_curator_review(user_id: str) -> dict:
         "info", f"进入 LLM 评审的候选技能:共 {candidate_line_count} 个", user_id=user_id
     )
 
-    llm_text = _run_llm_review(candidate_list, user_id=user_id)
+    if candidate_line_count == 0:
-    suggestions = _parse_structured_summary(llm_text)
+        # 没有候选技能 —— 跳过 LLM 评审,省掉无意义的模型调用。
-    curator_log.record(
+        # 年龄流转(上面的 apply_automatic_transitions)已完成,且不依赖 LLM。
-        "info",
+        curator_log.record("info", "无候选技能,跳过 LLM 评审", user_id=user_id)
-        (
+        suggestions = {"consolidations": [], "prunings": []}
-            f"LLM 建议已解析:建议合并 {len(suggestions['consolidations'])} 项,"
+        tool_actions: list[dict] = []
-            f"建议删减 {len(suggestions['prunings'])} 项"
+    else:
-        ),
+        llm_text, tool_actions = _run_llm_review(candidate_list, user_id=user_id)
-        user_id=user_id,
+        suggestions = _parse_structured_summary(llm_text)
-    )
+        curator_log.record(
+            "info",
+            (
+                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 []:
@@ -718,15 +828,26 @@ def run_curator_review(user_id: str) -> dict:
             user_id=user_id,
         )
 
+    # 上面是 LLM"自述"的建议;下面记录它"实际"调用了哪些工具,供对照审计。
+    curator_log.record("info", f"LLM 实际调用工具 {len(tool_actions)} 次", user_id=user_id)
+    for act in tool_actions:
+        target = act.get("args", {}).get("name", "")
+        curator_log.record(
+            "info",
+            f"  实际调用 [{act.get('tool')}]{f' [{target}]' if target else ''} → {act.get('status')}",
+            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"自动流转:标记陈旧 {transition_counters['marked_stale']} 个、"
         f"归档 {transition_counters['archived']} 个;"
-        f"LLM:建议合并 {len(suggestions['consolidations'])} 项"
+        f"LLM:建议合并 {len(suggestions['consolidations'])} 项、实际工具调用 {len(tool_actions)} 次"
     )
-    state["last_consolidation_suggestions"] = suggestions
+    state["last_consolidation_suggestions"] = suggestions  # LLM 自述
+    state["last_actual_actions"] = tool_actions  # 真实工具调用记录
     _write_curator_state(state, user_id)
 
     logger.info("Curator[%s]: run complete — %s", user_id, state["last_run_summary"])
@@ -768,10 +889,11 @@ def _curator_thread_target(user_id: str) -> None:
 
 
 async def maybe_run_curator() -> None:
-    """Async startup entry — kicks off a thread per user whose interval elapsed.
+    """周期调度入口 —— 给每个 interval 已到期的用户起一个后台策展线程。
 
-    Skips users with no DB-tracked entries (a fresh deployment) and never
+    由 app 的 curator 调度器周期性调用(也用于手动触发)。无候选技能的用户,其
-    blocks lifespan startup on failure.
+    LLM 评审会在 run_curator_review 内被跳过 —— 只做不耗费 LLM 的年龄流转。
+    任何失败都不会阻断调用方。
     """
     try:
         from sqlalchemy import select
@@ -819,7 +941,7 @@ def get_curator_status(user_id: str) -> dict:
         s = entry.get("state", STATE_ACTIVE)
         counts[s] = counts.get(s, 0) + 1
 
-    cfg = _get_curator_config()
+    cfg = _get_curator_config(user_id)
     return {
         "enabled": cfg is not None and getattr(cfg, "enabled", True),
         "run_count": state.get("run_count", 0),

offline-backend-20260512/backend/packages/harness/deerflow/skills/usage.py

@@ -6,6 +6,7 @@ import json
 import logging
 import os
 import tempfile
+import threading
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -13,6 +14,11 @@ logger = logging.getLogger(__name__)
 
 _USAGE_FILENAME = ".usage.json"
 
+# 串行化对 .usage.json 的"读-改-写",避免并发(尤其 curator 多线程)丢更新。
+# .usage.json 是全局单文件;此锁覆盖进程内并发。leader 选举已使 curator 只在
+# 单进程跑,故进程内锁即可保证状态流转不丢。
+_usage_lock = threading.Lock()
+
 STATE_ACTIVE = "active"
 STATE_STALE = "stale"
 STATE_ARCHIVED = "archived"
@@ -75,13 +81,20 @@ def _now_iso() -> str:
     return datetime.now(UTC).isoformat()
 
 
-def _update_entry(name: str, updates: dict) -> None:
+def _update_entry(name: str, updates) -> None:
+    """更新一个 usage 条目。``updates`` 可以是 dict,或 ``(entry) -> dict`` 回调。
+
+    用回调形式可让"读当前值 -> 算增量"也落在锁内(例如计数自增),
+    避免并发下的 lost update。
+    """
     try:
-        data = _read_all()
+        with _usage_lock:
-        entry = {**_DEFAULT_ENTRY, **data.get(name, {})}
+            data = _read_all()
-        entry.update(updates)
+            entry = {**_DEFAULT_ENTRY, **data.get(name, {})}
-        data[name] = entry
+            resolved = updates(entry) if callable(updates) else updates
-        _write_all(data)
+            entry.update(resolved)
+            data[name] = entry
+            _write_all(data)
     except Exception:
         logger.debug("Failed to update usage entry for %r", name, exc_info=True)
 
@@ -103,18 +116,16 @@ def agent_created_entries() -> dict:
 
 
 def bump_view(name: str) -> None:
-    entry = get_entry(name)
+    # 用回调:读当前计数 + 自增在同一把锁内完成,避免并发丢计数。
-    _update_entry(name, {"view_count": entry.get("view_count", 0) + 1, "last_viewed_at": _now_iso()})
+    _update_entry(name, lambda e: {"view_count": e.get("view_count", 0) + 1, "last_viewed_at": _now_iso()})
 
 
 def bump_use(name: str) -> None:
-    entry = get_entry(name)
+    _update_entry(name, lambda e: {"use_count": e.get("use_count", 0) + 1, "last_used_at": _now_iso()})
-    _update_entry(name, {"use_count": entry.get("use_count", 0) + 1, "last_used_at": _now_iso()})
 
 
 def bump_patch(name: str) -> None:
-    entry = get_entry(name)
+    _update_entry(name, lambda e: {"patch_count": e.get("patch_count", 0) + 1, "last_patched_at": _now_iso()})
-    _update_entry(name, {"patch_count": entry.get("patch_count", 0) + 1, "last_patched_at": _now_iso()})
 
 
 def set_state(name: str, state: str) -> None:
@@ -132,9 +143,10 @@ def set_pinned(name: str, pinned: bool) -> None:
 
 def forget(name: str) -> None:
     try:
-        data = _read_all()
+        with _usage_lock:
-        data.pop(name, None)
+            data = _read_all()
-        _write_all(data)
+            data.pop(name, None)
+            _write_all(data)
     except Exception:
         logger.debug("Failed to forget usage entry for %r", name, exc_info=True)