cleveragents-core/robot/helper_tool_env_preferences.py

"""Robot Framework helper for tool-level execution environment preferences."""

from __future__ import annotations

import importlib
import sys
from pathlib import Path
from typing import Any
from unittest.mock import create_autospec

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.domain.models.core.execution_environment_preference import (  # noqa: E402
    EnvironmentPreferenceMode,
    ExecutionEnvironmentPreference,
)
from cleveragents.domain.models.core.tool import Tool, ToolSource  # noqa: E402
from cleveragents.tool.registry import ToolRegistry  # noqa: E402
from cleveragents.tool.runner import ToolRunner  # noqa: E402
from cleveragents.tool.runtime import ToolResult, ToolSpec  # noqa: E402


def _make_handler(inputs: dict[str, Any]) -> dict[str, Any]:
    return {"result": "ok"}


def _run_default_preference() -> None:
    """Default preference is NONE."""
    pref = ExecutionEnvironmentPreference()
    assert pref.mode == EnvironmentPreferenceMode.NONE
    assert pref.target_resource is None
    print("default-preference-ok")


def _run_required_mode() -> None:
    """Required mode validates correctly."""
    pref = ExecutionEnvironmentPreference(mode=EnvironmentPreferenceMode.REQUIRED)
    assert pref.mode == EnvironmentPreferenceMode.REQUIRED
    assert pref.target_resource is None
    print("required-mode-ok")


def _run_preferred_mode() -> None:
    """Preferred mode validates correctly."""
    pref = ExecutionEnvironmentPreference(mode=EnvironmentPreferenceMode.PREFERRED)
    assert pref.mode == EnvironmentPreferenceMode.PREFERRED
    assert pref.target_resource is None
    print("preferred-mode-ok")


def _run_specific_mode() -> None:
    """Specific mode requires target_resource."""
    pref = ExecutionEnvironmentPreference(
        mode=EnvironmentPreferenceMode.SPECIFIC,
        target_resource="local/api-dev",
    )
    assert pref.mode == EnvironmentPreferenceMode.SPECIFIC
    assert pref.target_resource == "local/api-dev"

    # Should fail without target_resource
    try:
        ExecutionEnvironmentPreference(mode=EnvironmentPreferenceMode.SPECIFIC)
        print("FAIL: expected ValueError")
        sys.exit(1)
    except ValueError:
        pass

    print("specific-mode-ok")


def _run_specific_mode_validates_target() -> None:
    """Specific mode validates that target_resource is present."""
    try:
        ExecutionEnvironmentPreference(mode=EnvironmentPreferenceMode.SPECIFIC)
        print("FAIL: expected ValueError for missing target")
        sys.exit(1)
    except ValueError as exc:
        assert "'specific' mode requires target_resource" in str(exc)

    # target_resource not allowed with non-specific modes
    try:
        ExecutionEnvironmentPreference(
            mode=EnvironmentPreferenceMode.REQUIRED,
            target_resource="local/api-dev",
        )
        print("FAIL: expected ValueError for invalid target")
        sys.exit(1)
    except ValueError as exc:
        assert "target_resource only valid" in str(exc)

    print("specific-validates-target-ok")


def _run_toolspec_field() -> None:
    """ToolSpec has execution_environment field."""
    spec = ToolSpec(
        name="test/tool",
        description="A test tool",
        handler=_make_handler,
    )
    assert spec.execution_environment.mode == EnvironmentPreferenceMode.NONE

    spec2 = ToolSpec(
        name="test/tool2",
        description="A required tool",
        handler=_make_handler,
        execution_environment=ExecutionEnvironmentPreference(
            mode=EnvironmentPreferenceMode.REQUIRED,
        ),
    )
    assert spec2.execution_environment.mode == EnvironmentPreferenceMode.REQUIRED
    print("toolspec-field-ok")


def _run_domain_tool_field() -> None:
    """Domain Tool has execution_environment field."""
    tool = Tool(
        name="local/test-tool",
        description="A test tool",
        source=ToolSource.BUILTIN,
    )
    assert tool.execution_environment.mode == EnvironmentPreferenceMode.NONE
    print("domain-tool-field-ok")


