cleveractors-core/features/steps/smoke_steps.py

"""Behave step definitions for the library smoke feature."""

from __future__ import annotations

import asyncio
from typing import Any

from behave import given, then, when

from cleveractors.actor.compiler import compile_actor
from cleveractors.actor.schema import ActorConfigSchema
from cleveractors.agents.base import Agent
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
from cleveractors.langgraph.state import GraphState
from cleveractors.ports.provider_registry import ProviderRegistryPort
from cleveractors.templates.secure_renderer import SecureTemplateRenderer


def _minimal_graph_actor() -> dict[str, Any]:
    return {
        "name": "local/smoke",
        "type": "graph",
        "provider": "openai",
        "model": "gpt-4o-mini",
        "description": "smoke test actor",
        "route": {
            "entry_node": "start",
            "exit_nodes": ["end"],
            "nodes": [
                {
                    "id": "start",
                    "name": "start",
                    "type": "agent",
                    "description": "entry",
                },
                {
                    "id": "end",
                    "name": "end",
                    "type": "agent",
                    "description": "exit",
                },
            ],
            "edges": [
                {"from_node": "start", "to_node": "end"},
            ],
        },
    }


@given("a v3 actor configuration with a start node and an exit node")
def given_minimal_actor(context: Any) -> None:
    context.actor_raw = _minimal_graph_actor()


@given("a v3 actor configuration missing the provider field")
def given_actor_missing_provider(context: Any) -> None:
    raw = _minimal_graph_actor()
    raw.pop("provider")
    context.actor_raw = raw


@given('a simple template "{template}"')
def given_template(context: Any, template: str) -> None:
    context.template = template
    context.context_data = {}


@given('a context with greeting "{greeting}" and name "{name}"')
def given_context(context: Any, greeting: str, name: str) -> None:
    context.context_data = {"greeting": greeting, "name": name}


@when("the actor configuration is compiled")
def when_compiled(context: Any) -> None:
    context.compile_exc = None
    try:
        # NB: ``context.config`` is a reserved attribute on Behave's
        # Context; using ``actor_config`` keeps us out of its way.
        context.actor_config = ActorConfigSchema.model_validate(context.actor_raw)
        context.compiled = compile_actor(context.actor_config)
    except Exception as exc:
        context.compile_exc = exc


@when("the actor configuration is validated")
def when_validated(context: Any) -> None:
    context.validation_exc = None
    try:
        ActorConfigSchema.model_validate(context.actor_raw)
    except Exception as exc:
        context.validation_exc = exc


@when("the template is rendered")
def when_rendered(context: Any) -> None:
    renderer = SecureTemplateRenderer()
    context.rendered = renderer.render(context.template, context.context_data)


@then("the compile succeeds without errors")
def then_compile_ok(context: Any) -> None:
    assert context.compile_exc is None, f"unexpected: {context.compile_exc!r}"


@then("the compiled metadata lists both node ids")
def then_node_ids(context: Any) -> None:
    ids = set(context.compiled.metadata.node_ids)
    assert ids == {"start", "end"}, f"got {ids}"


@then("a ValidationError is raised")
def then_validation_error(context: Any) -> None:
    assert context.validation_exc is not None, "expected an exception"
    name = type(context.validation_exc).__name__
    assert "Validation" in name, f"expected validation error, got {name}"


@then('the result is "{expected}"')
def then_result_equals(context: Any, expected: str) -> None:
    assert context.rendered == expected, (
        f"expected {expected!r}, got {context.rendered!r}"
    )


# ---------------------------------------------------------------------------
# ProviderRegistryPort scenario helpers
# ---------------------------------------------------------------------------


class _StubAgent(Agent):
    """Minimal Agent stub for testing ProviderRegistryPort resolution."""

    def __init__(self) -> None:
        super().__init__(name="stub-agent")

    async def process_message(
        self, message: Any, context: dict[str, Any] | None = None
    ) -> Any:
        return "stub response"

    def get_capabilities(self) -> list[str]:
        return ["stub"]


class _StubProviderRegistry:
    """Minimal ProviderRegistryPort implementation for testing."""

    def __init__(self, provider: str, model: str) -> None:
        self._provider = provider
        self._model = model
        self.calls: list[tuple[str, str]] = []

    def get(self, provider: str, model: str) -> Agent | None:
        self.calls.append((provider, model))
        if provider == self._provider and model == self._model:
            return _StubAgent()
        return None


