CoreRasurae
a8f68b8262
CI / lint (pull_request) Successful in 50s
CI / typecheck (pull_request) Successful in 55s
CI / security (pull_request) Successful in 53s
CI / quality (pull_request) Successful in 38s
CI / unit_tests (pull_request) Successful in 3m15s
CI / integration_tests (pull_request) Successful in 1m14s
CI / build (pull_request) Successful in 39s
CI / coverage (pull_request) Successful in 3m16s
CI / status-check (pull_request) Successful in 3s
CI / lint (push) Successful in 37s
CI / typecheck (push) Successful in 56s
CI / security (push) Successful in 54s
CI / quality (push) Successful in 37s
CI / unit_tests (push) Successful in 3m15s
CI / integration_tests (push) Successful in 1m14s
CI / build (push) Successful in 39s
CI / coverage (push) Successful in 3m16s
CI / status-check (push) Successful in 3s
Enforces the USD budget (max_cost_usd) as a hard pre-flight gate before every node execution and at each main-agent ainvoke round inside the multi-turn tool loop, with pruning passes silently skipped when the budget is exhausted. Pre-flight gate (pure_graph.py): _check_budget_pre_flight() raises ExecutionError(kind='cost', reason='budget_exhausted') immediately before node execution in both _execute_from_node and all three branches of _stream_from_node. In-node budget gating (llm.py): _check_budget_before_invoke() gates every _retry_ainvoke() call inside _execute_tool_loop, process_message, and stream_message. _should_skip_pruning() skips pruning passes when budget exhausted. _accumulate_cost_after_invoke() computes USD cost from token counts using the pricing table and advances current_accumulated_cost so subsequent budget checks see the updated cost. ContextVar plumbing (retry.py, pure_graph.py): current_accumulated_cost, current_max_cost_usd, and current_pricing ContextVars carry per-node budget state from PureLangGraph into LLMAgent. Set before node.execute() via _set_budget_context_for_node, reset in finally blocks via _reset_budget_context. Reviewer fixes (rui.hu #80): - Move cost accumulation inside try block in terminal streaming branch to prevent ContextVar leak (was set after finally reset). - Standardize log precision to $%.6f across all budget messages. - Remove orphaned 'already exhausted' step (criterion unreachable). - Add dedicated streaming budget enforcement scenario. - Remove redundant = None default on _reset_budget_context pricing_token. ISSUES CLOSED: #76
5878 lines
220 KiB
Python
5878 lines
220 KiB
Python
"""Step definitions for execute_stream BDD tests.
|
|
|
|
Covers Executor.execute_stream() and related streaming infrastructure:
|
|
- LLMAgent.stream_message()
|
|
- Node.stream_agent()
|
|
- PureLangGraph.execute_stream()
|
|
- _execute_llm_stream() / _execute_graph_stream() dispatch
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from behave import given, then, when
|
|
from behave.api.async_step import async_run_until_complete
|
|
|
|
from cleveractors.agents.llm import LLMAgent, last_token_usage_var
|
|
from cleveractors.core.exceptions import ConfigurationError, ExecutionError
|
|
from cleveractors.result import ActorResult, NodeUsage
|
|
from cleveractors.runtime import Executor, create_executor
|
|
from cleveractors.templates.renderer import TemplateRenderer
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Background
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the execute_stream test context is initialised")
|
|
def step_es_init(context: Any) -> None:
|
|
context.es_executor = None
|
|
context.es_tokens = None
|
|
context.es_error = None
|
|
context.es_agent = None
|
|
context.es_mock_astream = None
|
|
context.es_token_iter = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given: Executor configurations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an Executor with a basic llm config (stream)")
|
|
def step_es_llm_executor(context: Any) -> None:
|
|
config = {
|
|
"type": "llm",
|
|
"name": "stream_test_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with timeout_ms 50 and a slow mock astream (stream)")
|
|
def step_es_llm_timeout_slow(context: Any) -> None:
|
|
"""LLM executor with timeout_ms=50ms and a mock astream that sleeps 200ms.
|
|
|
|
Exercises the M1 fix: timeout_ms wraps the stream in asyncio.wait_for on
|
|
the LLM streaming path, converting asyncio.TimeoutError to
|
|
ExecutionError(kind="timeout").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "timeout_test_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": 50},
|
|
pricing={},
|
|
)
|
|
|
|
async def _slow_astream(*args: Any, **kwargs: Any) -> Any:
|
|
await asyncio.sleep(0.2) # 200ms > 50ms timeout
|
|
yield MagicMock(content="token", usage_metadata=None)
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _slow_astream
|
|
context.es_mock_model = mock_model
|
|
|
|
|
|
@given("an LLM Executor with bool_true timeout_ms (stream)")
|
|
def step_es_llm_bool_timeout(context: Any) -> None:
|
|
"""LLM executor with timeout_ms=True (invalid bool).
|
|
|
|
Exercises the M1 fix: bool timeout_ms raises ExecutionError(kind="timeout")
|
|
before the stream starts.
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "bool_timeout_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": True},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with zero timeout_ms (stream)")
|
|
def step_es_llm_zero_timeout(context: Any) -> None:
|
|
"""LLM executor with timeout_ms=0 (invalid: must be positive).
|
|
|
|
Exercises the M1 fix: zero timeout_ms raises ExecutionError(kind="timeout").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "zero_timeout_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": 0},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with string timeout_ms (stream)")
|
|
def step_es_llm_string_timeout(context: Any) -> None:
|
|
"""LLM executor with timeout_ms="not_a_number" (invalid non-numeric string).
|
|
|
|
Exercises the M1 fix: non-numeric timeout_ms raises
|
|
ExecutionError(kind="timeout").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "string_timeout_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": "not_a_number"},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with timeout_ms 2000 and a fast mock astream (stream)")
|
|
def step_es_llm_timeout_fast(context: Any) -> None:
|
|
"""LLM executor with timeout_ms=2000ms and a fast mock astream.
|
|
|
|
Exercises the M1 fix: stream completes within timeout, no error raised.
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "fast_timeout_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": 2000},
|
|
pricing={},
|
|
)
|
|
context.es_mock_model = _make_async_chunks(["fast_token"], None)
|
|
|
|
|
|
@given(
|
|
"an LLM Executor with max_cost_usd 0.0 and pricing"
|
|
" and a token-bearing mock astream (stream)"
|
|
)
|
|
def step_es_llm_cost_exceeded(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=0.0 and a mock astream that yields tokens
|
|
with usage_metadata (prompt=100, completion=100).
|
|
|
|
Exercises the M1 fix: max_cost_usd is enforced after the stream completes
|
|
on the LLM streaming path, raising ExecutionError(kind="cost",
|
|
reason="budget_exhausted").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "cost_test_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 0.0},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 2.0, "completion": 2.0}}},
|
|
)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["CostToken"], {"input_tokens": 100, "output_tokens": 100}
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with bool max_cost_usd and pricing (stream)")
|
|
def step_es_llm_bool_cost(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=True (invalid bool).
|
|
|
|
Exercises the M1 fix: bool max_cost_usd raises ExecutionError(kind="cost").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "bool_cost_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": True},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 2.0, "completion": 2.0}}},
|
|
)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["BoolCostToken"], {"input_tokens": 100, "output_tokens": 100}
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with non-numeric max_cost_usd and pricing (stream)")
|
|
def step_es_llm_nonnumeric_cost(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd="bad" (invalid non-numeric string).
|
|
|
|
Exercises the M1 fix: non-numeric max_cost_usd raises
|
|
ExecutionError(kind="cost").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "nonnumeric_cost_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": "bad"},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 2.0, "completion": 2.0}}},
|
|
)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["BadCostToken"], {"input_tokens": 100, "output_tokens": 100}
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with max_cost_usd 1.0 and missing provider pricing (stream)")
|
|
def step_es_llm_missing_provider_pricing(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=1.0 but no pricing entry for 'openai'.
|
|
|
|
Exercises the M1 fix: missing provider pricing raises
|
|
ExecutionError(kind="cost", reason="missing_pricing_entry").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "missing_provider_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={"anthropic": {"claude-3": {"prompt": 3.0, "completion": 15.0}}},
|
|
)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["MissingProviderToken"], {"input_tokens": 100, "output_tokens": 100}
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with max_cost_usd 1.0 and missing model pricing (stream)")
|
|
def step_es_llm_missing_model_pricing(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=1.0 but no pricing entry for 'gpt-3.5-turbo'.
|
|
|
|
Exercises the M1 fix: missing model pricing raises
|
|
ExecutionError(kind="cost", reason="missing_pricing_entry").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "missing_model_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={"openai": {"gpt-4": {"prompt": 30.0, "completion": 60.0}}},
|
|
)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["MissingModelToken"], {"input_tokens": 100, "output_tokens": 100}
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with max_cost_usd 1.0 and incomplete pricing entry (stream)")
|
|
def step_es_llm_incomplete_pricing(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=1.0 but pricing entry missing 'completion' key.
|
|
|
|
Exercises the M1 fix: incomplete pricing entry raises
|
|
ExecutionError(kind="cost", reason="missing_pricing_entry").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "incomplete_pricing_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 2.0}}}, # missing 'completion'
|
|
)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["IncompletePricingToken"], {"input_tokens": 100, "output_tokens": 100}
|
|
)
|
|
|
|
|
|
@given("an LLM Executor with max_cost_usd 1.0 and invalid pricing rate (stream)")
|
|
def step_es_llm_invalid_pricing_rate(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=1.0 but pricing rate is a non-numeric string.
|
|
|
|
Exercises the M1 fix: invalid pricing rate raises
|
|
ExecutionError(kind="cost", reason="missing_pricing_entry").
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "invalid_rate_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={
|
|
"openai": {"gpt-3.5-turbo": {"prompt": "not_a_number", "completion": 2.0}}
|
|
},
|
|
)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["InvalidRateToken"], {"input_tokens": 100, "output_tokens": 100}
|
|
)
|
|
|
|
|
|
@given(
|
|
"an LLM Executor with max_cost_usd 1.0 and pricing"
|
|
" and a token-bearing mock astream (stream)"
|
|
)
|
|
def step_es_llm_cost_within_limit(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=1.0 and a mock astream that yields tokens
|
|
with usage_metadata (prompt=1, completion=1) — cost well within limit.
|
|
|
|
Exercises the M1 fix: stream completes normally when cost is within limit.
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "within_cost_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 2.0, "completion": 2.0}}},
|
|
)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["WithinCostToken"], {"input_tokens": 1, "output_tokens": 1}
|
|
)
|
|
|
|
|
|
@given("an LLM Executor where create_agent raises ConfigurationError (stream)")
|
|
def step_es_llm_create_agent_raises(context: Any) -> None:
|
|
"""LLM executor where AgentFactory.create_agent() raises ConfigurationError.
|
|
|
|
Exercises the m1 fix: executor.last_result is populated with a <no_llm>
|
|
placeholder before re-raising, mirroring the graph path's N5 fix.
|
|
Uses the existing es_factory_should_raise mechanism so the When-step
|
|
patches create_agent to raise ConfigurationError.
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "create_agent_error_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
# Signal to the When-step that create_agent should raise ConfigurationError
|
|
context.es_factory_should_raise = "ConfigurationError"
|
|
|
|
|
|
@given(
|
|
"an LLM Executor where build_chat_model raises ConfigurationError during lazy init (stream)"
|
|
)
|
|
def step_es_llm_build_chat_model_raises(context: Any) -> None:
|
|
"""LLM executor where build_chat_model() raises ConfigurationError during lazy init.
|
|
|
|
Exercises the m2 billing-integrity guarantee: when the LangChain client
|
|
construction fails inside stream_message() (lazy init path), the
|
|
ConfigurationError propagates through _execute_llm_stream()'s
|
|
except (ConfigurationError, ...) handler, which populates
|
|
executor.last_result with a partial ActorResult (token counts = 0)
|
|
before re-raising.
|
|
|
|
The agent is created normally by create_agent() but build_chat_model is
|
|
patched to raise when the chat_model property is first accessed inside
|
|
stream_message().
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "lazy_init_error_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
# Signal to the When-step that build_chat_model should raise during lazy init.
|
|
context.es_build_chat_model_should_raise = True
|
|
|
|
|
|
@given("an Executor with a graph config that has a single agent node (stream)")
|
|
def step_es_graph_executor(context: Any) -> None:
|
|
config = {
|
|
"name": "stream_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {
|
|
"type": "agent",
|
|
"agent": "agent1",
|
|
}
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a graph config that has a terminal agent node"
|
|
" with conditional edge (stream)"
|
|
)
|
|
def step_es_graph_conditional_edge_executor(context: Any) -> None:
|
|
"""Graph with a single agent node whose only edge to 'end' has a condition.
|
|
|
|
This exercises the _all_edges_unconditional=False path in _stream_from_node:
|
|
the node is statically terminal (all successors are END), but the edge has
|
|
a condition, so tokens must be buffered until full_response is available for
|
|
edge-condition evaluation.
|
|
"""
|
|
config = {
|
|
"name": "conditional_edge_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {
|
|
"type": "agent",
|
|
"agent": "agent1",
|
|
}
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
# Conditional edge: agent1 → end (only when output contains "ok")
|
|
{
|
|
"source": "agent1",
|
|
"target": "end",
|
|
"condition": {"type": "content_contains", "text": "Buf"},
|
|
},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and bool_true timeout_ms (stream)")
|
|
def step_es_graph_bool_timeout(context: Any) -> None:
|
|
config = {
|
|
"name": "bool_timeout_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {"agent1": {"type": "agent", "agent": "agent1"}},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": True}, # bool is invalid
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and zero timeout_ms (stream)")
|
|
def step_es_graph_zero_timeout(context: Any) -> None:
|
|
config = {
|
|
"name": "zero_timeout_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {"agent1": {"type": "agent", "agent": "agent1"}},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": 0}, # zero is invalid
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a tool-node graph and max_tool_calls {n:d} (stream)")
|
|
def step_es_tool_node_graph(context: Any, n: int) -> None:
|
|
"""Graph with a TOOL type node to test TOOL limit enforcement in _stream_from_node."""
|
|
config = {
|
|
"name": "tool_node_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"tool1": {"type": "tool", "tools": ["echo"]},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "tool1"},
|
|
{"source": "tool1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={"max_tool_calls": n},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and timeout_ms {ms:d} (stream)")
|
|
def step_es_graph_timeout_executor(context: Any, ms: int) -> None:
|
|
config = {
|
|
"name": "stream_timeout_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {
|
|
"type": "agent",
|
|
"agent": "agent1",
|
|
}
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": ms},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and string timeout_ms (stream)")
|
|
def step_es_graph_string_timeout(context: Any) -> None:
|
|
config = {
|
|
"name": "string_timeout_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {"agent1": {"type": "agent", "agent": "agent1"}},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"timeout_ms": "not_a_number"}, # non-numeric string
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and max_depth {n:d} (stream)")
|
|
def step_es_graph_max_depth(context: Any, n: int) -> None:
|
|
config = {
|
|
"name": "depth_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {"agent1": {"type": "agent", "agent": "agent1"}},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_depth": n},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and bool max_depth (stream)")
|
|
def step_es_graph_bool_max_depth(context: Any) -> None:
|
|
config = {
|
|
"name": "bool_depth_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {"agent1": {"type": "agent", "agent": "agent1"}},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_depth": True}, # bool is invalid
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and string max_depth (stream)")
|
|
def step_es_graph_string_max_depth(context: Any) -> None:
|
|
"""Exercises the non-numeric max_depth error path in _collect_stream_tokens."""
|
|
config = {
|
|
"name": "string_depth_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {"agent1": {"type": "agent", "agent": "agent1"}},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_depth": "not_a_number"}, # non-numeric string is invalid
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a parallel AGENT graph config (stream)")
|
|
def step_es_parallel_agent_graph(context: Any) -> None:
|
|
"""Graph: agent_start → {agent_a, agent_b} (parallel) → END.
|
|
Exercises the AGENT intermediate parallel path in _stream_from_node.
|
|
"""
|
|
config = {
|
|
"name": "parallel_agent_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent_start": {"type": "agent", "agent": "agent_start"},
|
|
"agent_a": {"type": "agent", "agent": "agent_a"},
|
|
"agent_b": {"type": "agent", "agent": "agent_b"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent_start"},
|
|
{"source": "agent_start", "target": "agent_a"},
|
|
{"source": "agent_start", "target": "agent_b"},
|
|
{"source": "agent_a", "target": "end"},
|
|
{"source": "agent_b", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
"parallel_execution": True,
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent_start": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_a": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_b": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config with actor context (stream)")
|
|
def step_es_graph_with_context(context: Any) -> None:
|
|
config = {
|
|
"name": "context_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {"agent1": {"type": "agent", "agent": "agent1"}},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
"context": {
|
|
"global": {"actor_mode": "streaming", "version": "2.0"},
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an LLMAgent with template config and mock renderer (stream)")
|
|
def step_es_llm_template_agent(context: Any) -> None:
|
|
config = {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"template": "test_template",
|
|
}
|
|
renderer = TemplateRenderer()
|
|
# Patch render to return a fixed string so we don't need an actual template file
|
|
original_render = renderer.render
|
|
renderer.render = lambda name, vars: "rendered_message"
|
|
agent = LLMAgent(name="template_agent", config=config, template_renderer=renderer)
|
|
renderer.render = original_render # restore
|
|
# Re-patch for tests
|
|
renderer.render = lambda name, vars: "rendered_message"
|
|
context.es_mock_model = _make_async_chunks(["template_token"], None)
|
|
agent.chat_model = context.es_mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given("the mock factory raises ConfigurationError on create_agent (stream)")
|
|
def step_es_factory_raises_config_error(context: Any) -> None:
|
|
context.es_factory_should_raise = "ConfigurationError"
|
|
|
|
|
|
@given("the mock factory raises RuntimeError on create_agent (stream)")
|
|
def step_es_factory_raises_runtime_error(context: Any) -> None:
|
|
context.es_factory_should_raise = "RuntimeError"
|
|
|
|
|
|
@given("a PureLangGraph already in running state (stream)")
|
|
def step_es_graph_already_running(context: Any) -> None:
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="test_graph",
|
|
nodes={},
|
|
edges=[],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(config=pg_config, agents={}, limits={}, pricing={})
|
|
graph.is_running = True # Simulate running state
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@given("a PureLangGraph with a dangling edge to nonexistent node (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_dangling_edge_graph(context: Any) -> None:
|
|
"""Graph with an edge pointing to 'missing_node' which is not in nodes dict."""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_agent", config=config, template_renderer=renderer)
|
|
context.es_mock_model = _make_async_chunks([], None)
|
|
agent.chat_model = context.es_mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="dangling_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="missing_node"), # Points to nonexistent node
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
context.es_pure_graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("a PureLangGraph with AGENT node that has no agent configured (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_bad_agent_graph(context: Any) -> None:
|
|
"""Graph with an AGENT node where agent=None (raises ValueError in stream_agent)."""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
# Node with type=AGENT but no agent name (agent=None)
|
|
pg_config = PureGraphConfig(
|
|
name="bad_agent_graph",
|
|
nodes={
|
|
"bad_node": NodeConfig(name="bad_node", type=NodeType.AGENT, agent=None),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="bad_node"),
|
|
Edge(source="bad_node", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
context.es_pure_graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with system_prompt at top level llm config (stream)")
|
|
def step_es_top_level_sp(context: Any) -> None:
|
|
config = {
|
|
"type": "llm",
|
|
"name": "sp_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"system_prompt": "You are a helpful streaming assistant.", # top-level system_prompt
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an LLMAgent with failing render_string mock (stream)")
|
|
def step_es_llm_failing_render_string(context: Any) -> None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo", "system_prompt": "test"}
|
|
renderer = TemplateRenderer()
|
|
original_rs = renderer.render_string
|
|
renderer.render_string = lambda *args, **kw: (_ for _ in ()).throw(
|
|
RuntimeError("render_string failed")
|
|
)
|
|
agent = LLMAgent(name="fail_render", config=config, template_renderer=renderer)
|
|
renderer.render_string = original_rs # restore for safety
|
|
# Patch for the actual call
|
|
agent.template_renderer.render_string = MagicMock(
|
|
side_effect=RuntimeError("render_string failed")
|
|
)
|
|
context.es_mock_model = _make_async_chunks(["render_ex_tok"], None)
|
|
agent.chat_model = context.es_mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given("a mock astream that raises ExecutionError (stream)")
|
|
def step_es_exec_error_astream(context: Any) -> None:
|
|
from cleveractors.core.exceptions import ExecutionError as ExecError
|
|
|
|
async def _raising_exec_astream(_messages: Any) -> Any:
|
|
raise ExecError("stream ExecutionError")
|
|
yield # make it a generator
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _raising_exec_astream
|
|
mock_model.temperature = 0.7
|
|
context.es_mock_model = mock_model
|
|
|
|
|
|
@given("a PureLangGraph with context_manager set (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_graph_with_context_manager(context: Any) -> None:
|
|
from unittest.mock import MagicMock
|
|
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="cm_agent", config=config, template_renderer=renderer)
|
|
context.es_mock_model = getattr(context, "es_mock_model", None)
|
|
if context.es_mock_model is None:
|
|
context.es_mock_model = _make_async_chunks(["CmToken"], None)
|
|
agent.chat_model = context.es_mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="cm_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
|
|
# Create a mock context manager
|
|
mock_ctx_mgr = MagicMock()
|
|
mock_ctx_mgr.get_global_context = MagicMock(return_value={"ctx_key": "ctx_val"})
|
|
mock_ctx_mgr.save_global_context = MagicMock()
|
|
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
context_manager=mock_ctx_mgr,
|
|
)
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@given("a PureLangGraph with string timeout_ms limit (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_graph_string_timeout_limit(context: Any) -> None:
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="str_timeout_graph",
|
|
nodes={
|
|
"fn1": NodeConfig(name="fn1", type=NodeType.FUNCTION, function="summarize"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="fn1"),
|
|
Edge(source="fn1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={},
|
|
limits={"timeout_ms": "not_a_number"}, # non-numeric string
|
|
pricing={},
|
|
)
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@given("a PureLangGraph with last_context populated (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_graph_with_last_context(context: Any) -> None:
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="ctx_agent", config=config, template_renderer=renderer)
|
|
context.es_mock_model = getattr(
|
|
context, "es_mock_model", None
|
|
) or _make_async_chunks(["CtxToken"], None)
|
|
agent.chat_model = context.es_mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="last_ctx_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
# Populate _last_context (triggers the elif self._last_context: branch)
|
|
graph._last_context = {"previous_key": "previous_value"}
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@given("a Node stream_agent with no agent configured (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_node_no_agent_config(context: Any) -> None:
|
|
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
|
|
node_config = NodeConfig(name="test_node", type=NodeType.AGENT, agent=None)
|
|
node = Node(config=node_config, agents={})
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "test"}]
|
|
state.metadata = {"current_message": "test"}
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in node._stream_agent(state):
|
|
tokens.append(token)
|
|
context.es_stream_agent_tokens2 = tokens
|
|
context.es_stream_agent_error2 = None
|
|
except ValueError as e:
|
|
context.es_stream_agent_error2 = e
|
|
context.es_stream_agent_tokens2 = []
|
|
except Exception as e:
|
|
context.es_stream_agent_error2 = e
|
|
context.es_stream_agent_tokens2 = []
|
|
|
|
|
|
@given("a Node stream_agent with agent key missing from agents dict (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_node_agent_not_found(context: Any) -> None:
|
|
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
|
|
node_config = NodeConfig(
|
|
name="test_node", type=NodeType.AGENT, agent="missing_agent"
|
|
)
|
|
node = Node(config=node_config, agents={}) # Empty agents dict
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "test"}]
|
|
state.metadata = {"current_message": "test"}
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in node._stream_agent(state):
|
|
tokens.append(token)
|
|
context.es_stream_agent_tokens3 = tokens
|
|
context.es_stream_agent_error3 = None
|
|
except ValueError as e:
|
|
context.es_stream_agent_error3 = e
|
|
context.es_stream_agent_tokens3 = []
|
|
|
|
|
|
@given("a Node stream_agent with empty state messages (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_node_empty_messages(context: Any) -> None:
|
|
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_agent", config=config, template_renderer=renderer)
|
|
mock_model = _make_async_chunks(["empty_msg_tok"], None)
|
|
agent.chat_model = mock_model
|
|
|
|
node_config = NodeConfig(name="test_node", type=NodeType.AGENT, agent="test_agent")
|
|
node = Node(config=node_config, agents={"test_agent": agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [] # Empty messages
|
|
state.metadata = {}
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in node._stream_agent(state):
|
|
tokens.append(token)
|
|
context.es_empty_msg_tokens = tokens
|
|
context.es_empty_msg_error = None
|
|
except Exception as e:
|
|
context.es_empty_msg_error = e
|
|
context.es_empty_msg_tokens = []
|
|
|
|
|
|
@given("a Node stream_agent with long conversation history (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_node_long_history(context: Any) -> None:
|
|
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_agent", config=config, template_renderer=renderer)
|
|
mock_model = _make_async_chunks(["truncated_tok"], None)
|
|
agent.chat_model = mock_model
|
|
|
|
node_config = NodeConfig(name="test_node", type=NodeType.AGENT, agent="test_agent")
|
|
node = Node(config=node_config, agents={"test_agent": agent})
|
|
|
|
state = GraphState()
|
|
# Create 25 messages (more than MAX_HISTORY_MESSAGES=20)
|
|
state.messages = [
|
|
{"role": "user" if i % 2 == 0 else "assistant", "content": f"msg {i}"}
|
|
for i in range(25)
|
|
]
|
|
state.metadata = {"current_message": "latest"}
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in node._stream_agent(state):
|
|
tokens.append(token)
|
|
context.es_trunc_tokens = tokens
|
|
context.es_trunc_error = None
|
|
except Exception as e:
|
|
context.es_trunc_error = e
|
|
context.es_trunc_tokens = []
|
|
|
|
|
|
@given("a Node stream_agent with nested context in metadata (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_node_nested_context(context: Any) -> None:
|
|
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_agent", config=config, template_renderer=renderer)
|
|
mock_model = _make_async_chunks(["nested_tok"], None)
|
|
agent.chat_model = mock_model
|
|
|
|
node_config = NodeConfig(name="test_node", type=NodeType.AGENT, agent="test_agent")
|
|
node = Node(config=node_config, agents={"test_agent": agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "test"}]
|
|
state.metadata = {
|
|
"current_message": "test",
|
|
"context": {"nested_key": "nested_value"}, # Nested context dict
|
|
}
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in node._stream_agent(state):
|
|
tokens.append(token)
|
|
context.es_nested_tokens = tokens
|
|
context.es_nested_error = None
|
|
except Exception as e:
|
|
context.es_nested_error = e
|
|
context.es_nested_tokens = []
|
|
|
|
|
|
@given("an Executor with a graph config and max_model_calls {n:d} (stream)")
|
|
def step_es_graph_max_model_calls(context: Any, n: int) -> None:
|
|
config = {
|
|
"name": "stream_model_calls_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {
|
|
"type": "agent",
|
|
"agent": "agent1",
|
|
}
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_model_calls": n},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a tool actor config (stream)")
|
|
def step_es_tool_executor(context: Any) -> None:
|
|
config = {
|
|
"type": "tool",
|
|
"name": "stream_tool",
|
|
"tools": ["echo"],
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a multi_actor config (stream)")
|
|
def step_es_multi_executor(context: Any) -> None:
|
|
config = {
|
|
"actors": {
|
|
"default": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
}
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a pure function graph config (stream)")
|
|
def step_es_function_graph_executor(context: Any) -> None:
|
|
"""Graph with only a FUNCTION node (no LLM) — exercises non-AGENT terminal path."""
|
|
config = {
|
|
"name": "function_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"fn1": {
|
|
"type": "function",
|
|
"function": "summarize",
|
|
}
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "fn1"},
|
|
{"source": "fn1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a two-agent sequential graph config (stream)")
|
|
def step_es_two_agent_sequential(context: Any) -> None:
|
|
"""Graph: START → agent_a (AGENT, intermediate) → agent_b (AGENT, terminal) → END."""
|
|
config = {
|
|
"name": "two_agent_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent_a": {"type": "agent", "agent": "agent_a"},
|
|
"agent_b": {"type": "agent", "agent": "agent_b"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent_a"},
|
|
{"source": "agent_a", "target": "agent_b"},
|
|
{"source": "agent_b", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
"parallel_execution": False,
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent_a": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_b": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a parallel function graph config (stream)")
|
|
def step_es_parallel_function_graph(context: Any) -> None:
|
|
"""Graph: fn1 → {fn2, fn3} both connected to END (parallel non-AGENT path)."""
|
|
config = {
|
|
"name": "parallel_fn_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"fn1": {"type": "function", "function": "summarize"},
|
|
"fn2": {"type": "function", "function": "summarize"},
|
|
"fn3": {"type": "function", "function": "summarize"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "fn1"},
|
|
{"source": "fn1", "target": "fn2"},
|
|
{"source": "fn1", "target": "fn3"},
|
|
{"source": "fn2", "target": "end"},
|
|
{"source": "fn3", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
"parallel_execution": True,
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a config_block format llm config (stream)")
|
|
def step_es_config_block_llm(context: Any) -> None:
|
|
"""LLM config using config_block format (no top-level provider/model)."""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "block_llm",
|
|
"config": {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"temperature": 0.5,
|
|
"max_tokens": 500,
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with invalid temperature llm config (stream)")
|
|
def step_es_invalid_temp_llm(context: Any) -> None:
|
|
config = {
|
|
"type": "llm",
|
|
"name": "bad_temp_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"temperature": "not_a_number",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with invalid max_tokens llm config (stream)")
|
|
def step_es_invalid_max_tokens_llm(context: Any) -> None:
|
|
config = {
|
|
"type": "llm",
|
|
"name": "bad_tokens_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"max_tokens": "also_not_a_number",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an LLMAgent with memory_enabled config and mock astream (stream)")
|
|
def step_es_llm_memory_enabled(context: Any) -> None:
|
|
config = {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"memory_enabled": True,
|
|
}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="mem_agent", config=config, template_renderer=renderer)
|
|
context.es_mock_model = _make_async_chunks(["mem_token"], None)
|
|
agent.chat_model = context.es_mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given("an Executor with a function-then-agent graph config (stream)")
|
|
def step_es_function_then_agent_graph(context: Any) -> None:
|
|
"""Graph: START → FUNCTION → AGENT → END, tests non-AGENT intermediate path."""
|
|
config = {
|
|
"name": "fn_agent_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"fn1": {
|
|
"type": "function",
|
|
"function": "summarize",
|
|
},
|
|
"agent1": {
|
|
"type": "agent",
|
|
"agent": "agent1",
|
|
},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "fn1"},
|
|
{"source": "fn1", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given: Mock astream configurations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_async_chunks(
|
|
chunks: list[str],
|
|
last_usage_metadata: dict[str, Any] | None = None,
|
|
) -> MagicMock:
|
|
"""Build a MagicMock chat model whose astream() returns an async generator
|
|
and whose ainvoke() returns a response object (for intermediate nodes that
|
|
use the ainvoke path after the M5 fix).
|
|
|
|
Args:
|
|
chunks: List of token strings to yield.
|
|
last_usage_metadata: Optional usage metadata for the final chunk.
|
|
"""
|
|
|
|
async def _astream_gen(_messages: Any) -> Any:
|
|
for i, text in enumerate(chunks):
|
|
chunk = MagicMock()
|
|
chunk.content = text
|
|
if i == len(chunks) - 1 and last_usage_metadata is not None:
|
|
chunk.usage_metadata = last_usage_metadata
|
|
else:
|
|
chunk.usage_metadata = None
|
|
yield chunk
|
|
|
|
# Build a combined response for ainvoke() (used by intermediate AGENT nodes
|
|
# after the M5 fix). The response content is the concatenation of all chunks
|
|
# so that intermediate nodes produce a sensible full_response.
|
|
combined_content = "".join(chunks) if chunks else ""
|
|
ainvoke_response = MagicMock()
|
|
ainvoke_response.content = combined_content
|
|
ainvoke_response.usage_metadata = last_usage_metadata
|
|
ainvoke_response.response_metadata = None
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _astream_gen
|
|
mock_model.ainvoke = AsyncMock(return_value=ainvoke_response)
|
|
mock_model.temperature = 0.7
|
|
return mock_model
|
|
|
|
|
|
@given(
|
|
"a mock astream yielding single token {token_str} with usage"
|
|
" prompt={p:d} completion={c:d} (stream)"
|
|
)
|
|
def step_es_mock_astream_with_usage(
|
|
context: Any, token_str: str, p: int, c: int
|
|
) -> None:
|
|
import ast
|
|
|
|
chunks = [ast.literal_eval(token_str.strip())]
|
|
usage_metadata = {"input_tokens": p, "output_tokens": c}
|
|
context.es_mock_model = _make_async_chunks(chunks, usage_metadata)
|
|
|
|
|
|
@given("a mock astream yielding tokens {chunks_str} with no usage (stream)")
|
|
def step_es_mock_astream_no_usage_multi(context: Any, chunks_str: str) -> None:
|
|
import ast
|
|
|
|
parts = [p.strip() for p in chunks_str.split(",")]
|
|
chunks = [ast.literal_eval(p) for p in parts]
|
|
context.es_mock_model = _make_async_chunks(chunks, None)
|
|
|
|
|
|
@given("a mock astream yielding single token {token_str} with no usage (stream)")
|
|
def step_es_mock_astream_single_no_usage(context: Any, token_str: str) -> None:
|
|
import ast
|
|
|
|
chunks = [ast.literal_eval(token_str.strip())]
|
|
context.es_mock_model = _make_async_chunks(chunks, None)
|
|
|
|
|
|
@given("the mock agent cleanup raises RuntimeError (stream)")
|
|
def step_es_cleanup_raises(context: Any) -> None:
|
|
context.es_cleanup_should_raise = True
|
|
|
|
|
|
@given("a prior successful execute_stream has populated last_result (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_prior_successful_stream(context: Any) -> None:
|
|
"""Run a complete stream to populate last_result, proving it was set before
|
|
the subsequent partial stream clears it. This makes the 'last_result is
|
|
None mid-stream' assertion non-tautological (n2 fix)."""
|
|
executor = context.es_executor
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
if mock_model is not None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="stream_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
agent.chat_model = mock_model
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
|
|
try:
|
|
tokens = []
|
|
async for token in executor.execute_stream("prior"):
|
|
tokens.append(token)
|
|
except Exception: # pylint: disable=broad-exception-caught
|
|
pass # Ignore errors from the prior stream
|
|
|
|
# Verify last_result was actually populated by the prior stream so the
|
|
# subsequent assertion is meaningful.
|
|
assert executor.last_result is not None, (
|
|
"Prior stream did not populate last_result — test setup is broken"
|
|
)
|
|
# Reset mock_model so the next stream step can set it fresh
|
|
context.es_mock_model = _make_async_chunks(["Hello", " World"], None)
|
|
|
|
|
|
@given("a mock astream that yields nothing (just for setup) (stream)")
|
|
def step_es_empty_astream_setup(context: Any) -> None:
|
|
context.es_mock_model = _make_async_chunks([], None)
|
|
|
|
|
|
@given("the intermediate agent returns a non-dict last_msg (stream)")
|
|
def step_es_intermediate_non_dict_last_msg(context: Any) -> None:
|
|
"""Override ainvoke to return a result where messages[-1] is not a dict.
|
|
Covers the 'else: full_response = str(last_msg)' branch in the intermediate
|
|
AGENT ainvoke path (M5 coverage)."""
|
|
context.es_intermediate_override = "non_dict_last_msg"
|
|
|
|
|
|
@given("the intermediate agent returns a result with no messages key (stream)")
|
|
def step_es_intermediate_no_messages(context: Any) -> None:
|
|
"""Override ainvoke to return a result dict with no 'messages' key.
|
|
Covers the 'else: full_response = result.get(...)' branch (M5 coverage)."""
|
|
context.es_intermediate_override = "no_messages"
|
|
|
|
|
|
@given("the intermediate agent returns a non-dict ainvoke result (stream)")
|
|
def step_es_intermediate_non_dict_result(context: Any) -> None:
|
|
"""Override ainvoke to return a non-dict value.
|
|
Covers the 'else: full_response = str(result)' branch (M5 coverage)."""
|
|
context.es_intermediate_override = "non_dict_result"
|
|
|
|
|
|
@given("the intermediate agent ainvoke raises RuntimeError (stream)")
|
|
def step_es_intermediate_ainvoke_raises(context: Any) -> None:
|
|
"""Override ainvoke to raise RuntimeError.
|
|
Covers the 'except Exception' branch in the intermediate AGENT path (M5 coverage)."""
|
|
context.es_intermediate_override = "raises"
|
|
|
|
|
|
@given("the function node returns a non-dict last_msg in messages (stream)")
|
|
def step_es_fn_non_dict_last_msg(context: Any) -> None:
|
|
"""Override Node.execute for function nodes to return non-dict last_msg.
|
|
Covers the non-AGENT 'else: output_message = str(last_msg)' branch."""
|
|
context.es_fn_override = "non_dict_last_msg"
|
|
|
|
|
|
@given("the function node returns a result with no messages key (stream)")
|
|
def step_es_fn_no_messages(context: Any) -> None:
|
|
"""Override Node.execute for function nodes to return no 'messages' key.
|
|
Covers the non-AGENT 'else: output_message = result.get(...)' branch."""
|
|
context.es_fn_override = "no_messages"
|
|
|
|
|
|
@given("the function node returns a non-dict result (stream)")
|
|
def step_es_fn_non_dict_result(context: Any) -> None:
|
|
"""Override Node.execute for function nodes to return a non-dict value.
|
|
Covers the non-AGENT 'else: output_message = result' branch."""
|
|
context.es_fn_override = "non_dict_result"
|
|
|
|
|
|
@given("the function node raises RuntimeError (stream)")
|
|
def step_es_fn_raises(context: Any) -> None:
|
|
"""Override Node.execute for function nodes to raise RuntimeError.
|
|
Covers the non-AGENT 'except Exception' branch."""
|
|
context.es_fn_override = "raises"
|
|
|
|
|
|
@given("the function node returns a result with _node_token_usage (stream)")
|
|
def step_es_fn_token_usage(context: Any) -> None:
|
|
"""Override Node.execute for function nodes to include _node_token_usage.
|
|
Covers the non-AGENT token usage branch (lines 1630-1642)."""
|
|
context.es_fn_override = "token_usage"
|
|
|
|
|
|
@given("an LLMAgent with memory_enabled and long history for truncation test (stream)")
|
|
def step_es_llm_agent_memory_long_history(context: Any) -> None:
|
|
"""Set up an LLMAgent with memory_enabled and a history that exceeds max_history.
|
|
Covers the history truncation branch in stream_message() (M2 coverage)."""
|
|
config = {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"memory_enabled": True,
|
|
"max_history": 2, # Small limit to force truncation
|
|
}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
context.es_mock_model = _make_async_chunks(["mem_trunc_tok"], None)
|
|
agent.chat_model = context.es_mock_model
|
|
|
|
# Pre-populate memory with more entries than max_history
|
|
async def _pre_populate() -> None:
|
|
await agent.update_memory(
|
|
"conversation_history",
|
|
[
|
|
{"role": "user", "content": "msg1"},
|
|
{"role": "assistant", "content": "resp1"},
|
|
{"role": "user", "content": "msg2"},
|
|
{"role": "assistant", "content": "resp2"},
|
|
{"role": "user", "content": "msg3"},
|
|
{"role": "assistant", "content": "resp3"},
|
|
],
|
|
)
|
|
|
|
asyncio.get_event_loop().run_until_complete(_pre_populate())
|
|
context.es_agent = agent
|
|
|
|
|
|
@given("an LLMAgent with memory_enabled that raises on update_memory (stream)")
|
|
def step_es_llm_agent_memory_raises(context: Any) -> None:
|
|
"""Set up an LLMAgent with memory_enabled where update_memory raises after
|
|
astream() completes. Covers the billing-integrity except branch in
|
|
stream_message() (m1 coverage): _captured_prompt is set so tokens are preserved."""
|
|
config = {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"memory_enabled": True,
|
|
}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["tok"], {"input_tokens": 5, "output_tokens": 10}
|
|
)
|
|
agent.chat_model = context.es_mock_model
|
|
|
|
# Patch update_memory to raise after astream() completes
|
|
original_update_memory = agent.update_memory
|
|
|
|
async def _raising_update_memory(key: str, value: Any) -> None:
|
|
raise RuntimeError("memory write failed")
|
|
|
|
agent.update_memory = _raising_update_memory # type: ignore[method-assign]
|
|
context.es_agent = agent
|
|
|
|
|
|
@given("a PureLangGraph with a single agent node and initial_state (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_pure_graph_initial_state(context: Any) -> None:
|
|
"""Set up a PureLangGraph for testing execute() with initial_state."""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_agent", config=config, template_renderer=renderer)
|
|
agent.chat_model = _make_async_chunks(["result"], None)
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="test_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
context.es_pure_graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
context.es_initial_state = {"restored_key": "restored_value", "stage": "resumed"}
|
|
|
|
|
|
@given("a mock astream that raises RuntimeError (stream)")
|
|
def step_es_raising_astream(context: Any) -> None:
|
|
async def _raising_astream(_messages: Any) -> Any:
|
|
raise RuntimeError("Stream intentionally failed")
|
|
yield # make it an async generator
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _raising_astream
|
|
mock_model.temperature = 0.7
|
|
context.es_mock_model = mock_model
|
|
|
|
|
|
@given("a mock astream that takes too long (stream)")
|
|
def step_es_slow_astream(context: Any) -> None:
|
|
async def _slow_astream(_messages: Any) -> Any:
|
|
await asyncio.sleep(10) # Much longer than timeout
|
|
chunk = MagicMock()
|
|
chunk.content = "slow"
|
|
chunk.usage_metadata = None
|
|
yield chunk
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _slow_astream
|
|
mock_model.temperature = 0.7
|
|
context.es_mock_model = mock_model
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given: LLMAgent-specific setups
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given('an LLMAgent with a mock astream that yields "tok1", "tok2", "tok3"')
|
|
def step_es_llm_agent_three_tokens(context: Any) -> None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
context.es_mock_model = _make_async_chunks(["tok1", "tok2", "tok3"], None)
|
|
agent.chat_model = context.es_mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given(
|
|
"an LLMAgent with a mock astream whose last chunk has usage_metadata"
|
|
" prompt={p:d} completion={c:d}"
|
|
)
|
|
def step_es_llm_agent_with_usage(context: Any, p: int, c: int) -> None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
context.es_mock_model = _make_async_chunks(
|
|
["response"], {"input_tokens": p, "output_tokens": c}
|
|
)
|
|
agent.chat_model = context.es_mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given("an LLMAgent with stale _last_token_usage (100, 200)")
|
|
def step_es_llm_agent_stale_usage(context: Any) -> None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
# Only set the instance attribute (not the ContextVar which is synchronous-context
|
|
# global and would pollute later async tests by leaking into their inherited context).
|
|
# The async stream_message() step will set last_token_usage_var inside the task context.
|
|
agent._last_token_usage = (100, 200)
|
|
context.es_agent = agent
|
|
context.es_stale_usage = (100, 200)
|
|
|
|
|
|
@given("a mock astream yielding empty sequence (stream-agent)")
|
|
def step_es_empty_chunks(context: Any) -> None:
|
|
context.es_mock_model = _make_async_chunks([], None)
|
|
if context.es_agent is not None:
|
|
context.es_agent.chat_model = context.es_mock_model
|
|
|
|
|
|
@given("an LLMAgent with a mock astream that yields a chunk with no usage_metadata")
|
|
def step_es_llm_agent_no_usage(context: Any) -> None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
context.es_mock_model = _make_async_chunks(["hello"], None)
|
|
agent.chat_model = context.es_mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given(
|
|
"an LLMAgent with a mock astream that yields a chunk with empty usage_metadata dict"
|
|
)
|
|
def step_es_llm_agent_empty_usage_metadata(context: Any) -> None:
|
|
"""N1: usage_metadata is present but is an empty dict {}.
|
|
|
|
The LLMAgent._CAUSE_USAGE_METADATA_EMPTY branch should fire and log a
|
|
warning, leaving _last_token_usage at (0, 0).
|
|
"""
|
|
|
|
async def _astream_empty_usage(_messages: Any) -> Any:
|
|
chunk = MagicMock()
|
|
chunk.content = "hello"
|
|
chunk.usage_metadata = {} # present but empty — triggers _CAUSE_USAGE_METADATA_EMPTY
|
|
yield chunk
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _astream_empty_usage
|
|
mock_model.temperature = 0.7
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
agent.chat_model = mock_model
|
|
context.es_mock_model = mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given(
|
|
"an LLMAgent with a mock astream that yields a chunk with non-dict response_metadata"
|
|
)
|
|
def step_es_llm_agent_non_dict_response_metadata(context: Any) -> None:
|
|
"""N2: response_metadata is present but not a dict (e.g. a string).
|
|
|
|
The LLMAgent._CAUSE_RESPONSE_METADATA_NOT_DICT branch should fire.
|
|
"""
|
|
|
|
async def _astream_non_dict_rm(_messages: Any) -> Any:
|
|
chunk = MagicMock()
|
|
chunk.content = "hello"
|
|
chunk.usage_metadata = (
|
|
None # no usage_metadata → fall through to response_metadata
|
|
)
|
|
chunk.response_metadata = "not-a-dict" # non-dict triggers the branch
|
|
yield chunk
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _astream_non_dict_rm
|
|
mock_model.temperature = 0.7
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
agent.chat_model = mock_model
|
|
context.es_mock_model = mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given(
|
|
"an LLMAgent with a mock astream that yields a chunk"
|
|
" with response_metadata missing token_usage"
|
|
)
|
|
def step_es_llm_agent_response_metadata_no_token_usage(context: Any) -> None:
|
|
"""N2: response_metadata is a dict but has no 'token_usage' key.
|
|
|
|
The LLMAgent._CAUSE_RESPONSE_METADATA_NO_TOKEN_USAGE branch should fire.
|
|
"""
|
|
|
|
async def _astream_no_token_usage(_messages: Any) -> Any:
|
|
chunk = MagicMock()
|
|
chunk.content = "hello"
|
|
chunk.usage_metadata = (
|
|
None # no usage_metadata → fall through to response_metadata
|
|
)
|
|
chunk.response_metadata = {
|
|
"model": "gpt-3.5-turbo"
|
|
} # dict but no token_usage key
|
|
yield chunk
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _astream_no_token_usage
|
|
mock_model.temperature = 0.7
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
agent.chat_model = mock_model
|
|
context.es_mock_model = mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given(
|
|
"an LLMAgent with a mock astream that yields a chunk"
|
|
" with no response_metadata attribute"
|
|
)
|
|
def step_es_llm_agent_no_response_metadata_attr(context: Any) -> None:
|
|
"""n5 fix: exercise _CAUSE_RESPONSE_METADATA_MISSING branch.
|
|
|
|
Uses a SimpleNamespace chunk that has only a ``content`` attribute — no
|
|
``usage_metadata`` and no ``response_metadata``. The LLMAgent code path
|
|
falls through to the final ``else`` branch and logs a warning with
|
|
_CAUSE_RESPONSE_METADATA_MISSING.
|
|
"""
|
|
from types import SimpleNamespace
|
|
|
|
async def _astream_no_rm(_messages: Any) -> Any:
|
|
# SimpleNamespace exposes only the attributes explicitly set, so
|
|
# hasattr(chunk, "response_metadata") is False — triggering the
|
|
# _CAUSE_RESPONSE_METADATA_MISSING branch.
|
|
chunk = SimpleNamespace(content="hello")
|
|
yield chunk
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _astream_no_rm
|
|
mock_model.temperature = 0.7
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
agent.chat_model = mock_model
|
|
context.es_mock_model = mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
@given("an LLMAgent with temperature 0.7 and mock astream for override test (stream)")
|
|
def step_es_llm_agent_temp_override_setup(context: Any) -> None:
|
|
"""Set up an LLMAgent with a known temperature for override testing (n5/M1)."""
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
# Track temperatures seen during streaming so the Then step can verify them.
|
|
temperatures_seen: list[float] = []
|
|
|
|
async def _astream_recording(messages: Any) -> Any:
|
|
"""Record the temperature at the time astream() is called."""
|
|
temperatures_seen.append(agent._chat_model.temperature) # type: ignore[union-attr]
|
|
chunk = MagicMock()
|
|
chunk.content = "override_tok"
|
|
chunk.usage_metadata = None
|
|
yield chunk
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _astream_recording
|
|
mock_model.temperature = 0.7
|
|
agent.chat_model = mock_model
|
|
context.es_agent = agent
|
|
context.es_temperatures_seen = temperatures_seen
|
|
|
|
|
|
@given(
|
|
"an LLMAgent with a mock astream whose final chunk has"
|
|
" response_metadata token_usage (stream)"
|
|
)
|
|
def step_es_llm_agent_response_metadata_usage(context: Any) -> None:
|
|
"""Set up an LLMAgent whose final chunk has response_metadata but no
|
|
usage_metadata, to test the tier-2 fallback (n6/M3)."""
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
|
|
async def _astream_rm(messages: Any) -> Any:
|
|
chunk = MagicMock()
|
|
chunk.content = "rm_tok"
|
|
chunk.usage_metadata = None # No usage_metadata — forces tier-2 fallback
|
|
chunk.response_metadata = {
|
|
"token_usage": {"prompt_tokens": 3, "completion_tokens": 7}
|
|
}
|
|
yield chunk
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _astream_rm
|
|
mock_model.temperature = 0.7
|
|
agent.chat_model = mock_model
|
|
context.es_agent = agent
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _run_stream(coro: Any) -> Any:
|
|
"""Run an async coroutine synchronously."""
|
|
return asyncio.get_event_loop().run_until_complete(coro)
|
|
|
|
|
|
@when("I call execute_stream with message {msg} (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream(context: Any, msg: str) -> None:
|
|
msg = msg.strip('"')
|
|
executor = context.es_executor
|
|
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
cleanup_should_raise = getattr(context, "es_cleanup_should_raise", False)
|
|
intermediate_override = getattr(context, "es_intermediate_override", None)
|
|
fn_override = getattr(context, "es_fn_override", None)
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
if mock_model is not None:
|
|
# Build a fake LLMAgent with our mock model
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="stream_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
agent.chat_model = mock_model
|
|
if cleanup_should_raise:
|
|
agent.cleanup = AsyncMock(side_effect=RuntimeError("cleanup failed"))
|
|
|
|
# Apply intermediate node override for M5 branch coverage tests
|
|
if intermediate_override == "non_dict_last_msg":
|
|
# Return a result where messages[-1] is not a dict
|
|
non_dict_response = MagicMock()
|
|
non_dict_response.content = "intermediate_response"
|
|
non_dict_response.usage_metadata = None
|
|
non_dict_response.response_metadata = None
|
|
agent.chat_model.ainvoke = AsyncMock(return_value=non_dict_response)
|
|
# Make the result have a non-dict last message
|
|
orig_ainvoke = agent.chat_model.ainvoke
|
|
|
|
async def _ainvoke_non_dict_last_msg(messages: Any) -> Any:
|
|
r = MagicMock()
|
|
r.content = "intermediate_response"
|
|
r.usage_metadata = None
|
|
r.response_metadata = None
|
|
return r
|
|
|
|
agent.chat_model.ainvoke = _ainvoke_non_dict_last_msg
|
|
# Patch Node._execute_agent to return non-dict last_msg
|
|
from cleveractors.langgraph import nodes as _nodes_mod
|
|
|
|
_orig_execute_agent = _nodes_mod.Node._execute_agent
|
|
|
|
async def _patched_execute_agent(
|
|
self_node: Any, state: Any
|
|
) -> dict[str, Any]:
|
|
result = await _orig_execute_agent(self_node, state)
|
|
if isinstance(result, dict) and "messages" in result:
|
|
# Replace last message with a non-dict value
|
|
result["messages"] = [*result["messages"][:-1], "non_dict_msg"]
|
|
return result
|
|
|
|
_nodes_mod.Node._execute_agent = _patched_execute_agent # type: ignore[method-assign]
|
|
context.es_patched_execute_agent = (_nodes_mod, _orig_execute_agent)
|
|
|
|
elif intermediate_override == "no_messages":
|
|
# Return a result dict with no 'messages' key
|
|
from cleveractors.langgraph import nodes as _nodes_mod
|
|
|
|
_orig_execute_agent = _nodes_mod.Node._execute_agent
|
|
|
|
async def _patched_no_messages(
|
|
self_node: Any, state: Any
|
|
) -> dict[str, Any]:
|
|
result = await _orig_execute_agent(self_node, state)
|
|
if isinstance(result, dict):
|
|
result.pop("messages", None)
|
|
result["output"] = "intermediate_output"
|
|
return result
|
|
|
|
_nodes_mod.Node._execute_agent = _patched_no_messages # type: ignore[method-assign]
|
|
context.es_patched_execute_agent = (_nodes_mod, _orig_execute_agent)
|
|
|
|
elif intermediate_override == "non_dict_result":
|
|
# Return a non-dict value from node.execute()
|
|
from cleveractors.langgraph import nodes as _nodes_mod
|
|
|
|
_orig_execute = _nodes_mod.Node.execute
|
|
|
|
async def _patched_non_dict_result(self_node: Any, state: Any) -> Any:
|
|
# Return a string instead of a dict for the intermediate node
|
|
if self_node.name == "agent_a":
|
|
return "string_result"
|
|
return await _orig_execute(self_node, state)
|
|
|
|
_nodes_mod.Node.execute = _patched_non_dict_result # type: ignore[method-assign]
|
|
context.es_patched_execute = (_nodes_mod, _orig_execute)
|
|
|
|
elif intermediate_override == "raises":
|
|
# Make ainvoke raise RuntimeError for the intermediate node
|
|
from cleveractors.langgraph import nodes as _nodes_mod
|
|
|
|
_orig_execute = _nodes_mod.Node.execute
|
|
|
|
async def _patched_raises(self_node: Any, state: Any) -> Any:
|
|
if self_node.name == "agent_a":
|
|
raise RuntimeError("intermediate ainvoke failed")
|
|
return await _orig_execute(self_node, state)
|
|
|
|
_nodes_mod.Node.execute = _patched_raises # type: ignore[method-assign]
|
|
context.es_patched_execute = (_nodes_mod, _orig_execute)
|
|
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
|
|
# Apply function node override for non-AGENT branch coverage tests
|
|
if fn_override is not None:
|
|
from cleveractors.langgraph import nodes as _fn_nodes_mod
|
|
|
|
_fn_orig_execute = _fn_nodes_mod.Node.execute
|
|
|
|
if fn_override == "non_dict_last_msg":
|
|
|
|
async def _fn_patched_non_dict_last_msg(
|
|
self_node: Any, state: Any
|
|
) -> Any:
|
|
result = await _fn_orig_execute(self_node, state)
|
|
if self_node.config.type.value == "function" and isinstance(
|
|
result, dict
|
|
):
|
|
# Add a messages list with a non-dict last entry to
|
|
# exercise the 'else: output_message = str(last_msg)' branch
|
|
result["messages"] = ["non_dict_msg"]
|
|
return result
|
|
|
|
_fn_nodes_mod.Node.execute = _fn_patched_non_dict_last_msg # type: ignore[method-assign]
|
|
|
|
elif fn_override == "no_messages":
|
|
|
|
async def _fn_patched_no_messages(self_node: Any, state: Any) -> Any:
|
|
result = await _fn_orig_execute(self_node, state)
|
|
if self_node.config.type.value == "function" and isinstance(
|
|
result, dict
|
|
):
|
|
result.pop("messages", None)
|
|
result["output"] = "fn_output"
|
|
return result
|
|
|
|
_fn_nodes_mod.Node.execute = _fn_patched_no_messages # type: ignore[method-assign]
|
|
|
|
elif fn_override == "non_dict_result":
|
|
|
|
async def _fn_patched_non_dict_result(
|
|
self_node: Any, state: Any
|
|
) -> Any:
|
|
if self_node.config.type.value == "function":
|
|
return "string_fn_result"
|
|
return await _fn_orig_execute(self_node, state)
|
|
|
|
_fn_nodes_mod.Node.execute = _fn_patched_non_dict_result # type: ignore[method-assign]
|
|
|
|
elif fn_override == "raises":
|
|
|
|
async def _fn_patched_raises(self_node: Any, state: Any) -> Any:
|
|
if self_node.config.type.value == "function":
|
|
raise RuntimeError("function node failed")
|
|
return await _fn_orig_execute(self_node, state)
|
|
|
|
_fn_nodes_mod.Node.execute = _fn_patched_raises # type: ignore[method-assign]
|
|
|
|
elif fn_override == "token_usage":
|
|
|
|
async def _fn_patched_token_usage(self_node: Any, state: Any) -> Any:
|
|
result = await _fn_orig_execute(self_node, state)
|
|
if self_node.config.type.value == "function" and isinstance(
|
|
result, dict
|
|
):
|
|
# Add _node_token_usage to exercise the token-usage branch
|
|
result["_node_token_usage"] = {
|
|
"node_id": self_node.name,
|
|
"provider": "test",
|
|
"model": "test-model",
|
|
"prompt_tokens": 2,
|
|
"completion_tokens": 3,
|
|
}
|
|
return result
|
|
|
|
_fn_nodes_mod.Node.execute = _fn_patched_token_usage # type: ignore[method-assign]
|
|
|
|
context.es_patched_fn_execute = (_fn_nodes_mod, _fn_orig_execute)
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in executor.execute_stream(msg):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
finally:
|
|
# Restore any patched methods
|
|
if hasattr(context, "es_patched_execute_agent"):
|
|
_mod, _orig = context.es_patched_execute_agent
|
|
_mod.Node._execute_agent = _orig
|
|
del context.es_patched_execute_agent
|
|
if hasattr(context, "es_patched_execute"):
|
|
_mod, _orig = context.es_patched_execute
|
|
_mod.Node.execute = _orig
|
|
del context.es_patched_execute
|
|
if hasattr(context, "es_patched_fn_execute"):
|
|
_mod, _orig = context.es_patched_fn_execute
|
|
_mod.Node.execute = _orig
|
|
del context.es_patched_fn_execute
|
|
|
|
|
|
@when("I call execute_stream on graph with context manager (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_context_manager(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
if mock_model is not None:
|
|
for ag in graph.agents.values():
|
|
if hasattr(ag, "chat_model"):
|
|
ag.chat_model = mock_model
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph.execute_stream("test"):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I call execute_stream with empty content history entry (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_empty_history(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
if mock_model is not None:
|
|
for ag in graph.agents.values():
|
|
if hasattr(ag, "chat_model"):
|
|
ag.chat_model = mock_model
|
|
try:
|
|
tokens: list[str] = []
|
|
# Pass history with an empty content entry (should be skipped)
|
|
history = [
|
|
{"role": "user", "content": ""}, # Empty content - should be skipped
|
|
{"role": "user", "content": "real message"},
|
|
]
|
|
async for token in graph.execute_stream("test", conversation_history=history):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I call execute on the pure graph with a message (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_pure_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
try:
|
|
result, state, usages = await graph.execute("test")
|
|
context.es_graph_result = result
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_graph_result = None
|
|
|
|
|
|
@when("I call execute_stream on the running graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_running_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph.execute_stream("test"):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except RuntimeError as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I call execute_stream on the dangling graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_dangling(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph.execute_stream("test"):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I call execute_stream on the bad-agent graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_bad_agent(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph.execute_stream("test"):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I call execute_stream on graph with last context (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_last_context(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
if mock_model is not None:
|
|
# Update the agent's chat model if available
|
|
for agent in graph.agents.values():
|
|
if hasattr(agent, "chat_model"):
|
|
agent.chat_model = mock_model
|
|
try:
|
|
tokens: list[str] = []
|
|
# No global_context passed (triggers last_context usage)
|
|
async for token in graph.execute_stream("test", global_context=None):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I execute the graph with initial_state and a message (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_graph_initial_state(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
initial_state = context.es_initial_state
|
|
try:
|
|
result, state_out, _ = await graph.execute(
|
|
input_message="test",
|
|
initial_state=initial_state,
|
|
)
|
|
context.es_graph_result = result
|
|
context.es_graph_state_out = state_out
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_graph_result = None
|
|
context.es_graph_state_out = {}
|
|
|
|
|
|
@when("I run execute_stream with initial_state on graph executor (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_with_state(context: Any) -> None:
|
|
executor = context.es_executor
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
if mock_model is not None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="stream_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
agent.chat_model = mock_model
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
initial_state = {"some_key": "some_value", "conversation_stage": "greeting"}
|
|
async for token in executor.execute_stream("test", state=initial_state):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I run execute_stream on graph with messages param (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_graph_with_messages(context: Any) -> None:
|
|
executor = context.es_executor
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
if mock_model is not None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="stream_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
agent.chat_model = mock_model
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
messages = [
|
|
{"role": "user", "content": "What is AI?"},
|
|
{"role": "assistant", "content": "AI is..."},
|
|
]
|
|
async for token in executor.execute_stream("follow up", messages=messages):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I run execute_stream with history on executor (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_with_history(context: Any) -> None:
|
|
executor = context.es_executor
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
if mock_model is not None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="stream_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
agent.chat_model = mock_model
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
history = [
|
|
{"role": "user", "content": "previous"},
|
|
{"role": "assistant", "content": "yes"},
|
|
]
|
|
async for token in executor.execute_stream("Hi", messages=history):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I run execute_stream with messages param on executor (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_with_messages_param(context: Any) -> None:
|
|
executor = context.es_executor
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
if mock_model is not None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="stream_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
agent.chat_model = mock_model
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
messages = [
|
|
{"role": "user", "content": "hello"},
|
|
{"role": "assistant", "content": "hi"},
|
|
]
|
|
async for token in executor.execute_stream("question", messages=messages):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I attempt execute_stream expecting an error with message {msg} (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_expecting_error(context: Any, msg: str) -> None:
|
|
msg = msg.strip('"')
|
|
executor = context.es_executor
|
|
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
factory_should_raise = getattr(context, "es_factory_should_raise", None)
|
|
graph_raises_runtime_error = getattr(
|
|
context, "es_graph_raises_runtime_error", False
|
|
)
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
if factory_should_raise == "ConfigurationError":
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(
|
|
side_effect=ConfigurationError("mock factory config error")
|
|
)
|
|
mock_factory.return_value = mock_factory_instance
|
|
elif factory_should_raise == "RuntimeError":
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(
|
|
side_effect=RuntimeError("unexpected factory error")
|
|
)
|
|
mock_factory.return_value = mock_factory_instance
|
|
else:
|
|
# Use a pre-built agent if provided (e.g. M4 fix: agent with patched
|
|
# update_memory to test billing-integrity on post-stream failure).
|
|
# Otherwise, build a fresh LLMAgent from the mock model.
|
|
pre_built_agent = getattr(context, "es_agent", None)
|
|
build_chat_model_should_raise = getattr(
|
|
context, "es_build_chat_model_should_raise", False
|
|
)
|
|
if pre_built_agent is not None:
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(
|
|
return_value=pre_built_agent
|
|
)
|
|
mock_factory.return_value = mock_factory_instance
|
|
elif build_chat_model_should_raise:
|
|
# m2 fix: create an agent without injecting chat_model so that
|
|
# lazy init is triggered inside stream_message(). Patch
|
|
# build_chat_model to raise ConfigurationError so the lazy-init
|
|
# path raises before any tokens are streamed.
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="lazy_init_error_llm",
|
|
config=config,
|
|
template_renderer=renderer,
|
|
)
|
|
# Do NOT set agent.chat_model — leave it None so lazy init fires.
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
# Patch build_chat_model to raise ConfigurationError.
|
|
with patch(
|
|
"cleveractors.agents.llm.build_chat_model",
|
|
side_effect=ConfigurationError("mock build_chat_model failure"),
|
|
):
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in executor.execute_stream(msg):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except (ExecutionError, ConfigurationError) as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
return # early return — execution already done inside the patch
|
|
else:
|
|
# Use slow mock if provided, otherwise use a fast one for limit testing
|
|
if mock_model is None:
|
|
mock_model = _make_async_chunks(["token"], None)
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="stream_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
agent.chat_model = mock_model
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
|
|
# M1 fix: patch PureLangGraph.execute_stream to raise RuntimeError so
|
|
# the bare except-Exception block in _execute_graph_stream is exercised.
|
|
if graph_raises_runtime_error:
|
|
from cleveractors.langgraph import pure_graph as _pg_mod
|
|
|
|
broken_state_capture = getattr(
|
|
context, "es_graph_broken_state_capture", False
|
|
)
|
|
|
|
async def _raising_execute_stream(*_args: Any, **_kwargs: Any) -> Any:
|
|
raise RuntimeError("unexpected graph error")
|
|
yield # make it an async generator
|
|
|
|
if broken_state_capture:
|
|
# Also make _last_stream_state raise to exercise the inner
|
|
# except block in the M1 fix (state-capture failure path).
|
|
# We use a custom execute_stream that sets _last_stream_state
|
|
# to a broken object before raising, so the except block
|
|
# tries to call dict() on it and gets an exception.
|
|
|
|
class _BrokenState:
|
|
"""Object that raises when dict() is called on it."""
|
|
|
|
def keys(self) -> None:
|
|
raise RuntimeError("state capture failed")
|
|
|
|
async def _raising_execute_stream_broken(
|
|
self_graph: Any, *_args: Any, **_kwargs: Any
|
|
) -> Any:
|
|
# Set _last_stream_state to a broken object so that
|
|
# dict(graph._last_stream_state) raises in the M1 handler.
|
|
self_graph._last_stream_state = _BrokenState()
|
|
raise RuntimeError("unexpected graph error")
|
|
yield # make it an async generator
|
|
|
|
with patch.object(
|
|
_pg_mod.PureLangGraph,
|
|
"execute_stream",
|
|
new=_raising_execute_stream_broken,
|
|
):
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in executor.execute_stream(msg):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except (ExecutionError, ConfigurationError) as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
else:
|
|
with patch.object(
|
|
_pg_mod.PureLangGraph,
|
|
"execute_stream",
|
|
new=_raising_execute_stream,
|
|
):
|
|
try:
|
|
tokens = []
|
|
async for token in executor.execute_stream(msg):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except (ExecutionError, ConfigurationError) as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
else:
|
|
try:
|
|
tokens = []
|
|
async for token in executor.execute_stream(msg):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except (ExecutionError, ConfigurationError) as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I call stream_message with a string message")
|
|
@async_run_until_complete
|
|
async def step_es_stream_message(context: Any) -> None:
|
|
import logging
|
|
|
|
agent = context.es_agent
|
|
if hasattr(context, "es_mock_model") and context.es_mock_model is not None:
|
|
agent.chat_model = context.es_mock_model
|
|
|
|
# Capture warning log records emitted during stream_message so that
|
|
# Then-steps can assert that the no-usage warning was emitted.
|
|
captured_warnings: list[logging.LogRecord] = []
|
|
|
|
class _WarningCapture(logging.Handler):
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
if record.levelno >= logging.WARNING:
|
|
captured_warnings.append(record)
|
|
|
|
_handler = _WarningCapture()
|
|
_root_logger = logging.getLogger()
|
|
_root_logger.addHandler(_handler)
|
|
|
|
# Also patch _log_no_usage_metadata to record calls directly;
|
|
# this is robust even when slipcover wraps the logging machinery
|
|
# (coverage_report session).
|
|
_cause_records: list[str] = []
|
|
_original = agent._log_no_usage_metadata
|
|
|
|
def _tracking_log(self: Any, cause: str) -> None:
|
|
_cause_records.append(cause)
|
|
_original(cause)
|
|
|
|
agent._log_no_usage_metadata = _tracking_log.__get__(agent, type(agent)) # type: ignore[assignment]
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in agent.stream_message("Hello", None):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
# Capture last_token_usage_var inside the async context (ContextVar is
|
|
# task-local; it cannot be read from a synchronous then-step).
|
|
context.es_captured_usage_var = last_token_usage_var.get((0, 0))
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
context.es_captured_usage_var = (0, 0)
|
|
finally:
|
|
_root_logger.removeHandler(_handler)
|
|
agent._log_no_usage_metadata = _original # type: ignore[method-assign]
|
|
|
|
context.es_captured_warnings = captured_warnings
|
|
context.es_log_causes = _cause_records
|
|
|
|
|
|
@when("I call stream_message with _temperature_override {override} in context (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_stream_message_temp_override(context: Any, override: str) -> None:
|
|
"""Call stream_message with a _temperature_override in the context dict (n5/M1)."""
|
|
agent = context.es_agent
|
|
# Parse the override value — may be a number or a string like "bad"
|
|
try:
|
|
import ast
|
|
|
|
override_val: Any = ast.literal_eval(override)
|
|
except (ValueError, SyntaxError):
|
|
override_val = override # keep as raw string for the invalid-type test
|
|
|
|
ctx: dict[str, Any] = {"_temperature_override": override_val}
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in agent.stream_message("Hello", ctx):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@when("I start iterating execute_stream and stop after first token (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_partial_stream(context: Any) -> None:
|
|
executor = context.es_executor
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
context.es_agent = None # will be set below if a mock agent is created
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
if mock_model is not None:
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="stream_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
agent.chat_model = mock_model
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
context.es_agent = agent # capture for Then-step assertions
|
|
|
|
try:
|
|
gen = executor.execute_stream("test")
|
|
first_token = await gen.__anext__()
|
|
context.es_tokens = [first_token]
|
|
context.es_error = None
|
|
# Don't exhaust the generator (simulate abandonment)
|
|
except StopAsyncIteration:
|
|
context.es_tokens = []
|
|
context.es_error = None
|
|
except Exception as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("executor.last_result should be None (stream)")
|
|
def step_es_last_result_none(context: Any) -> None:
|
|
assert context.es_executor is not None
|
|
assert context.es_executor.last_result is None, (
|
|
f"Expected last_result to be None, got {context.es_executor.last_result!r}"
|
|
)
|
|
|
|
|
|
@then("the collected tokens should be {expected_str} (stream)")
|
|
def step_es_assert_tokens(context: Any, expected_str: str) -> None:
|
|
import ast
|
|
|
|
expected = ast.literal_eval(expected_str)
|
|
assert context.es_tokens is not None, "No tokens were collected"
|
|
assert context.es_tokens == expected, (
|
|
f"Expected tokens {expected!r}, got {context.es_tokens!r}"
|
|
)
|
|
|
|
|
|
@then("executor.last_result should be an ActorResult (stream)")
|
|
def step_es_last_result_actor_result(context: Any) -> None:
|
|
assert context.es_executor is not None
|
|
assert context.es_executor.last_result is not None, "last_result is None"
|
|
assert isinstance(context.es_executor.last_result, ActorResult), (
|
|
f"Expected ActorResult, got {type(context.es_executor.last_result)}"
|
|
)
|
|
|
|
|
|
@then("executor.last_result.response should equal {expected_str} (stream)")
|
|
def step_es_last_result_response(context: Any, expected_str: str) -> None:
|
|
import ast
|
|
|
|
expected = ast.literal_eval(expected_str)
|
|
result = context.es_executor.last_result
|
|
assert result is not None
|
|
assert result.response == expected, (
|
|
f"Expected response {expected!r}, got {result.response!r}"
|
|
)
|
|
|
|
|
|
@then("executor.last_result.prompt_tokens should be {n:d} (stream)")
|
|
def step_es_prompt_tokens(context: Any, n: int) -> None:
|
|
result = context.es_executor.last_result
|
|
assert result is not None
|
|
assert result.prompt_tokens == n, (
|
|
f"Expected prompt_tokens={n}, got {result.prompt_tokens}"
|
|
)
|
|
|
|
|
|
@then("executor.last_result.completion_tokens should be {n:d} (stream)")
|
|
def step_es_completion_tokens(context: Any, n: int) -> None:
|
|
result = context.es_executor.last_result
|
|
assert result is not None
|
|
assert result.completion_tokens == n, (
|
|
f"Expected completion_tokens={n}, got {result.completion_tokens}"
|
|
)
|
|
|
|
|
|
@then("executor.last_result should have at least one NodeUsage entry (stream)")
|
|
def step_es_node_usages(context: Any) -> None:
|
|
result = context.es_executor.last_result
|
|
assert result is not None
|
|
assert len(result.nodes) >= 1, f"Expected at least 1 NodeUsage, got {result.nodes}"
|
|
|
|
|
|
@then("the yielded tokens should be {expected_str}")
|
|
def step_es_yielded_tokens(context: Any, expected_str: str) -> None:
|
|
import ast
|
|
|
|
expected = ast.literal_eval(expected_str)
|
|
assert context.es_tokens is not None
|
|
assert context.es_tokens == expected, (
|
|
f"Expected tokens {expected!r}, got {context.es_tokens!r}"
|
|
)
|
|
|
|
|
|
@then("_last_token_usage should be ({p:d}, {c:d}) after stream_message")
|
|
def step_es_last_token_usage(context: Any, p: int, c: int) -> None:
|
|
agent = context.es_agent
|
|
assert agent._last_token_usage == (p, c), (
|
|
f"Expected _last_token_usage=({p},{c}), got {agent._last_token_usage}"
|
|
)
|
|
|
|
|
|
@then("last_token_usage_var was ({p:d}, {c:d}) inside the async step")
|
|
def step_es_last_token_usage_var(context: Any, p: int, c: int) -> None:
|
|
# last_token_usage_var is captured inside the async step's context
|
|
# (ContextVar values are task-local and cannot be read from a sync step).
|
|
val = getattr(context, "es_captured_usage_var", (0, 0))
|
|
assert val == (p, c), (
|
|
f"Expected last_token_usage_var captured as ({p},{c}), got {val}"
|
|
)
|
|
|
|
|
|
@then("_last_token_usage should be zero after stream_message")
|
|
def step_es_last_token_usage_zero(context: Any) -> None:
|
|
agent = context.es_agent
|
|
assert agent._last_token_usage == (0, 0), (
|
|
f"Expected _last_token_usage=(0,0), got {agent._last_token_usage}"
|
|
)
|
|
|
|
|
|
@then("a no-usage warning was emitted during stream_message")
|
|
def step_es_no_usage_warning_emitted(context: Any) -> None:
|
|
"""Verify that a warning with one of the _CAUSE_* strings was logged.
|
|
|
|
The LLMAgent._log_no_usage_metadata() method logs at WARNING level with
|
|
a message containing the cause string. This step checks that at least
|
|
one such warning was captured during the stream_message call.
|
|
"""
|
|
from cleveractors.agents.llm import (
|
|
_CAUSE_RESPONSE_METADATA_MISSING,
|
|
_CAUSE_RESPONSE_METADATA_NO_TOKEN_USAGE,
|
|
_CAUSE_RESPONSE_METADATA_NOT_DICT,
|
|
_CAUSE_USAGE_METADATA_EMPTY,
|
|
)
|
|
|
|
_cause_strings = (
|
|
_CAUSE_USAGE_METADATA_EMPTY,
|
|
_CAUSE_RESPONSE_METADATA_MISSING,
|
|
_CAUSE_RESPONSE_METADATA_NO_TOKEN_USAGE,
|
|
_CAUSE_RESPONSE_METADATA_NOT_DICT,
|
|
)
|
|
# Primary check: the tracking wrapper recorded calls to _log_no_usage_metadata.
|
|
# This is robust across normal behave and slipcover (coverage_report).
|
|
_log_causes = getattr(context, "es_log_causes", None)
|
|
if _log_causes:
|
|
return # at least one cause was recorded — warning was emitted
|
|
|
|
# Fallback (pre-existing): check the logging handler captured a warning record.
|
|
captured = getattr(context, "es_captured_warnings", [])
|
|
matching = [
|
|
r for r in captured if any(cause in r.getMessage() for cause in _cause_strings)
|
|
]
|
|
assert matching, (
|
|
"Expected a no-usage-metadata warning to be emitted during stream_message, "
|
|
f"but none was found. Captured warnings: {[r.getMessage() for r in captured]}"
|
|
)
|
|
|
|
|
|
@then(
|
|
"the temperature was {override:g} during streaming and"
|
|
" restored to {original:g} afterwards (stream)"
|
|
)
|
|
def step_es_temp_override_verified(
|
|
context: Any, override: float, original: float
|
|
) -> None:
|
|
"""Verify _temperature_override was applied during streaming and restored (n5/M1)."""
|
|
assert context.es_error is None, f"Unexpected error: {context.es_error}"
|
|
assert context.es_tokens is not None, "No tokens were yielded"
|
|
assert len(context.es_tokens) >= 1, f"Expected tokens, got: {context.es_tokens}"
|
|
|
|
# Verify the temperature seen inside astream() was the override value
|
|
temps = getattr(context, "es_temperatures_seen", [])
|
|
assert len(temps) >= 1, "astream() was never called — no temperature recorded"
|
|
assert temps[0] == override, (
|
|
f"Expected temperature {override} during streaming, got {temps[0]}"
|
|
)
|
|
|
|
# Verify the temperature was restored after streaming
|
|
agent = context.es_agent
|
|
actual_temp = agent._chat_model.temperature # type: ignore[union-attr]
|
|
assert actual_temp == original, (
|
|
f"Expected temperature restored to {original}, got {actual_temp}"
|
|
)
|
|
|
|
|
|
@then("a ConfigurationError is raised from stream_message (stream)")
|
|
def step_es_config_error_stream_message(context: Any) -> None:
|
|
"""Verify ConfigurationError is raised for invalid _temperature_override (n5/M1)."""
|
|
assert context.es_error is not None, "Expected ConfigurationError but none raised"
|
|
assert isinstance(context.es_error, ConfigurationError), (
|
|
f"Expected ConfigurationError, got {type(context.es_error).__name__}: "
|
|
f"{context.es_error}"
|
|
)
|
|
|
|
|
|
@then('an ExecutionError with kind "{kind}" should be raised (stream)')
|
|
def step_es_execution_error_kind(context: Any, kind: str) -> None:
|
|
assert context.es_error is not None, "Expected an error but none was raised"
|
|
assert isinstance(context.es_error, ExecutionError), (
|
|
f"Expected ExecutionError, got {type(context.es_error)}"
|
|
)
|
|
assert context.es_error.kind == kind, (
|
|
f"Expected kind={kind!r}, got {context.es_error.kind!r}"
|
|
)
|
|
|
|
|
|
@then("executor.last_result token counts should be zero (stream)")
|
|
def step_es_last_result_zero_tokens(context: Any) -> None:
|
|
"""Verify that executor.last_result has zero prompt and completion tokens.
|
|
|
|
Used by the m2 billing-integrity test: when build_chat_model() raises
|
|
ConfigurationError during lazy init, no tokens were consumed, so both
|
|
prompt_tokens and completion_tokens must be 0.
|
|
"""
|
|
result = context.es_executor.last_result
|
|
assert result is not None, "last_result is None — expected a partial ActorResult"
|
|
assert result.prompt_tokens == 0, (
|
|
f"Expected prompt_tokens=0, got {result.prompt_tokens}"
|
|
)
|
|
assert result.completion_tokens == 0, (
|
|
f"Expected completion_tokens=0, got {result.completion_tokens}"
|
|
)
|
|
|
|
|
|
@then("a ConfigurationError should be raised about unsupported streaming type (stream)")
|
|
def step_es_config_error_type(context: Any) -> None:
|
|
assert context.es_error is not None, (
|
|
"Expected a ConfigurationError but none was raised"
|
|
)
|
|
assert isinstance(context.es_error, ConfigurationError), (
|
|
f"Expected ConfigurationError, got {type(context.es_error)}"
|
|
)
|
|
|
|
|
|
@then("executor.last_result should have a no_llm placeholder node (stream)")
|
|
def step_es_no_llm_placeholder(context: Any) -> None:
|
|
result = context.es_executor.last_result
|
|
assert result is not None, "last_result is None"
|
|
assert len(result.nodes) >= 1, "Expected at least one node in last_result.nodes"
|
|
# The placeholder node has provider="graph" and model="<no_llm>"
|
|
placeholder_found = any(n.model == "<no_llm>" for n in result.nodes)
|
|
assert placeholder_found, (
|
|
f"Expected no_llm placeholder node but got: {[(n.node_id, n.model) for n in result.nodes]}"
|
|
)
|
|
|
|
|
|
@then("executor.last_result.nodes should contain a no_llm placeholder (stream)")
|
|
def step_es_nodes_contain_no_llm_placeholder(context: Any) -> None:
|
|
"""Verify that executor.last_result.nodes contains a <no_llm> placeholder.
|
|
|
|
Used by the m1 fix test: when create_agent() raises on the LLM streaming
|
|
path, executor.last_result is populated with a placeholder node whose
|
|
model is "<no_llm>", mirroring the graph path's N5 fix.
|
|
"""
|
|
result = context.es_executor.last_result
|
|
assert result is not None, (
|
|
"last_result is None — expected a placeholder ActorResult"
|
|
)
|
|
assert len(result.nodes) >= 1, "Expected at least one node in last_result.nodes"
|
|
placeholder_found = any(n.model == "<no_llm>" for n in result.nodes)
|
|
assert placeholder_found, (
|
|
f"Expected <no_llm> placeholder node in last_result.nodes but got: "
|
|
f"{[(n.node_id, n.model) for n in result.nodes]}"
|
|
)
|
|
|
|
|
|
@given("a Node stream_agent with a ToolAgent (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_node_stream_tool_agent(context: Any) -> None:
|
|
"""Test Node.stream_agent() with a ToolAgent (non-LLM fallback path)."""
|
|
from cleveractors.agents.tool import ToolAgent
|
|
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
|
|
tool_agent = ToolAgent(
|
|
name="echo_tool",
|
|
config={"tools": ["echo"]},
|
|
template_renderer=TemplateRenderer(),
|
|
)
|
|
tool_agent.process_message = AsyncMock(return_value="tool_response")
|
|
|
|
node_config = NodeConfig(name="tool_node", type=NodeType.AGENT, agent="echo_tool")
|
|
node = Node(config=node_config, agents={"echo_tool": tool_agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "test tool"}]
|
|
state.metadata = {"current_message": "test tool"}
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in node._stream_agent(state):
|
|
tokens.append(token)
|
|
context.es_tool_agent_tokens = tokens
|
|
context.es_tool_agent_error = None
|
|
except Exception as e:
|
|
context.es_tool_agent_error = e
|
|
context.es_tool_agent_tokens = []
|
|
|
|
|
|
@given("a Node stream_agent without current_message in state (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_node_stream_no_current_msg(context: Any) -> None:
|
|
"""Test Node.stream_agent() when state has messages but no current_message."""
|
|
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_agent", config=config, template_renderer=renderer)
|
|
agent.chat_model = _make_async_chunks(["hello_no_cm"], None)
|
|
|
|
node_config = NodeConfig(name="test_node", type=NodeType.AGENT, agent="test_agent")
|
|
node = Node(config=node_config, agents={"test_agent": agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [
|
|
{"role": "user", "content": "what is the question?"},
|
|
{"role": "assistant", "content": "the answer"},
|
|
]
|
|
state.metadata = {} # No current_message
|
|
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in node._stream_agent(state):
|
|
tokens.append(token)
|
|
context.es_node_tokens = tokens
|
|
context.es_node_error = None
|
|
except Exception as e:
|
|
context.es_node_error = e
|
|
context.es_node_tokens = []
|
|
|
|
|
|
@given("a Node stream_agent that raises during streaming (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_node_stream_error(context: Any) -> None:
|
|
"""Test Node.stream_agent() error path: agent raises during stream."""
|
|
from cleveractors.langgraph.nodes import Node, NodeConfig, NodeType
|
|
from cleveractors.langgraph.state import GraphState
|
|
|
|
# Create an LLMAgent whose astream raises
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="error_agent", config=config, template_renderer=renderer)
|
|
|
|
async def _raise_astream(_messages: Any) -> Any:
|
|
raise RuntimeError("astream failed")
|
|
yield # make it a generator
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _raise_astream
|
|
mock_model.temperature = 0.7
|
|
agent.chat_model = mock_model
|
|
|
|
node_config = NodeConfig(name="test_node", type=NodeType.AGENT, agent="test_agent")
|
|
node = Node(config=node_config, agents={"test_agent": agent})
|
|
|
|
state = GraphState()
|
|
state.messages = [{"role": "user", "content": "test"}]
|
|
state.metadata = {"current_message": "test"}
|
|
|
|
try:
|
|
error_tokens: list[str] = []
|
|
async for token in node._stream_agent(state):
|
|
error_tokens.append(token)
|
|
context.es_stream_agent_tokens = error_tokens
|
|
context.es_stream_agent_error = None
|
|
except Exception as e:
|
|
context.es_stream_agent_error = e
|
|
context.es_stream_agent_tokens = []
|
|
|
|
|
|
@then("the execution result should contain the restored state key (stream)")
|
|
def step_es_initial_state_check(context: Any) -> None:
|
|
state_out = getattr(context, "es_graph_state_out", {})
|
|
# The initial_state was {"restored_key": "restored_value", "stage": "resumed"}
|
|
# These should be merged into the graph metadata context
|
|
assert context.es_error is None, f"Expected no error, got {context.es_error}"
|
|
# The graph executed successfully
|
|
result = getattr(context, "es_graph_result", None)
|
|
assert result is not None, "Expected a result from graph.execute()"
|
|
# Verify that the initial_state keys were actually restored into the graph
|
|
# metadata and are present in the captured state_out dict.
|
|
assert "restored_key" in state_out, (
|
|
f"Expected 'restored_key' in state_out (initial_state restoration), "
|
|
f"got keys: {list(state_out.keys())}"
|
|
)
|
|
assert state_out["restored_key"] == "restored_value", (
|
|
f"Expected state_out['restored_key'] == 'restored_value', "
|
|
f"got {state_out['restored_key']!r}"
|
|
)
|
|
|
|
|
|
@then("a ConfigurationError should be raised (stream)")
|
|
def step_es_config_error(context: Any) -> None:
|
|
from cleveractors.core.exceptions import ConfigurationError as CE
|
|
|
|
assert context.es_error is not None, "Expected ConfigurationError but none raised"
|
|
assert isinstance(context.es_error, CE), (
|
|
f"Expected ConfigurationError, got {type(context.es_error).__name__}: {context.es_error}"
|
|
)
|
|
|
|
|
|
@then("the exception should be propagated from stream_agent (stream)")
|
|
def step_es_stream_agent_error_check(context: Any) -> None:
|
|
"""After the M1 fix, _stream_agent() re-raises exceptions rather than
|
|
converting them to error tokens. The contract is that an exception is
|
|
always raised — yielding error tokens is no longer acceptable behaviour.
|
|
|
|
Note: stream_message() wraps unexpected exceptions in ExecutionError, so
|
|
the exception that propagates from _stream_agent() is ExecutionError (not
|
|
the original RuntimeError from the mock astream).
|
|
"""
|
|
from cleveractors.core.exceptions import ExecutionError as ExecError
|
|
|
|
error = getattr(context, "es_stream_agent_error", None)
|
|
assert error is not None, (
|
|
"Expected an exception to be raised from _stream_agent(), but none was captured"
|
|
)
|
|
assert isinstance(error, ExecError), (
|
|
f"Expected ExecutionError from _stream_agent() (stream_message wraps exceptions), "
|
|
f"got {type(error).__name__}: {error}"
|
|
)
|
|
|
|
|
|
@then("a RuntimeError should be raised from execute_stream (stream)")
|
|
def step_es_runtime_error(context: Any) -> None:
|
|
assert context.es_error is not None, "Expected RuntimeError but none raised"
|
|
assert isinstance(context.es_error, RuntimeError), (
|
|
f"Expected RuntimeError, got {type(context.es_error).__name__}"
|
|
)
|
|
|
|
|
|
@then("the stream completes without error (stream)")
|
|
def step_es_no_error(context: Any) -> None:
|
|
assert context.es_error is None, (
|
|
f"Expected no error, got {type(context.es_error).__name__}: {context.es_error}"
|
|
)
|
|
|
|
|
|
@then("ValueError is raised from stream_agent (stream)")
|
|
def step_es_valueerror_stream_agent(context: Any) -> None:
|
|
err2 = getattr(context, "es_stream_agent_error2", None)
|
|
err3 = getattr(context, "es_stream_agent_error3", None)
|
|
err = err2 or err3
|
|
assert err is not None, "Expected ValueError from stream_agent but none raised"
|
|
assert isinstance(err, ValueError), (
|
|
f"Expected ValueError, got {type(err).__name__}: {err}"
|
|
)
|
|
|
|
|
|
@then("stream_agent yields from empty input (stream)")
|
|
def step_es_empty_msg_yield(context: Any) -> None:
|
|
err = getattr(context, "es_empty_msg_error", None)
|
|
tokens = getattr(context, "es_empty_msg_tokens", [])
|
|
assert err is None, f"Unexpected error: {err}"
|
|
# With empty messages, agent_input="" and the stream should yield something
|
|
assert len(tokens) >= 1, f"Expected at least one token, got: {tokens}"
|
|
|
|
|
|
@then("stream_agent processes truncated history (stream)")
|
|
def step_es_truncated_history(context: Any) -> None:
|
|
err = getattr(context, "es_trunc_error", None)
|
|
tokens = getattr(context, "es_trunc_tokens", [])
|
|
assert err is None, f"Unexpected error: {err}"
|
|
assert len(tokens) >= 1, (
|
|
f"Expected at least one token from truncated history, got: {tokens}"
|
|
)
|
|
|
|
|
|
@then("stream_agent processes nested context tokens (stream)")
|
|
def step_es_nested_context_tokens(context: Any) -> None:
|
|
err = getattr(context, "es_nested_error", None)
|
|
tokens = getattr(context, "es_nested_tokens", [])
|
|
assert err is None, f"Unexpected error: {err}"
|
|
assert len(tokens) >= 1, (
|
|
f"Expected at least one token from nested context, got: {tokens}"
|
|
)
|
|
|
|
|
|
@then("the stream completes with result (stream)")
|
|
def step_es_stream_completes_with_result(context: Any) -> None:
|
|
assert context.es_error is None, f"Unexpected error: {context.es_error}"
|
|
# Stream completed successfully
|
|
assert context.es_tokens is not None, "No tokens collected"
|
|
|
|
|
|
@then("an ExecutionError should be raised from execute (stream)")
|
|
def step_es_execution_error_from_execute(context: Any) -> None:
|
|
assert context.es_error is not None, "Expected ExecutionError but none raised"
|
|
assert isinstance(context.es_error, ExecutionError), (
|
|
f"Expected ExecutionError, got {type(context.es_error).__name__}: {context.es_error}"
|
|
)
|
|
|
|
|
|
@then("stream_agent yields the ToolAgent response (stream)")
|
|
def step_es_stream_agent_tool_check(context: Any) -> None:
|
|
tokens = getattr(context, "es_tool_agent_tokens", None)
|
|
error = getattr(context, "es_tool_agent_error", None)
|
|
assert error is None, f"Unexpected error: {error}"
|
|
assert tokens is not None, "No tokens from stream_agent with ToolAgent"
|
|
assert len(tokens) >= 1, f"Expected tokens from ToolAgent, got: {tokens}"
|
|
assert "tool_response" in tokens, (
|
|
f"Expected 'tool_response' in tokens, got: {tokens}"
|
|
)
|
|
|
|
|
|
@then("stream_agent yields tokens from last user message context (stream)")
|
|
def step_es_stream_agent_no_cm(context: Any) -> None:
|
|
tokens = getattr(context, "es_node_tokens", None)
|
|
error = getattr(context, "es_node_error", None)
|
|
assert error is None, f"Unexpected error: {error}"
|
|
assert tokens is not None, "No tokens from stream_agent"
|
|
# Should have yielded at least one token
|
|
assert len(tokens) >= 1, f"Expected at least one token, got: {tokens}"
|
|
|
|
|
|
@then("an ExecutionError should be raised from _execute_llm_stream (stream)")
|
|
def step_es_llm_stream_error(context: Any) -> None:
|
|
assert context.es_error is not None, "Expected error but none raised"
|
|
assert isinstance(context.es_error, ExecutionError), (
|
|
f"Expected ExecutionError, got {type(context.es_error).__name__}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# C1: auto_finish_active bypass in streaming loop detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with auto_finish_active set in state"
|
|
" and a repeated-visit agent node (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_auto_finish_graph(context: Any) -> None:
|
|
"""Graph: start → agent1 → end.
|
|
We directly call _stream_from_node after seeding _node_message_visits with
|
|
a count of 2 (so the loop detector would normally fire) and setting
|
|
auto_finish_active=True in state metadata so the bypass kicks in.
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
mock_model = getattr(context, "es_mock_model", None) or _make_async_chunks(
|
|
["SectionToken"], None
|
|
)
|
|
agent.chat_model = mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="auto_finish_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
# Initialise graph state (mirrors what execute_stream does internally)
|
|
init_payload: dict[str, Any] = {
|
|
"messages": [{"role": "user", "content": "test"}],
|
|
"metadata": {"auto_finish_active": True},
|
|
}
|
|
graph.state_manager.update_state(init_payload, node_id="input")
|
|
|
|
# Seed _node_message_visits so the loop detector fires on the next visit
|
|
graph._node_message_visits = {("agent1", "test"[:200]): 2}
|
|
graph._execution_path = []
|
|
graph._node_usages = []
|
|
graph._model_call_count = 0
|
|
graph._tool_call_count = 0
|
|
graph._accumulated_cost = 0.0
|
|
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@when("I call execute_stream on the auto_finish graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_auto_finish_graph(context: Any) -> None:
|
|
"""Directly call _stream_from_node to exercise the C1 bypass without
|
|
going through execute_stream (which resets _node_message_visits)."""
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph._stream_from_node("agent1", "test", depth=0):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# C2: Router-agent ping-pong detection in streaming path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a PureLangGraph with router-agent ping-pong setup for streaming (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_ping_pong_graph(context: Any) -> None:
|
|
"""Graph that would ping-pong between a router and an agent.
|
|
We pre-populate _execution_path to simulate the ping-pong pattern so the
|
|
guard fires immediately on the next visit of agent1.
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
mock_model = _make_async_chunks(["PingPongToken"], None)
|
|
agent.chat_model = mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="ping_pong_graph",
|
|
nodes={
|
|
"router": NodeConfig(name="router", type=NodeType.FUNCTION),
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
# Initialise graph state
|
|
init_payload: dict[str, Any] = {
|
|
"messages": [{"role": "user", "content": "test"}],
|
|
"metadata": {},
|
|
}
|
|
graph.state_manager.update_state(init_payload, node_id="input")
|
|
|
|
# Pre-populate _execution_path to simulate the ping-pong pattern:
|
|
# router → agent1 → router → (agent1 is about to be visited again)
|
|
graph._execution_path = ["router", "agent1", "router"]
|
|
graph._node_message_visits = {}
|
|
graph._node_usages = []
|
|
graph._model_call_count = 0
|
|
graph._tool_call_count = 0
|
|
graph._accumulated_cost = 0.0
|
|
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@when("I call execute_stream on the ping-pong graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_ping_pong_graph(context: Any) -> None:
|
|
"""Directly call _stream_from_node to exercise the C2 ping-pong guard."""
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph._stream_from_node("agent1", "test", depth=0):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# C3: No routing command → return to user guard in streaming path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"an Executor with an agent-then-router graph config and no routing command (stream)"
|
|
)
|
|
def step_es_agent_then_router_graph(context: Any) -> None:
|
|
"""Graph: start → agent1 → router → end.
|
|
The agent output has no routing prefix, so the C3 guard should short-circuit
|
|
and return the agent output directly to the user.
|
|
"""
|
|
config = {
|
|
"name": "agent_router_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
"router": {"type": "function"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "router"},
|
|
{"source": "router", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M2: executor.last_result populated on exception path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an Executor with a two-agent sequential graph and max_model_calls 1 (stream)")
|
|
def step_es_two_agent_max_model_calls_1(context: Any) -> None:
|
|
"""Graph with two agent nodes and max_model_calls=1.
|
|
The first agent call succeeds; the second raises ExecutionError(kind='model_calls').
|
|
This exercises the M2 fix: executor.last_result should be populated even on error.
|
|
"""
|
|
config = {
|
|
"name": "stream_model_calls_1_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
"agent2": {"type": "agent", "agent": "agent2"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "agent2"},
|
|
{"source": "agent2", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent2": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_model_calls": 1},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@then(
|
|
"executor.last_result should be an ActorResult with token counts after the error (stream)"
|
|
)
|
|
def step_es_last_result_after_error(context: Any) -> None:
|
|
"""M2 fix: executor.last_result should be populated with token counts even on error.
|
|
|
|
The mock has usage prompt=10 completion=20. The first agent node (agent1)
|
|
is an intermediate node so it uses ainvoke(); its token counts are captured
|
|
from the ainvoke response. After the model_calls limit fires before agent2
|
|
executes, last_result must reflect the partial billing data from agent1.
|
|
"""
|
|
executor = context.es_executor
|
|
assert executor.last_result is not None, (
|
|
"executor.last_result should be populated even on exception path (M2 fix)"
|
|
)
|
|
assert isinstance(executor.last_result, ActorResult), (
|
|
f"Expected ActorResult, got {type(executor.last_result).__name__}"
|
|
)
|
|
# Verify that token counts from the completed agent1 node are preserved.
|
|
# agent1 used ainvoke() (intermediate node); the mock ainvoke_response has
|
|
# usage_metadata = {"input_tokens": 10, "output_tokens": 20}.
|
|
result = executor.last_result
|
|
assert len(result.nodes) >= 1, (
|
|
f"Expected at least one NodeUsage entry, got {len(result.nodes)}"
|
|
)
|
|
assert result.nodes[0].prompt_tokens == 10, (
|
|
f"Expected prompt_tokens=10 (M2 billing integrity), got {result.nodes[0].prompt_tokens}"
|
|
)
|
|
assert result.nodes[0].completion_tokens == 20, (
|
|
f"Expected completion_tokens=20 (M2 billing integrity), got {result.nodes[0].completion_tokens}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M2 fix: LLM path — executor.last_result populated on exception path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"a mock astream that raises ExecutionError mid-stream"
|
|
" with usage prompt={p:d} completion={c:d} (stream)"
|
|
)
|
|
def step_es_mock_astream_raises_exec_error_with_usage(
|
|
context: Any, p: int, c: int
|
|
) -> None:
|
|
"""Mock astream that completes successfully (yielding a usage-bearing chunk)
|
|
then raises during the post-stream memory update.
|
|
|
|
M4 fix: the previous implementation raised before yielding any chunk, so
|
|
_captured_prompt stayed None and the billing-integrity branch that preserves
|
|
non-zero counts was never exercised — the test trivially passed with (0, 0).
|
|
|
|
The corrected mock:
|
|
1. Yields one chunk whose usage_metadata carries the expected (p, c) counts.
|
|
2. Enables memory_enabled so stream_message() calls update_memory() after
|
|
the stream loop completes.
|
|
3. Patches update_memory() to raise RuntimeError, triggering the
|
|
except-Exception block with _captured_prompt already set to p and
|
|
_captured_completion already set to c.
|
|
|
|
This exercises the partial-billing preservation contract: when a post-stream
|
|
step raises after the LLM has already charged for tokens, the captured counts
|
|
are preserved in executor.last_result.
|
|
"""
|
|
config = {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"memory_enabled": True, # enables post-stream update_memory() call
|
|
}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="stream_test_llm", config=config, template_renderer=renderer)
|
|
|
|
# Build a mock that yields one chunk with the expected usage counts.
|
|
usage_metadata = {"input_tokens": p, "output_tokens": c}
|
|
agent.chat_model = _make_async_chunks(["partial_token"], usage_metadata)
|
|
|
|
# Patch update_memory to raise after the stream loop completes.
|
|
# This triggers the except-Exception branch with _captured_prompt already set.
|
|
async def _raising_update_memory(_key: str, _value: Any) -> None:
|
|
raise RuntimeError("post-stream memory write failed")
|
|
|
|
agent.update_memory = _raising_update_memory # type: ignore[method-assign]
|
|
|
|
context.es_agent = agent
|
|
context.es_mock_model = agent.chat_model
|
|
# Store expected counts so the Then step can verify them.
|
|
context.es_expected_m2_llm_prompt = p
|
|
context.es_expected_m2_llm_completion = c
|
|
|
|
|
|
@then("an ExecutionError should be raised (stream)")
|
|
def step_es_execution_error_raised(context: Any) -> None:
|
|
assert context.es_error is not None, "Expected ExecutionError but none raised"
|
|
assert isinstance(context.es_error, ExecutionError), (
|
|
f"Expected ExecutionError, got {type(context.es_error).__name__}: {context.es_error}"
|
|
)
|
|
|
|
|
|
@then(
|
|
"executor.last_result should be an ActorResult with llm token counts after the error (stream)"
|
|
)
|
|
def step_es_last_result_llm_after_error(context: Any) -> None:
|
|
"""M2 fix (LLM path): executor.last_result should be populated even when
|
|
ExecutionError is raised from _execute_llm_stream.
|
|
"""
|
|
executor = context.es_executor
|
|
assert executor.last_result is not None, (
|
|
"executor.last_result should be populated on LLM exception path (M2 fix)"
|
|
)
|
|
assert isinstance(executor.last_result, ActorResult), (
|
|
f"Expected ActorResult, got {type(executor.last_result).__name__}"
|
|
)
|
|
result = executor.last_result
|
|
assert len(result.nodes) >= 1, (
|
|
f"Expected at least one NodeUsage entry, got {len(result.nodes)}"
|
|
)
|
|
# M4 fix: the mock now yields a usage-bearing chunk before raising in a
|
|
# post-stream step, so _captured_prompt is set to the expected counts.
|
|
# The billing-integrity branch in stream_message() preserves these counts.
|
|
expected_p = getattr(context, "es_expected_m2_llm_prompt", 0)
|
|
expected_c = getattr(context, "es_expected_m2_llm_completion", 0)
|
|
assert result.nodes[0].prompt_tokens == expected_p, (
|
|
f"Expected prompt_tokens={expected_p} (partial billing preserved), "
|
|
f"got {result.nodes[0].prompt_tokens}"
|
|
)
|
|
assert result.nodes[0].completion_tokens == expected_c, (
|
|
f"Expected completion_tokens={expected_c} (partial billing preserved), "
|
|
f"got {result.nodes[0].completion_tokens}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M3: Cost-limit enforcement in streaming path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an Executor with a graph config and max_cost_usd 0.0 with pricing (stream)")
|
|
def step_es_graph_max_cost(context: Any) -> None:
|
|
"""Graph with a single agent node and max_cost_usd=0.0.
|
|
Any token usage will exceed the zero budget, triggering ExecutionError(kind='cost').
|
|
"""
|
|
config = {
|
|
"name": "stream_cost_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 0.0},
|
|
pricing={
|
|
"openai": {
|
|
"gpt-3.5-turbo": {"prompt": 0.5, "completion": 1.5},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and invalid pricing rate (stream)")
|
|
def step_es_graph_invalid_pricing_rate(context: Any) -> None:
|
|
"""Graph with a single agent node and a pricing entry whose rate is a non-numeric string.
|
|
|
|
This exercises the except (TypeError, ValueError) branch in the M3 cost-enforcement
|
|
block of _stream_from_node when float(rate) fails.
|
|
"""
|
|
config = {
|
|
"name": "stream_invalid_rate_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 100.0},
|
|
pricing={
|
|
"openai": {
|
|
# "not-a-number" will cause float() to raise ValueError
|
|
"gpt-3.5-turbo": {"prompt": "not-a-number", "completion": 1.5},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and non-numeric max_cost_usd (stream)")
|
|
def step_es_graph_non_numeric_max_cost(context: Any) -> None:
|
|
"""Graph with a single agent node and a non-numeric max_cost_usd string.
|
|
|
|
This exercises the except (TypeError, ValueError) branch in the M3 cost-enforcement
|
|
block of _stream_from_node when float(max_cost_usd) fails.
|
|
"""
|
|
config = {
|
|
"name": "stream_bad_cost_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": "not-a-number"}, # non-numeric string
|
|
pricing={
|
|
"openai": {
|
|
"gpt-3.5-turbo": {"prompt": 0.5, "completion": 1.5},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M4: _collect_stream_tokens forwards depth — parallel AGENT with max_depth 0
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with repeated-visit agent node and no auto_finish_active (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_loop_stop_graph(context: Any) -> None:
|
|
"""Graph: agent1 → end.
|
|
Seed _node_message_visits with count 2 and auto_finish_active=False so the
|
|
loop detector fires and stops execution (C1 stop path).
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
mock_model = _make_async_chunks(["LoopStopToken"], None)
|
|
agent.chat_model = mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="loop_stop_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
# Initialise graph state with auto_finish_active=False (default)
|
|
init_payload: dict[str, Any] = {
|
|
"messages": [{"role": "user", "content": "test"}],
|
|
"metadata": {}, # no auto_finish_active → loop detector fires
|
|
}
|
|
graph.state_manager.update_state(init_payload, node_id="input")
|
|
|
|
# Seed _node_message_visits so the loop detector fires on the next visit
|
|
graph._node_message_visits = {("agent1", "test"[:200]): 2}
|
|
graph._execution_path = []
|
|
graph._node_usages = []
|
|
graph._model_call_count = 0
|
|
graph._tool_call_count = 0
|
|
graph._accumulated_cost = 0.0
|
|
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@when("I call _stream_from_node on the repeated-visit graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_loop_stop_graph(context: Any) -> None:
|
|
"""Directly call _stream_from_node to exercise the C1 loop stop path."""
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph._stream_from_node("agent1", "test", depth=0):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with intermediate agent then router and no routing command (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_intermediate_router_no_cmd_graph(context: Any) -> None:
|
|
"""Graph: agent1 (intermediate) → router → end.
|
|
agent1 is statically intermediate (has non-END successor 'router').
|
|
The agent output has no routing prefix, so the C3 guard should short-circuit
|
|
and yield the response directly.
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
mock_model = getattr(context, "es_mock_model", None) or _make_async_chunks(
|
|
["IntermediateAnswer"], None
|
|
)
|
|
agent.chat_model = mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="intermediate_router_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
"router": NodeConfig(name="router", type=NodeType.FUNCTION),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="router"),
|
|
Edge(source="router", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@when("I call execute_stream on the intermediate-router graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_intermediate_router_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph.execute_stream("test"):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@given(
|
|
"an Executor with a graph config and max_cost_usd 1.0 with missing pricing (stream)"
|
|
)
|
|
def step_es_graph_max_cost_missing_pricing(context: Any) -> None:
|
|
"""Graph with a single agent node and max_cost_usd=1.0, but pricing table
|
|
has no entry for the 'openai' provider — triggers missing_pricing_entry error.
|
|
"""
|
|
config = {
|
|
"name": "stream_cost_missing_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={
|
|
# Missing 'openai' provider entry — triggers missing_pricing_entry error
|
|
"anthropic": {
|
|
"claude-3": {"prompt": 3.0, "completion": 15.0},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@given("a PureLangGraph with terminal agent and conditional router edge (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_terminal_conditional_router_graph(context: Any) -> None:
|
|
"""Graph: agent1 → end (static), agent1 → router (conditional: always).
|
|
agent1 has edges to both "end" (static) and "router" (conditional: always).
|
|
Because "router" is a non-END static successor, _statically_terminal=False
|
|
for agent1, so this exercises the intermediate AGENT branch (ainvoke path).
|
|
The agent output has no routing prefix, so the C3 guard fires in the
|
|
intermediate AGENT branch and returns the output to the user directly.
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
mock_model = getattr(context, "es_mock_model", None) or _make_async_chunks(
|
|
["ConditionalAnswer"], None
|
|
)
|
|
agent.chat_model = mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="terminal_conditional_router_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
"router": NodeConfig(name="router", type=NodeType.FUNCTION),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
# Static edge to END (makes agent1 statically terminal)
|
|
Edge(source="agent1", target="end"),
|
|
# Conditional edge to router (fires at runtime via always condition)
|
|
Edge(
|
|
source="agent1",
|
|
target="router",
|
|
condition={"type": "always"},
|
|
),
|
|
Edge(source="router", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@when("I call execute_stream on the conditional-router graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_conditional_router_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph.execute_stream("test"):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with auto_finish_active in nested context"
|
|
" and repeated-visit node (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_auto_finish_nested_context_graph(context: Any) -> None:
|
|
"""Graph: agent1 → end.
|
|
Set auto_finish_active=True in nested context dict (state.metadata['context'])
|
|
and seed _node_message_visits with count 2 to trigger the C1 bypass via
|
|
the nested context path.
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
mock_model = _make_async_chunks(["NestedCtxToken"], None)
|
|
agent.chat_model = mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="auto_finish_nested_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
# Set auto_finish_active in nested context (not direct metadata)
|
|
init_payload: dict[str, Any] = {
|
|
"messages": [{"role": "user", "content": "test"}],
|
|
"metadata": {
|
|
"context": {"auto_finish_active": True}, # nested context
|
|
},
|
|
}
|
|
graph.state_manager.update_state(init_payload, node_id="input")
|
|
|
|
# Seed _node_message_visits so the loop detector fires on the next visit
|
|
graph._node_message_visits = {("agent1", "test"[:200]): 2}
|
|
graph._execution_path = []
|
|
graph._node_usages = []
|
|
graph._model_call_count = 0
|
|
graph._tool_call_count = 0
|
|
graph._accumulated_cost = 0.0
|
|
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with auto_finish_active nested context and ping-pong setup (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_auto_finish_nested_ping_pong_graph(context: Any) -> None:
|
|
"""Graph: agent1 → end.
|
|
Set auto_finish_active=True in nested context and pre-populate _execution_path
|
|
with the ping-pong pattern so the C2 bypass fires via the nested context path.
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
mock_model = _make_async_chunks(["NestedPingPongToken"], None)
|
|
agent.chat_model = mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="auto_finish_nested_pp_graph",
|
|
nodes={
|
|
"router": NodeConfig(name="router", type=NodeType.FUNCTION),
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
# Set auto_finish_active in nested context
|
|
init_payload: dict[str, Any] = {
|
|
"messages": [{"role": "user", "content": "test"}],
|
|
"metadata": {
|
|
"context": {"auto_finish_active": True}, # nested context
|
|
},
|
|
}
|
|
graph.state_manager.update_state(init_payload, node_id="input")
|
|
|
|
# Pre-populate _execution_path with ping-pong pattern
|
|
graph._execution_path = ["router", "agent1", "router"]
|
|
graph._node_message_visits = {}
|
|
graph._node_usages = []
|
|
graph._model_call_count = 0
|
|
graph._tool_call_count = 0
|
|
graph._accumulated_cost = 0.0
|
|
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@given("a PureLangGraph with intermediate agent and conditional edge to END (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_intermediate_conditional_end_graph(context: Any) -> None:
|
|
"""Graph: agent1 (intermediate) → agent2 (conditional: never fires) → end.
|
|
agent1 has a non-END static successor (agent2), making it statically
|
|
intermediate. At runtime, the conditional edge to agent2 does NOT fire
|
|
(condition requires 'NEVER_PRESENT_XYZ' in output), leaving
|
|
content_next_nodes empty. This exercises the dynamically-terminal path in
|
|
PureLangGraph._stream_from_node() (intermediate AGENT branch, the
|
|
``if not content_next_nodes:`` guard that yields the response and returns
|
|
early when all runtime successors resolve to END).
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent1 = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
mock_model = _make_async_chunks(["DynTerminalToken"], None)
|
|
agent1.chat_model = mock_model
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="intermediate_cond_end_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
"agent2": NodeConfig(name="agent2", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
# Conditional edge to agent2 that only fires if output contains NEVER_PRESENT
|
|
Edge(
|
|
source="agent1",
|
|
target="agent2",
|
|
condition={"type": "content_contains", "text": "NEVER_PRESENT_XYZ"},
|
|
),
|
|
Edge(source="agent2", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent1},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
context.es_pure_graph = graph
|
|
|
|
|
|
@when("I call execute_stream on the conditional-end graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_conditional_end_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph.execute_stream("test"):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@given("an Executor with a graph config and string max_model_calls (stream)")
|
|
def step_es_graph_string_max_model_calls(context: Any) -> None:
|
|
"""Graph with string max_model_calls — triggers the invalid (non-numeric) guard."""
|
|
config = {
|
|
"name": "stream_string_model_calls_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_model_calls": "not_a_number"}, # string is invalid
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a tool-node graph and string max_tool_calls (stream)")
|
|
def step_es_tool_node_string_max_tool_calls(context: Any) -> None:
|
|
"""Graph with string max_tool_calls — triggers the invalid (non-numeric) guard."""
|
|
config = {
|
|
"name": "tool_node_string_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"tool1": {"type": "tool", "tools": ["echo"]},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "tool1"},
|
|
{"source": "tool1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={"max_tool_calls": "not_a_number"}, # string is invalid
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a graph config and max_cost_usd 1.0 with incomplete pricing (stream)"
|
|
)
|
|
def step_es_graph_max_cost_incomplete_pricing(context: Any) -> None:
|
|
"""Graph with max_cost_usd=1.0 and pricing that has the provider and model
|
|
but is missing the 'prompt' or 'completion' rate key.
|
|
Triggers the incomplete pricing entry error in M3 cost enforcement.
|
|
"""
|
|
config = {
|
|
"name": "stream_cost_incomplete_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={
|
|
"openai": {
|
|
"gpt-3.5-turbo": {"prompt": 0.5}, # Missing 'completion' key
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and bool max_cost_usd with pricing (stream)")
|
|
def step_es_graph_bool_max_cost(context: Any) -> None:
|
|
"""Graph with bool max_cost_usd — triggers the bool guard in M3 cost enforcement."""
|
|
config = {
|
|
"name": "stream_bool_cost_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": True}, # bool is invalid
|
|
pricing={
|
|
"openai": {
|
|
"gpt-3.5-turbo": {"prompt": 0.5, "completion": 1.5},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@given("an Executor with a parallel intermediate AGENT graph config (stream)")
|
|
def step_es_parallel_intermediate_agent_graph(context: Any) -> None:
|
|
"""Graph: agent_start (intermediate) → {agent_a, agent_b} (parallel) → END.
|
|
agent_start is intermediate (has non-END successors agent_a, agent_b).
|
|
Exercises the intermediate AGENT parallel path in _stream_from_node.
|
|
"""
|
|
config = {
|
|
"name": "parallel_intermediate_agent_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent_start": {"type": "agent", "agent": "agent_start"},
|
|
"agent_a": {"type": "agent", "agent": "agent_a"},
|
|
"agent_b": {"type": "agent", "agent": "agent_b"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent_start"},
|
|
{"source": "agent_start", "target": "agent_a"},
|
|
{"source": "agent_start", "target": "agent_b"},
|
|
{"source": "agent_a", "target": "end"},
|
|
{"source": "agent_b", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
"parallel_execution": True,
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent_start": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_a": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_b": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config and bool max_model_calls (stream)")
|
|
def step_es_graph_bool_max_model_calls(context: Any) -> None:
|
|
"""Graph with bool max_model_calls — triggers the bool guard in _stream_from_node."""
|
|
config = {
|
|
"name": "stream_bool_model_calls_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_model_calls": True}, # bool is invalid
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a tool-node graph and bool max_tool_calls (stream)")
|
|
def step_es_tool_node_bool_max_tool_calls(context: Any) -> None:
|
|
"""Graph with bool max_tool_calls — triggers the bool guard in _stream_from_node."""
|
|
config = {
|
|
"name": "tool_node_bool_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"tool1": {"type": "tool", "tools": ["echo"]},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "tool1"},
|
|
{"source": "tool1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials=None,
|
|
limits={"max_tool_calls": True}, # bool is invalid
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a graph config and max_cost_usd 1.0 with missing model pricing (stream)"
|
|
)
|
|
def step_es_graph_max_cost_missing_model_pricing(context: Any) -> None:
|
|
"""Graph with max_cost_usd=1.0 and pricing that has the provider but not the model.
|
|
Triggers the missing model pricing entry error in M3 cost enforcement.
|
|
"""
|
|
config = {
|
|
"name": "stream_cost_missing_model_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={
|
|
"openai": {
|
|
# Missing 'gpt-3.5-turbo' entry — triggers missing model pricing error
|
|
"gpt-4": {"prompt": 30.0, "completion": 60.0},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@given("an Executor with a parallel AGENT graph config and max_depth 1 (stream)")
|
|
def step_es_parallel_agent_max_depth_1(context: Any) -> None:
|
|
"""Graph: start → agent_start → {agent_a, agent_b} (parallel) → end.
|
|
|
|
max_depth=1 so agent_start (depth 1) is within the limit, but agent_a and
|
|
agent_b (depth 2) exceed it. This exercises the M4 fix: _collect_stream_tokens
|
|
receives depth+1 from the parallel block in _stream_from_node (intermediate
|
|
AGENT branch), so the depth error fires for the parallel children rather than
|
|
for agent_start itself.
|
|
|
|
The graph is set up so agent_start is an intermediate AGENT node (it has
|
|
non-END successors agent_a and agent_b), which routes through the intermediate
|
|
AGENT branch (ainvoke path) in _stream_from_node. The parallel children
|
|
agent_a and agent_b are terminal AGENT nodes (→ end), so they are reached
|
|
via _collect_stream_tokens with depth=2, triggering the depth limit.
|
|
"""
|
|
config = {
|
|
"name": "parallel_agent_depth1_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent_start": {"type": "agent", "agent": "agent_start"},
|
|
"agent_a": {"type": "agent", "agent": "agent_a"},
|
|
"agent_b": {"type": "agent", "agent": "agent_b"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent_start"},
|
|
{"source": "agent_start", "target": "agent_a"},
|
|
{"source": "agent_start", "target": "agent_b"},
|
|
{"source": "agent_a", "target": "end"},
|
|
{"source": "agent_b", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
"parallel_execution": True,
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent_start": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_a": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_b": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_depth": 1},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Major #1 fix: max_cost_usd enforced for intermediate AGENT nodes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and max_cost_usd 0.0 with pricing (stream)"
|
|
)
|
|
def step_es_two_agent_max_cost(context: Any) -> None:
|
|
"""Graph: START → agent_a (AGENT, intermediate) → agent_b (AGENT, terminal) → END.
|
|
|
|
max_cost_usd=0.0 so any token usage on the intermediate node (agent_a)
|
|
will exceed the budget, triggering ExecutionError(kind='cost'). This
|
|
exercises the Major #1 fix: cost enforcement in the intermediate AGENT
|
|
branch of _stream_from_node.
|
|
"""
|
|
config = {
|
|
"name": "two_agent_cost_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent_a": {"type": "agent", "agent": "agent_a"},
|
|
"agent_b": {"type": "agent", "agent": "agent_b"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent_a"},
|
|
{"source": "agent_a", "target": "agent_b"},
|
|
{"source": "agent_b", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
"parallel_execution": False,
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent_a": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_b": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 0.0},
|
|
pricing={
|
|
"openai": {
|
|
"gpt-3.5-turbo": {"prompt": 0.5, "completion": 1.5},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
def _two_agent_sequential_config() -> dict[str, Any]:
|
|
"""Return a base two-agent sequential graph config (no limits/pricing)."""
|
|
return {
|
|
"name": "two_agent_cost_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent_a": {"type": "agent", "agent": "agent_a"},
|
|
"agent_b": {"type": "agent", "agent": "agent_b"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent_a"},
|
|
{"source": "agent_a", "target": "agent_b"},
|
|
{"source": "agent_b", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
"parallel_execution": False,
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent_a": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
"agent_b": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
},
|
|
},
|
|
}
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and"
|
|
" max_cost_usd 0.40 with pricing (stream)"
|
|
)
|
|
def step_es_two_agent_max_cost_040(context: Any) -> None:
|
|
"""Graph: START → agent_a (AGENT, intermediate) → agent_b (AGENT, terminal) → END.
|
|
|
|
max_cost_usd=0.40 with prompt pricing at 0.50/M so that a first node
|
|
producing 3M prompt tokens costs $1.50 and exhausts the budget. The
|
|
post-node cost check catches the exhaustion; the pre-flight gate is also
|
|
exercised as a no-op before the first node. This exercises the streaming
|
|
path cost enforcement for a non-zero budget (issue #76, AC4).
|
|
"""
|
|
context.es_executor = create_executor(
|
|
config_dict=_two_agent_sequential_config(),
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 0.40},
|
|
pricing={
|
|
"openai": {
|
|
"gpt-3.5-turbo": {"prompt": 0.50, "completion": 1.50},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and"
|
|
" max_cost_usd 0.30 at 0.15/M prompt pricing (stream)"
|
|
)
|
|
def step_es_two_agent_max_cost_030_exact(context: Any) -> None:
|
|
"""Graph: START → agent_a (AGENT, intermediate) → agent_b (AGENT, terminal) → END.
|
|
|
|
max_cost_usd=0.30 with prompt pricing at 0.15/M so that a first node
|
|
producing 2M prompt tokens costs exactly $0.30. The post-node check uses
|
|
``>`` so the first node completes (0.30 is NOT > 0.30). The second node's
|
|
pre-flight gate uses ``>=`` so 0.30 >= 0.30 blocks it, exercising the
|
|
intentional asymmetry between post-node and pre-flight checks
|
|
(Graa review PR #80).
|
|
"""
|
|
context.es_executor = create_executor(
|
|
config_dict=_two_agent_sequential_config(),
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 0.30},
|
|
pricing={
|
|
"openai": {
|
|
"gpt-3.5-turbo": {"prompt": 0.15, "completion": 0.60},
|
|
}
|
|
},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and max_cost_usd 1.0"
|
|
" with missing provider pricing (stream)"
|
|
)
|
|
def step_es_two_agent_missing_provider_pricing(context: Any) -> None:
|
|
"""Exercises the missing-provider-pricing branch in the intermediate AGENT
|
|
cost block (Major #1 fix).
|
|
|
|
The pricing dict is non-empty (so self._pricing is truthy) but does not
|
|
contain an entry for 'openai', triggering the missing-provider-pricing error.
|
|
"""
|
|
context.es_executor = create_executor(
|
|
config_dict=_two_agent_sequential_config(),
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
# Non-empty pricing dict with a different provider — openai entry is absent
|
|
pricing={"anthropic": {"claude-3": {"prompt": 0.5, "completion": 1.5}}},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and max_cost_usd 1.0"
|
|
" with missing model pricing (stream)"
|
|
)
|
|
def step_es_two_agent_missing_model_pricing(context: Any) -> None:
|
|
"""Exercises the missing-model-pricing branch in the intermediate AGENT
|
|
cost block (Major #1 fix)."""
|
|
context.es_executor = create_executor(
|
|
config_dict=_two_agent_sequential_config(),
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={"openai": {}}, # provider present but no model entry
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and max_cost_usd 1.0"
|
|
" with incomplete pricing (stream)"
|
|
)
|
|
def step_es_two_agent_incomplete_pricing(context: Any) -> None:
|
|
"""Exercises the incomplete-pricing branch (missing 'prompt' or 'completion' key)
|
|
in the intermediate AGENT cost block (Major #1 fix)."""
|
|
context.es_executor = create_executor(
|
|
config_dict=_two_agent_sequential_config(),
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 0.5}}}, # missing completion
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and max_cost_usd 1.0"
|
|
" with invalid pricing rate (stream)"
|
|
)
|
|
def step_es_two_agent_invalid_pricing_rate(context: Any) -> None:
|
|
"""Exercises the invalid-rate branch (non-numeric rate string) in the
|
|
intermediate AGENT cost block (Major #1 fix)."""
|
|
context.es_executor = create_executor(
|
|
config_dict=_two_agent_sequential_config(),
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={
|
|
"openai": {"gpt-3.5-turbo": {"prompt": "not-a-number", "completion": 1.5}}
|
|
},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and bool max_cost_usd with pricing (stream)"
|
|
)
|
|
def step_es_two_agent_bool_max_cost(context: Any) -> None:
|
|
"""Exercises the bool-max_cost_usd branch in the intermediate AGENT cost
|
|
block (Major #1 fix)."""
|
|
context.es_executor = create_executor(
|
|
config_dict=_two_agent_sequential_config(),
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": True},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 0.5, "completion": 1.5}}},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a two-agent sequential graph and non-numeric max_cost_usd"
|
|
" with pricing (stream)"
|
|
)
|
|
def step_es_two_agent_nonnumeric_max_cost(context: Any) -> None:
|
|
"""Exercises the non-numeric-max_cost_usd branch in the intermediate AGENT
|
|
cost block (Major #1 fix)."""
|
|
context.es_executor = create_executor(
|
|
config_dict=_two_agent_sequential_config(),
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": "not-a-number"},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 0.5, "completion": 1.5}}},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Major #2 fix: state not polluted on agent failure
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a PureLangGraph with terminal AGENT node that fails during streaming (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_failing_agent_graph(context: Any) -> None:
|
|
"""Graph with a single terminal AGENT node whose stream_message() raises.
|
|
|
|
The Major #2 fix ensures that when _stream_agent() raises a non-ExecutionError
|
|
exception, the streaming path does NOT call state_manager.update_state() with
|
|
an assistant message containing the user's input. Only last_output is set.
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="failing_agent_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
|
|
# Create a mock LLMAgent whose stream_message() raises RuntimeError
|
|
from cleveractors.agents.llm import LLMAgent
|
|
from cleveractors.templates.renderer import TemplateRenderer
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
|
|
async def _failing_astream(_messages: Any) -> Any:
|
|
raise RuntimeError("simulated streaming failure")
|
|
yield # make it a generator
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _failing_astream
|
|
mock_model.temperature = 0.7
|
|
agent.chat_model = mock_model
|
|
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
context.es_pure_graph = graph
|
|
context.es_input_message = "user input that must not appear as assistant"
|
|
|
|
|
|
@when("I call execute_stream on the failing-agent graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_failing_agent(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in graph.execute_stream(context.es_input_message):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@then(
|
|
"the graph state messages should not contain the user input as an assistant message (stream)"
|
|
)
|
|
def step_es_state_not_polluted(context: Any) -> None:
|
|
"""Major #2 fix: verify that state.messages does not contain the user's input
|
|
as an assistant-role message after a streaming failure.
|
|
|
|
The non-streaming path (_execute_from_node) does NOT update state.messages on
|
|
failure. The streaming path must mirror this behaviour.
|
|
"""
|
|
graph = context.es_pure_graph
|
|
user_input = context.es_input_message
|
|
state = graph.state_manager.get_state()
|
|
for msg in state.messages:
|
|
if isinstance(msg, dict):
|
|
role = msg.get("role", "")
|
|
content = msg.get("content", "")
|
|
assert not (role == "assistant" and content == user_input), (
|
|
f"State pollution detected: user input {user_input!r} was "
|
|
f"persisted as an assistant message in graph state after "
|
|
f"streaming failure. state.messages={state.messages!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M2 state-capture failure path: exercises the except Exception block inside
|
|
# the M2 fix's "if graph is not None:" guard in _execute_graph_stream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an Executor with a graph that raises and has broken state capture (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_broken_state_capture(context: Any) -> None:
|
|
"""Set up an Executor whose graph raises ExecutionError (max_model_calls=0)
|
|
and whose _last_stream_state attribute raises AttributeError when accessed.
|
|
|
|
This exercises the `except Exception as _state_err` block inside the M2 fix's
|
|
`if graph is not None:` guard in `_execute_graph_stream`, which logs a debug
|
|
message and continues (so executor.last_result is still populated with empty
|
|
node usages).
|
|
|
|
We patch PureLangGraph so that _last_stream_state is a property that raises,
|
|
ensuring the inner try/except in the M2 fix is exercised.
|
|
"""
|
|
from cleveractors.langgraph.pure_graph import PureLangGraph
|
|
|
|
config = {
|
|
"name": "broken_state_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
"agent1": {"type": "agent", "agent": "agent1"},
|
|
},
|
|
"edges": [
|
|
{"source": "start", "target": "agent1"},
|
|
{"source": "agent1", "target": "end"},
|
|
],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"agents": {
|
|
"agent1": {
|
|
"type": "llm",
|
|
"provider": "openai",
|
|
"config": {"provider": "openai", "model": "gpt-3.5-turbo"},
|
|
}
|
|
},
|
|
}
|
|
# max_model_calls=0 ensures the graph raises ExecutionError before any LLM call
|
|
executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_model_calls": 0},
|
|
pricing={},
|
|
)
|
|
context.es_executor = executor
|
|
|
|
# Patch PureLangGraph._last_stream_state to be a property that raises.
|
|
# This is applied as a class-level patch so it affects the instance created
|
|
# inside _execute_graph_stream.
|
|
original_prop = PureLangGraph.__dict__.get("_last_stream_state")
|
|
|
|
def _raising_getter(self: Any) -> None:
|
|
raise AttributeError("simulated broken state capture")
|
|
|
|
# Store the original so the When step can restore it after the test
|
|
context.es_broken_state_original = original_prop
|
|
context.es_broken_state_class = PureLangGraph
|
|
PureLangGraph._last_stream_state = property(_raising_getter) # type: ignore[method-assign]
|
|
|
|
|
|
@when("I attempt execute_stream with broken state capture expecting an error (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_broken_state(context: Any) -> None:
|
|
"""Attempt execute_stream with the broken _last_stream_state patch active,
|
|
then restore the original attribute regardless of outcome."""
|
|
executor = context.es_executor
|
|
broken_class = context.es_broken_state_class
|
|
original_prop = context.es_broken_state_original
|
|
|
|
mock_model = _make_async_chunks(["token"], None)
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
agent.chat_model = mock_model
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
|
|
try:
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory_cls,
|
|
):
|
|
mock_factory_cls.return_value = mock_factory_instance
|
|
try:
|
|
tokens: list[str] = []
|
|
async for token in executor.execute_stream("test"):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except (ExecutionError, ConfigurationError) as e:
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
finally:
|
|
# Restore the original _last_stream_state attribute
|
|
if original_prop is None:
|
|
# It was a plain instance attribute, not a class-level descriptor
|
|
if hasattr(broken_class, "_last_stream_state"):
|
|
try:
|
|
delattr(broken_class, "_last_stream_state")
|
|
except AttributeError:
|
|
pass
|
|
else:
|
|
broken_class._last_stream_state = original_prop # type: ignore[method-assign]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M2 fix: graph path — executor.last_result populated when AgentCreationError
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the mock graph factory raises ConfigurationError on create_agent (stream)")
|
|
def step_es_graph_factory_raises_config_error(context: Any) -> None:
|
|
"""Set the flag so the attempt-execute-stream step makes the factory raise
|
|
ConfigurationError. This exercises the graph-path M2 fix: when agent
|
|
creation fails before the graph is built, executor.last_result should still
|
|
be populated with a synthetic <no_llm> placeholder.
|
|
"""
|
|
context.es_factory_should_raise = "ConfigurationError"
|
|
|
|
|
|
@then("executor.last_result should be an ActorResult with no_llm placeholder (stream)")
|
|
def step_es_last_result_no_llm_placeholder(context: Any) -> None:
|
|
"""M2 fix (graph path): executor.last_result should be populated with a
|
|
synthetic <no_llm> placeholder when agent creation fails before the graph
|
|
is built.
|
|
"""
|
|
executor = context.es_executor
|
|
assert executor.last_result is not None, (
|
|
"executor.last_result should be populated on graph exception path (M2 fix)"
|
|
)
|
|
assert isinstance(executor.last_result, ActorResult), (
|
|
f"Expected ActorResult, got {type(executor.last_result).__name__}"
|
|
)
|
|
result = executor.last_result
|
|
assert len(result.nodes) >= 1, (
|
|
f"Expected at least one NodeUsage entry, got {len(result.nodes)}"
|
|
)
|
|
placeholder_found = any(n.model == "<no_llm>" for n in result.nodes)
|
|
assert placeholder_found, (
|
|
f"Expected <no_llm> placeholder node but got: "
|
|
f"{[(n.node_id, n.model) for n in result.nodes]}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# m4 fix: empty astream with memory_enabled=True stores empty assistant entry
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an LLMAgent with memory_enabled and empty astream (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_llm_agent_memory_empty_astream(context: Any) -> None:
|
|
"""Set up an LLMAgent with memory_enabled=True and an empty astream.
|
|
|
|
When the stream yields nothing, stream_message() should call
|
|
update_memory("last_response", "") with an empty string — not the user's
|
|
input. This verifies that the assistant memory entry has empty content
|
|
rather than the user's input (m4 fix).
|
|
"""
|
|
config = {
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
"memory_enabled": True,
|
|
}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="test_stream", config=config, template_renderer=renderer)
|
|
agent.chat_model = _make_async_chunks([], None)
|
|
|
|
# Pre-populate memory so we can verify the assistant entry after streaming.
|
|
await agent.update_memory("conversation_history", [])
|
|
|
|
context.es_agent = agent
|
|
|
|
|
|
@then("the assistant memory entry should have empty content (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_assistant_memory_empty(context: Any) -> None:
|
|
"""Verify that after an empty stream with memory_enabled=True, the
|
|
conversation history has an assistant entry with empty content — not the
|
|
user's input.
|
|
"""
|
|
agent = context.es_agent
|
|
history: list[dict[str, str]] = await agent.get_memory("conversation_history", [])
|
|
assistant_entries = [m for m in history if m.get("role") == "assistant"]
|
|
assert len(assistant_entries) >= 1, (
|
|
f"Expected at least one assistant entry in conversation history, got: {history}"
|
|
)
|
|
last_assistant = assistant_entries[-1]
|
|
assert last_assistant.get("content") == "", (
|
|
f"Expected empty assistant content after empty stream, "
|
|
f"got: {last_assistant.get('content')!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# m5 fix: partial-stream abandonment leaves _last_token_usage at (0, 0)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("the agent _last_token_usage should be (0, 0) after abandonment (stream)")
|
|
def step_es_agent_last_token_usage_zero_after_abandonment(context: Any) -> None:
|
|
"""Verify that abandoning a stream before exhaustion leaves _last_token_usage
|
|
at (0, 0) — the documented contract for partial-stream abandonment.
|
|
|
|
The agent reference is captured in context.es_agent by the When-step so we
|
|
can directly assert _last_token_usage == (0, 0) on the LLMAgent instance.
|
|
stream_message() resets _last_token_usage to (0, 0) at the start of each
|
|
call; since the generator was not exhausted, the final-chunk update never
|
|
ran, so _last_token_usage remains (0, 0).
|
|
"""
|
|
executor = context.es_executor
|
|
assert executor.last_result is None, (
|
|
"executor.last_result should be None after stream abandonment"
|
|
)
|
|
agent = getattr(context, "es_agent", None)
|
|
if agent is not None:
|
|
actual_usage = getattr(agent, "_last_token_usage", None)
|
|
assert actual_usage == (0, 0), (
|
|
f"Expected agent._last_token_usage == (0, 0) after abandonment, "
|
|
f"got {actual_usage!r}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# m6 fix: executor.last_result.state verified in graph streaming success path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("executor.last_result.state should be populated with initial_state keys (stream)")
|
|
def step_es_last_result_state_populated(context: Any) -> None:
|
|
"""m6 fix: verify that executor.last_result.state is not None and contains
|
|
the initial_state keys passed to execute_stream(). A regression that drops
|
|
state from last_result in streaming mode would silently break stateless
|
|
resumption.
|
|
"""
|
|
executor = context.es_executor
|
|
assert executor.last_result is not None, "executor.last_result is None"
|
|
result = executor.last_result
|
|
assert result.state is not None, (
|
|
"executor.last_result.state should not be None after graph streaming "
|
|
"with initial_state"
|
|
)
|
|
# The initial_state passed in step_es_execute_stream_with_state contains
|
|
# "some_key" and "conversation_stage".
|
|
expected_keys = {"some_key", "conversation_stage"}
|
|
missing = expected_keys - set(result.state.keys())
|
|
assert not missing, (
|
|
f"executor.last_result.state is missing expected keys: {missing}. "
|
|
f"Got state keys: {set(result.state.keys())}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# M1 fix: graph path — executor.last_result populated for unexpected exceptions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("the graph execute_stream raises RuntimeError unexpectedly (stream)")
|
|
def step_es_graph_execute_stream_raises_runtime_error(context: Any) -> None:
|
|
"""Set a flag so the attempt-execute-stream step patches PureLangGraph to
|
|
raise a RuntimeError from execute_stream(). This exercises the M1 fix:
|
|
the bare except-Exception block in _execute_graph_stream must set
|
|
executor.last_result before re-raising as ExecutionError.
|
|
"""
|
|
context.es_graph_raises_runtime_error = True
|
|
|
|
|
|
@given(
|
|
"the graph execute_stream raises RuntimeError with broken state capture (stream)"
|
|
)
|
|
def step_es_graph_raises_runtime_error_broken_state(context: Any) -> None:
|
|
"""Set flags so the attempt-execute-stream step patches PureLangGraph to
|
|
raise a RuntimeError from execute_stream() AND makes _last_stream_state
|
|
raise an exception when accessed. This exercises the inner except block
|
|
in the M1 fix that handles state-capture failures gracefully.
|
|
"""
|
|
context.es_graph_raises_runtime_error = True
|
|
context.es_graph_broken_state_capture = True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Issue 2: billing-integrity for early config-validation errors in
|
|
# _execute_graph_stream (node/edge validation loop fires before try/except)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an Executor with a graph config that has an invalid node definition (stream)")
|
|
def step_es_graph_invalid_node_def(context: Any) -> None:
|
|
"""Graph config with a node definition that is not a dict (missing 'id').
|
|
|
|
This triggers the ConfigurationError in the node-validation loop inside
|
|
_execute_graph_stream, which fires BEFORE the main try/except block.
|
|
The billing-integrity wrapper must populate executor.last_result before
|
|
re-raising.
|
|
"""
|
|
config = {
|
|
"type": "graph",
|
|
"name": "invalid_node_graph",
|
|
"routes": {
|
|
"main": {
|
|
"nodes": {
|
|
# Intentionally invalid: value is not a dict
|
|
},
|
|
"edges": [],
|
|
"entry_point": "start",
|
|
}
|
|
},
|
|
"actors": {},
|
|
}
|
|
# Inject a raw list node (not a dict with 'id') via the route key so the
|
|
# validation loop fires. We use the 'route' key (list-of-dicts format)
|
|
# to inject a non-dict node entry.
|
|
config2 = {
|
|
"type": "graph",
|
|
"name": "invalid_node_graph",
|
|
"route": {
|
|
"nodes": ["not_a_dict"], # invalid: not a dict, no 'id'
|
|
"edges": [],
|
|
"entry_node": "start",
|
|
},
|
|
"actors": {},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config2,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
@given("an Executor with a graph config that has a duplicate node ID (stream)")
|
|
def step_es_graph_duplicate_node_id(context: Any) -> None:
|
|
"""Graph config with two node definitions sharing the same ID.
|
|
|
|
This triggers the ConfigurationError for duplicate node IDs in the
|
|
node-validation loop inside _execute_graph_stream, which fires BEFORE
|
|
the main try/except block. The billing-integrity wrapper must populate
|
|
executor.last_result before re-raising.
|
|
"""
|
|
config = {
|
|
"type": "graph",
|
|
"name": "dup_node_graph",
|
|
"route": {
|
|
"nodes": [
|
|
{"id": "agent1", "type": "agent"},
|
|
{"id": "agent1", "type": "agent"}, # duplicate
|
|
],
|
|
"edges": [],
|
|
"entry_node": "start",
|
|
},
|
|
"actors": {},
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Issue 3: GOTO_/ROUTE_ routing commands parsed in streaming path
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with terminal AGENT node that emits a GOTO_ routing command (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_terminal_goto_graph(context: Any) -> None:
|
|
"""Graph: start → agent1 → end. agent1 is statically terminal (only END
|
|
successor). Its mock astream yields a single token containing a GOTO_
|
|
routing command. After streaming, state.metadata["next_node"] must be set.
|
|
"""
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="goto_terminal_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
|
|
# The agent output contains a GOTO_ routing command.
|
|
# Format: "GOTO_NODENAME:rest of message"
|
|
goto_token = "GOTO_NEXTNODE:some additional text"
|
|
agent.chat_model = _make_async_chunks([goto_token], None)
|
|
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
context.es_pure_graph = graph
|
|
context.es_input_message = "route me"
|
|
|
|
|
|
@when("I call execute_stream on the GOTO-routing graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_goto_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
tokens: list[str] = []
|
|
try:
|
|
async for token in graph.execute_stream(context.es_input_message):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@then("state.metadata next_node should be set from the GOTO_ command (stream)")
|
|
def step_es_assert_next_node_from_goto(context: Any) -> None:
|
|
assert context.es_error is None, f"Unexpected error: {context.es_error}"
|
|
graph = context.es_pure_graph
|
|
state = graph.state_manager.get_state()
|
|
next_node = state.metadata.get("next_node")
|
|
assert next_node == "nextnode", (
|
|
f"Expected next_node='nextnode' from GOTO_ command, got {next_node!r}. "
|
|
f"state.metadata={state.metadata}"
|
|
)
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with intermediate AGENT node that emits a GOTO_ routing command (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_intermediate_goto_graph(context: Any) -> None:
|
|
"""Graph: start → agent1 → agent2 → end. agent1 is intermediate (non-END
|
|
successor agent2). Its mock process_message returns a GOTO_ routing command.
|
|
After streaming, state.metadata["next_node"] must be set.
|
|
"""
|
|
from unittest.mock import AsyncMock
|
|
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="goto_intermediate_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
"agent2": NodeConfig(name="agent2", type=NodeType.AGENT, agent="agent2"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="agent2"),
|
|
Edge(source="agent2", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
|
|
# agent1 is intermediate: process_message returns a GOTO_ routing command.
|
|
# The intermediate branch calls node.execute() → _execute_agent() →
|
|
# agent.process_message(). We mock process_message directly.
|
|
agent1 = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
goto_response = "GOTO_TARGETNODE:routing instruction"
|
|
agent1.process_message = AsyncMock(return_value=goto_response) # type: ignore[method-assign]
|
|
|
|
# agent2 is terminal: astream yields a simple token.
|
|
agent2 = LLMAgent(name="agent2", config=config, template_renderer=renderer)
|
|
agent2.chat_model = _make_async_chunks(["terminal_response"], None)
|
|
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent1, "agent2": agent2},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
context.es_pure_graph = graph
|
|
context.es_input_message = "route me via intermediate"
|
|
|
|
|
|
@when("I call execute_stream on the intermediate GOTO-routing graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_intermediate_goto_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
tokens: list[str] = []
|
|
try:
|
|
async for token in graph.execute_stream(context.es_input_message):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@then(
|
|
"state.metadata next_node should be set from the intermediate GOTO_ command (stream)"
|
|
)
|
|
def step_es_assert_next_node_from_intermediate_goto(context: Any) -> None:
|
|
assert context.es_error is None, f"Unexpected error: {context.es_error}"
|
|
graph = context.es_pure_graph
|
|
state = graph.state_manager.get_state()
|
|
next_node = state.metadata.get("next_node")
|
|
assert next_node == "targetnode", (
|
|
f"Expected next_node='targetnode' from intermediate GOTO_ command, "
|
|
f"got {next_node!r}. state.metadata={state.metadata}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Issue 1 (review round 7): GOTO_/ROUTE_ parsing in non-AGENT branch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with non-AGENT function node that emits a GOTO_ routing command (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_non_agent_goto_graph(context: Any) -> None:
|
|
"""Graph: start → fn1 (FUNCTION) → end. fn1's execute() returns a dict
|
|
whose messages[-1]["content"] contains a GOTO_ routing command. After
|
|
streaming, state.metadata["next_node"] must be set — mirroring the
|
|
_execute_from_node() behaviour for all node types.
|
|
"""
|
|
from unittest.mock import AsyncMock
|
|
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="goto_non_agent_graph",
|
|
nodes={
|
|
"fn1": NodeConfig(name="fn1", type=NodeType.FUNCTION, function="fn1"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="fn1"),
|
|
Edge(source="fn1", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
|
|
# The function node returns a dict whose messages[-1]["content"] contains a
|
|
# GOTO_ routing command. The non-AGENT branch of _stream_from_node() must
|
|
# parse this and set state.metadata["next_node"].
|
|
goto_content = "GOTO_FNROUTE:some payload"
|
|
mock_node_execute = AsyncMock(
|
|
return_value={"messages": [{"content": goto_content}]}
|
|
)
|
|
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
# Patch the fn1 node's execute method to return the GOTO_ content.
|
|
graph.nodes["fn1"].execute = mock_node_execute # type: ignore[method-assign]
|
|
context.es_pure_graph = graph
|
|
context.es_input_message = "route via function"
|
|
|
|
|
|
@when("I call execute_stream on the non-AGENT GOTO-routing graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_non_agent_goto_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
tokens: list[str] = []
|
|
try:
|
|
async for token in graph.execute_stream(context.es_input_message):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@then(
|
|
"state.metadata next_node should be set from the non-AGENT GOTO_ command (stream)"
|
|
)
|
|
def step_es_assert_next_node_from_non_agent_goto(context: Any) -> None:
|
|
assert context.es_error is None, f"Unexpected error: {context.es_error}"
|
|
graph = context.es_pure_graph
|
|
state = graph.state_manager.get_state()
|
|
next_node = state.metadata.get("next_node")
|
|
assert next_node == "fnroute", (
|
|
f"Expected next_node='fnroute' from non-AGENT GOTO_ command, "
|
|
f"got {next_node!r}. state.metadata={state.metadata}"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Issue 2 (review round 7): str(None) guard in intermediate AGENT branch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given(
|
|
"a PureLangGraph with intermediate AGENT node that returns None content (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_intermediate_none_content_graph(context: Any) -> None:
|
|
"""Graph: start → agent1 (intermediate AGENT) → agent2 (terminal AGENT) → end.
|
|
agent1.process_message returns None (simulating last_msg["content"] = None).
|
|
The intermediate AGENT branch must yield "" instead of the literal "None".
|
|
"""
|
|
from unittest.mock import AsyncMock
|
|
|
|
from cleveractors.langgraph.nodes import Edge, NodeConfig, NodeType
|
|
from cleveractors.langgraph.pure_graph import PureGraphConfig, PureLangGraph
|
|
|
|
pg_config = PureGraphConfig(
|
|
name="none_content_graph",
|
|
nodes={
|
|
"agent1": NodeConfig(name="agent1", type=NodeType.AGENT, agent="agent1"),
|
|
"agent2": NodeConfig(name="agent2", type=NodeType.AGENT, agent="agent2"),
|
|
},
|
|
edges=[
|
|
Edge(source="start", target="agent1"),
|
|
Edge(source="agent1", target="agent2"),
|
|
Edge(source="agent2", target="end"),
|
|
],
|
|
entry_point="start",
|
|
)
|
|
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
|
|
# agent1 is intermediate: node.execute() returns a dict with messages where
|
|
# the last message's content is None — simulating a provider that returns
|
|
# None content (e.g. a metadata-only response).
|
|
agent1 = LLMAgent(name="agent1", config=config, template_renderer=renderer)
|
|
# Mock node.execute to return a dict with None content in messages
|
|
agent1_result: dict[str, Any] = {"messages": [{"content": None}]}
|
|
agent1.process_message = AsyncMock(return_value=None) # type: ignore[method-assign]
|
|
|
|
# agent2 is terminal: astream yields a simple token.
|
|
agent2 = LLMAgent(name="agent2", config=config, template_renderer=renderer)
|
|
agent2.chat_model = _make_async_chunks(["terminal_ok"], None)
|
|
|
|
graph = PureLangGraph(
|
|
config=pg_config,
|
|
agents={"agent1": agent1, "agent2": agent2},
|
|
limits={},
|
|
pricing={},
|
|
)
|
|
# Patch agent1's node.execute to return the None-content dict directly,
|
|
# bypassing the normal agent execution path.
|
|
graph.nodes["agent1"].execute = AsyncMock( # type: ignore[method-assign]
|
|
return_value=agent1_result
|
|
)
|
|
context.es_pure_graph = graph
|
|
context.es_input_message = "trigger None content"
|
|
|
|
|
|
@when("I call execute_stream on the None-content intermediate graph (stream)")
|
|
@async_run_until_complete
|
|
async def step_es_execute_stream_none_content_graph(context: Any) -> None:
|
|
graph = context.es_pure_graph
|
|
tokens: list[str] = []
|
|
try:
|
|
async for token in graph.execute_stream(context.es_input_message):
|
|
tokens.append(token)
|
|
context.es_tokens = tokens
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@then('the collected tokens should not contain the literal string "None" (stream)')
|
|
def step_es_assert_no_none_token(context: Any) -> None:
|
|
assert context.es_error is None, f"Unexpected error: {context.es_error}"
|
|
tokens = context.es_tokens
|
|
assert tokens is not None, "No tokens were collected"
|
|
assert "None" not in tokens, (
|
|
f"Literal string 'None' found in tokens {tokens!r}. "
|
|
"The str(None) guard is missing — full_response=None must yield '' not 'None'."
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Issue 5 (review round 7): LangChainException arm in stream_message()
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("a mock astream that raises LangChainException (stream)")
|
|
def step_es_langchain_exception_astream(context: Any) -> None:
|
|
"""Inject a mock astream that raises LangChainException.
|
|
|
|
This exercises the dedicated ``except LangChainException`` handler in
|
|
``stream_message()`` added in review round 5. The handler must:
|
|
(a) raise ExecutionError, and
|
|
(b) log "LangChain streaming error" (distinct from the generic "streaming failed").
|
|
"""
|
|
from cleveractors.agents.llm import LangChainException as _LangChainException
|
|
|
|
# LangChainException may be None in environments without langchain_core.
|
|
# Fall back to a plain Exception subclass so the test still exercises the
|
|
# except-arm (the arm catches whatever LangChainException resolves to).
|
|
_exc_class = _LangChainException if _LangChainException is not None else Exception
|
|
|
|
async def _raising_astream(_messages: Any) -> Any:
|
|
raise _exc_class("Simulated LangChain streaming error")
|
|
yield # make it an async generator
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.astream = _raising_astream
|
|
mock_model.temperature = 0.7
|
|
context.es_mock_model = mock_model
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Issue 4 (review round 7): resource-leak test verifies cleanup() called
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when(
|
|
"I start iterating execute_stream and stop after first token verifying cleanup (stream)"
|
|
)
|
|
@async_run_until_complete
|
|
async def step_es_partial_stream_verify_cleanup(context: Any) -> None:
|
|
"""Simulate stream abandonment and verify that agent.cleanup() was awaited.
|
|
|
|
Unlike the existing abandonment step (which only checks last_result and
|
|
_last_token_usage), this step replaces agent.cleanup with an AsyncMock so
|
|
that the Then-step can assert it was called.
|
|
"""
|
|
executor = context.es_executor
|
|
mock_model = getattr(context, "es_mock_model", None)
|
|
|
|
with (
|
|
patch("cleveractors.runtime_dispatch.TemplateRenderer"),
|
|
patch("cleveractors.runtime_dispatch.AgentFactory") as mock_factory,
|
|
):
|
|
config = {"provider": "openai", "model": "gpt-3.5-turbo"}
|
|
renderer = TemplateRenderer()
|
|
agent = LLMAgent(
|
|
name="cleanup_test_llm", config=config, template_renderer=renderer
|
|
)
|
|
if mock_model is not None:
|
|
agent.chat_model = mock_model
|
|
# Replace cleanup with an AsyncMock so we can assert it was called.
|
|
agent.cleanup = AsyncMock() # type: ignore[method-assign]
|
|
mock_factory_instance = MagicMock()
|
|
mock_factory_instance.create_agent = MagicMock(return_value=agent)
|
|
mock_factory.return_value = mock_factory_instance
|
|
context.es_agent = agent # capture for Then-step assertions
|
|
|
|
try:
|
|
gen = executor.execute_stream("test")
|
|
_first_token = await gen.__anext__()
|
|
context.es_tokens = [_first_token]
|
|
context.es_error = None
|
|
# Don't exhaust the generator — simulate abandonment.
|
|
# The execute_stream() try/finally must call gen.aclose() which
|
|
# propagates GeneratorExit into stream_message(), triggering
|
|
# agent.cleanup() via the finally block.
|
|
await gen.aclose()
|
|
except StopAsyncIteration:
|
|
context.es_tokens = []
|
|
context.es_error = None
|
|
except Exception as e: # pylint: disable=broad-exception-caught
|
|
context.es_error = e
|
|
context.es_tokens = None
|
|
|
|
|
|
@then("agent.cleanup should have been called (stream)")
|
|
def step_es_assert_cleanup_called(context: Any) -> None:
|
|
"""Verify that agent.cleanup() was awaited after stream abandonment.
|
|
|
|
This is the key assertion that the existing abandonment test was missing:
|
|
it confirms that the try/finally block in execute_stream() actually calls
|
|
gen.aclose(), which propagates GeneratorExit into stream_message() and
|
|
triggers agent.cleanup() via the finally block.
|
|
"""
|
|
agent = getattr(context, "es_agent", None)
|
|
assert agent is not None, "No agent captured in context.es_agent"
|
|
cleanup_mock = getattr(agent, "cleanup", None)
|
|
assert cleanup_mock is not None, "agent.cleanup was not replaced with AsyncMock"
|
|
assert cleanup_mock.called, (
|
|
"agent.cleanup() was NOT called after stream abandonment. "
|
|
"The try/finally block in execute_stream() must call gen.aclose() to "
|
|
"ensure agent.cleanup() is invoked promptly."
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Issue 8 (review round 7): ExecutionError.reason asserted in cost scenarios
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("an Executor with a basic llm config and zero cost limit (stream)")
|
|
def step_es_llm_zero_cost_limit(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=0.0 and real pricing.
|
|
|
|
Any token usage will exceed the zero budget, triggering
|
|
ExecutionError(kind='cost', reason='budget_exhausted').
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "zero_cost_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 0.0},
|
|
pricing={"openai": {"gpt-3.5-turbo": {"prompt": 2.0, "completion": 2.0}}},
|
|
)
|
|
|
|
|
|
@given(
|
|
"an Executor with a basic llm config and pricing but missing provider entry (stream)"
|
|
)
|
|
def step_es_llm_missing_provider_for_reason(context: Any) -> None:
|
|
"""LLM executor with max_cost_usd=1.0 but no pricing entry for 'openai'.
|
|
|
|
Triggers ExecutionError(kind='cost', reason='missing_pricing_entry').
|
|
"""
|
|
config = {
|
|
"type": "llm",
|
|
"name": "missing_provider_llm",
|
|
"provider": "openai",
|
|
"model": "gpt-3.5-turbo",
|
|
}
|
|
context.es_executor = create_executor(
|
|
config_dict=config,
|
|
credentials={"openai": {"api_key": "test-key"}},
|
|
limits={"max_cost_usd": 1.0},
|
|
pricing={"anthropic": {"claude-3": {"prompt": 1.0, "completion": 1.0}}},
|
|
)
|
|
|
|
|
|
@then(
|
|
'an ExecutionError with kind "{kind}" and reason "{reason}" should be raised (stream kind+reason)'
|
|
)
|
|
def step_es_execution_error_kind_and_reason(
|
|
context: Any, kind: str, reason: str
|
|
) -> None:
|
|
"""Assert both error.kind and error.reason on the raised ExecutionError.
|
|
|
|
This extends the existing kind-only assertion to also verify the reason
|
|
field, preventing regressions where the wrong reason is set.
|
|
"""
|
|
assert context.es_error is not None, "Expected an error but none was raised"
|
|
assert isinstance(context.es_error, ExecutionError), (
|
|
f"Expected ExecutionError, got {type(context.es_error)}"
|
|
)
|
|
assert context.es_error.kind == kind, (
|
|
f"Expected kind={kind!r}, got {context.es_error.kind!r}"
|
|
)
|
|
assert context.es_error.reason == reason, (
|
|
f"Expected reason={reason!r}, got {context.es_error.reason!r}"
|
|
)
|