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