feat(tool): add tool-level execution environment preferences #970

Merged
brent.edwards merged 2 commits from feature/m5-tool-env-prefs into master 2026-03-20 23:51:47 +00:00
11 changed files with 1211 additions and 7 deletions
+3
View File
@@ -2,6 +2,9 @@
## Unreleased
- Added tool-level execution environment preferences with NONE, REQUIRED,
PREFERRED, and SPECIFIC modes. ToolRunner routes tool execution based on
preference mode with caller-override precedence. (#879)
- Added TDD bug-capture tests for #969`plan correct` expects `decision_id`
but M3 acceptance test passes `plan_id`. Behave BDD scenarios (revert and
append modes) and Robot Framework integration tests verify that
@@ -0,0 +1,476 @@
"""Step definitions for tool-level execution environment preference tests."""
from __future__ import annotations
from enum import StrEnum
from typing import Any
from unittest.mock import create_autospec
from behave import given, then, when
from behave.runner import Context
# ---------------------------------------------------------------------------
# EnvironmentPreferenceMode enum steps
# ---------------------------------------------------------------------------
@given("I import EnvironmentPreferenceMode")
def step_import_pref_mode(context: Context) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
)
context.pref_mode_cls = EnvironmentPreferenceMode
@then('the NONE preference value should be "{value}"')
def step_none_pref_value(context: Context, value: str) -> None:
assert context.pref_mode_cls.NONE.value == value
@then('the REQUIRED preference value should be "{value}"')
def step_required_pref_value(context: Context, value: str) -> None:
assert context.pref_mode_cls.REQUIRED.value == value
@then('the PREFERRED preference value should be "{value}"')
def step_preferred_pref_value(context: Context, value: str) -> None:
assert context.pref_mode_cls.PREFERRED.value == value
@then('the SPECIFIC preference value should be "{value}"')
def step_specific_pref_value(context: Context, value: str) -> None:
assert context.pref_mode_cls.SPECIFIC.value == value
@then("EnvironmentPreferenceMode should be a StrEnum subclass")
def step_pref_mode_is_strenum(context: Context) -> None:
assert issubclass(context.pref_mode_cls, StrEnum)
# ---------------------------------------------------------------------------
# ExecutionEnvironmentPreference creation steps
# ---------------------------------------------------------------------------
@given("I create a default ExecutionEnvironmentPreference")
def step_create_default_pref(context: Context) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
ExecutionEnvironmentPreference,
)
context.pref = ExecutionEnvironmentPreference()
@given('I create an ExecutionEnvironmentPreference with mode "{mode}"')
def step_create_pref_with_mode(context: Context, mode: str) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
context.pref = ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode(mode),
)
@given('I create a specific ExecutionEnvironmentPreference targeting "{target}"')
def step_create_specific_pref(context: Context, target: str) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
context.pref = ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode.SPECIFIC,
target_resource=target,
)
@when(
'I try to create an ExecutionEnvironmentPreference with mode "{mode}"'
' and target "{target}"'
)
def step_try_create_pref_with_target(context: Context, mode: str, target: str) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
try:
ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode(mode),
target_resource=target,
)
context.pref_error = None
except ValueError as exc:
context.pref_error = str(exc)
@when(
'I try to create an ExecutionEnvironmentPreference with mode "specific"'
" and no target"
)
def step_try_create_specific_no_target(context: Context) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
try:
ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode.SPECIFIC,
)
context.pref_error = None
except ValueError as exc:
context.pref_error = str(exc)
@then('the preference creation should fail with "{message}"')
def step_pref_creation_failed(context: Context, message: str) -> None:
assert context.pref_error is not None, "Expected creation to fail"
assert message in context.pref_error, (
f"Expected '{message}' in error, got: {context.pref_error}"
)
@then('the preference mode should be "{mode}"')
def step_pref_mode_is(context: Context, mode: str) -> None:
assert context.pref.mode.value == mode
@then("the preference target_resource should be None")
def step_pref_target_none(context: Context) -> None:
assert context.pref.target_resource is None
@then('the preference target_resource should be "{target}"')
def step_pref_target_is(context: Context, target: str) -> None:
assert context.pref.target_resource == target
# ---------------------------------------------------------------------------
# Frozen model steps
# ---------------------------------------------------------------------------
@when("I try to mutate the preference mode")
def step_try_mutate_pref(context: Context) -> None:
from pydantic import ValidationError
try:
context.pref.mode = "required" # type: ignore[misc]
context.mutation_failed = False
except (ValidationError, TypeError):
context.mutation_failed = True
@then("the mutation should fail")
def step_mutation_failed(context: Context) -> None:
assert context.mutation_failed is True
# ---------------------------------------------------------------------------
# ToolSpec field steps
# ---------------------------------------------------------------------------
def _make_handler(inputs: dict[str, Any]) -> dict[str, Any]:
return {"result": "ok"}
@given("I create a ToolSpec with default execution_environment")
def step_toolspec_default_env(context: Context) -> None:
from cleveragents.tool.runtime import ToolSpec
context.tool_spec = ToolSpec(
name="test/tool",
description="A test tool",
handler=_make_handler,
)
@given("I create a ToolSpec with required execution_environment")
def step_toolspec_required_env(context: Context) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
from cleveragents.tool.runtime import ToolSpec
context.tool_spec = ToolSpec(
name="test/tool",
description="A test tool",
handler=_make_handler,
execution_environment=ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode.REQUIRED,
),
)
@then('the ToolSpec execution_environment mode should be "{mode}"')
def step_toolspec_env_mode(context: Context, mode: str) -> None:
assert context.tool_spec.execution_environment.mode.value == mode
# ---------------------------------------------------------------------------
# Domain Tool field steps
# ---------------------------------------------------------------------------
@given("I create a domain Tool with default execution_environment")
def step_domain_tool_default_env(context: Context) -> None:
from cleveragents.domain.models.core.tool import Tool, ToolSource
context.domain_tool = Tool(
name="local/test-tool",
description="A test tool",
source=ToolSource.BUILTIN,
)
@given("I create a domain Tool with required execution_environment")
def step_domain_tool_required_env(context: Context) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
from cleveragents.domain.models.core.tool import Tool, ToolSource
context.domain_tool = Tool(
name="local/test-tool",
description="A test tool",
source=ToolSource.BUILTIN,
execution_environment=ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode.REQUIRED,
),
)
@then('the domain Tool execution_environment mode should be "{mode}"')
def step_domain_tool_env_mode(context: Context, mode: str) -> None:
assert context.domain_tool.execution_environment.mode.value == mode
# ---------------------------------------------------------------------------
# Tool.from_config steps
# ---------------------------------------------------------------------------
@given('I have a tool config dict with execution_environment mode "{mode}"')
def step_config_with_env_mode(context: Context, mode: str) -> None:
context.tool_config = {
"name": "local/test-tool",
"description": "A test tool",
"source": "builtin",
"execution_environment": {"mode": mode},
}
@given('I have a tool config dict with execution_environment targeting "{target}"')
def step_config_with_env_mode_target(context: Context, target: str) -> None:
context.tool_config = {
"name": "local/test-tool",
"description": "A test tool",
"source": "builtin",
"execution_environment": {"mode": "specific", "target_resource": target},
}
@given("I have a tool config dict without execution_environment")
def step_config_without_env(context: Context) -> None:
context.tool_config = {
"name": "local/test-tool",
"description": "A test tool",
"source": "builtin",
}
@when("I create a Tool from the config")
def step_tool_from_config(context: Context) -> None:
from cleveragents.domain.models.core.tool import Tool
context.domain_tool = Tool.from_config(context.tool_config)
@then('the tool execution_environment mode should be "{mode}"')
def step_tool_env_mode(context: Context, mode: str) -> None:
assert context.domain_tool.execution_environment.mode.value == mode
@then('the tool execution_environment target_resource should be "{target}"')
def step_tool_env_target(context: Context, target: str) -> None:
assert context.domain_tool.execution_environment.target_resource == target
# ---------------------------------------------------------------------------
# as_cli_dict steps
# ---------------------------------------------------------------------------
@when("I call as_cli_dict on the tool")
def step_call_cli_dict(context: Context) -> None:
context.cli_dict = context.domain_tool.as_cli_dict()
@then('the cli dict should not contain "{key}"')
def step_cli_dict_no_key(context: Context, key: str) -> None:
assert key not in context.cli_dict
@then('the cli dict should contain execution_environment with mode "{mode}"')
def step_cli_dict_env_mode(context: Context, mode: str) -> None:
assert "execution_environment" in context.cli_dict
assert context.cli_dict["execution_environment"]["mode"] == mode
# ---------------------------------------------------------------------------
# ToolRunner preference routing steps
# ---------------------------------------------------------------------------
@given("I have a ToolRunner with a tool with NONE preference")
def step_runner_none_pref(context: Context) -> None:
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner
from cleveragents.tool.runtime import ToolSpec
registry = ToolRegistry()
spec = ToolSpec(
name="test/none-pref",
description="Tool with no env preference",
handler=_make_handler,
)
registry.register(spec)
context.runner = ToolRunner(registry=registry)
context.runner.activate("test/none-pref")
context.tool_name = "test/none-pref"
@when("I execute the tool with no caller env")
def step_execute_no_caller_env(context: Context) -> None:
context.result = context.runner.execute(context.tool_name, {"input": "value"})
@then("the tool preference routing should succeed on host")
def step_tool_success_host(context: Context) -> None:
assert context.result.success is True
@given("I have a ToolRunner with a tool with REQUIRED preference and no container")
def step_runner_required_no_container(context: Context) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner
from cleveragents.tool.runtime import ToolSpec
registry = ToolRegistry()
spec = ToolSpec(
name="test/required-pref",
description="Tool requiring container",
handler=_make_handler,
execution_environment=ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode.REQUIRED,
),
)
registry.register(spec)
context.runner = ToolRunner(registry=registry)
context.runner.activate("test/required-pref")
context.tool_name = "test/required-pref"
@when("I execute the required-preference tool")
def step_execute_required(context: Context) -> None:
context.result = context.runner.execute(context.tool_name, {"input": "value"})
@then("the execution should fail with container required error")
def step_fail_container_required(context: Context) -> None:
assert context.result.success is False
assert "requires container execution" in (context.result.error or "")
@given("I have a ToolRunner with a tool with PREFERRED preference and no container")
def step_runner_preferred_no_container(context: Context) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner
from cleveragents.tool.runtime import ToolSpec
registry = ToolRegistry()
spec = ToolSpec(
name="test/preferred-pref",
description="Tool preferring container",
handler=_make_handler,
execution_environment=ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode.PREFERRED,
),
)
registry.register(spec)
context.runner = ToolRunner(registry=registry)
context.runner.activate("test/preferred-pref")
context.tool_name = "test/preferred-pref"
@when("I execute the preferred-preference tool")
def step_execute_preferred(context: Context) -> None:
context.result = context.runner.execute(context.tool_name, {"input": "value"})
@given('I have a ToolRunner with a tool with SPECIFIC preference targeting "{target}"')
def step_runner_specific_pref(context: Context, target: str) -> None:
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner
from cleveragents.tool.runtime import ToolSpec
registry = ToolRegistry()
spec = ToolSpec(
name="test/specific-pref",
description="Tool targeting specific container",
handler=_make_handler,
execution_environment=ExecutionEnvironmentPreference(
mode=EnvironmentPreferenceMode.SPECIFIC,
target_resource=target,
),
)
registry.register(spec)
# Provide a mock container executor so container routing works
from cleveragents.tool.container_executor import ContainerToolExecutor
from cleveragents.tool.runtime import ToolResult
mock_executor = create_autospec(ContainerToolExecutor, instance=True)
mock_executor.execute_tool.return_value = ToolResult(
success=True,
output={"result": "container-ok"},
duration_ms=1.0,
)
context.runner = ToolRunner(registry=registry, container_executor=mock_executor)
context.runner.activate("test/specific-pref")
context.tool_name = "test/specific-pref"
@when("I execute the specific-preference tool with container available")
def step_execute_specific_container(context: Context) -> None:
context.result = context.runner.execute(
context.tool_name,
{"input": "value"},
linked_resource_types=["container-instance"],
project_name="test-project",
)
@then("the tool preference routing should succeed in container")
def step_tool_container_mode(context: Context) -> None:
assert context.result.success is True
+151
View File
@@ -0,0 +1,151 @@
Feature: Tool-level execution environment preferences
As the tool execution engine
I need tools to declare intrinsic execution environment preferences
So that execution routing respects tool requirements
# EnvironmentPreferenceMode enum
Scenario: EnvironmentPreferenceMode has NONE value
Given I import EnvironmentPreferenceMode
Then the NONE preference value should be "none"
Scenario: EnvironmentPreferenceMode has REQUIRED value
Given I import EnvironmentPreferenceMode
Then the REQUIRED preference value should be "required"
Scenario: EnvironmentPreferenceMode has PREFERRED value
Given I import EnvironmentPreferenceMode
Then the PREFERRED preference value should be "preferred"
Scenario: EnvironmentPreferenceMode has SPECIFIC value
Given I import EnvironmentPreferenceMode
Then the SPECIFIC preference value should be "specific"
Scenario: EnvironmentPreferenceMode is a StrEnum
Given I import EnvironmentPreferenceMode
Then EnvironmentPreferenceMode should be a StrEnum subclass
# ── Default preference ─────────────────────────────────────────────
Scenario: Default preference is NONE
Given I create a default ExecutionEnvironmentPreference
Then the preference mode should be "none"
And the preference target_resource should be None
# ── Required mode ──────────────────────────────────────────────────
Scenario: Required mode with no target_resource
Given I create an ExecutionEnvironmentPreference with mode "required"
Then the preference mode should be "required"
And the preference target_resource should be None
Scenario: Required mode rejects target_resource
When I try to create an ExecutionEnvironmentPreference with mode "required" and target "local/api-dev"
Then the preference creation should fail with "target_resource only valid"
# ── Preferred mode ─────────────────────────────────────────────────
Scenario: Preferred mode with no target_resource
Given I create an ExecutionEnvironmentPreference with mode "preferred"
Then the preference mode should be "preferred"
And the preference target_resource should be None
Scenario: Preferred mode rejects target_resource
When I try to create an ExecutionEnvironmentPreference with mode "preferred" and target "local/api-dev"
Then the preference creation should fail with "target_resource only valid"
# ── Specific mode ──────────────────────────────────────────────────
Scenario: Specific mode requires target_resource
When I try to create an ExecutionEnvironmentPreference with mode "specific" and no target
Then the preference creation should fail with "'specific' mode requires target_resource"
Scenario: Specific mode with target_resource
Given I create a specific ExecutionEnvironmentPreference targeting "local/api-dev"
Then the preference mode should be "specific"
And the preference target_resource should be "local/api-dev"
# ── None mode ──────────────────────────────────────────────────────
Scenario: None mode rejects target_resource
When I try to create an ExecutionEnvironmentPreference with mode "none" and target "local/api-dev"
Then the preference creation should fail with "target_resource only valid"
# ── Frozen model ───────────────────────────────────────────────────
Scenario: ExecutionEnvironmentPreference is frozen
Given I create a default ExecutionEnvironmentPreference
When I try to mutate the preference mode
Then the mutation should fail
# ── ToolSpec field ─────────────────────────────────────────────────
Scenario: ToolSpec has default execution_environment
Given I create a ToolSpec with default execution_environment
Then the ToolSpec execution_environment mode should be "none"
Scenario: ToolSpec accepts execution_environment preference
Given I create a ToolSpec with required execution_environment
Then the ToolSpec execution_environment mode should be "required"
# ── Domain Tool field ──────────────────────────────────────────────
Scenario: Domain Tool has default execution_environment
Given I create a domain Tool with default execution_environment
Then the domain Tool execution_environment mode should be "none"
Scenario: Domain Tool accepts execution_environment preference
Given I create a domain Tool with required execution_environment
Then the domain Tool execution_environment mode should be "required"
# ── Tool YAML config parsing ───────────────────────────────────────
Scenario: Tool.from_config parses execution_environment with required mode
Given I have a tool config dict with execution_environment mode "required"
When I create a Tool from the config
Then the tool execution_environment mode should be "required"
Scenario: Tool.from_config parses execution_environment with specific mode
Given I have a tool config dict with execution_environment targeting "local/api-dev"
When I create a Tool from the config
Then the tool execution_environment mode should be "specific"
And the tool execution_environment target_resource should be "local/api-dev"
Scenario: Tool.from_config defaults execution_environment to none
Given I have a tool config dict without execution_environment
When I create a Tool from the config
Then the tool execution_environment mode should be "none"
# ── as_cli_dict ────────────────────────────────────────────────────
Scenario: as_cli_dict omits execution_environment when none
Given I create a domain Tool with default execution_environment
When I call as_cli_dict on the tool
Then the cli dict should not contain "execution_environment"
Scenario: as_cli_dict includes execution_environment when non-none
Given I create a domain Tool with required execution_environment
When I call as_cli_dict on the tool
Then the cli dict should contain execution_environment with mode "required"
# ── ToolRunner preference routing ──────────────────────────────────
Scenario: ToolRunner with NONE preference uses caller-supplied env
Given I have a ToolRunner with a tool with NONE preference
When I execute the tool with no caller env
Then the tool preference routing should succeed on host
Scenario: ToolRunner with REQUIRED preference fails without container
Given I have a ToolRunner with a tool with REQUIRED preference and no container
When I execute the required-preference tool
Then the execution should fail with container required error
Scenario: ToolRunner with PREFERRED preference falls back to host
Given I have a ToolRunner with a tool with PREFERRED preference and no container
When I execute the preferred-preference tool
Then the tool preference routing should succeed on host
Scenario: ToolRunner with SPECIFIC preference overrides tool_env
Given I have a ToolRunner with a tool with SPECIFIC preference targeting "local/api-dev"
When I execute the specific-preference tool with container available
Then the tool preference routing should succeed in container
+283
View File
@@ -0,0 +1,283 @@
"""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)
+83
View File
@@ -0,0 +1,83 @@
*** Settings ***
Documentation Integration tests for tool-level execution environment preferences
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_tool_env_preferences.py
*** Test Cases ***
Default Preference Is NONE
[Documentation] Default ExecutionEnvironmentPreference has mode=none
${result}= Run Process ${PYTHON} ${HELPER} default-preference cwd=${WORKSPACE} on_timeout=kill timeout=30s
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} default-preference-ok
Required Mode Validates
[Documentation] Required mode creates correctly with no target_resource
${result}= Run Process ${PYTHON} ${HELPER} required-mode cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} required-mode-ok
Preferred Mode Validates
[Documentation] Preferred mode creates correctly with no target_resource
${result}= Run Process ${PYTHON} ${HELPER} preferred-mode cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} preferred-mode-ok
Specific Mode With Target
[Documentation] Specific mode requires and accepts target_resource
${result}= Run Process ${PYTHON} ${HELPER} specific-mode cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} specific-mode-ok
Specific Mode Validates Target Required
[Documentation] Specific mode validates target_resource is present
${result}= Run Process ${PYTHON} ${HELPER} specific-validates-target cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} specific-validates-target-ok
ToolSpec Has Execution Environment Field
[Documentation] ToolSpec model includes execution_environment field
${result}= Run Process ${PYTHON} ${HELPER} toolspec-field cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} toolspec-field-ok
Domain Tool Has Execution Environment Field
[Documentation] Domain Tool model includes execution_environment field
${result}= Run Process ${PYTHON} ${HELPER} domain-tool-field cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} domain-tool-field-ok
Tool From Config Parses Execution Environment
[Documentation] Tool.from_config parses execution_environment from YAML dict
${result}= Run Process ${PYTHON} ${HELPER} tool-from-config cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} tool-from-config-ok
Runner NONE Preference Uses Caller Env
[Documentation] ToolRunner with NONE preference uses caller-supplied env
${result}= Run Process ${PYTHON} ${HELPER} runner-none-preference cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} runner-none-preference-ok
Runner Required Fails Without Container
[Documentation] ToolRunner with REQUIRED preference fails without container
${result}= Run Process ${PYTHON} ${HELPER} runner-required-no-container cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} runner-required-no-container-ok
Runner Preferred Falls Back To Host
[Documentation] ToolRunner with PREFERRED preference falls back to host
${result}= Run Process ${PYTHON} ${HELPER} runner-preferred-fallback cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} runner-preferred-fallback-ok
Runner Specific Targets Container
[Documentation] ToolRunner with SPECIFIC preference targets named container
${result}= Run Process ${PYTHON} ${HELPER} runner-specific-container cwd=${WORKSPACE} on_timeout=kill timeout=30s
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} runner-specific-container-ok
@@ -129,6 +129,10 @@ from cleveragents.domain.models.core.escalation import (
HistoricalOutcome,
OperationContext,
)
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
from cleveragents.domain.models.core.invariant import (
Invariant,
InvariantEnforcementRecord,
@@ -368,6 +372,7 @@ __all__ = [
"DoDResult",
"DoDStatus",
"DoDSummary",
"EnvironmentPreferenceMode",
"ErrorCategory",
"ErrorHistory",
"ErrorPattern",
@@ -375,6 +380,7 @@ __all__ = [
"ErrorRecoveryPolicy",
"EscalationDecision",
"ExecutionEnvironment",
"ExecutionEnvironmentPreference",
"FileRecord",
"FragmentProvenance",
"GuardResult",
@@ -0,0 +1,78 @@
"""Execution environment preference for tools.
Defines intrinsic execution environment preferences that are part of
a tool's definition — not caller-supplied. These preferences are
consulted during execution routing to determine *where* a tool should
run.
## Preference Modes
| Mode | Behavior |
|---------------|----------------------------------------------------|
| ``none`` | No preference (default); use caller-supplied env |
| ``required`` | Must run in container; fail if unavailable |
| ``preferred`` | Try container, fall back to host |
| ``specific`` | Target a named container resource |
Based on issue #879 — Tool-level execution environment preferences.
"""
from __future__ import annotations
from enum import StrEnum
from typing import Self
from pydantic import BaseModel, ConfigDict, Field, model_validator
__all__ = [
"EnvironmentPreferenceMode",
"ExecutionEnvironmentPreference",
]
class EnvironmentPreferenceMode(StrEnum):
"""Mode for tool execution environment preference.
Controls how the tool runner interprets the tool's environment
preference during execution routing.
"""
NONE = "none"
REQUIRED = "required"
PREFERRED = "preferred"
SPECIFIC = "specific"
class ExecutionEnvironmentPreference(BaseModel):
"""Intrinsic execution environment preference for a tool.
Set at tool definition time (e.g. in YAML config) and consulted
by ``ToolRunner.execute()`` before delegating to the environment
resolver.
When ``mode`` is ``specific``, ``target_resource`` must name the
container resource to target (e.g. ``local/api-dev``).
"""
model_config = ConfigDict(frozen=True, str_strip_whitespace=True)
mode: EnvironmentPreferenceMode = Field(
default=EnvironmentPreferenceMode.NONE,
description="Execution environment preference mode",
)
target_resource: str | None = Field(
default=None,
description="Named container resource for 'specific' mode",
)
@model_validator(mode="after")
def _validate_target(self) -> Self:
"""Enforce target_resource rules based on mode."""
if self.mode == EnvironmentPreferenceMode.SPECIFIC and not self.target_resource:
raise ValueError("'specific' mode requires target_resource")
if (
self.mode != EnvironmentPreferenceMode.SPECIFIC
and self.target_resource is not None
):
raise ValueError("target_resource only valid with 'specific' mode")
return self
@@ -68,6 +68,11 @@ from typing import Any
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
ExecutionEnvironmentPreference,
)
# ---------------------------------------------------------------------------
# Regex patterns
# ---------------------------------------------------------------------------
@@ -366,6 +371,12 @@ class Tool(BaseModel):
# Execution
timeout: int = Field(300, ge=1, description="Execution timeout in seconds")
# Execution environment preference (intrinsic to tool definition)
execution_environment: ExecutionEnvironmentPreference = Field(
default_factory=ExecutionEnvironmentPreference,
description="Intrinsic execution environment preference",
)
# -- Name format validation ------------------------------------------------
@field_validator("name")
@@ -439,6 +450,14 @@ class Tool(BaseModel):
lc_data = config.get("lifecycle")
lifecycle = ToolLifecycle.model_validate(lc_data) if lc_data else None
# Build execution environment preference
ee_data = config.get("execution_environment")
exec_env = (
ExecutionEnvironmentPreference.model_validate(ee_data)
if ee_data
else ExecutionEnvironmentPreference()
)
return cls(
name=config["name"],
description=config["description"],
@@ -454,6 +473,7 @@ class Tool(BaseModel):
resource_slots=resource_slots,
lifecycle=lifecycle,
timeout=config.get("timeout", 300),
execution_environment=exec_env,
)
def as_cli_dict(self) -> OrderedDict[str, Any]:
@@ -510,6 +530,13 @@ class Tool(BaseModel):
result["timeout"] = self.timeout
ee = self.execution_environment
if ee.mode != EnvironmentPreferenceMode.NONE:
ee_dict: dict[str, str] = {"mode": ee.mode.value}
if ee.target_resource:
ee_dict["target_resource"] = ee.target_resource
result["execution_environment"] = ee_dict
return result
model_config = ConfigDict(
+91 -7
View File
@@ -27,11 +27,18 @@ import threading
import time
from typing import TYPE_CHECKING, Any
import structlog
from cleveragents.domain.models.core.execution_environment_preference import (
EnvironmentPreferenceMode,
)
from cleveragents.domain.models.core.plan import ExecutionEnvironment
from cleveragents.tool.container_executor import ContainerToolExecutor
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runtime import ToolError, ToolResult, ToolSpec
logger = structlog.get_logger(__name__)
if TYPE_CHECKING:
from cleveragents.application.services.execution_environment_resolver import (
ExecutionEnvironmentResolver,
@@ -157,22 +164,99 @@ class ToolRunner:
ContainerUnavailableError,
)
# Consult tool-level execution environment preference.
# Caller override takes precedence over tool preference —
# PREFERRED only upgrades when no caller override is set.
pref = spec.execution_environment
effective_tool_env = tool_env
if pref.mode == EnvironmentPreferenceMode.SPECIFIC:
# Route to container; the named target_resource identifies
# which container to use.
effective_tool_env = "container"
# TODO: pass pref.target_resource to the resolver /
# container executor so the specific named resource is
# selected. The current resolver and ContainerToolExecutor
# APIs do not accept a target_resource parameter.
logger.debug(
"tool_env_preference.specific_override",
tool=tool_name,
target_resource=pref.target_resource,
)
elif pref.mode == EnvironmentPreferenceMode.REQUIRED:
# Force container execution
effective_tool_env = "container"
logger.info(
"tool_env_preference.required_container",
tool=tool_name,
)
elif (
pref.mode == EnvironmentPreferenceMode.PREFERRED
and effective_tool_env is None
):
# Try container if available, otherwise fall back
effective_tool_env = "container"
logger.debug(
"tool_env_preference.preferred_upgrade",
tool=tool_name,
)
# Resolve execution environment
try:
env = self._env_resolver.resolve_and_validate(
linked_resource_types=linked_resource_types or [],
project_name=project_name,
tool_env=tool_env,
tool_env=effective_tool_env,
plan_env=plan_env,
project_env=project_env,
)
except ContainerUnavailableError as exc:
return ToolResult(
success=False,
output={},
error=str(exc),
duration_ms=0.0,
)
if pref.mode == EnvironmentPreferenceMode.REQUIRED:
logger.info(
"tool_env_preference.required_failed",
tool=tool_name,
error=str(exc),
)
return ToolResult(
success=False,
output={},
error=(
f"Tool '{tool_name}' requires container execution "
f"but no container is available: {exc}"
),
duration_ms=0.0,
)
if pref.mode == EnvironmentPreferenceMode.PREFERRED:
# Fall back to host
logger.info(
"tool_env_preference.preferred_fallback_to_host",
tool=tool_name,
reason=str(exc),
)
env = ExecutionEnvironment.HOST
elif pref.mode == EnvironmentPreferenceMode.SPECIFIC:
logger.info(
"tool_env_preference.specific_target_unavailable",
tool=tool_name,
target_resource=pref.target_resource,
error=str(exc),
)
return ToolResult(
success=False,
output={},
error=(
f"Target container resource "
f"'{pref.target_resource}' unavailable: {exc}"
),
duration_ms=0.0,
)
else:
return ToolResult(
success=False,
output={},
error=str(exc),
duration_ms=0.0,
)
except Exception as exc:
return ToolResult(
success=False,
+7
View File
@@ -33,6 +33,9 @@ from typing import Any
from pydantic import BaseModel, ConfigDict, Field, model_validator
from cleveragents.domain.models.core.execution_environment_preference import (
ExecutionEnvironmentPreference,
)
from cleveragents.domain.models.core.tool import ToolCapability
@@ -87,6 +90,10 @@ class ToolSpec(BaseModel):
default_factory=dict,
description=("Provenance details for the tool (e.g. path, server URI)"),
)
execution_environment: ExecutionEnvironmentPreference = Field(
default_factory=ExecutionEnvironmentPreference,
description="Intrinsic execution environment preference for the tool",
)
model_config = ConfigDict(
str_strip_whitespace=True,
+6
View File
@@ -594,6 +594,12 @@ _MAX_TERMINAL_TRACKERS # noqa: B018, F821 — used in devcontainer_lifecycle_st
host_workspace_path # noqa: B018, F821 — F15 fix: new tracker field for host-side workspace path
_ACTIVATABLE_STATES # noqa: B018, F821 — F16 fix: class attribute on DevcontainerHandler
# Tool execution environment preferences — public API (#879)
EnvironmentPreferenceMode # noqa: B018, F821
ExecutionEnvironmentPreference # noqa: B018, F821
target_resource # noqa: B018, F821
_validate_target # noqa: B018, F821
# Execution environment routing — public API (#512)
ExecutionEnvironment # noqa: B018, F821
ExecutionEnvironmentResolver # noqa: B018, F821