CoreRasurae
227e2feece
CI / build (pull_request) Successful in 45s
CI / quality (pull_request) Successful in 58s
CI / lint (pull_request) Successful in 1m15s
CI / typecheck (pull_request) Successful in 1m15s
CI / security (pull_request) Successful in 1m15s
CI / integration_tests (pull_request) Successful in 1m24s
CI / unit_tests (pull_request) Successful in 3m25s
CI / coverage (pull_request) Successful in 3m10s
CI / status-check (pull_request) Successful in 6s
CI / lint (push) Successful in 42s
CI / quality (push) Successful in 48s
CI / typecheck (push) Successful in 51s
CI / security (push) Successful in 49s
CI / build (push) Successful in 47s
CI / integration_tests (push) Successful in 1m10s
CI / unit_tests (push) Successful in 3m9s
CI / coverage (push) Successful in 3m7s
CI / status-check (push) Successful in 3s
Implements the complete LLM agent tool-calling pipeline — the tool-calling path was broken in multiple ways: tools from agent config were not passed to the LLM, tool execution was single-pass (the model could not see tool results or make follow-up calls), tool errors were discarded, and shell/ python_exec tools were never registered. Key changes: - Multi-turn tool-call loop: passes tools to the LLM via ainvoke(tools=...) and re-invokes after each batch of ToolResults, with a configurable round limit (tool_max_rounds config / TOOL_MAX_ROUNDS env var, default 20). - Tool schema module (llm_tools.py): normalize_tool_entry() converts string tool names and config dicts to OpenAI function-calling format with full parameter schemas and LLM-facing guidance (max_chars for file_read, sandbox builtins for python_exec, etc.). - file_read enhancements: directory listing with file sizes and type indicators, max_chars truncation to prevent context overflow, shell command detection with redirect to shell tool, file-not-found recovery hints with parent directory listing. - python_exec sandbox: stdout capture so print() produces visible output, NameError guidance directing LLM to file_read/file_write instead of using open(). - shell/python_exec tool registration in builtin_tools when allow_shell / exec_python is enabled. - create_subprocess_shell for shell commands (was create_subprocess_exec, which blocks pipes/redirections/chains). - Tool error propagation: ExecutionError/ConfigurationError details included in ToolMessages for LLM self-correction. - Stuck-model recovery: injects synthesizing prompt when model exhausts tool rounds without producing content. - LLM provider error details preserved in ExecutionError messages instead of generic "LLM processing failed". - Empty message filtering in _prepare_conversation_history() prevents whitespace-only messages from polluting downstream conversation history in multi-agent pipelines. Closes #59
501 lines
20 KiB
Python
501 lines
20 KiB
Python
"""Robot Framework keyword library for email categorization actor graph tests.
|
|
|
|
Provides keywords for loading YAML configs, creating executors,
|
|
executing graph actors, and validating results against the email
|
|
categorization use case.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
from cleveractors.config_utils import merge_configs
|
|
from cleveractors.core.exceptions import (
|
|
AgentCreationError,
|
|
ConfigurationError,
|
|
ExecutionError,
|
|
)
|
|
from cleveractors.result import ActorResult
|
|
from cleveractors.runtime import create_executor
|
|
|
|
_FIXTURES = Path(__file__).resolve().parent.parent / "tests" / "fixtures"
|
|
|
|
|
|
class EmailGraphLib:
|
|
"""Keyword library for email categorization graph integration tests.
|
|
|
|
Wraps the cleveractors runtime API (create_executor, execute) with
|
|
keywords specific to the email actor graph scenario.
|
|
"""
|
|
|
|
ROBOT_LIBRARY_SCOPE = "TEST SUITE"
|
|
|
|
def __init__(self) -> None:
|
|
self._executor: Any = None
|
|
self._last_result: ActorResult | None = None
|
|
|
|
# -- config loading -------------------------------------------------------
|
|
|
|
def load_email_graph_config(self, graph_name: str) -> Any:
|
|
"""Load an email graph YAML config fixture by name.
|
|
|
|
Valid names: ``local_graph``, ``remote_graph``, ``mixed_graph``.
|
|
Returns the parsed YAML as a dict.
|
|
"""
|
|
path = _FIXTURES / "email_graph" / f"{graph_name}.yaml"
|
|
with open(path, encoding="utf-8") as fh:
|
|
return yaml.safe_load(fh)
|
|
|
|
def load_config_file(self, config_path: str) -> Any:
|
|
"""Load a YAML config file from ``tests/fixtures/``."""
|
|
path = _FIXTURES / config_path
|
|
with open(path, encoding="utf-8") as fh:
|
|
return yaml.safe_load(fh)
|
|
|
|
# -- executor creation ----------------------------------------------------
|
|
|
|
def create_executor_for_email_graph(self, graph_name: str) -> None:
|
|
"""Create an Executor from the named email graph config."""
|
|
config = self.load_email_graph_config(graph_name)
|
|
self._executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
def create_executor_from_config_file(self, config_path: str) -> None:
|
|
"""Create an Executor from a fixture config file."""
|
|
config = self.load_config_file(config_path)
|
|
self._executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
def create_executor_from_merged_configs(
|
|
self, graph_name: str, overlay_path: str
|
|
) -> None:
|
|
"""Create an Executor from a merged graph + overlay config."""
|
|
graph = self.load_email_graph_config(graph_name)
|
|
overlay = self.load_config_file(overlay_path)
|
|
merged = merge_configs(graph, overlay)
|
|
self._executor = create_executor(
|
|
config_dict=merged,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
def create_executor_with_limits(
|
|
self, graph_name: str, max_depth: int | None = None
|
|
) -> None:
|
|
"""Create an Executor with execution limits."""
|
|
config = self.load_email_graph_config(graph_name)
|
|
limits: dict[str, Any] = {}
|
|
if max_depth is not None:
|
|
limits["max_depth"] = max_depth
|
|
self._executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits=limits,
|
|
pricing={},
|
|
)
|
|
|
|
def create_executor_from_dict(self, config_dict_str: str) -> None:
|
|
"""Create an Executor from a JSON string dict."""
|
|
config = json.loads(config_dict_str)
|
|
self._executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
def create_executor_from_config_dict(self, config_dict: dict[str, Any]) -> None:
|
|
"""Create an Executor directly from a Python dict."""
|
|
self._executor = create_executor(
|
|
config_dict=config_dict,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
# -- execution ------------------------------------------------------------
|
|
|
|
async def _execute_async(self, message: str) -> ActorResult:
|
|
result = await self._executor.execute(message)
|
|
self._last_result = result
|
|
return result
|
|
|
|
def execute_email_graph(self, email_json: str) -> None:
|
|
"""Execute the email graph with an email JSON input."""
|
|
self._last_result = asyncio.run(self._execute_async(email_json))
|
|
|
|
async def _execute_stream_async(self, message: str) -> str:
|
|
chunks: list[str] = []
|
|
async for token in self._executor.execute_stream(message):
|
|
chunks.append(token)
|
|
return "".join(chunks)
|
|
|
|
def execute_email_graph_stream(self, email_json: str) -> None:
|
|
"""Execute via streaming path and collect tokens."""
|
|
response = asyncio.run(self._execute_stream_async(email_json))
|
|
self._last_result = self._executor.last_result
|
|
|
|
# -- result validation ----------------------------------------------------
|
|
|
|
def result_is_valid_actor_result(self) -> None:
|
|
"""Assert last_result is an ActorResult instance."""
|
|
assert isinstance(self._last_result, ActorResult), (
|
|
f"Expected ActorResult, got {type(self._last_result)}"
|
|
)
|
|
|
|
def result_response_contains(self, text: str) -> None:
|
|
"""Assert the response string contains the given text."""
|
|
assert self._last_result is not None
|
|
assert text in self._last_result.response, (
|
|
f"Expected '{text}' in response, got: {self._last_result.response[:200]}"
|
|
)
|
|
|
|
def result_response_contains_json_key(self, key: str) -> None:
|
|
"""Assert the response contains JSON with the given top-level key."""
|
|
assert self._last_result is not None
|
|
try:
|
|
parsed = json.loads(self._last_result.response)
|
|
except json.JSONDecodeError:
|
|
raise AssertionError(
|
|
f"Response is not valid JSON: {self._last_result.response[:200]}"
|
|
)
|
|
assert key in parsed, (
|
|
f"Key '{key}' not found in response JSON. Keys: {list(parsed.keys())}"
|
|
)
|
|
|
|
def result_json_key_equals(self, key: str, expected: str) -> None:
|
|
"""Assert a JSON key in the response has an expected value."""
|
|
assert self._last_result is not None
|
|
parsed = json.loads(self._last_result.response)
|
|
actual = parsed.get(key)
|
|
assert str(actual) == expected, (
|
|
f"Key '{key}': expected '{expected}', got '{actual}'"
|
|
)
|
|
|
|
def result_has_nodes(self) -> None:
|
|
"""Assert the result has at least one NodeUsage entry."""
|
|
assert self._last_result is not None
|
|
assert len(self._last_result.nodes) > 0, "Expected at least one node in result"
|
|
|
|
def result_node_count_equals(self, count: int) -> None:
|
|
"""Assert result has exactly count NodeUsage entries."""
|
|
assert self._last_result is not None
|
|
actual = len(self._last_result.nodes)
|
|
assert actual == count, f"Expected {count} nodes, got {actual}"
|
|
|
|
def result_has_state(self) -> None:
|
|
"""Assert the result has a non-None state field."""
|
|
assert self._last_result is not None
|
|
assert self._last_result.state is not None, "Expected non-None state"
|
|
|
|
def result_state_contains_key(self, key: str) -> None:
|
|
"""Assert the result state dict contains the given key."""
|
|
assert self._last_result is not None
|
|
assert self._last_result.state is not None
|
|
assert key in self._last_result.state, (
|
|
f"Key '{key}' not in state: {list(self._last_result.state.keys())}"
|
|
)
|
|
|
|
# -- error validation (negative tests) ------------------------------------
|
|
|
|
def executor_creation_should_fail(self, graph_name: str) -> None:
|
|
"""Assert creating an executor fails with ConfigurationError."""
|
|
try:
|
|
self.create_executor_for_email_graph(graph_name)
|
|
raise AssertionError("Expected ConfigurationError but no exception raised")
|
|
except (ConfigurationError, AgentCreationError):
|
|
pass
|
|
|
|
def execution_should_raise(self, email_json: str, error_type: str) -> None:
|
|
"""Assert executing with the given input raises a specific error."""
|
|
try:
|
|
asyncio.run(self._execute_async(email_json))
|
|
raise AssertionError(f"Expected {error_type} but no exception raised")
|
|
except (ConfigurationError, AgentCreationError, ExecutionError) as e:
|
|
actual_type = type(e).__name__
|
|
if error_type not in actual_type:
|
|
raise AssertionError(
|
|
f"Expected error type containing '{error_type}', got '{actual_type}: {e}'"
|
|
)
|
|
|
|
def executor_from_dict_should_fail(self, config_dict_str: str) -> None:
|
|
"""Assert creating executor from dict raises ConfigurationError."""
|
|
try:
|
|
self.create_executor_from_dict(config_dict_str)
|
|
raise AssertionError("Expected ConfigurationError but no exception raised")
|
|
except (ConfigurationError, AgentCreationError):
|
|
pass
|
|
|
|
def executor_from_python_dict_should_fail(
|
|
self, config_dict: dict[str, Any]
|
|
) -> None:
|
|
"""Assert creating executor from a Python dict raises ConfigurationError."""
|
|
try:
|
|
self.create_executor_from_config_dict(config_dict)
|
|
raise AssertionError("Expected ConfigurationError but no exception raised")
|
|
except (ConfigurationError, AgentCreationError):
|
|
pass
|
|
|
|
def executor_non_dict_should_fail(self, non_dict_value: str) -> None:
|
|
"""Assert non-dict value raises ConfigurationError."""
|
|
try:
|
|
create_executor(
|
|
config_dict=non_dict_value,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
raise AssertionError("Expected ConfigurationError but no exception raised")
|
|
except (ConfigurationError, AgentCreationError):
|
|
pass
|
|
|
|
# -- negative test configs ------------------------------------------------
|
|
|
|
def load_invalid_graph_no_routes(self) -> dict[str, Any]:
|
|
"""Return a config dict missing the routes key."""
|
|
return {"actors": {}}
|
|
|
|
def load_invalid_graph_no_actors(self) -> dict[str, Any]:
|
|
"""Return a config dict missing the actors key."""
|
|
return {"routes": {"main": {"nodes": {}, "edges": []}}}
|
|
|
|
def load_invalid_graph_duplicate_nodes(self) -> dict[str, Any]:
|
|
"""Return a graph config with duplicate node IDs."""
|
|
return {
|
|
"routes": {
|
|
"main": {
|
|
"entry_point": "start",
|
|
"nodes": {
|
|
"n1": {"id": "start", "type": "agent", "agent": "echo"},
|
|
"n2": {"id": "start", "type": "agent", "agent": "echo"},
|
|
},
|
|
"edges": [],
|
|
}
|
|
},
|
|
"actors": {},
|
|
}
|
|
|
|
def load_invalid_graph_missing_agent_ref(self) -> dict[str, Any]:
|
|
"""Return a graph config referencing a nonexistent agent."""
|
|
return {
|
|
"routes": {
|
|
"main": {
|
|
"entry_point": "start",
|
|
"nodes": {
|
|
"n1": {
|
|
"id": "start",
|
|
"type": "agent",
|
|
"agent": "nonexistent_agent",
|
|
}
|
|
},
|
|
"edges": [],
|
|
}
|
|
},
|
|
"actors": {"echo": {"type": "tool", "config": {}}},
|
|
}
|
|
|
|
def load_invalid_config_nondict(self) -> str:
|
|
"""Return a non-dict config for type validation."""
|
|
return "not-a-dict"
|
|
|
|
# -- merge_configs keywords -----------------------------------------------
|
|
|
|
def merge_email_graph_configs(self, graph_name: str, overlay_name: str) -> Any:
|
|
"""Merge two email graph configs and return the result."""
|
|
base = self.load_email_graph_config(graph_name)
|
|
overlay = self.load_email_graph_config(overlay_name)
|
|
return merge_configs(base, overlay)
|
|
|
|
def merge_email_graph_configs_invalid(self, graph_name: str) -> Any:
|
|
"""Attempt merge_configs with one valid and one non-existent name.
|
|
|
|
This tests that passing an invalid (non-existant) file to the
|
|
file loader raises FileNotFoundError before merge_configs is called."""
|
|
base = self.load_email_graph_config(graph_name)
|
|
return merge_configs(base, "not_a_valid_dict")
|
|
|
|
def merged_config_has_nested_key(self, merged: Any, *keys: str) -> None:
|
|
"""Assert nested keys exist in a merged config dict."""
|
|
current = merged
|
|
for key in keys:
|
|
assert isinstance(current, dict), (
|
|
f"Expected dict at key path, got {type(current)}"
|
|
)
|
|
assert key in current, (
|
|
f"Key '{key}' not found. Available: {list(current.keys())[:10]}"
|
|
)
|
|
current = current[key]
|
|
|
|
def merged_config_key_count(self, merged: Any, section: str, minimum: int) -> None:
|
|
"""Assert a config section has at least minimum keys."""
|
|
assert section in merged, f"Section '{section}' not in merged config"
|
|
actual = len(merged[section])
|
|
assert actual >= minimum, (
|
|
f"Expected >= {minimum} keys in '{section}', got {actual}"
|
|
)
|
|
|
|
# -- component-based graph assembly ---------------------------------------
|
|
|
|
def load_all_email_components(self) -> Any:
|
|
"""Load all 7 component YAML files and merge them into one config.
|
|
|
|
Returns the fully assembled graph config ready for Executor.
|
|
"""
|
|
base = _FIXTURES / "email_graph" / "components"
|
|
component_files = [
|
|
"email_categorizer.yaml",
|
|
"component_orders.yaml",
|
|
"personal_summarizer.yaml",
|
|
"business_email.yaml",
|
|
"supplier.yaml",
|
|
"client.yaml",
|
|
"graph_routes.yaml",
|
|
]
|
|
configs = []
|
|
for fname in component_files:
|
|
with open(base / fname, encoding="utf-8") as fh:
|
|
configs.append(yaml.safe_load(fh))
|
|
result: dict[str, Any] = {}
|
|
for cfg in configs:
|
|
result = merge_configs(result, cfg)
|
|
return result
|
|
|
|
def create_executor_from_components(self) -> None:
|
|
"""Create an Executor from the assembled component graph."""
|
|
config = self.load_all_email_components()
|
|
self._executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
def component_graph_actor_count_equals(self, expected: int) -> None:
|
|
"""Assert the assembled component graph has exactly expected actors."""
|
|
config = self.load_all_email_components()
|
|
actual = len(config.get("actors", {}))
|
|
assert actual == expected, f"Expected {expected} actors, got {actual}"
|
|
|
|
# -- namespaced component graph with reference resolver -------------------
|
|
|
|
def assemble_namespaced_graph(self) -> None:
|
|
"""Assemble the namespaced email graph by resolving all local: refs.
|
|
|
|
Uses LocalPackageStore + ReferenceResolver to resolve each
|
|
``agent_ref: local:...`` in the graph routes into fully inlined
|
|
actor configs, then creates an Executor from the assembled config.
|
|
"""
|
|
from cleveractors.registry.local_store import LocalPackageStore
|
|
from cleveractors.registry.resolver import ReferenceResolver
|
|
|
|
base_dir = _FIXTURES / "email_graph" / "namespaced"
|
|
store = LocalPackageStore(base_dir)
|
|
|
|
# Resolve each local: reference via the store
|
|
graph_path = base_dir / "email_graph.yaml"
|
|
with open(graph_path, encoding="utf-8") as fh:
|
|
graph_config = yaml.safe_load(fh)
|
|
|
|
resolver = ReferenceResolver(local_store=store)
|
|
|
|
actors: dict[str, Any] = {}
|
|
nodes = graph_config.get("routes", {}).get("main", {}).get("nodes", {})
|
|
for _node_id, node_def in nodes.items():
|
|
agent_ref = node_def.get("agent_ref", "")
|
|
if agent_ref.startswith("local:"):
|
|
ref_name = agent_ref[len("local:") :]
|
|
pkg = store.resolve_package(ref_name)
|
|
content = pkg.content
|
|
actor_name = content.get("name", ref_name)
|
|
actors[actor_name] = {
|
|
"type": content.get("type", "tool"),
|
|
"config": content.get("config", {}),
|
|
}
|
|
|
|
graph_config["actors"] = actors
|
|
for node_def in (
|
|
graph_config.get("routes", {}).get("main", {}).get("nodes", {}).values()
|
|
):
|
|
if "agent_ref" in node_def and node_def["agent_ref"].startswith("local:"):
|
|
ref_name = node_def["agent_ref"][len("local:") :]
|
|
pkg = store.resolve_package(ref_name)
|
|
node_def["agent"] = pkg.content.get("name", ref_name)
|
|
del node_def["agent_ref"]
|
|
|
|
self._executor = create_executor(
|
|
config_dict=graph_config,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
def namespaced_graph_actor_count_equals(self, expected: int) -> None:
|
|
"""Assert the namespaced graph resolves to exactly expected actors."""
|
|
base = _FIXTURES / "email_graph" / "namespaced"
|
|
with open(base / "email_graph.yaml", encoding="utf-8") as fh:
|
|
graph_config = yaml.safe_load(fh)
|
|
ref_count = 0
|
|
nodes = graph_config.get("routes", {}).get("main", {}).get("nodes", {})
|
|
for _nid, ndef in nodes.items():
|
|
if ndef.get("agent_ref", "").startswith("local:"):
|
|
ref_count += 1
|
|
assert ref_count == expected, (
|
|
f"Expected {expected} agent_ref references in graph, got {ref_count}"
|
|
)
|
|
|
|
# -- agent factory keywords -----------------------------------------------
|
|
|
|
def create_agents_from_config(self, graph_name: str) -> int:
|
|
"""Create all agents from the named graph config and return count."""
|
|
from cleveractors.agents.factory import AgentFactory
|
|
from cleveractors.templates.renderer import TemplateRenderer
|
|
|
|
config = self.load_email_graph_config(graph_name)
|
|
config.setdefault("agents", {})
|
|
config["agents"].update(config.get("actors", {}))
|
|
renderer = TemplateRenderer()
|
|
factory = AgentFactory(config=config, template_renderer=renderer)
|
|
agents = factory.create_agents_from_config()
|
|
return len(agents)
|
|
|
|
def factory_validates_configuration(self, graph_name: str) -> None:
|
|
"""Assert AgentFactory.validate_configuration succeeds."""
|
|
from cleveractors.agents.factory import AgentFactory
|
|
from cleveractors.templates.renderer import TemplateRenderer
|
|
|
|
config = self.load_email_graph_config(graph_name)
|
|
config.setdefault("agents", {})
|
|
config["agents"].update(config.get("actors", {}))
|
|
renderer = TemplateRenderer()
|
|
factory = AgentFactory(config=config, template_renderer=renderer)
|
|
factory.validate_configuration()
|
|
|
|
def factory_agent_has_metadata_key(
|
|
self, graph_name: str, agent_name: str, key: str
|
|
) -> None:
|
|
"""Assert an agent's metadata contains a key."""
|
|
from cleveractors.agents.factory import AgentFactory
|
|
from cleveractors.templates.renderer import TemplateRenderer
|
|
|
|
config = self.load_email_graph_config(graph_name)
|
|
config.setdefault("agents", {})
|
|
config["agents"].update(config.get("actors", {}))
|
|
renderer = TemplateRenderer()
|
|
factory = AgentFactory(config=config, template_renderer=renderer)
|
|
meta = factory.get_agent_metadata(agent_name)
|
|
assert key in meta, f"Key '{key}' not in metadata. Keys: {list(meta.keys())}"
|