cleveragents-core/robot/helper_skill_context.py

"""Robot Framework helper for skill context and registry smoke tests.

Provides a CLI-style interface for Robot to invoke context creation,
resource resolution, change tracking, write guard, and registry operations.
Exit code 0 = success, 1 = failure.

Usage:
    python robot/helper_skill_context.py create-context
    python robot/helper_skill_context.py resolve-resource
    python robot/helper_skill_context.py track-invocation
    python robot/helper_skill_context.py write-guard
    python robot/helper_skill_context.py registry-crud
    python robot/helper_skill_context.py registry-list
    python robot/helper_skill_context.py registry-resolve
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Any

# 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,
    SkillResolver,
)
from cleveragents.skills.context import SkillContext, SkillExecutionError  # noqa: E402
from cleveragents.skills.protocol import (  # noqa: E402
    SkillDefinition,
    SkillMetadata,
)
from cleveragents.skills.registry import SkillRegistry  # noqa: E402


def _make_defn(name: str, **kw: Any) -> SkillDefinition:
    """Create a SkillDefinition with defaults."""
    skill = Skill(name=name, description=f"Test {name}", **kw)
    resolver = SkillResolver()
    resolved = resolver.resolve_tools(skill, {})
    meta = SkillMetadata.from_skill(skill, resolved=resolved)
    return SkillDefinition(skill=skill, resolved_tools=resolved, metadata=meta)


def _cmd_create_context() -> int:
    """Test SkillContext creation."""
    ctx = SkillContext(
        plan_id="plan-robot",
        project_id="proj-robot",
        sandbox_path=Path("/tmp/robot-sandbox"),
    )
    if ctx.plan_id != "plan-robot":
        print(f"skill-context-fail: unexpected plan_id {ctx.plan_id}")
        return 1
    if ctx.is_read_only():
        print("skill-context-fail: should not be read-only")
        return 1
    print(f"skill-context-ok: plan={ctx.plan_id}")
    return 0


def _cmd_resolve_resource() -> int:
    """Test resource resolution."""
    ctx = SkillContext(
        plan_id="plan-res",
        project_id="proj-res",
        sandbox_path=Path("/tmp/res-sandbox"),
        resource_bindings={"git-checkout": "/repo/path"},
    )
    result = ctx.resolve_resource("git-checkout")
    if result != "/repo/path":
        print(f"skill-context-fail: unexpected resource {result}")
        return 1
    print(f"skill-context-resolve-ok: {result}")
    return 0


def _cmd_track_invocation() -> int:
    """Test change tracking."""
    ctx = SkillContext(
        plan_id="plan-track",
        project_id="proj-track",
        sandbox_path=Path("/tmp/track-sandbox"),
    )
    ctx.register_tool_invocation("local/test-tool", {"a": 1}, {"b": 2}, 15.5)
    if len(ctx.change_tracker) != 1:
        print(f"skill-context-fail: expected 1 record, got {len(ctx.change_tracker)}")
        return 1
    if ctx.change_tracker[0]["tool_name"] != "local/test-tool":
        print("skill-context-fail: wrong tool_name in tracker")
        return 1
    print("skill-context-track-ok")
    return 0


def _cmd_write_guard() -> int:
    """Test write guard enforcement."""
    ctx = SkillContext(
        plan_id="plan-guard",
        project_id="proj-guard",
        sandbox_path=Path("/tmp/guard-sandbox"),
        read_only=True,
    )
    try:
        ctx.enforce_write_guard("local/write-tool")
        print("skill-context-fail: expected SkillError")
        return 1
    except SkillExecutionError as e:
        if e.error_type.value != "permission_denied":
            print(f"skill-context-fail: unexpected error type {e.error_type}")
            return 1
    print("skill-context-guard-ok")
    return 0


def _cmd_registry_crud() -> int:
    """Test registry register/get/unregister."""
    reg = SkillRegistry()
    defn = _make_defn("local/robot-reg")
    reg.register(defn)
    got = reg.get("local/robot-reg")
    if got.skill.name != "local/robot-reg":
        print(f"skill-registry-fail: unexpected name {got.skill.name}")
        return 1
    reg.unregister("local/robot-reg")
    try:
        reg.get("local/robot-reg")
        print("skill-registry-fail: expected not-found error")
        return 1
    except SkillExecutionError:
        pass
    print("skill-registry-crud-ok")
    return 0


def _cmd_registry_list() -> int:
    """Test registry list_all."""
    reg = SkillRegistry()
    reg.register(_make_defn("local/list-a"))
    reg.register(_make_defn("local/list-b"))
    items = reg.list_all()
    if len(items) != 2:
        print(f"skill-registry-fail: expected 2, got {len(items)}")
        return 1
    print(f"skill-registry-list-ok: {len(items)} skills")
    return 0


def _cmd_registry_resolve() -> int:
    """Test registry tool resolution."""
    reg = SkillRegistry()
    defn = _make_defn("local/resolve-skill", tool_refs=["local/tool-x"])
    reg.register(defn)
    tools = reg.resolve_tools("local/resolve-skill")
    if len(tools) != 1:
        print(f"skill-registry-fail: expected 1 tool, got {len(tools)}")
        return 1
    print(f"skill-registry-resolve-ok: {len(tools)} tools")
    return 0


_COMMANDS = {
    "create-context": _cmd_create_context,
    "resolve-resource": _cmd_resolve_resource,
    "track-invocation": _cmd_track_invocation,
    "write-guard": _cmd_write_guard,
    "registry-crud": _cmd_registry_crud,
    "registry-list": _cmd_registry_list,
    "registry-resolve": _cmd_registry_resolve,
}


def main() -> int:
    """Entry point called by Robot Framework ``Run Process``."""
    if len(sys.argv) < 2:
        print(
            "Usage: helper_skill_context.py "
            "<create-context|resolve-resource|track-invocation|write-guard"
            "|registry-crud|registry-list|registry-resolve>"
        )
        return 1

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

    return handler()


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