cleveragents-core/robot/helper_skill_flatten.py

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

Provides a CLI-style interface for Robot to invoke skill flattening,
capability summary computation, validate_plan(), and the tools() method.

Usage:
    python robot/helper_skill_flatten.py flatten-simple
    python robot/helper_skill_flatten.py flatten-deep
    python robot/helper_skill_flatten.py flatten-wide
    python robot/helper_skill_flatten.py flatten-inline
    python robot/helper_skill_flatten.py flatten-cycle
    python robot/helper_skill_flatten.py flatten-include-overrides
    python robot/helper_skill_flatten.py flatten-non-overridable
    python robot/helper_skill_flatten.py capability-summary
    python robot/helper_skill_flatten.py tools-method
    python robot/helper_skill_flatten.py validate-plan-ok
    python robot/helper_skill_flatten.py validate-plan-missing
"""

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,
    SkillResolver,
)
from cleveragents.domain.models.core.tool import (  # noqa: E402
    ToolCapability,
    ToolSource,
)
from cleveragents.skills.protocol import (  # noqa: E402
    SkillDefinition,
    SkillMetadata,
)
from cleveragents.skills.registry import SkillRegistry  # noqa: E402


def _make_defn(skill: Skill) -> SkillDefinition:
    """Build a SkillDefinition from a Skill with default metadata."""
    resolver = SkillResolver()
    resolved = resolver.resolve_tools(skill, {})
    meta = SkillMetadata.from_skill(skill, resolved=resolved)
    return SkillDefinition(skill=skill, resolved_tools=resolved, metadata=meta)


def _flatten_simple() -> int:
    """Flatten a skill with 5 tool refs."""
    skill = Skill(
        name="local/simple",
        description="Simple skill",
        tool_refs=[f"local/tool-{i}" for i in range(5)],
    )
    resolver = SkillResolver()
    result = resolver.resolve_tools(skill, {})
    names = [e.name for e in result]
    expected = [f"local/tool-{i}" for i in range(5)]
    if names != expected:
        print(f"flatten-fail: expected {expected}, got {names}")
        return 1
    print(f"flatten-simple-ok: {json.dumps(names)}")
    return 0


def _flatten_deep() -> int:
    """Flatten a skill with 3 levels of includes."""
    base = Skill(name="local/base", description="Base", tool_refs=["local/base-t"])
    mid = Skill(
        name="local/mid",
        description="Mid",
        tool_refs=["local/mid-t"],
        includes=[SkillInclude(name="local/base")],
    )
    top = Skill(
        name="local/top",
        description="Top",
        tool_refs=["local/top-t"],
        includes=[SkillInclude(name="local/mid")],
    )
    registry = {
        "local/base": base,
        "local/mid": mid,
        "local/top": top,
    }
    resolver = SkillResolver()
    result = resolver.resolve_tools(top, registry)
    names = [e.name for e in result]
    expected = ["local/base-t", "local/mid-t", "local/top-t"]
    if names != expected:
        print(f"flatten-fail: expected {expected}, got {names}")
        return 1
    print(f"flatten-deep-ok: {json.dumps(names)}")
    return 0


def _flatten_wide() -> int:
    """Flatten a skill that includes 10 other skills."""
    children: dict[str, Skill] = {}
    includes: list[SkillInclude] = []
    for i in range(10):
        name = f"local/child-{i}"
        children[name] = Skill(
            name=name,
            description=f"Child {i}",
            tool_refs=[f"local/child-{i}-tool"],
        )
        includes.append(SkillInclude(name=name))
    top = Skill(name="local/wide", description="Wide", includes=includes)
    registry = {**children, "local/wide": top}
    resolver = SkillResolver()
    result = resolver.resolve_tools(top, registry)
    if len(result) != 10:
        print(f"flatten-fail: expected 10 tools, got {len(result)}")
        return 1
    print(f"flatten-wide-ok: {len(result)} tools")
    return 0


def _flatten_inline() -> int:
    """Flatten a skill with inline tools."""
    skill = Skill(
        name="local/inline",
        description="Inline",
        anonymous_tools=[
            SkillInlineTool(
                description=f"Inline {i}",
                source=ToolSource.CUSTOM,
                code=f"return {i}",
            )
            for i in range(3)
        ],
    )
    resolver = SkillResolver()
    result = resolver.resolve_tools(skill, {})
    if len(result) != 3:
        print(f"flatten-fail: expected 3, got {len(result)}")
        return 1
    for i, entry in enumerate(result):
        if not entry.is_inline:
            print(f"flatten-fail: entry {i} not inline")
            return 1
    print(f"flatten-inline-ok: {len(result)} inline tools")
    return 0


def _flatten_cycle() -> int:
    """Detect a cycle and verify error."""
    a = Skill(
        name="local/cyc-a",
        description="A",
        includes=[SkillInclude(name="local/cyc-b")],
    )
    b = Skill(
        name="local/cyc-b",
        description="B",
        includes=[SkillInclude(name="local/cyc-a")],
    )
    registry = {"local/cyc-a": a, "local/cyc-b": b}
    resolver = SkillResolver()
    try:
        resolver.resolve_tools(a, registry)
        print("flatten-fail: expected cycle error")
        return 1
    except ValueError as exc:
        if "ycle" not in str(exc):
            print(f"flatten-fail: wrong error: {exc}")
            return 1
        print(f"flatten-cycle-ok: {exc}")
        return 0


def _flatten_include_overrides() -> int:
    """Apply per-include overrides."""
    child = Skill(
        name="local/child",
        description="Child",
        tool_refs=["local/tool-x"],
    )
    parent = Skill(
        name="local/parent",
        description="Parent",
        includes=[
            SkillInclude(
                name="local/child",
                overrides={"timeout": 600},
            )
        ],
    )
    registry = {"local/child": child, "local/parent": parent}
    resolver = SkillResolver()
    result = resolver.resolve_tools(parent, registry)
    entry = result[0]
    if entry.overrides.get("timeout") != 600:
        print(f"flatten-fail: override not applied: {entry.overrides}")
        return 1
    print(f"flatten-include-overrides-ok: {entry.overrides}")
    return 0


def _flatten_non_overridable() -> int:
    """Reject non-overridable fields."""
    child = Skill(
        name="local/child-nr",
        description="Child",
        tool_refs=["local/tool-y"],
    )
    parent = Skill(
        name="local/parent-nr",
        description="Parent",
        includes=[
            SkillInclude(
                name="local/child-nr",
                overrides={"name": "bad"},
            )
        ],
    )
    registry = {"local/child-nr": child, "local/parent-nr": parent}
    resolver = SkillResolver()
    try:
        resolver.resolve_tools(parent, registry)
        print("flatten-fail: expected override rejection error")
        return 1
    except ValueError as exc:
        if "Non-overridable" not in str(exc):
            print(f"flatten-fail: wrong error: {exc}")
            return 1
        print(f"flatten-non-overridable-ok: {exc}")
        return 0


def _capability_summary() -> int:
    """Compute capability summary with diverse inline tools."""
    skill = Skill(
        name="local/cap",
        description="Cap test",
        anonymous_tools=[
            SkillInlineTool(
                description="RO",
                source=ToolSource.CUSTOM,
                code="pass",
                capability=ToolCapability(read_only=True),
            ),
            SkillInlineTool(
                description="Writer",
                source=ToolSource.CUSTOM,
                code="pass",
                capability=ToolCapability(writes=True, side_effects=["fs"]),
            ),
            SkillInlineTool(
                description="Checkpoint",
                source=ToolSource.CUSTOM,
                code="pass",
                capability=ToolCapability(checkpointable=True),
            ),
        ],
    )
    resolver = SkillResolver()
    entries = resolver.resolve_tools(skill, {})
    summary = resolver.compute_capability_summary(skill, entries)
    checks = [
        summary.total_tools == 3,
        summary.read_only_tools == 1,
        summary.write_tools == 1,
        # Writer tool now defaults to checkpointable=True (writes=True
        # AND NOT read_only), plus the explicit Checkpoint tool = 2.
        summary.checkpointable_tools == 2,
        summary.has_side_effects is True,
    ]
    if not all(checks):
        print(f"flatten-fail: summary checks failed: {summary}")
        return 1
    print("flatten-capability-summary-ok")
    return 0


def _tools_method() -> int:
    """Call SkillRegistry.tools()."""
    skill = Skill(
        name="local/tm",
        description="Tools method test",
        tool_refs=["local/t1", "local/t2"],
    )
    defn = _make_defn(skill)
    sr = SkillRegistry()
    sr.register(defn)
    entries, summary = sr.tools("local/tm")
    if len(entries) != 2:
        print(f"flatten-fail: expected 2 entries, got {len(entries)}")
        return 1
    if summary.total_tools != 2:
        print(f"flatten-fail: summary total={summary.total_tools}")
        return 1
    print("flatten-tools-method-ok")
    return 0


def _validate_plan_ok() -> int:
    """validate_plan with valid plan."""
    skill = Skill(
        name="local/vp",
        description="VP test",
        tool_refs=["local/t1"],
    )
    defn = _make_defn(skill)
    sr = SkillRegistry()
    sr.register(defn)
    errors = sr.validate_plan({"skills": ["local/vp"]})
    if errors:
        print(f"flatten-fail: unexpected errors: {errors}")
        return 1
    print("flatten-validate-plan-ok")
    return 0


def _validate_plan_missing() -> int:
    """validate_plan with missing skill."""
    sr = SkillRegistry()
    errors = sr.validate_plan({"skills": ["local/ghost"]})
    if not errors:
        print("flatten-fail: expected error for missing skill")
        return 1
    if "local/ghost" not in errors[0]:
        print(f"flatten-fail: wrong error: {errors[0]}")
        return 1
    print(f"flatten-validate-plan-missing-ok: {errors[0]}")
    return 0


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print("Usage: helper_skill_flatten.py <command>")
        return 1

    commands = {
        "flatten-simple": _flatten_simple,
        "flatten-deep": _flatten_deep,
        "flatten-wide": _flatten_wide,
        "flatten-inline": _flatten_inline,
        "flatten-cycle": _flatten_cycle,
        "flatten-include-overrides": _flatten_include_overrides,
        "flatten-non-overridable": _flatten_non_overridable,
        "capability-summary": _capability_summary,
        "tools-method": _tools_method,
        "validate-plan-ok": _validate_plan_ok,
        "validate-plan-missing": _validate_plan_missing,
    }

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

    return handler()


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