From 3e6e3c08d5713b1241738417b90ffbad3e43f6e8 Mon Sep 17 00:00:00 2001 From: xuhd Date: Thu, 14 May 2026 18:49:22 +0800 Subject: [PATCH] =?UTF-8?q?feat(skills):=20curator=20=E4=B8=89=E6=9D=A1?= =?UTF-8?q?=E4=BB=B6=E8=BF=87=E6=BB=A4=20=E2=80=94=20agent=20=E5=88=9B?= =?UTF-8?q?=E5=BB=BA=20+=20=E6=9C=AA=E5=8F=91=E5=B8=83=20+=20=E6=97=A0?= =?UTF-8?q?=E5=BC=95=E7=94=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Curator 之前只过滤 source=='agent'，导致两类 skill 仍会被自动归档： 1. owner 已 published 的 skill（user B 在用，结果被 curator 收走） 2. 加进 agent.skills 但 .usage.json 活动不够频繁的 skill 新规则：curator transitions 和 LLM 合并候选**只**作用于同时满足 三条件的 skill —— source=='agent' + published==False + 不被任何 agent.config.yaml 的 skills 字段引用。 curator.py - 新 helper _collect_referenced_skill_names()：扫所有 .deer-flow/agents/*/config.yaml 把 skills 字段聚合成 set，一次 curator run 内只调一次 - 新 helper _is_skill_published(name)：通过 SkillStore 查 published 状态，失败时返回 False（保守降级到旧行为） - 新 helper _eligible_for_curation(name, entry, owned_set, referenced_set) 三条件合一判定 - apply_automatic_transitions 应用三条件过滤；返回的 counters 加 skipped_referenced / skipped_published 计数 - _render_candidate_list 应用同一过滤，LLM 看不到 protected skill - _make_skill_archive_tool safety net：即使 LLM 编造名字调 archive， 工具内部再做一次"被引用 + published"检查，违规拒绝并记 warning 测试 tests/test_skills_multitenancy.py 新增 5 用例： - test_curator_skips_skill_referenced_by_any_agent - test_curator_skips_published_skill - test_curator_archives_orphan_unpublished_skill - test_skill_archive_tool_refuses_referenced_or_published - test_collect_referenced_skill_names_aggregates_from_disk 跑测试：tests/test_skills_multitenancy.py 17/17，全套 36/36 全过。 Co-Authored-By: Claude Opus 4.7 (1M context) --- .../harness/deerflow/skills/curator.py | 182 ++++++++++++++++-- .../backend/tests/test_skills_multitenancy.py | 168 ++++++++++++++++ 2 files changed, 336 insertions(+), 14 deletions(-) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py index 08491e0e..1ab81cac 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/curator.py @@ -109,6 +109,107 @@ def _list_user_owned_skill_names(user_id: str) -> set[str] | None: return None +def _collect_referenced_skill_names() -> set[str]: + """Scan every agent's config.yaml and return the union of their ``skills`` + fields. + + Used as the "do not touch" set during curator transitions and LLM + candidate selection — any skill currently bound to *any* agent + (including the owner's own) is exempt from automatic stale/archive + and from being suggested for merging. + + Returns an empty set on filesystem errors (best-effort, non-fatal). + """ + import yaml + + from deerflow.config.paths import get_paths + + referenced: set[str] = set() + try: + agents_dir = get_paths().agents_dir + except Exception: + return referenced + if not agents_dir.exists(): + return referenced + + for entry in agents_dir.iterdir(): + if not entry.is_dir(): + continue + config_file = entry / "config.yaml" + if not config_file.exists(): + continue + try: + with config_file.open("r", encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + except Exception: + logger.debug("_collect_referenced_skill_names: failed to parse %s", config_file, exc_info=True) + continue + skills = data.get("skills") + if isinstance(skills, list): + for s in skills: + if isinstance(s, str): + referenced.add(s) + return referenced + + +def _is_skill_published(skill_name: str) -> bool: + """Return True if ``skill_name`` is marked ``published=True`` in SkillStore. + + Best-effort — returns False on any error so a transient DB hiccup never + causes a published skill to be wrongly archived (the caller treats + "published=True" as exempt, so False = "fail open to processing", which + only puts a skill *back* into curator scope when DB is unavailable; in + that degraded mode curator legacy behaviour applies). + """ + 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 False + store = make_skill_store(factory) + + async def _run() -> dict | None: + return await store.get_any(skill_name) + + record = asyncio.run(_run()) + return bool(record and record.get("published", False)) + except Exception: + logger.debug("_is_skill_published(%s) failed", skill_name, exc_info=True) + return False + + +def _eligible_for_curation( + name: str, + entry: dict, + *, + owned_set: set[str] | None, + referenced_set: set[str], +) -> bool: + """Three-condition gate for whether curator may touch a skill. + + Curator only acts on skills that ALL of: + 1. source == "agent" (caller is expected to have already filtered, but + we double-check here for safety) + 2. NOT referenced by any agent's skills config + 3. NOT published (queried via SkillStore) + + ``owned_set`` is the set of skill names owned by the running user. When + None, the legacy "process every agent-created skill" fallback applies + for ownership only (the other two conditions still hold). + """ + if entry.get("source") != "agent": + return False + if owned_set is not None and name not in owned_set: + return False + if name in referenced_set: + return False + if _is_skill_published(name): + return False + return True + + def should_run_now(user_id: str) -> bool: """Return True if the curator interval has elapsed for ``user_id`` and it is not paused.""" @@ -132,8 +233,13 @@ def should_run_now(user_id: str) -> bool: def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> dict: """Pure age-based state transitions — no LLM involved. - Only processes agent-created skills owned by ``user_id``. Skips pinned - skills. Returns counters: {marked_stale, archived, checked}. + Only processes agent-created skills that are ALL of: + - owned by ``user_id`` + - not referenced by any agent's ``skills`` config + - not currently published + + Skips pinned skills. Returns counters: {marked_stale, archived, checked, + skipped_referenced, skipped_published}. """ from deerflow.skills.usage import STATE_ACTIVE, STATE_ARCHIVED, STATE_STALE, agent_created_entries, latest_activity_at, set_state @@ -145,12 +251,32 @@ def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> di archive_days = getattr(curator_cfg, "archive_after_days", 90) if curator_cfg else 90 owned = _list_user_owned_skill_names(user_id) - entries = agent_created_entries() - if owned is not None: - entries = {name: entry for name, entry in entries.items() if name in owned} - counters = {"marked_stale": 0, "archived": 0, "checked": len(entries)} + referenced = _collect_referenced_skill_names() + raw_entries = agent_created_entries() - for name, entry in entries.items(): + skipped_referenced = 0 + skipped_published = 0 + eligible: dict[str, dict] = {} + for name, entry in raw_entries.items(): + if owned is not None and name not in owned: + continue + if name in referenced: + skipped_referenced += 1 + continue + if _is_skill_published(name): + skipped_published += 1 + continue + eligible[name] = entry + + counters = { + "marked_stale": 0, + "archived": 0, + "checked": len(eligible), + "skipped_referenced": skipped_referenced, + "skipped_published": skipped_published, + } + + for name, entry in eligible.items(): if entry.get("pinned"): continue last_activity_str = latest_activity_at(name) @@ -177,16 +303,24 @@ def apply_automatic_transitions(user_id: str, now: datetime | None = None) -> di def _render_candidate_list(user_id: str) -> str: + """List of skill candidates handed to the LLM for merge/archive review. + + Same three-condition gate as ``apply_automatic_transitions``: only + agent-created, not-referenced-by-any-agent, not-published skills owned + by ``user_id`` show up here. The LLM literally cannot see (and thus + cannot suggest action on) anything outside that set. + """ from deerflow.skills.usage import STATE_ARCHIVED, all_entries owned = _list_user_owned_skill_names(user_id) - entries = { - k: v - for k, v in all_entries().items() - if v.get("source") == "agent" and v.get("state") != STATE_ARCHIVED - } - if owned is not None: - entries = {k: v for k, v in entries.items() if k in owned} + referenced = _collect_referenced_skill_names() + entries: dict[str, dict] = {} + for k, v in all_entries().items(): + if not _eligible_for_curation(k, v, owned_set=owned, referenced_set=referenced): + continue + if v.get("state") == STATE_ARCHIVED: + continue + entries[k] = v if not entries: return "No agent-created active skills found." lines = [] @@ -325,6 +459,26 @@ def _make_skill_archive_tool(user_id: str): if get_entry(name).get("pinned"): curator_log.record("info", f"skill_archive: '{name}' is pinned — skipping", user_id=user_id) return f"Skill '{name}' is pinned — skipping." + + # Safety net: even if the LLM hallucinated a name not in the + # candidate list, refuse to archive anything that is published or + # currently referenced by an agent. + if _is_skill_published(name): + curator_log.record( + "warning", + f"skill_archive: '{name}' is published — refused (curator must not touch published skills)", + user_id=user_id, + ) + return f"Refused: skill '{name}' is published; manual unpublish is required first." + referenced = _collect_referenced_skill_names() + if name in referenced: + curator_log.record( + "warning", + f"skill_archive: '{name}' is referenced by an agent's skills config — refused", + 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) diff --git a/offline-backend-20260512/backend/tests/test_skills_multitenancy.py b/offline-backend-20260512/backend/tests/test_skills_multitenancy.py index 26c18741..dbd5cb1a 100644 --- a/offline-backend-20260512/backend/tests/test_skills_multitenancy.py +++ b/offline-backend-20260512/backend/tests/test_skills_multitenancy.py @@ -430,3 +430,171 @@ def test_curator_log_buckets_records_per_user() -> None: assert curator_log.get_recent(user_id="user-c") == [] curator_log.clear_all() + + +# --------------------------------------------------------------------------- +# Increment #1: Curator three-condition eligibility filter +# --------------------------------------------------------------------------- + + +def _seed_old_agent_skill(name: str) -> None: + """Helper: stamp an agent-created skill as 90 days idle (archive-eligible).""" + from datetime import UTC, datetime, timedelta + + from deerflow.skills import usage + + old_iso = (datetime.now(UTC) - timedelta(days=90)).isoformat() + usage._update_entry( + name, + { + "source": usage.SOURCE_AGENT, + "created_at": old_iso, + "last_used_at": old_iso, + "state": usage.STATE_ACTIVE, + }, + ) + + +def _isolated_skills_root(isolated_paths, monkeypatch) -> None: + """Helper: point the on-disk skills root at the temp directory.""" + from deerflow.skills.storage import get_or_new_skill_storage, reset_skill_storage + + reset_skill_storage() + monkeypatch.setattr( + "deerflow.skills.storage._default_skill_storage", + get_or_new_skill_storage(skills_path=str(isolated_paths / "skills")), + raising=False, + ) + + +def test_curator_skips_skill_referenced_by_any_agent( + isolated_paths, monkeypatch +) -> None: + """A skill listed in any agent's config.yaml ``skills`` field must be + exempt from automatic archive, even if it has been idle for ages.""" + from datetime import UTC, datetime + + from deerflow.skills import curator, usage + + monkeypatch.setattr(curator, "_list_user_owned_skill_names", lambda uid: {"linked-skill"}) + monkeypatch.setattr(curator, "_is_skill_published", lambda name: False) + # Pretend one agent's config references linked-skill. + monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: {"linked-skill"}) + _isolated_skills_root(isolated_paths, monkeypatch) + _seed_old_agent_skill("linked-skill") + + counters = curator.apply_automatic_transitions("user-a", datetime.now(UTC)) + + # Should be excluded from the eligible set entirely. + assert counters["checked"] == 0 + assert counters["archived"] == 0 + assert counters["skipped_referenced"] == 1 + assert usage.get_entry("linked-skill")["state"] == usage.STATE_ACTIVE + + +def test_curator_skips_published_skill(isolated_paths, monkeypatch) -> None: + """A skill marked ``published`` in SkillStore must never be auto-archived + even when nothing references it.""" + from datetime import UTC, datetime + + from deerflow.skills import curator, usage + + monkeypatch.setattr(curator, "_list_user_owned_skill_names", lambda uid: {"public-tool"}) + monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: set()) + monkeypatch.setattr(curator, "_is_skill_published", lambda name: name == "public-tool") + _isolated_skills_root(isolated_paths, monkeypatch) + _seed_old_agent_skill("public-tool") + + counters = curator.apply_automatic_transitions("user-a", datetime.now(UTC)) + + assert counters["checked"] == 0 + assert counters["archived"] == 0 + assert counters["skipped_published"] == 1 + assert usage.get_entry("public-tool")["state"] == usage.STATE_ACTIVE + + +def test_curator_archives_orphan_unpublished_skill(isolated_paths, monkeypatch) -> None: + """The "default happy path": owner's own agent-created skill, unpublished, + not referenced anywhere — IS archived when idle past the threshold.""" + from datetime import UTC, datetime + + from deerflow.skills import curator, usage + + monkeypatch.setattr(curator, "_list_user_owned_skill_names", lambda uid: {"orphan-skill"}) + monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: set()) + monkeypatch.setattr(curator, "_is_skill_published", lambda name: False) + _isolated_skills_root(isolated_paths, monkeypatch) + _seed_old_agent_skill("orphan-skill") + + counters = curator.apply_automatic_transitions("user-a", datetime.now(UTC)) + + assert counters["checked"] == 1 + assert counters["archived"] == 1 + assert counters["skipped_referenced"] == 0 + assert counters["skipped_published"] == 0 + assert usage.get_entry("orphan-skill")["state"] == usage.STATE_ARCHIVED + + +def test_skill_archive_tool_refuses_referenced_or_published( + isolated_paths, monkeypatch +) -> None: + """Safety net: even if the LLM bypasses the candidate filter and invokes + skill_archive with a protected name, the tool itself refuses.""" + from deerflow.skills import curator + + # Make the on-disk lookup believe the skill exists and is not pinned. + class _FakeStorage: + def custom_skill_exists(self, _name: str) -> bool: + return True + + class _FakeEntry: + def get(self, key, default=None): + return False # pinned=False + + monkeypatch.setattr("deerflow.skills.storage.get_or_new_skill_storage", lambda *_a, **_kw: _FakeStorage()) + monkeypatch.setattr("deerflow.skills.usage.get_entry", lambda _n: _FakeEntry()) + + tool = curator._make_skill_archive_tool("user-a") + + # Case A: target is referenced — must refuse. + monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: {"linked"}) + monkeypatch.setattr(curator, "_is_skill_published", lambda name: False) + out_a = tool.invoke({"name": "linked"}) + assert "Refused" in out_a + assert "referenced" in out_a + + # Case B: target is published — must refuse. + monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: set()) + monkeypatch.setattr(curator, "_is_skill_published", lambda name: name == "pub") + out_b = tool.invoke({"name": "pub"}) + assert "Refused" in out_b + assert "published" in out_b + + +def test_collect_referenced_skill_names_aggregates_from_disk( + isolated_paths, monkeypatch +) -> None: + """End-to-end of the scanner helper: it walks ``.deer-flow/agents/*/config.yaml`` + and returns the union of every ``skills`` list.""" + from deerflow.skills import curator + + # Build two agents on disk, each referencing one skill. + agents_root = isolated_paths / "agents" + (agents_root / "agent-1").mkdir(parents=True, exist_ok=True) + (agents_root / "agent-2").mkdir(parents=True, exist_ok=True) + (agents_root / "agent-1" / "config.yaml").write_text( + yaml.safe_dump({"id": "agent-1", "name": "A", "skills": ["alpha", "beta"]}), + encoding="utf-8", + ) + (agents_root / "agent-2" / "config.yaml").write_text( + yaml.safe_dump({"id": "agent-2", "name": "B", "skills": ["beta", "gamma"]}), + encoding="utf-8", + ) + + # Point get_paths().agents_dir at the temp tree. + class _FakePaths: + agents_dir = agents_root + + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: _FakePaths()) + + assert curator._collect_referenced_skill_names() == {"alpha", "beta", "gamma"}