"""Step definitions for actor YAML schema validation. Tests for features/actor_schema.feature — validates the ActorConfigSchema Pydantic model, YAML loading, graph topology validation, tool definitions, and error messages. """ from __future__ import annotations import tempfile from pathlib import Path from behave import given, then, when # type: ignore[import-untyped] from behave.runner import Context # type: ignore[import-untyped] from pydantic import ValidationError from cleveragents.actor.schema import ActorConfigSchema, NodeType, actor_role_warnings # ──────────────────────────────────────────────────────────── # Test YAML Templates # ──────────────────────────────────────────────────────────── _MINIMAL_LLM_YAML = """\ name: assistants/simple type: llm description: A simple LLM actor provider: openai model: gpt-4 """ _LLM_WITH_PROMPT_YAML = """\ name: assistants/expert type: llm description: An expert assistant provider: openai model: gpt-4 system_prompt: "You are an expert Python developer" """ _LLM_WITH_TOOLS_YAML = """\ name: assistants/helper type: llm description: Helper with tools provider: openai model: gpt-4 tools: - files/read_file - files/write_file """ _LLM_WITH_MEMORY_YAML = """\ name: assistants/chatbot type: llm description: Chatbot with memory provider: openai model: gpt-4 memory: enabled: true max_messages: 50 max_tokens: 4000 """ _LLM_WITH_CONTEXT_YAML = """\ name: assistants/analyzer type: llm description: Code analyzer provider: openai model: gpt-4 context: include_files: - README.md - pyproject.toml include_dirs: - src/ exclude_patterns: - "**/__pycache__/**" """ _TOOL_MINIMAL_YAML = """\ name: utilities/helpers type: tool description: Utility tools tools: - files/read_file """ _TOOL_INLINE_YAML = """\ name: utilities/custom type: tool description: Custom tools tools: - name: utils/count_lines description: Count lines in a file parameters: - name: file_path type: str description: Path to file required: true code: | def count_lines(file_path: str) -> int: with open(file_path, 'r') as f: return len(f.readlines()) """ _GRAPH_MINIMAL_YAML = """\ name: workflows/simple type: graph description: Simple workflow provider: openai model: gpt-4 route: nodes: - id: start type: agent name: Starter description: Start node config: prompt: "Begin processing" edges: [] entry_node: start exit_nodes: - start """ _GRAPH_LINEAR_YAML = """\ name: workflows/linear type: graph description: Linear workflow provider: openai model: gpt-4 route: nodes: - id: extract type: tool name: Extractor description: Extract data config: tool_name: data/extract - id: process type: agent name: Processor description: Process data config: prompt: "Process the data" - id: save type: tool name: Saver description: Save results config: tool_name: data/save edges: - from_node: extract to_node: process - from_node: process to_node: save entry_node: extract exit_nodes: - save """ _GRAPH_CONDITIONAL_YAML = """\ name: workflows/conditional type: graph description: Workflow with conditionals provider: openai model: gpt-4 route: nodes: - id: start type: agent name: Starter description: Start config: prompt: "Start" - id: checker type: conditional name: Checker description: Check condition config: conditions: - check: "state.get('ok') == True" route_to: success - check: "state.get('ok') == False" route_to: failure - id: success type: agent name: Success description: Success path config: prompt: "Success" - id: failure type: agent name: Failure description: Failure path config: prompt: "Failure" edges: - from_node: start to_node: checker entry_node: start exit_nodes: - success - failure """ _GRAPH_SUBGRAPH_YAML = """\ name: workflows/composed type: graph description: Workflow with subgraph provider: openai model: gpt-4 route: nodes: - id: main type: agent name: Main description: Main processor config: prompt: "Process" - id: review type: subgraph name: Reviewer description: Review subgraph config: actor_path: actors/reviewer.yaml edges: - from_node: main to_node: review entry_node: main exit_nodes: - review """ # ──────────────────────────────────────────────────────────── # Given Steps # ──────────────────────────────────────────────────────────── @given("an actor YAML string with minimal LLM configuration") def step_given_minimal_llm(context: Context) -> None: """Provide minimal LLM actor YAML.""" context.actor_yaml_string = _MINIMAL_LLM_YAML @given("an actor YAML string with LLM and system prompt") def step_given_llm_with_prompt(context: Context) -> None: """Provide LLM actor with system prompt.""" context.actor_yaml_string = _LLM_WITH_PROMPT_YAML @given("an actor YAML string with LLM and tools") def step_given_llm_with_tools(context: Context) -> None: """Provide LLM actor with tools.""" context.actor_yaml_string = _LLM_WITH_TOOLS_YAML @given("an actor YAML string with memory configuration") def step_given_llm_with_memory(context: Context) -> None: """Provide LLM actor with memory config.""" context.actor_yaml_string = _LLM_WITH_MEMORY_YAML @given("an actor YAML string with context configuration") def step_given_llm_with_context(context: Context) -> None: """Provide LLM actor with context config.""" context.actor_yaml_string = _LLM_WITH_CONTEXT_YAML @given("an actor YAML string with invalid response_format missing type") def step_given_invalid_response_format_missing_type(context: Context) -> None: """Provide estimation actor with invalid response_format missing type.""" context.actor_yaml_string = """\ name: local/invalid-response-format type: llm description: Invalid response format model: gpt-4 role_hint: estimation context_view: strategist response_format: title: EstimationReport """ @given("an actor YAML string with invalid role_hint") def step_given_invalid_role_hint(context: Context) -> None: """Provide estimation actor with invalid role_hint value.""" context.actor_yaml_string = """\ name: local/invalid-role-hint type: llm description: Invalid role hint model: gpt-4 role_hint: estmation """ @given("an actor YAML string with non-dict response_format") def step_given_nondict_response_format(context: Context) -> None: """Provide estimation actor with non-dict response_format value.""" context.actor_yaml_string = """\ name: local/non-dict-response-format type: llm description: Non-dict response format model: gpt-4 role_hint: estimation response_format: - not-a-dict """ @given("an ActorConfigSchema estimation actor with executor context_view") def step_given_actor_model_for_role_warning(context: Context) -> None: """Provide ActorConfigSchema instance for actor_role_warnings model-input path.""" context.actor_config = ActorConfigSchema( name="local/model-warning-actor", type="llm", description="Model-input warnings path", provider="openai", model="gpt-4", role_hint="estimation", context_view="executor", response_format={"type": "object"}, ) @given("an ActorConfigSchema estimation actor without response_format") def step_given_actor_model_without_response_format(context: Context) -> None: """Provide ActorConfigSchema estimation actor to exercise model missing-schema warning.""" context.actor_config = ActorConfigSchema( name="local/model-warning-no-schema", type="llm", description="Model-input missing response_format", provider="openai", model="gpt-4", role_hint="estimation", context_view="strategist", ) @given("an estimation actor payload with unrecognized context_view") def step_given_payload_with_unrecognized_context_view(context: Context) -> None: """Provide dict payload using invalid context_view to verify warning path.""" context.actor_payload = { "name": "local/payload-warning-context", "type": "llm", "model": "gpt-4", "role_hint": "estimation", "context_view": "plannerish", "response_format": {"type": "object"}, } @given("an estimation actor payload with uppercase role_hint") def step_given_payload_with_uppercase_role_hint(context: Context) -> None: """Provide dict payload with uppercase role_hint to validate case-insensitive coercion.""" context.actor_payload = { "name": "local/payload-uppercase-role", "type": "llm", "model": "gpt-4", "role_hint": "ESTIMATION", "context_view": "strategist", } @given("an actor YAML string with TOOL type and tools") def step_given_tool_minimal(context: Context) -> None: """Provide minimal TOOL actor.""" context.actor_yaml_string = _TOOL_MINIMAL_YAML @given("an actor YAML string with inline tool definition") def step_given_tool_inline(context: Context) -> None: """Provide TOOL actor with inline tool.""" context.actor_yaml_string = _TOOL_INLINE_YAML @given("an actor YAML string with minimal GRAPH configuration") def step_given_graph_minimal(context: Context) -> None: """Provide minimal GRAPH actor.""" context.actor_yaml_string = _GRAPH_MINIMAL_YAML @given("an actor YAML string with linear graph topology") def step_given_graph_linear(context: Context) -> None: """Provide linear GRAPH actor.""" context.actor_yaml_string = _GRAPH_LINEAR_YAML @given("an actor YAML string with conditional routing") def step_given_graph_conditional(context: Context) -> None: """Provide GRAPH with conditional node.""" context.actor_yaml_string = _GRAPH_CONDITIONAL_YAML @given("an actor YAML string with subgraph node") def step_given_graph_subgraph(context: Context) -> None: """Provide GRAPH with subgraph node.""" context.actor_yaml_string = _GRAPH_SUBGRAPH_YAML @given('an actor YAML string with name "{name}"') def step_given_actor_with_name(context: Context, name: str) -> None: """Provide actor YAML with specific name.""" context.actor_yaml_string = f"""\ name: {name} type: llm description: Test actor provider: openai model: gpt-4 """ @given("an actor YAML string with LLM type but no model") def step_given_llm_no_model(context: Context) -> None: """Provide LLM actor without model field.""" context.actor_yaml_string = """\ name: assistants/broken type: llm description: Missing model """ @given("an actor YAML string with TOOL type but no tools") def step_given_tool_no_tools(context: Context) -> None: """Provide TOOL actor without tools field.""" context.actor_yaml_string = """\ name: utilities/broken type: tool description: Missing tools """ @given("an actor YAML string with TOOL type and empty tools") def step_given_tool_empty_tools(context: Context) -> None: """Provide TOOL actor with empty tools list.""" context.actor_yaml_string = """\ name: utilities/broken type: tool description: Empty tools tools: [] """ @given("an actor YAML string with GRAPH type but no model") def step_given_graph_no_model(context: Context) -> None: """Provide GRAPH actor without model field.""" context.actor_yaml_string = """\ name: workflows/broken type: graph description: Missing model route: nodes: [] edges: [] entry_node: start exit_nodes: [] """ @given("an actor YAML string with GRAPH type but no route") def step_given_graph_no_route(context: Context) -> None: """Provide GRAPH actor without route field.""" context.actor_yaml_string = """\ name: workflows/broken type: graph description: Missing route model: gpt-4 """ @given("an actor YAML string with duplicate node IDs") def step_given_duplicate_nodes(context: Context) -> None: """Provide GRAPH with duplicate node IDs.""" context.actor_yaml_string = """\ name: workflows/duplicate type: graph description: Duplicate nodes model: gpt-4 route: nodes: - id: node1 type: agent name: First description: First node config: prompt: "First" - id: node1 type: agent name: Second description: Duplicate ID config: prompt: "Second" edges: [] entry_node: node1 exit_nodes: - node1 """ @given("an actor YAML string with non-existent entry node") def step_given_invalid_entry_node(context: Context) -> None: """Provide GRAPH with invalid entry node.""" context.actor_yaml_string = """\ name: workflows/bad_entry type: graph description: Bad entry node model: gpt-4 route: nodes: - id: actual_node type: agent name: Node description: The only node config: prompt: "Process" edges: [] entry_node: missing_node exit_nodes: - actual_node """ @given("an actor YAML string with non-existent exit node") def step_given_invalid_exit_node(context: Context) -> None: """Provide GRAPH with invalid exit node.""" context.actor_yaml_string = """\ name: workflows/bad_exit type: graph description: Bad exit node model: gpt-4 route: nodes: - id: actual_node type: agent name: Node description: The only node config: prompt: "Process" edges: [] entry_node: actual_node exit_nodes: - missing_node """ @given("an actor YAML string with invalid edge from_node") def step_given_invalid_edge_from(context: Context) -> None: """Provide GRAPH with invalid from_node in edge.""" context.actor_yaml_string = """\ name: workflows/bad_edge type: graph description: Bad edge from_node model: gpt-4 route: nodes: - id: node_a type: agent name: Node A description: First node config: prompt: "A" - id: node_b type: agent name: Node B description: Second node config: prompt: "B" edges: - from_node: missing_node to_node: node_b entry_node: node_a exit_nodes: - node_b """ @given("an actor YAML string with invalid edge to_node") def step_given_invalid_edge_to(context: Context) -> None: """Provide GRAPH with invalid to_node in edge.""" context.actor_yaml_string = """\ name: workflows/bad_edge type: graph description: Bad edge to_node model: gpt-4 route: nodes: - id: node_a type: agent name: Node A description: First node config: prompt: "A" - id: node_b type: agent name: Node B description: Second node config: prompt: "B" edges: - from_node: node_a to_node: missing_node entry_node: node_a exit_nodes: - node_b """ @given("an actor YAML string with cyclic graph") def step_given_cyclic_graph(context: Context) -> None: """Provide GRAPH with cycle.""" context.actor_yaml_string = """\ name: workflows/cyclic type: graph description: Graph with cycle model: gpt-4 route: nodes: - id: node_a type: agent name: Node A description: First config: prompt: "A" - id: node_b type: agent name: Node B description: Second config: prompt: "B" - id: node_c type: agent name: Node C description: Third config: prompt: "C" edges: - from_node: node_a to_node: node_b - from_node: node_b to_node: node_c - from_node: node_c to_node: node_a entry_node: node_a exit_nodes: - node_c """ @given("an actor YAML string with unreachable node") def step_given_unreachable_node(context: Context) -> None: """Provide GRAPH where a node has no path from the entry node.""" context.actor_yaml_string = """\ name: workflows/unreachable type: graph description: Graph with an isolated node model: gpt-4 route: nodes: - id: node_a type: agent name: Node A description: Entry node config: prompt: "A" - id: node_b type: agent name: Node B description: Reachable from entry config: prompt: "B" - id: node_c type: agent name: Node C description: Isolated — no edge points here from entry path config: prompt: "C" edges: - from_node: node_a to_node: node_b entry_node: node_a exit_nodes: - node_b """ @given('an actor YAML string with inline tool name "{name}"') def step_given_inline_tool_name(context: Context, name: str) -> None: """Provide actor with inline tool having specific name.""" context.actor_yaml_string = f"""\ name: utilities/test type: tool description: Test tool tools: - name: {name} description: Test tool parameters: [] code: "def test(): pass" """ @given("an actor YAML string with invalid tool parameter name") def step_given_invalid_param_name(context: Context) -> None: """Provide inline tool with invalid parameter name.""" context.actor_yaml_string = """\ name: utilities/bad_param type: tool description: Bad parameter tools: - name: utils/bad_tool description: Tool with bad param parameters: - name: invalid-name! type: str required: true code: "def bad_tool(): pass" """ @given("an actor YAML string with node ID containing spaces") def step_given_invalid_node_id(context: Context) -> None: """Provide GRAPH with invalid node ID.""" context.actor_yaml_string = """\ name: workflows/bad_id type: graph description: Bad node ID model: gpt-4 route: nodes: - id: "node with spaces" type: agent name: Bad Node description: Node with invalid ID config: prompt: "Process" edges: [] entry_node: "node with spaces" exit_nodes: - "node with spaces" """ @given('an actor YAML string with node ID "{node_id}"') def step_given_specific_node_id(context: Context, node_id: str) -> None: """Provide GRAPH with specific node ID.""" context.actor_yaml_string = f"""\ name: workflows/test type: graph description: Test workflow provider: openai model: gpt-4 route: nodes: - id: {node_id} type: agent name: Test Node description: Test node config: prompt: "Test" edges: [] entry_node: {node_id} exit_nodes: - {node_id} """ @given('an actor YAML string with context_view "{view}"') def step_given_context_view(context: Context, view: str) -> None: """Provide actor with specific context view.""" context.actor_yaml_string = f"""\ name: assistants/test type: llm description: Test actor provider: openai model: gpt-4 context_view: {view} """ @given("an actor YAML string with env_vars") def step_given_env_vars(context: Context) -> None: """Provide actor with environment variables.""" context.actor_yaml_string = """\ name: assistants/test type: llm description: Test actor provider: openai model: gpt-4 env_vars: LOG_LEVEL: info WORK_DIR: /tmp/work """ @given('the actor YAML file "{file_path}"') def step_given_actor_yaml_file(context: Context, file_path: str) -> None: """Store actor YAML file path for loading.""" context.actor_yaml_file = file_path @given("a valid actor configuration object") def step_given_valid_actor_object(context: Context) -> None: """Create a valid actor configuration object.""" context.actor_config = ActorConfigSchema( name="test/actor", type="llm", description="Test actor", provider="openai", model="gpt-4", ) @given("a non-existent actor YAML file path") def step_given_nonexistent_file(context: Context) -> None: """Provide a non-existent file path.""" context.actor_yaml_file = "/nonexistent/path/actor.yaml" @given("an actor YAML string with edges having different priorities") def step_given_edge_priorities(context: Context) -> None: """Provide GRAPH with edge priorities.""" context.actor_yaml_string = """\ name: workflows/priorities type: graph description: Workflow with priorities provider: openai model: gpt-4 route: nodes: - id: start type: agent name: Start description: Start node config: prompt: "Start" - id: end type: agent name: End description: End node config: prompt: "End" edges: - from_node: start to_node: end priority: 10 entry_node: start exit_nodes: - end """ @given("an actor YAML string with memory enabled false") def step_given_memory_disabled(context: Context) -> None: """Provide actor with memory disabled.""" context.actor_yaml_string = """\ name: assistants/test type: llm description: Test actor provider: openai model: gpt-4 memory: enabled: false """ @given("an actor YAML string with max_messages {count:d}") def step_given_max_messages(context: Context, count: int) -> None: """Provide actor with max_messages limit.""" context.actor_yaml_string = f"""\ name: assistants/test type: llm description: Test actor provider: openai model: gpt-4 memory: max_messages: {count} """ @given("an actor YAML string with max_tokens {count:d}") def step_given_max_tokens(context: Context, count: int) -> None: """Provide actor with max_tokens limit.""" context.actor_yaml_string = f"""\ name: assistants/test type: llm description: Test actor provider: openai model: gpt-4 memory: max_tokens: {count} """ @given("an actor YAML string with summarize_old true") def step_given_summarize_old(context: Context) -> None: """Provide actor with summarize_old enabled.""" context.actor_yaml_string = """\ name: assistants/test type: llm description: Test actor provider: openai model: gpt-4 memory: summarize_old: true """ # ──────────────────────────────────────────────────────────── # When Steps # ──────────────────────────────────────────────────────────── @when("I validate the actor schema") def step_when_validate_schema(context: Context) -> None: """Validate actor YAML string.""" import yaml try: data = yaml.safe_load(context.actor_yaml_string) context.actor_config = ActorConfigSchema.model_validate(data) context.validation_error = None context.error = None # For compatibility with service_steps except (ValidationError, ValueError) as e: context.actor_config = None context.validation_error = e context.error = e # For compatibility with service_steps @when("I validate the actor schema from file") def step_when_validate_from_file(context: Context) -> None: """Validate actor YAML from file.""" try: context.actor_config = ActorConfigSchema.from_yaml_file(context.actor_yaml_file) context.validation_error = None context.error = None # For compatibility with service_steps except (ValidationError, ValueError, FileNotFoundError) as e: context.actor_config = None context.validation_error = e context.error = e # For compatibility with service_steps @when("I save the actor to YAML file") def step_when_save_to_file(context: Context) -> None: """Save actor configuration to temporary YAML file.""" with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as temp_file: context.temp_file_name = temp_file.name context.actor_config.to_yaml_file(context.temp_file_name) @when("I reload the actor from YAML file") def step_when_reload_from_file(context: Context) -> None: """Reload actor from saved YAML file.""" context.reloaded_actor = ActorConfigSchema.from_yaml_file(context.temp_file_name) Path(context.temp_file_name).unlink() # Clean up @when("I attempt to load the actor from file") def step_when_attempt_load(context: Context) -> None: """Attempt to load actor from file (may fail).""" try: context.actor_config = ActorConfigSchema.from_yaml_file(context.actor_yaml_file) context.validation_error = None context.error = None # For compatibility with service_steps except FileNotFoundError as e: context.actor_config = None context.validation_error = e context.error = e # For compatibility with service_steps @when("I evaluate actor_role_warnings for the actor model") def step_when_actor_role_warnings_on_model(context: Context) -> None: """Evaluate actor_role_warnings on an ActorConfigSchema object.""" assert context.actor_config is not None context.actor_role_warnings = actor_role_warnings(context.actor_config) @when("I evaluate actor_role_warnings for the actor payload") def step_when_actor_role_warnings_on_payload(context: Context) -> None: """Evaluate actor_role_warnings on a raw dict payload.""" payload = getattr(context, "actor_payload", None) assert isinstance(payload, dict), "actor payload not set" context.actor_role_warnings = actor_role_warnings(payload) # ──────────────────────────────────────────────────────────── # Then Steps # ──────────────────────────────────────────────────────────── @then("the actor schema validation should succeed") def step_then_validation_succeeds(context: Context) -> None: """Assert validation succeeded.""" assert context.validation_error is None, ( f"Expected validation to succeed, but got error: {context.validation_error}" ) assert context.actor_config is not None @then("the actor schema validation should fail") def step_then_validation_fails(context: Context) -> None: """Assert validation failed.""" assert context.validation_error is not None, ( "Expected validation to fail, but it succeeded" ) assert context.actor_config is None @then('the actor config name should be "{expected_name}"') def step_then_name_matches(context: Context, expected_name: str) -> None: """Assert actor name matches.""" assert context.actor_config is not None assert context.actor_config.name == expected_name @then('the actor config type should be "{expected_type}"') def step_then_type_matches(context: Context, expected_type: str) -> None: """Assert actor type matches.""" assert context.actor_config is not None assert context.actor_config.type.value == expected_type @then('the actor config model should be "{expected_model}"') def step_then_model_matches(context: Context, expected_model: str) -> None: """Assert actor model matches.""" assert context.actor_config is not None assert context.actor_config.model == expected_model @then('the actor config system_prompt should contain "{text}"') def step_then_prompt_contains(context: Context, text: str) -> None: """Assert system prompt contains text.""" assert context.actor_config is not None assert context.actor_config.system_prompt is not None assert text in context.actor_config.system_prompt @then("the actor config should have {count:d} tools") def step_then_tool_count(context: Context, count: int) -> None: """Assert tool count matches.""" assert context.actor_config is not None assert len(context.actor_config.tools) == count @then("the actor config should have at least {count:d} tool") def step_then_at_least_tools(context: Context, count: int) -> None: """Assert at least N tools.""" assert context.actor_config is not None assert len(context.actor_config.tools) >= count @then("the actor config should have at least {count:d} tools") def step_then_at_least_tools_plural(context: Context, count: int) -> None: """Assert at least N tools (plural).""" assert context.actor_config is not None assert len(context.actor_config.tools) >= count @then("the actor config should have {count:d} inline tool") def step_then_inline_tool_count(context: Context, count: int) -> None: """Assert inline tool count.""" assert context.actor_config is not None inline_count = sum( 1 for tool in context.actor_config.tools if not isinstance(tool, str) ) assert inline_count == count @then("the actor memory enabled should be {expected:w}") def step_then_memory_enabled(context: Context, expected: str) -> None: """Assert memory enabled state.""" assert context.actor_config is not None expected_bool = expected.lower() == "true" assert context.actor_config.memory.enabled == expected_bool @then("the actor memory max_messages should be {expected:d}") def step_then_memory_max_messages(context: Context, expected: int) -> None: """Assert memory max_messages value.""" assert context.actor_config is not None assert context.actor_config.memory.max_messages == expected @then("the actor memory max_tokens should be {expected:d}") def step_then_memory_max_tokens(context: Context, expected: int) -> None: """Assert memory max_tokens value.""" assert context.actor_config is not None assert context.actor_config.memory.max_tokens == expected @then("the actor memory summarize_old should be {expected:w}") def step_then_memory_summarize(context: Context, expected: str) -> None: """Assert memory summarize_old state.""" assert context.actor_config is not None expected_bool = expected.lower() == "true" assert context.actor_config.memory.summarize_old == expected_bool @then("the actor context should include {count:d} files") def step_then_context_files(context: Context, count: int) -> None: """Assert context includes N files.""" assert context.actor_config is not None assert len(context.actor_config.context.include_files) == count @then("the actor context should include {count:d} directory") def step_then_context_dirs(context: Context, count: int) -> None: """Assert context includes N directories.""" assert context.actor_config is not None assert len(context.actor_config.context.include_dirs) == count @then('the actor route should have entry_node "{node_id}"') def step_then_entry_node(context: Context, node_id: str) -> None: """Assert route entry node matches.""" assert context.actor_config is not None assert context.actor_config.route is not None assert context.actor_config.route.entry_node == node_id @then("the actor route should have {count:d} nodes") def step_then_node_count(context: Context, count: int) -> None: """Assert route node count.""" assert context.actor_config is not None assert context.actor_config.route is not None assert len(context.actor_config.route.nodes) == count @then("the actor route should have {count:d} edges") def step_then_edge_count(context: Context, count: int) -> None: """Assert route edge count.""" assert context.actor_config is not None assert context.actor_config.route is not None assert len(context.actor_config.route.edges) == count @then("the actor route should have {count:d} conditional node") def step_then_conditional_count(context: Context, count: int) -> None: """Assert conditional node count.""" assert context.actor_config is not None assert context.actor_config.route is not None conditional_count = sum( 1 for node in context.actor_config.route.nodes if node.type == NodeType.CONDITIONAL ) assert conditional_count == count @then("the actor route should have {count:d} subgraph node") def step_then_subgraph_count(context: Context, count: int) -> None: """Assert subgraph node count.""" assert context.actor_config is not None assert context.actor_config.route is not None subgraph_count = sum( 1 for node in context.actor_config.route.nodes if node.type == NodeType.SUBGRAPH ) assert subgraph_count == count @then('the validation error should contain "{text}"') def step_then_error_contains(context: Context, text: str) -> None: """Assert error message contains text.""" assert context.validation_error is not None error_str = str(context.validation_error) assert text in error_str, f"Expected '{text}' in error: {error_str}" @then('the actor context_view should be "{expected_view}"') def step_then_context_view_matches(context: Context, expected_view: str) -> None: """Assert context view matches.""" assert context.actor_config is not None assert context.actor_config.context_view is not None assert context.actor_config.context_view.value == expected_view @then('the actor role_hint should be "{expected_hint}"') def step_then_role_hint_matches(context: Context, expected_hint: str) -> None: """Assert role hint matches.""" assert context.actor_config is not None assert context.actor_config.role_hint is not None assert context.actor_config.role_hint.value == expected_hint @then('the actor response_format title should be "{expected_title}"') def step_then_response_format_title_matches( context: Context, expected_title: str ) -> None: """Assert response format title matches.""" assert context.actor_config is not None assert context.actor_config.response_format is not None title = context.actor_config.response_format.get("title") assert title == expected_title @then('the actor response_format should include key "{expected_key}"') def step_then_response_format_includes_key(context: Context, expected_key: str) -> None: """Assert response_format includes the expected top-level key.""" assert context.actor_config is not None assert context.actor_config.response_format is not None assert expected_key in context.actor_config.response_format, ( f"Expected key '{expected_key}' in response_format, " f"got keys {list(context.actor_config.response_format.keys())}" ) @then("the actor config should have {count:d} skills") def step_then_skill_count(context: Context, count: int) -> None: """Assert skill count matches.""" assert context.actor_config is not None assert len(context.actor_config.skills) == count @then("the actor should have {count:d} env_vars") def step_then_env_var_count(context: Context, count: int) -> None: """Assert env_vars count.""" assert context.actor_config is not None assert len(context.actor_config.env_vars) == count @then("the reloaded actor should match the original") def step_then_reloaded_matches(context: Context) -> None: """Assert reloaded actor matches original.""" assert context.actor_config.name == context.reloaded_actor.name assert context.actor_config.type == context.reloaded_actor.type assert context.actor_config.model == context.reloaded_actor.model @then("a FileNotFoundError should be raised") def step_then_file_not_found(context: Context) -> None: """Assert FileNotFoundError was raised.""" assert isinstance(context.validation_error, FileNotFoundError) @then("the highest priority edge should be {priority:d}") def step_then_highest_priority(context: Context, priority: int) -> None: """Assert highest edge priority.""" assert context.actor_config is not None assert context.actor_config.route is not None max_priority = max(edge.priority for edge in context.actor_config.route.edges) assert max_priority == priority @then('actor role warnings should include "{text}"') def step_then_actor_role_warnings_include(context: Context, text: str) -> None: """Assert actor role warnings include expected text fragment.""" warnings = getattr(context, "actor_role_warnings", None) assert isinstance(warnings, list), "No actor role warnings recorded" assert any(text in warning for warning in warnings), ( f"Expected '{text}' in warnings, got: {warnings}" )