feat(skills): curator 三条件过滤 — agent 创建 + 未发布 + 无引用
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) <noreply@anthropic.com>
This commit is contained in:
xuhdandClaude Opus 4.7 committed
1 parent
6155a2ec2b
commit
3e6e3c08d5
2 files changed
+336
-14
No files matched your search
@@ -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)
|
||||
|
||||
@@ -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"}
|
||||
Reference in new issue
Block a user