cleveragents-core/robot/helper_tool_router.py

"""Helper script for Robot Framework tool router smoke tests."""

from __future__ import annotations

import json
import sys
from pathlib import Path
from typing import Any

# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.router import (
    NormalizedToolCallResult,
    ProviderFormat,
    StreamingToolUpdate,
    ToolCallRouter,
    classify_tool_error,
    detect_provider_format,
    generate_tool_call_id,
    normalize_tool_call,
    normalize_tool_schema_for_provider,
)
from cleveragents.tool.runner import ToolRunner
from cleveragents.tool.runtime import ToolSpec


def _echo(inputs: dict[str, Any]) -> dict[str, Any]:
    return dict(inputs)


def _fail(inputs: dict[str, Any]) -> dict[str, Any]:
    raise RuntimeError("boom")


def test_detect_openai() -> None:
    fmt = detect_provider_format({"name": "x/y", "arguments": '{"a": 1}'})
    assert fmt == ProviderFormat.OPENAI
    print("detect-openai-ok")


def test_detect_anthropic() -> None:
    fmt = detect_provider_format({"name": "x/y", "input": {"a": 1}})
    assert fmt == ProviderFormat.ANTHROPIC
    print("detect-anthropic-ok")


def test_detect_langchain() -> None:
    fmt = detect_provider_format({"name": "x/y", "type": "tool_call", "args": {}})
    assert fmt == ProviderFormat.LANGCHAIN
    print("detect-langchain-ok")


def test_normalize_openai() -> None:
    req = normalize_tool_call({"name": "x/y", "arguments": '{"a": 1}'})
    assert req.tool_name == "x/y"
    assert req.arguments["a"] == 1
    assert req.provider_format == ProviderFormat.OPENAI
    print("normalize-openai-ok")


def test_stable_id() -> None:
    id1 = generate_tool_call_id("plan-1", 0)
    id2 = generate_tool_call_id("plan-1", 0)
    assert id1 == id2
    assert id1.startswith("tc_")
    print("stable-id-ok")


def test_route_openai() -> None:
    registry = ToolRegistry()
    spec = ToolSpec(name="smoke/echo", description="Echo", handler=_echo)
    registry.register(spec)
    runner = ToolRunner(registry)
    router = ToolCallRouter(registry=registry, runner=runner, plan_id="plan-1")
    result = router.route({"name": "smoke/echo", "arguments": '{"msg": "hi"}'})
    assert result.result.success
    assert result.provider_format == ProviderFormat.OPENAI
    print("route-openai-ok")


def test_route_anthropic() -> None:
    registry = ToolRegistry()
    spec = ToolSpec(name="smoke/echo", description="Echo", handler=_echo)
    registry.register(spec)
    runner = ToolRunner(registry)
    router = ToolCallRouter(registry=registry, runner=runner, plan_id="plan-2")
    result = router.route({"name": "smoke/echo", "input": {"msg": "hi"}})
    assert result.result.success
    assert result.provider_format == ProviderFormat.ANTHROPIC
    print("route-anthropic-ok")


def test_route_error() -> None:
    registry = ToolRegistry()
    spec = ToolSpec(name="smoke/fail", description="Fail", handler=_fail)
    registry.register(spec)
    runner = ToolRunner(registry)
    router = ToolCallRouter(registry=registry, runner=runner, plan_id="plan-3")
    result = router.route({"name": "smoke/fail", "arguments": "{}"})
    assert not result.result.success
    print("route-error-ok")


def test_error_classify() -> None:
    cat = classify_tool_error("Operation timed out")
    assert cat.value == "timeout"
    cat2 = classify_tool_error("Permission denied")
    assert cat2.value == "permission"
    print("error-classify-ok")


def test_schema_normalize() -> None:
    spec = ToolSpec(name="smoke/echo", description="Echo tool", handler=_echo)
    schema = normalize_tool_schema_for_provider(spec, ProviderFormat.OPENAI)
    assert "parameters" in schema
    schema2 = normalize_tool_schema_for_provider(spec, ProviderFormat.ANTHROPIC)
    assert "input_schema" in schema2
    print("schema-normalize-ok")


def test_batch_route() -> None:
    registry = ToolRegistry()
    spec = ToolSpec(name="smoke/echo", description="Echo", handler=_echo)
    registry.register(spec)
    runner = ToolRunner(registry)
    router = ToolCallRouter(registry=registry, runner=runner, plan_id="plan-batch")
    payloads = [
        {"name": "smoke/echo", "arguments": json.dumps({"i": i})} for i in range(3)
    ]
    results = router.route_batch(payloads)
    assert len(results) == 3
    assert all(r.result.success for r in results)
    print("batch-route-ok")


def test_streaming() -> None:
    registry = ToolRegistry()
    spec = ToolSpec(name="smoke/echo", description="Echo", handler=_echo)
    registry.register(spec)
    runner = ToolRunner(registry)
    router = ToolCallRouter(registry=registry, runner=runner, plan_id="plan-stream")
    items = list(
        router.route_streaming({"name": "smoke/echo", "arguments": '{"msg": "s"}'})
    )
    updates = [i for i in items if isinstance(i, StreamingToolUpdate)]
    results = [i for i in items if isinstance(i, NormalizedToolCallResult)]
    assert len(updates) >= 2
    assert len(results) == 1
    assert results[0].result.success
    print("streaming-ok")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "detect_openai"
    commands = {
        "detect_openai": test_detect_openai,
        "detect_anthropic": test_detect_anthropic,
        "detect_langchain": test_detect_langchain,
        "normalize_openai": test_normalize_openai,
        "stable_id": test_stable_id,
        "route_openai": test_route_openai,
        "route_anthropic": test_route_anthropic,
        "route_error": test_route_error,
        "error_classify": test_error_classify,
        "schema_normalize": test_schema_normalize,
        "batch_route": test_batch_route,
        "streaming": test_streaming,
    }
    handler = commands.get(cmd)
    if handler is None:
        print(f"Unknown command: {cmd}", file=sys.stderr)
        sys.exit(1)
    handler()