cleveragents-core/robot/helper_actor_runtime.py

"""Helper script for Robot Framework actor runtime smoke tests."""

from __future__ import annotations

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.actor_context import ToolActorContext
from cleveragents.tool.actor_runtime import (
    LLMResponse,
    LLMToolCall,
    ToolCallingRuntime,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner
from cleveragents.tool.runtime import ToolSpec


def _echo(inputs: dict[str, Any]) -> dict[str, Any]:
    return {"echoed": inputs}


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


_captured: dict[str, Any] = {}


def _capture(inputs: dict[str, Any]) -> dict[str, Any]:
    _captured.clear()
    _captured.update(inputs)
    return {"ok": True}


class SingleToolLLM:
    """Requests one tool call then returns."""

    def __init__(self) -> None:
        self._n = 0

    def invoke(
        self,
        prompt: str,
        tool_schemas: list[dict[str, Any]],
        tool_results: list[dict[str, Any]] | None = None,
        actor_config: dict[str, Any] | None = None,
    ) -> LLMResponse:
        self._n += 1
        if self._n == 1:
            return LLMResponse(
                tool_calls=[LLMToolCall(name="test/echo", arguments={"a": 1})]
            )
        return LLMResponse(content="done")


class AlwaysToolLLM:
    """Always requests tool calls."""

    def invoke(
        self,
        prompt: str,
        tool_schemas: list[dict[str, Any]],
        tool_results: list[dict[str, Any]] | None = None,
        actor_config: dict[str, Any] | None = None,
    ) -> LLMResponse:
        return LLMResponse(
            tool_calls=[LLMToolCall(name="test/echo", arguments={"x": 1})]
        )


class FailToolLLM:
    """Requests failing tool then stops."""

    def __init__(self) -> None:
        self._n = 0

    def invoke(
        self,
        prompt: str,
        tool_schemas: list[dict[str, Any]],
        tool_results: list[dict[str, Any]] | None = None,
        actor_config: dict[str, Any] | None = None,
    ) -> LLMResponse:
        self._n += 1
        if self._n == 1:
            return LLMResponse(tool_calls=[LLMToolCall(name="test/fail", arguments={})])
        return LLMResponse(content="handled error")


class CaptureToolLLM:
    """Requests capturing tool then stops."""

    def __init__(self) -> None:
        self._n = 0

    def invoke(
        self,
        prompt: str,
        tool_schemas: list[dict[str, Any]],
        tool_results: list[dict[str, Any]] | None = None,
        actor_config: dict[str, Any] | None = None,
    ) -> LLMResponse:
        self._n += 1
        if self._n == 1:
            return LLMResponse(
                tool_calls=[LLMToolCall(name="test/capture", arguments={"d": 1})]
            )
        return LLMResponse(content="captured")


class ImmediateLLM:
    """Returns immediately without tool calls."""

    def invoke(
        self,
        prompt: str,
        tool_schemas: list[dict[str, Any]],
        tool_results: list[dict[str, Any]] | None = None,
        actor_config: dict[str, Any] | None = None,
    ) -> LLMResponse:
        return LLMResponse(content="immediate")


def _make_registry_with_echo() -> ToolRegistry:
    reg = ToolRegistry()
    reg.register(ToolSpec(name="test/echo", description="Echo", handler=_echo))
    return reg


def test_single_tool() -> None:
    reg = _make_registry_with_echo()
    runner = ToolRunner(reg)
    rt = ToolCallingRuntime(registry=reg, runner=runner, llm_caller=SingleToolLLM())
    result = rt.run_tool_loop(prompt="go")
    assert result.content == "done"
    assert len(result.tool_call_history) == 1
    assert not result.terminated_by_limit
    print("single-tool-ok")


def test_max_iterations() -> None:
    reg = _make_registry_with_echo()
    runner = ToolRunner(reg)
    rt = ToolCallingRuntime(
        registry=reg, runner=runner, llm_caller=AlwaysToolLLM(), max_iterations=3
    )
    result = rt.run_tool_loop(prompt="loop")
    assert result.terminated_by_limit
    assert result.iterations == 3
    print("max-iterations-ok")


def test_error_handling() -> None:
    reg = ToolRegistry()
    reg.register(ToolSpec(name="test/fail", description="Fail", handler=_fail))
    runner = ToolRunner(reg)
    rt = ToolCallingRuntime(registry=reg, runner=runner, llm_caller=FailToolLLM())
    result = rt.run_tool_loop(prompt="fail")
    assert len(result.tool_call_history) == 1
    assert not result.tool_call_history[0].success
    assert result.tool_call_history[0].error is not None
    print("error-handling-ok")


def test_context_props() -> None:
    ctx = ToolActorContext(plan_id="p1", phase="execute", sandbox_root="/sb")
    assert ctx.plan_id == "p1"
    assert ctx.phase == "execute"
    assert ctx.sandbox_root == "/sb"
    assert len(ctx.tool_call_history) == 0
    summary = ctx.as_summary()
    assert summary["plan_id"] == "p1"
    print("context-props-ok")


def test_sandbox_thread() -> None:
    reg = ToolRegistry()
    reg.register(ToolSpec(name="test/capture", description="Capture", handler=_capture))
    runner = ToolRunner(reg)
    ctx = ToolActorContext(plan_id="p2", phase="execute", sandbox_root="/sandbox/path")
    rt = ToolCallingRuntime(registry=reg, runner=runner, llm_caller=CaptureToolLLM())
    rt.run_tool_loop(prompt="capture", context=ctx)
    assert _captured.get("sandbox_root") == "/sandbox/path"
    print("sandbox-thread-ok")


def test_empty_registry() -> None:
    reg = ToolRegistry()
    runner = ToolRunner(reg)
    rt = ToolCallingRuntime(registry=reg, runner=runner, llm_caller=ImmediateLLM())
    result = rt.run_tool_loop(prompt="nothing")
    assert result.content == "immediate"
    assert len(result.tool_call_history) == 0
    print("empty-registry-ok")


COMMANDS = {
    "single_tool": test_single_tool,
    "max_iterations": test_max_iterations,
    "error_handling": test_error_handling,
    "context_props": test_context_props,
    "sandbox_thread": test_sandbox_thread,
    "empty_registry": test_empty_registry,
}

if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "single_tool"
    COMMANDS[cmd]()