cleveragents-core/robot/helper_skill_resolution.py

"""Robot Framework helper for skill resolution smoke tests.

Provides a CLI-style interface for Robot to invoke skill resolution
logic and validate resolver output ordering, de-duplication, and
inline tool handling.

Usage:
    python robot/helper_skill_resolution.py resolve-flat
    python robot/helper_skill_resolution.py resolve-dedup
    python robot/helper_skill_resolution.py resolve-inline
    python robot/helper_skill_resolution.py resolve-cycle
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

from cleveragents.domain.models.core.skill import (  # noqa: E402
    Skill,
    SkillInclude,
    SkillInlineTool,
    SkillMcpSource,
    SkillResolver,
)
from cleveragents.domain.models.core.tool import ToolSource  # noqa: E402


def _resolve_flat() -> int:
    """Resolve a skill with flat includes and verify output ordering."""
    base = Skill(
        name="local/base-tools",
        description="Base tools",
        tool_refs=["local/edit-file", "local/read-file"],
    )
    root = Skill(
        name="local/code-tools",
        description="Code editing tools",
        tool_refs=["local/format"],
        includes=[SkillInclude(name="local/base-tools")],
    )
    registry = {"local/base-tools": base, "local/code-tools": root}
    resolver = SkillResolver()
    resolved = resolver.resolve_tools(root, registry)
    names = [e.name for e in resolved]

    # Include tools should come first (depth-first), then own refs
    expected = ["local/edit-file", "local/read-file", "local/format"]
    if names != expected:
        print(f"skill-resolve-fail: expected {expected}, got {names}")
        return 1

    print(f"skill-resolve-flat-ok: {json.dumps(names)}")
    return 0


def _resolve_dedup() -> int:
    """Resolve overlapping includes and verify de-duplication."""
    skill_a = Skill(
        name="local/a",
        description="Skill A",
        tool_refs=["local/shared-tool", "local/a-only"],
    )
    skill_b = Skill(
        name="local/b",
        description="Skill B",
        tool_refs=["local/shared-tool", "local/b-only"],
    )
    root = Skill(
        name="local/top",
        description="Top skill",
        includes=[
            SkillInclude(name="local/a"),
            SkillInclude(name="local/b"),
        ],
    )
    registry = {
        "local/a": skill_a,
        "local/b": skill_b,
        "local/top": root,
    }
    resolver = SkillResolver()
    resolved = resolver.resolve_tools(root, registry)
    names = [e.name for e in resolved]

    # shared-tool appears once (position from first occurrence in a),
    # but entry metadata is from b (last-wins)
    if names.count("local/shared-tool") != 1:
        print(f"skill-resolve-fail: shared-tool should appear once, got {names}")
        return 1

    # The shared-tool entry should come from skill b (last-wins)
    shared_entry = next(e for e in resolved if e.name == "local/shared-tool")
    if shared_entry.source_skill != "local/b":
        print(
            f"skill-resolve-fail: shared-tool source should be local/b, "
            f"got {shared_entry.source_skill}"
        )
        return 1

    print(f"skill-resolve-dedup-ok: {json.dumps(names)}")
    return 0


def _resolve_inline() -> int:
    """Resolve a skill with inline tools and verify keying."""
    root = Skill(
        name="local/inline-skill",
        description="Skill with inline tools",
        tool_refs=["local/ref-tool"],
        anonymous_tools=[
            SkillInlineTool(
                description="First inline",
                source=ToolSource.CUSTOM,
                code="return 1",
                timeout=300,
            ),
            SkillInlineTool(
                description="Second inline",
                source=ToolSource.CUSTOM,
                code="return 2",
                timeout=300,
            ),
        ],
        mcp_servers=[
            SkillMcpSource(
                server="npx @mcp/test-server",
                tools=["mcp-read"],
            ),
        ],
    )
    resolver = SkillResolver()
    resolved = resolver.resolve_tools(root, {})
    names = [e.name for e in resolved]

    expected = [
        "local/ref-tool",
        "local/inline-skill/_anon_0",
        "local/inline-skill/_anon_1",
        "mcp:npx @mcp/test-server/mcp-read",
    ]
    if names != expected:
        print(f"skill-resolve-fail: expected {expected}, got {names}")
        return 1

    # Check that inline entries are marked correctly
    anon_entries = [e for e in resolved if e.is_inline]
    if len(anon_entries) != 2:
        print(f"skill-resolve-fail: expected 2 inline entries, got {len(anon_entries)}")
        return 1

    print(f"skill-resolve-inline-ok: {json.dumps(names)}")
    return 0


def _resolve_cycle() -> int:
    """Attempt cyclic resolution and verify error."""
    skill_a = Skill(
        name="local/cycle-a",
        description="Cycle A",
        includes=[SkillInclude(name="local/cycle-b")],
    )
    skill_b = Skill(
        name="local/cycle-b",
        description="Cycle B",
        includes=[SkillInclude(name="local/cycle-a")],
    )
    registry = {"local/cycle-a": skill_a, "local/cycle-b": skill_b}
    resolver = SkillResolver()
    try:
        resolver.resolve_tools(skill_a, registry)
        print("skill-resolve-fail: expected cycle error but resolution succeeded")
        return 1
    except ValueError as exc:
        if "ycle" not in str(exc):
            print(f"skill-resolve-fail: expected cycle error, got: {exc}")
            return 1
        print(f"skill-resolve-cycle-ok: {exc}")
        return 0


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(
            "Usage: helper_skill_resolution.py "
            "<resolve-flat|resolve-dedup|resolve-inline|resolve-cycle>"
        )
        return 1

    command = sys.argv[1]

    commands = {
        "resolve-flat": _resolve_flat,
        "resolve-dedup": _resolve_dedup,
        "resolve-inline": _resolve_inline,
        "resolve-cycle": _resolve_cycle,
    }

    handler = commands.get(command)
    if handler is None:
        print(f"Unknown command: {command}")
        return 1

    return handler()


if __name__ == "__main__":
    sys.exit(main())