"""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 ") 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())