cleveragents-core/robot/helper_skill_refresh.py

"""Robot Framework helper for Skill Refresh smoke tests.

Exercises ``SkillRegistry.refresh()``, ``SkillRegistry.refresh_all()``,
``SkillRefreshResult``, and ``MCPRefreshHook`` end-to-end from a single
process so Robot Framework can verify outputs without importing Python
objects directly.

Usage::

    python robot/helper_skill_refresh.py refresh-single-ok
    python robot/helper_skill_refresh.py refresh-single-missing-tool
    python robot/helper_skill_refresh.py refresh-single-no-registry
    python robot/helper_skill_refresh.py refresh-all-mixed
    python robot/helper_skill_refresh.py refresh-unknown-skill
    python robot/helper_skill_refresh.py result-summary
    python robot/helper_skill_refresh.py result-merge
    python robot/helper_skill_refresh.py hook-wiring
    python robot/helper_skill_refresh.py hook-debounce
    python robot/helper_skill_refresh.py hook-cancel
"""

from __future__ import annotations

import sys
import time
from pathlib import Path
from typing import Any

_ROOT = str(Path(__file__).resolve().parents[1])
_SRC = str(Path(__file__).resolve().parents[1] / "src")
for _p in (_SRC, _ROOT):
    if _p not in sys.path:
        sys.path.insert(0, _p)

from features.mocks.mock_mcp_transport import MockMCPTransport  # noqa: E402

from cleveragents.domain.models.core.skill import Skill  # noqa: E402
from cleveragents.mcp.adapter import MCPServerConfig, MCPToolAdapter  # noqa: E402
from cleveragents.mcp.refresh_hook import MCPRefreshHook  # noqa: E402
from cleveragents.skills.protocol import SkillDefinition, SkillMetadata  # noqa: E402
from cleveragents.skills.refresh import SkillRefreshResult  # noqa: E402
from cleveragents.skills.registry import SkillRegistry  # noqa: E402
from cleveragents.tool.registry import ToolRegistry  # noqa: E402

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_skill(name: str, tool_refs: list[str] | None = None) -> SkillDefinition:
    """Build a minimal SkillDefinition for testing."""
    skill = Skill(
        name=name,
        description=f"Test skill {name}",
        tool_refs=tool_refs or [],
    )
    return SkillDefinition(
        skill=skill,
        metadata=SkillMetadata.from_skill(skill),
    )


def _make_adapter(
    server_name: str = "test-server",
    tools: list[dict[str, Any]] | None = None,
) -> MCPToolAdapter:
    config = MCPServerConfig(name=server_name, transport="stdio", command="echo")
    transport = MockMCPTransport(tools=tools or [])
    adapter = MCPToolAdapter(config=config, transport=transport)
    adapter.connect()
    return adapter


def _make_registry_with_tools(tool_names: list[str]) -> ToolRegistry:
    """Create a ToolRegistry pre-populated with named tools."""
    from cleveragents.tool.runtime import ToolSpec

    registry = ToolRegistry()
    for name in tool_names:
        registry.register(
            ToolSpec(
                name=name,
                description=f"Tool {name}",
                handler=lambda **_kw: None,
            )
        )
    return registry


# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------


def _refresh_single_ok() -> int:
    """Refresh a skill that has all its tool refs resolved — expect success."""
    registry = SkillRegistry()
    skill_def = _make_skill("ns/skill-a", tool_refs=[])
    registry.register(skill_def)

    result = registry.refresh("ns/skill-a")
    print(f"refreshed-count: {result.total_refreshed}")
    print(f"failed-count: {result.total_failed}")
    print(f"skipped-count: {result.total_skipped}")
    print(f"name-in-refreshed: {'ns/skill-a' in result.refreshed}")
    print("refresh-single-ok")
    return 0


def _refresh_single_missing_tool() -> int:
    """Refresh when a tool ref is absent from the ToolRegistry — expect failure."""
    tool_reg = _make_registry_with_tools(["ns/tool-existing"])
    skill_reg = SkillRegistry(tool_registry=tool_reg)
    skill_def = _make_skill(
        "ns/skill-b", tool_refs=["ns/tool-existing", "ns/tool-missing"]
    )
    skill_reg.register(skill_def)

    result = skill_reg.refresh("ns/skill-b")
    print(f"failed-count: {result.total_failed}")
    print(f"has-error: {'ns/skill-b' in result.failed}")
    error_msg = result.failed.get("ns/skill-b", "")
    print(f"error-mentions-missing: {'ns/tool-missing' in error_msg}")
    print("refresh-missing-tool-ok")
    return 0


def _refresh_single_no_registry() -> int:
    """Refresh without a ToolRegistry — expect skill in refreshed (no validation)."""
    skill_reg = SkillRegistry()
    skill_def = _make_skill("ns/skill-c", tool_refs=["some/tool"])
    skill_reg.register(skill_def)

    result = skill_reg.refresh("ns/skill-c")
    print(f"refreshed-count: {result.total_refreshed}")
    print(f"failed-count: {result.total_failed}")
    print(f"name-in-refreshed: {'ns/skill-c' in result.refreshed}")
    print("refresh-no-registry-ok")
    return 0


def _refresh_all_mixed() -> int:
    """refresh_all() with a mix of success and failure skills."""
    tool_reg = _make_registry_with_tools(["ns/tool-x"])
    skill_reg = SkillRegistry(tool_registry=tool_reg)

    # skill-ok: all refs present
    skill_reg.register(_make_skill("ns/skill-ok", tool_refs=["ns/tool-x"]))
    # skill-bad: ref missing
    skill_reg.register(_make_skill("ns/skill-bad", tool_refs=["ns/tool-missing"]))

    result = skill_reg.refresh_all()
    print(f"total-refreshed: {result.total_refreshed}")
    print(f"total-failed: {result.total_failed}")
    print(f"ok-refreshed: {'ns/skill-ok' in result.refreshed}")
    print(f"bad-failed: {'ns/skill-bad' in result.failed}")
    summary = result.to_summary()
    print(f"summary: {summary}")
    print("refresh-all-mixed-ok")
    return 0


