cleveractors-core/features/steps/credential_composite_steps.py

"""Step definitions for composite agent credential threading.

Extracted from ``credential_injection_steps.py`` to keep that file
under 500 lines (CONTRIBUTING.md §General Principles).

Covers the ``components``-based composite path and the legacy
``"agents": [...]``-style composite path.
"""

from __future__ import annotations

from typing import Any

from behave import given, then, when
from features.mocks.credential_helpers import make_template_renderer

from cleveractors.agents.factory import AgentFactory
from cleveractors.agents.llm import LLMAgent

# ---------------------------------------------------------------------------
# Composite agent credential threading (m8)
# ---------------------------------------------------------------------------


@given("a composite agent config with a nested LLM component for credential test")
def step_composite_agent_config_for_creds(context: Any) -> None:
    """Set up a composite agent config with a nested LLM component."""
    context.actor_config = {
        "agents": {
            "my_composite": {
                "type": "composite",
                "config": {
                    "components": {
                        "agents": {
                            "nested_llm": {
                                "type": "llm",
                                "config": {
                                    "provider": "openai",
                                    "model": "gpt-3.5-turbo",
                                },
                            }
                        }
                    }
                },
            }
        }
    }
    context.template_renderer = make_template_renderer()
    context.credentials = {"openai": {"api_key": "composite-injected-key"}}
    context.factory = AgentFactory(
        config=context.actor_config,
        template_renderer=context.template_renderer,
        credentials=context.credentials,
    )


@when("I call create_agent for the composite agent in credential test")
def step_create_composite_agent_for_creds(context: Any) -> None:
    """Create the composite agent via the factory."""
    context.created_agent = context.factory.create_agent("my_composite")


@then("the nested LLM agent within the composite should have the credentials dict set")
def step_nested_llm_has_credentials(context: Any) -> None:
    """Verify the nested LLM agent received the provider-specific credentials slice."""
    composite = context.created_agent
    assert hasattr(composite, "agents"), (
        "Composite agent should have 'agents' attribute"
    )
    agents = composite.agents
    assert "nested_llm" in agents, (
        f"Expected 'nested_llm' in composite agents, got {list(agents.keys())}"
    )
    nested = agents["nested_llm"]
    assert isinstance(nested, LLMAgent), (
        f"Expected LLMAgent, got {type(nested).__name__}"
    )
    # Factory extracts provider-specific slice: credentials.get("openai")
    assert nested._credentials == {"api_key": "composite-injected-key"}, (
        f"Expected _credentials {{'api_key': 'composite-injected-key'!r}}, "
        f"got {nested._credentials!r}"
    )


# ---------------------------------------------------------------------------
# Legacy composite agent credential threading (m2)
# ---------------------------------------------------------------------------


@given("a legacy composite agent config with nested agents list for credential test")
def step_legacy_composite_agent_config_for_creds(context: Any) -> None:
    """Set up a legacy composite agent config using the ``"agents": [...]`` format.

    This tests the fix for the legacy composite path where the cache is bypassed
    when credentials are set (factory.py:227-233).
    """
    context.actor_config = {
        "agents": {
            "my_legacy_composite": {
                "type": "composite",
                "config": {
                    "agents": ["nested_llm"],
                },
            },
            "nested_llm": {
                "type": "llm",
                "config": {
                    "provider": "openai",
                    "model": "gpt-3.5-turbo",
                },
            },
        }
    }
    context.template_renderer = make_template_renderer()
    context.credentials = {"openai": {"api_key": "legacy-composite-key"}}
    context.factory = AgentFactory(
        config=context.actor_config,
        template_renderer=context.template_renderer,
        credentials=context.credentials,
    )


@when("I call create_agent for the legacy composite agent in credential test")
def step_create_legacy_composite_agent_for_creds(context: Any) -> None:
    """Create the legacy composite agent via the factory."""
    context.created_agent = context.factory.create_agent("my_legacy_composite")


@then(
    "the nested LLM agent within the legacy composite should have the credentials dict set"
)
def step_nested_legacy_llm_has_credentials(context: Any) -> None:
    """Verify the nested LLM agent (legacy path) received the factory credentials."""
    composite = context.created_agent
    assert hasattr(composite, "agents"), (
        "Composite agent should have 'agents' attribute"
    )
    agents = composite.agents
    assert "nested_llm" in agents, (
        f"Expected 'nested_llm' in legacy composite agents, got {list(agents.keys())}"
    )
    nested = agents["nested_llm"]
    assert isinstance(nested, LLMAgent), (
        f"Expected LLMAgent, got {type(nested).__name__}"
    )
    # Factory extracts provider-specific slice: credentials.get("openai")
    assert nested._credentials == {"api_key": "legacy-composite-key"}, (
        f"Expected _credentials {{'api_key': 'legacy-composite-key'!r}}, "
        f"got {nested._credentials!r}"
    )


# ---------------------------------------------------------------------------
# m6: legacy composite path without credentials
# ---------------------------------------------------------------------------


@given("a legacy composite agent config with nested agents list without credentials")
def step_legacy_composite_no_credentials(context: Any) -> None:
    """Set up a legacy composite agent config without a credentials dict.

    Tests that the legacy composite path works when credentials is None
    (no credential injection needed).
    """
    context.actor_config = {
        "agents": {
            "my_legacy_composite": {
                "type": "composite",
                "config": {
                    "agents": ["nested_llm"],
                },
            },
            "nested_llm": {
                "type": "llm",
                "config": {
                    "provider": "openai",
                    "model": "gpt-3.5-turbo",
                },
            },
        }
    }
    context.template_renderer = make_template_renderer()
    context.factory = AgentFactory(
        config=context.actor_config,
        template_renderer=context.template_renderer,
    )


@then(
    "the nested LLM agent within the legacy composite should have credentials "
    "equal to None"
)
def step_nested_legacy_llm_credentials_none(context: Any) -> None:
    """Verify the nested LLM agent (legacy path) has credentials=None."""
    composite = context.created_agent
    agents = composite.agents
    nested = agents["nested_llm"]
    assert nested._credentials is None, (
        f"Expected _credentials to be None, got {type(nested._credentials).__name__}"
    )