cleveragents-core/robot/helper_skill_discovery.py

"""Robot Framework helper for Agent Skills discovery tests."""

from __future__ import annotations

import shutil
import tempfile
from pathlib import Path
from typing import Any

from cleveragents.application.services.config_service import ConfigService
from cleveragents.skills.discovery import (
    DiscoveredAgentSkill,
    build_tool_spec,
    register_discovered_skills,
    scan_directory,
)
from cleveragents.skills.discovery import (
    parse_agent_skills_paths as _parse_paths,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runtime import ToolSpec


def _noop_handler(**_kwargs: Any) -> dict[str, Any]:
    return {"status": "placeholder"}


def _create_skill_md(folder: Path, name: str, description: str) -> None:
    folder.mkdir(parents=True, exist_ok=True)
    content = (
        "---\n"
        f"name: {name}\n"
        f"description: {description}\n"
        "---\n"
        f"\n# {name}\n\n{description}\n"
    )
    (folder / "SKILL.md").write_text(content, encoding="utf-8")


# ---------------------------------------------------------------------------
# Robot keywords
# ---------------------------------------------------------------------------


def parse_agent_skills_paths(raw_value: str) -> list[str]:
    """Parse comma-separated paths and return as string list."""
    paths = _parse_paths(raw_value)
    return [str(p) for p in paths]


def create_agent_skills_directory(count: int) -> str:
    """Create a temp directory with *count* agent skill folders."""
    tmpdir = Path(tempfile.mkdtemp())
    for i in range(int(count)):
        _create_skill_md(tmpdir / f"skill-{i}", f"skill-{i}", f"Skill {i}")
    return str(tmpdir)


def create_mixed_directory() -> str:
    """Create a directory with one valid skill and one without SKILL.md."""
    tmpdir = Path(tempfile.mkdtemp())
    _create_skill_md(tmpdir / "valid-skill", "valid-skill", "A valid skill")
    (tmpdir / "no-skill").mkdir()
    return str(tmpdir)


def scan_agent_skills_directory(directory: str) -> list[dict[str, Any]]:
    """Scan a directory and return discovered skills as dicts."""
    results = scan_directory(Path(directory))
    return [
        {"name": s.name, "description": s.description, "path": s.path} for s in results
    ]


def build_agent_skill_toolspec(name: str, path: str) -> ToolSpec:
    """Build a ToolSpec from a discovered agent skill."""
    skill = DiscoveredAgentSkill(
        name=name,
        description=f"Agent skill {name}",
        path=path,
    )
    return build_tool_spec(skill)


def register_agent_skills_in_fresh_registry(
    count: int,
) -> dict[str, int]:
    """Register discovered skills in a fresh ToolRegistry."""
    registry = ToolRegistry()
    discovered = [
        DiscoveredAgentSkill(
            name=f"tool-{i}",
            description=f"Tool {i}",
            path=f"/tmp/skills/tool-{i}",
        )
        for i in range(int(count))
    ]
    registered, conflicts = register_discovered_skills(
        discovered, registry, on_conflict="skip"
    )
    return {
        "registered_count": len(registered),
        "conflict_count": len(conflicts),
    }


def register_agent_skills_with_collision(
    strategy: str,
) -> dict[str, int]:
    """Register a skill that collides with an existing tool."""
    registry = ToolRegistry()
    existing = ToolSpec(
        name="agent_skills/collider",
        description="Existing tool",
        handler=_noop_handler,
    )
    registry.register(existing)

    discovered = [
        DiscoveredAgentSkill(
            name="collider",
            description="Conflicting skill",
            path="/tmp/skills/collider",
        )
    ]

    try:
        registered, conflicts = register_discovered_skills(
            discovered, registry, on_conflict=strategy
        )
    except ValueError:
        return {"registered_count": 0, "conflict_count": 1}

    return {
        "registered_count": len(registered),
        "conflict_count": len(conflicts),
    }


def get_config_entry(key: str) -> Any:
    """Get a config entry from the registry."""
    return ConfigService.get_entry(key)


def cleanup_temp_directory(path: str) -> None:
    """Remove a temporary directory."""
    shutil.rmtree(path, ignore_errors=True)