def _refresh_unknown_skill() -> int:
    """Refresh a non-existent skill — expect SkillExecutionError."""
    skill_reg = SkillRegistry()
    try:
        skill_reg.refresh("ns/nonexistent")
        print("unexpected-success")
        return 1
    except Exception as exc:
        err = str(exc).lower()
        print(f"error-type: {type(exc).__name__}")
        print(f"error-contains-name: {'nonexistent' in err}")
        print("refresh-unknown-ok")
        return 0


def _result_summary() -> int:
    """Verify SkillRefreshResult.to_summary() output format."""
    result = SkillRefreshResult(
        refreshed=["a", "b", "c"],
        failed={"d": "missing tool"},
        skipped=[],
    )
    summary = result.to_summary()
    print(f"summary: {summary}")
    print(f"has-refreshed: {'refreshed: 3' in summary}")
    print(f"has-failed: {'failed: 1' in summary}")
    print(f"has-skipped: {'skipped: 0' in summary}")
    print("result-summary-ok")
    return 0


def _result_merge() -> int:
    """Verify SkillRefreshResult.merge() correctly combines two results."""
    r1 = SkillRefreshResult(refreshed=["a"], failed={}, skipped=["b"])
    r2 = SkillRefreshResult(refreshed=["c"], failed={"d": "err"}, skipped=[])
    merged = r1.merge(r2)
    print(f"total-refreshed: {merged.total_refreshed}")
    print(f"total-failed: {merged.total_failed}")
    print(f"total-skipped: {merged.total_skipped}")
    print(f"a-in-refreshed: {'a' in merged.refreshed}")
    print(f"c-in-refreshed: {'c' in merged.refreshed}")
    print(f"b-in-skipped: {'b' in merged.skipped}")
    print(f"d-in-failed: {'d' in merged.failed}")
    print("result-merge-ok")
    return 0


def _hook_wiring() -> int:
    """MCPRefreshHook wires notification → refresh_all() correctly."""
    adapter = _make_adapter("hook-server")
    skill_reg = SkillRegistry()
    skill_reg.register(_make_skill("ns/hook-skill"))

    # Use zero debounce for deterministic test
    hook = MCPRefreshHook(
        adapter=adapter, skill_registry=skill_reg, debounce_seconds=0.0
    )

    # Dispatch the notification
    adapter.dispatch_notification("notifications/tools/list_changed")

    # Wait briefly for the daemon timer to fire
    deadline = time.monotonic() + 3.0
    while hook.refresh_count == 0 and time.monotonic() < deadline:
        time.sleep(0.05)

    print(f"refresh-count: {hook.refresh_count}")
    print(f"refreshed-at-least-once: {hook.refresh_count >= 1}")
    hook.cancel()
    print("hook-wiring-ok")
    return 0


def _hook_debounce() -> int:
    """Multiple notifications within debounce window collapse to one refresh."""
    adapter = _make_adapter("debounce-server")
    skill_reg = SkillRegistry()
    skill_reg.register(_make_skill("ns/debounce-skill"))

    hook = MCPRefreshHook(
        adapter=adapter, skill_registry=skill_reg, debounce_seconds=0.1
    )

    # Fire 5 notifications in rapid succession
    for _ in range(5):
        adapter.dispatch_notification("notifications/tools/list_changed")
        time.sleep(0.01)

    # Wait for the single coalesced refresh to complete
    deadline = time.monotonic() + 3.0
    while hook.refresh_count == 0 and time.monotonic() < deadline:
        time.sleep(0.05)

    # Allow a moment for any spurious second fire
    time.sleep(0.3)

    print(f"refresh-count: {hook.refresh_count}")
    print(f"debounced-to-one: {hook.refresh_count == 1}")
    hook.cancel()
    print("hook-debounce-ok")
    return 0


def _hook_cancel() -> int:
    """MCPRefreshHook.cancel() prevents the pending refresh from firing."""
    adapter = _make_adapter("cancel-server")
    skill_reg = SkillRegistry()
    skill_reg.register(_make_skill("ns/cancel-skill"))

    # Use a generous debounce so we can cancel before it fires
    hook = MCPRefreshHook(
        adapter=adapter, skill_registry=skill_reg, debounce_seconds=2.0
    )
    adapter.dispatch_notification("notifications/tools/list_changed")

    # Cancel before the timer fires
    hook.cancel()

    # Wait longer than the debounce — refresh should NOT have occurred
    time.sleep(0.2)

    print(f"refresh-count-after-cancel: {hook.refresh_count}")
    print(f"no-refresh-after-cancel: {hook.refresh_count == 0}")
    print("hook-cancel-ok")
    return 0


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------


def main() -> int:
    if len(sys.argv) < 2:
        print("Usage: helper_skill_refresh.py <command>")
        return 1

    dispatch = {
        "refresh-single-ok": _refresh_single_ok,
        "refresh-single-missing-tool": _refresh_single_missing_tool,
        "refresh-single-no-registry": _refresh_single_no_registry,
        "refresh-all-mixed": _refresh_all_mixed,
        "refresh-unknown-skill": _refresh_unknown_skill,
        "result-summary": _result_summary,
        "result-merge": _result_merge,
        "hook-wiring": _hook_wiring,
        "hook-debounce": _hook_debounce,
        "hook-cancel": _hook_cancel,
    }
    handler = dispatch.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())