@given(
    'a compiled graph actor with an agent node using provider "{provider}"'
    ' and model "{model}" and agent name "{agent_name}"'
)
def given_compiled_actor_with_agent_name(
    context: Any, provider: str, model: str, agent_name: str
) -> None:
    context.node_provider = provider
    context.node_model = model
    context.agent_name = agent_name
    context.node_config = NodeConfig(
        name="planner",
        type=NodeType.AGENT,
        agent=agent_name,
        metadata={"provider": provider, "model": model},
    )


@given(
    'a compiled graph actor with an agent node using provider "{provider}"'
    ' and model "{model}"'
)
def given_compiled_actor_with_agent_node(
    context: Any, provider: str, model: str
) -> None:
    context.node_provider = provider
    context.node_model = model
    # Build a NodeConfig that simulates what the compiler produces for an
    # AGENT node — provider and model are in metadata, agent name is None
    # (the common case when no explicit agent reference is configured).
    context.node_config = NodeConfig(
        name="planner",
        type=NodeType.AGENT,
        agent=None,
        metadata={"provider": provider, "model": model},
    )


@given(
    'a provider registry stub that returns a stub agent for "{provider}" and "{model}"'
)
def given_stub_provider_registry(context: Any, provider: str, model: str) -> None:
    context.provider_registry = _StubProviderRegistry(provider, model)


@when("the agent node is executed without a pre-resolved agents dict")
def when_node_executed(context: Any) -> None:
    node = Node(
        config=context.node_config,
        agents={},  # empty — forces ProviderRegistryPort lookup
        provider_registry=context.provider_registry,
    )
    state = GraphState(
        messages=[{"role": "user", "content": "hello"}],
        metadata={"current_message": "hello"},
    )
    context.node_result = asyncio.run(node.execute(state))


@then('the provider registry was called with provider "{provider}" and model "{model}"')
def then_registry_called(context: Any, provider: str, model: str) -> None:
    calls = context.provider_registry.calls
    assert (provider, model) in calls, (
        f"expected registry.get({provider!r}, {model!r}) to be called; "
        f"actual calls: {calls}"
    )


@then("the execution result contains an assistant message")
def then_has_assistant_message(context: Any) -> None:
    messages = context.node_result.get("messages", [])
    assistant_msgs = [m for m in messages if m.get("role") == "assistant"]
    assert assistant_msgs, f"no assistant message in result: {context.node_result}"


@then('the assistant message content is "{expected_content}"')
def then_assistant_message_content(context: Any, expected_content: str) -> None:
    messages = context.node_result.get("messages", [])
    assistant_msgs = [m for m in messages if m.get("role") == "assistant"]
    assert assistant_msgs, f"no assistant message in result: {context.node_result}"
    actual = assistant_msgs[0].get("content", "")
    assert actual == expected_content, f"expected {expected_content!r}, got {actual!r}"


@when("the agent node is executed without a provider registry")
def when_node_executed_no_registry(context: Any) -> None:
    node = Node(
        config=context.node_config,
        agents={},
        provider_registry=None,
    )
    state = GraphState(
        messages=[{"role": "user", "content": "hello"}],
        metadata={"current_message": "hello"},
    )
    context.node_result = asyncio.run(node.execute(state))


@when('the agent node is executed with a pre-resolved agent named "{agent_name}"')
def when_node_executed_with_agents_dict(context: Any, agent_name: str) -> None:
    node = Node(
        config=context.node_config,
        agents={agent_name: _StubAgent()},
        provider_registry=context.provider_registry,
    )
    state = GraphState(
        messages=[{"role": "user", "content": "hello"}],
        metadata={"current_message": "hello"},
    )
    context.node_result = asyncio.run(node.execute(state))


@then("the pre-resolved agents dict was used instead of the provider registry")
def then_agents_dict_precedence(context: Any) -> None:
    # The provider registry should not have been called — the agents dict
    # took priority.
    calls = context.provider_registry.calls
    assert not calls, (
        f"provider registry should not have been called; got calls: {calls}"
    )
    messages = context.node_result.get("messages", [])
    assistant_msgs = [m for m in messages if m.get("role") == "assistant"]
    assert assistant_msgs, f"no assistant message in result: {context.node_result}"
    actual = assistant_msgs[0].get("content", "")
    assert actual == "stub response", f"expected 'stub response', got {actual!r}"


@then("the stub registry satisfies ProviderRegistryPort")
def then_stub_satisfies_protocol(context: Any) -> None:
    assert isinstance(context.provider_registry, ProviderRegistryPort), (
        "stub registry must satisfy ProviderRegistryPort"
    )