def _run_tool_from_config() -> None:
    """Tool.from_config parses execution_environment."""
    config: dict[str, Any] = {
        "name": "local/test-tool",
        "description": "A test tool",
        "source": "builtin",
        "execution_environment": {"mode": "required"},
    }
    tool = Tool.from_config(config)
    assert tool.execution_environment.mode == EnvironmentPreferenceMode.REQUIRED

    # Without execution_environment
    config2: dict[str, Any] = {
        "name": "local/test-tool2",
        "description": "Another test tool",
        "source": "builtin",
    }
    tool2 = Tool.from_config(config2)
    assert tool2.execution_environment.mode == EnvironmentPreferenceMode.NONE

    # With specific mode and target
    config3: dict[str, Any] = {
        "name": "local/test-tool3",
        "description": "Specific test tool",
        "source": "builtin",
        "execution_environment": {
            "mode": "specific",
            "target_resource": "local/api-dev",
        },
    }
    tool3 = Tool.from_config(config3)
    assert tool3.execution_environment.mode == EnvironmentPreferenceMode.SPECIFIC
    assert tool3.execution_environment.target_resource == "local/api-dev"

    print("tool-from-config-ok")


def _run_runner_none_preference() -> None:
    """ToolRunner with NONE preference uses caller-supplied env."""
    registry = ToolRegistry()
    spec = ToolSpec(
        name="test/none-pref",
        description="Tool with no env preference",
        handler=_make_handler,
    )
    registry.register(spec)
    runner = ToolRunner(registry=registry)
    runner.activate("test/none-pref")
    result = runner.execute("test/none-pref", {"input": "value"})
    assert result.success is True
    print("runner-none-preference-ok")


def _run_runner_required_no_container() -> None:
    """ToolRunner with REQUIRED preference fails without container."""
    registry = ToolRegistry()
    spec = ToolSpec(
        name="test/required-pref",
        description="Tool requiring container",
        handler=_make_handler,
        execution_environment=ExecutionEnvironmentPreference(
            mode=EnvironmentPreferenceMode.REQUIRED,
        ),
    )
    registry.register(spec)
    runner = ToolRunner(registry=registry)
    runner.activate("test/required-pref")
    result = runner.execute("test/required-pref", {"input": "value"})
    assert result.success is False
    assert "requires container execution" in (result.error or "")
    print("runner-required-no-container-ok")


def _run_runner_preferred_fallback() -> None:
    """ToolRunner with PREFERRED preference falls back to host."""
    registry = ToolRegistry()
    spec = ToolSpec(
        name="test/preferred-pref",
        description="Tool preferring container",
        handler=_make_handler,
        execution_environment=ExecutionEnvironmentPreference(
            mode=EnvironmentPreferenceMode.PREFERRED,
        ),
    )
    registry.register(spec)
    runner = ToolRunner(registry=registry)
    runner.activate("test/preferred-pref")
    result = runner.execute("test/preferred-pref", {"input": "value"})
    assert result.success is True
    print("runner-preferred-fallback-ok")


def _run_runner_specific_container() -> None:
    """ToolRunner with SPECIFIC preference overrides tool_env."""
    registry = ToolRegistry()
    spec = ToolSpec(
        name="test/specific-pref",
        description="Tool targeting specific container",
        handler=_make_handler,
        execution_environment=ExecutionEnvironmentPreference(
            mode=EnvironmentPreferenceMode.SPECIFIC,
            target_resource="local/api-dev",
        ),
    )
    registry.register(spec)

    from cleveragents.tool.container_executor import ContainerToolExecutor

    mock_executor = create_autospec(ContainerToolExecutor, instance=True)
    mock_executor.execute_tool.return_value = ToolResult(
        success=True,
        output={"result": "container-ok"},
        duration_ms=1.0,
    )
    runner = ToolRunner(registry=registry, container_executor=mock_executor)
    runner.activate("test/specific-pref")
    result = runner.execute(
        "test/specific-pref",
        {"input": "value"},
        linked_resource_types=["container-instance"],
        project_name="test-project",
    )
    assert result.success is True
    print("runner-specific-container-ok")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "all"
    dispatch = {
        "default-preference": _run_default_preference,
        "required-mode": _run_required_mode,
        "preferred-mode": _run_preferred_mode,
        "specific-mode": _run_specific_mode,
        "specific-validates-target": _run_specific_mode_validates_target,
        "toolspec-field": _run_toolspec_field,
        "domain-tool-field": _run_domain_tool_field,
        "tool-from-config": _run_tool_from_config,
        "runner-none-preference": _run_runner_none_preference,
        "runner-required-no-container": _run_runner_required_no_container,
        "runner-preferred-fallback": _run_runner_preferred_fallback,
        "runner-specific-container": _run_runner_specific_container,
    }

    if cmd == "all":
        for fn in dispatch.values():
            fn()
    elif cmd in dispatch:
        dispatch[cmd]()
    else:
        print(f"Unknown command: {cmd}")
        sys.exit(1)