deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/skills/cascade.py

"""Cascade handling when a published skill is removed from circulation.

Called from the Gateway skills router whenever a published skill is
unpublished, deleted, or admin-taken-down. Walks every agent on disk that
referenced the skill, strips the skill from its ``config.yaml`` ``skills``
list, and emits an in-app notification to that agent's owner.

The harness/app firewall means we cannot import the FastAPI app side; the
``AgentStore`` instance must be passed in by the caller (the router gets
it from ``app.gateway.deps.get_agent_store``).
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any, Literal

import yaml

from deerflow.config.agents_config import load_agent_config
from deerflow.config.paths import get_paths
from deerflow.persistence.notifications import notify

logger = logging.getLogger(__name__)

CascadeOperation = Literal["unpublish", "delete", "admin_takedown"]


def _list_agent_dirs() -> list[Path]:
    agents_dir = get_paths().agents_dir
    if not agents_dir.exists():
        return []
    return [entry for entry in sorted(agents_dir.iterdir()) if entry.is_dir()]


def _remove_skill_from_yaml(config_file: Path, skill_name: str) -> bool:
    """Rewrite ``config.yaml`` removing ``skill_name`` from its ``skills`` list.

    Returns True if the file was modified, False otherwise.
    """
    if not config_file.exists():
        return False
    try:
        with config_file.open("r", encoding="utf-8") as f:
            data: dict[str, Any] = yaml.safe_load(f) or {}
    except yaml.YAMLError:
        logger.warning("Cascade: failed to parse %s, skipping", config_file)
        return False

    skills = data.get("skills")
    if not isinstance(skills, list) or skill_name not in skills:
        return False

    data["skills"] = [s for s in skills if s != skill_name]
    with config_file.open("w", encoding="utf-8") as f:
        yaml.safe_dump(data, f, default_flow_style=False, allow_unicode=True)
    return True


_OP_LABELS: dict[CascadeOperation, str] = {
    "unpublish": "取消发布",
    "delete": "删除",
    "admin_takedown": "管理员强制下架",
}


async def cascade_skill_unpublished_or_deleted(
    skill_name: str,
    *,
    actor_user_id: str,
    operation: CascadeOperation,
    skill_owner_user_id: str | None,
    agent_store: Any,
) -> dict[str, Any]:
    """Detach ``skill_name`` from every referencing agent and notify owners.

    Self-references (agent.user_id == actor_user_id) are still detached but
    the owner gets no in-app notification — the user just did the action,
    they know.

    On ``admin_takedown`` we additionally notify ``skill_owner_user_id``.

    Returns counters for testability: ``{checked, modified, notified}``.
    """
    counters = {"checked": 0, "modified": 0, "notified": 0}
    label = _OP_LABELS.get(operation, operation)

    for agent_dir in _list_agent_dirs():
        config_file = agent_dir / "config.yaml"
        if not config_file.exists():
            continue
        counters["checked"] += 1

        # Quick read to decide whether we need to mutate at all.
        try:
            agent_cfg = load_agent_config(agent_dir.name)
        except Exception:
            logger.debug("Cascade: failed to load %s, skipping", agent_dir.name, exc_info=True)
            continue
        if agent_cfg is None or not agent_cfg.skills or skill_name not in agent_cfg.skills:
            continue

        modified = _remove_skill_from_yaml(config_file, skill_name)
        if not modified:
            continue
        counters["modified"] += 1
        logger.info(
            "Cascade: removed skill '%s' from agent '%s/config.yaml' (op=%s)",
            skill_name,
            agent_dir.name,
            operation,
        )

        # Look up agent owner from DB to decide where to send the notification.
        record = await agent_store.get_any(agent_dir.name)
        agent_owner = record.get("user_id") if record else None
        if agent_owner and agent_owner != actor_user_id:
            agent_display = (record.get("name") if record else None) or agent_dir.name
            await notify(
                agent_owner,
                type="agent_skill_removed",
                title=f"专家 “{agent_display}” 的一个技能已失效",
                body=(
                    f"技能 “{skill_name}” 被 {label}，已从该专家的技能列表中自动移除。"
                ),
                payload={
                    "skill_name": skill_name,
                    "agent_id": agent_dir.name,
                    "agent_name": agent_display,
                    "operation": operation,
                    "actor_user_id": actor_user_id,
                },
            )
            counters["notified"] += 1

    if operation == "admin_takedown" and skill_owner_user_id and skill_owner_user_id != actor_user_id:
        await notify(
            skill_owner_user_id,
            type="skill_taken_down",
            title=f"你的技能 “{skill_name}” 已被管理员下架",
            body="管理员对该技能执行了强制下架。如有疑问请联系管理员。",
            payload={
                "skill_name": skill_name,
                "operation": operation,
                "actor_user_id": actor_user_id,
            },
        )
        counters["notified"] += 1

    return counters


def sync_agent_skills_to_available(available_skill_names: set[str]) -> dict[str, Any]:
    """Strip skill references that no longer exist on disk from every agent.

    Walks every agent dir and rewrites its ``config.yaml`` ``skills`` list to
    contain only names that appear in ``available_skill_names``. Used by the
    admin "agent-skills sync" endpoint to repair historical drift — skills
    deleted before the cascade existed, deleted via a path that skipped the
    cascade, or removed from the filesystem manually.

    Synchronous because it only touches local files and the admin endpoint
    runs it in a worker thread.

    Returns ``{checked, modified, removals: [{agent_id, removed_skills}]}``.
    """
    counters: dict[str, Any] = {"checked": 0, "modified": 0, "removals": []}

    for agent_dir in _list_agent_dirs():
        config_file = agent_dir / "config.yaml"
        if not config_file.exists():
            continue
        counters["checked"] += 1

        try:
            with config_file.open("r", encoding="utf-8") as f:
                data: dict[str, Any] = yaml.safe_load(f) or {}
        except yaml.YAMLError:
            logger.warning("Sync: failed to parse %s, skipping", config_file)
            continue

        skills = data.get("skills")
        if not isinstance(skills, list) or not skills:
            continue

        kept: list[str] = []
        removed: list[str] = []
        for name in skills:
            if isinstance(name, str) and name in available_skill_names:
                kept.append(name)
            else:
                removed.append(name if isinstance(name, str) else str(name))

        if not removed:
            continue

        data["skills"] = kept
        with config_file.open("w", encoding="utf-8") as f:
            yaml.safe_dump(data, f, default_flow_style=False, allow_unicode=True)

        counters["modified"] += 1
        counters["removals"].append({"agent_id": agent_dir.name, "removed_skills": removed})
        logger.info("Sync: removed orphan skills %s from agent '%s'", removed, agent_dir.name)

    return counters