"""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 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="" placeholder_found = any(n.model == "" 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 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 "", 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 == "" for n in result.nodes) assert placeholder_found, ( f"Expected 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 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 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 == "" for n in result.nodes) assert placeholder_found, ( f"Expected 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}" )