"""Step definitions for actor compiler full coverage tests. Tests for features/actor_compiler_coverage.feature — targets every code path in ``cleveragents.actor.compiler`` including edge-case branches in node mapping, edge mapping, LSP binding extraction, subgraph cycle detection, and defensive compile_actor error handling. """ from __future__ import annotations from typing import Any from unittest.mock import patch from behave import given, then, when # type: ignore[import-untyped] from behave.runner import Context # type: ignore[import-untyped] from cleveragents.actor.compiler import ( CompilationMetadata, CompiledActor, InvalidEntryExitError, MissingNodeError, SubgraphCycleError, compile_actor, ) from cleveragents.actor.schema import ( ActorConfigSchema, ActorType, EdgeDefinition, NodeDefinition, NodeType, RouteDefinition, ) from cleveragents.langgraph import nodes as lg_nodes # ──────────────────────────────────────────────────────────── # Helper builders # ──────────────────────────────────────────────────────────── def _build_graph_config( name: str, nodes: list[NodeDefinition], edges: list[EdgeDefinition], entry_node: str, exit_nodes: list[str], ) -> ActorConfigSchema: """Build a valid GRAPH ActorConfigSchema for testing.""" route = RouteDefinition( nodes=nodes, edges=edges, entry_node=entry_node, exit_nodes=exit_nodes, ) return ActorConfigSchema( name=name, type=ActorType.GRAPH, description="Coverage test graph actor", model="gpt-4", route=route, ) def _build_graph_config_raw( name: str, nodes: list[NodeDefinition], edges: list[EdgeDefinition], entry_node: str, exit_nodes: list[str], ) -> ActorConfigSchema: """Build a GRAPH ActorConfigSchema using model_construct to bypass validators. This allows constructing configs that would normally fail validation, enabling tests for the compiler's own defensive checks. """ route = RouteDefinition.model_construct( nodes=nodes, edges=edges, entry_node=entry_node, exit_nodes=exit_nodes, ) return ActorConfigSchema.model_construct( name=name, type=ActorType.GRAPH, description="Coverage test graph actor (raw)", model="gpt-4", version="1.0", tools=[], memory=None, context=None, context_view=None, system_prompt=None, route=route, env_vars={}, ) def _agent_node(node_id: str, agent: str = "default") -> NodeDefinition: """Shorthand for creating an AGENT NodeDefinition.""" return NodeDefinition( id=node_id, type=NodeType.AGENT, name=node_id.title(), description=f"Agent {node_id}", config={"agent": agent}, ) def _tool_node(node_id: str, tools: list[str] | None = None) -> NodeDefinition: """Shorthand for creating a TOOL NodeDefinition.""" return NodeDefinition( id=node_id, type=NodeType.TOOL, name=node_id.title(), description=f"Tool {node_id}", config={"tool_name": "t/run", "tools": tools or []}, ) def _conditional_node(node_id: str, function: str = "check") -> NodeDefinition: """Shorthand for creating a CONDITIONAL NodeDefinition.""" return NodeDefinition( id=node_id, type=NodeType.CONDITIONAL, name=node_id.title(), description=f"Conditional {node_id}", config={"function": function}, ) def _subgraph_node(node_id: str, actor_ref: str = "") -> NodeDefinition: """Shorthand for creating a SUBGRAPH NodeDefinition.""" return NodeDefinition( id=node_id, type=NodeType.SUBGRAPH, name=node_id.title(), description=f"Subgraph {node_id}", config={"actor_ref": actor_ref}, ) # ──────────────────────────────────────────────────────────── # Given steps — node type mapping # ──────────────────────────────────────────────────────────── @given("a GRAPH actor config containing AGENT, TOOL, CONDITIONAL, and SUBGRAPH nodes") def step_given_all_four_node_types(context: Context) -> None: nodes = [ _agent_node("agent_node", agent="a1"), _tool_node("tool_node"), _conditional_node("cond_node", function="fn"), _subgraph_node("sub_node", actor_ref="workflows/inner"), ] edges = [ EdgeDefinition(from_node="agent_node", to_node="tool_node"), EdgeDefinition(from_node="tool_node", to_node="cond_node"), EdgeDefinition(from_node="cond_node", to_node="sub_node"), ] context.actor_config = _build_graph_config( "workflows/all-types", nodes, edges, "agent_node", ["sub_node"] ) @given('a GRAPH actor config with an agent node configured as "{agent_name}"') def step_given_agent_config(context: Context, agent_name: str) -> None: nodes = [_agent_node("agent_node", agent=agent_name)] context.actor_config = _build_graph_config( "workflows/agent-cfg", nodes, [], "agent_node", ["agent_node"], ) @given('a GRAPH actor config with a conditional node configured with function "{fn}"') def step_given_conditional_config(context: Context, fn: str) -> None: nodes = [ _agent_node("start"), _conditional_node("cond_node", function=fn), ] edges = [EdgeDefinition(from_node="start", to_node="cond_node")] context.actor_config = _build_graph_config( "workflows/cond-cfg", nodes, edges, "start", ["cond_node"] ) @given('a GRAPH actor config with a tool node configured with tools "{tools_csv}"') def step_given_tool_config(context: Context, tools_csv: str) -> None: tools = [t.strip() for t in tools_csv.split(",")] nodes = [ _agent_node("start"), NodeDefinition( id="tool_node", type=NodeType.TOOL, name="Tools", description="Tool node", config={"tool_name": "t/run", "tools": tools}, ), ] edges = [EdgeDefinition(from_node="start", to_node="tool_node")] context.actor_config = _build_graph_config( "workflows/tool-cfg", nodes, edges, "start", ["tool_node"] ) @given('a GRAPH actor config with a subgraph node referencing actor "{actor_ref}"') def step_given_subgraph_config(context: Context, actor_ref: str) -> None: nodes = [ _agent_node("start"), _subgraph_node("sub_node", actor_ref=actor_ref), ] edges = [EdgeDefinition(from_node="start", to_node="sub_node")] context.actor_config = _build_graph_config( "workflows/sub-cfg", nodes, edges, "start", ["sub_node"] ) # ──────────────────────────────────────────────────────────── # Given steps — edge mapping # ──────────────────────────────────────────────────────────── @given('a GRAPH actor config with an unconditional edge from "{src}" to "{tgt}"') def step_given_unconditional_edge(context: Context, src: str, tgt: str) -> None: nodes = [_agent_node(src), _agent_node(tgt)] edges = [EdgeDefinition(from_node=src, to_node=tgt)] context.actor_config = _build_graph_config( "workflows/uncond-edge", nodes, edges, src, [tgt] ) @given( 'a GRAPH actor config with a conditional edge from "{src}" to "{tgt}" ' 'with condition "{cond}"' ) def step_given_conditional_edge( context: Context, src: str, tgt: str, cond: str ) -> None: nodes = [_agent_node(src), _agent_node(tgt)] edges = [EdgeDefinition(from_node=src, to_node=tgt, condition=cond)] context.actor_config = _build_graph_config( "workflows/cond-edge", nodes, edges, src, [tgt] ) @given( 'a GRAPH actor config with an edge from "{src}" to "{tgt}" with priority {pri:d}' ) def step_given_priority_edge(context: Context, src: str, tgt: str, pri: int) -> None: nodes = [_agent_node(src), _agent_node(tgt)] edges = [EdgeDefinition(from_node=src, to_node=tgt, priority=pri)] context.actor_config = _build_graph_config( "workflows/pri-edge", nodes, edges, src, [tgt] ) # ──────────────────────────────────────────────────────────── # Given steps — LSP binding edge cases # ──────────────────────────────────────────────────────────── def _single_node_with_lsp(lsp_bindings: Any) -> ActorConfigSchema: """Build a config with one agent node carrying arbitrary lsp_bindings.""" nodes = [ NodeDefinition( id="node_a", type=NodeType.AGENT, name="A", description="Agent A", config={"agent": "a", "lsp_bindings": lsp_bindings}, ), ] return _build_graph_config("workflows/lsp-test", nodes, [], "node_a", ["node_a"]) @given("a GRAPH actor config where lsp_bindings is a string instead of a list") def step_given_lsp_not_list(context: Context) -> None: context.actor_config = _single_node_with_lsp("not_a_list") @given("a GRAPH actor config where lsp_bindings contains a non-dict entry") def step_given_lsp_non_dict(context: Context) -> None: context.actor_config = _single_node_with_lsp(["not_a_dict", 42]) @given("a GRAPH actor config where an LSP binding has an empty server name") def step_given_lsp_empty_server(context: Context) -> None: context.actor_config = _single_node_with_lsp( [{"lsp_server_name": "", "languages": ["python"]}] ) @given("a GRAPH actor config where an LSP binding has a numeric server name") def step_given_lsp_numeric_server(context: Context) -> None: context.actor_config = _single_node_with_lsp( [{"lsp_server_name": 999, "languages": ["python"]}] ) @given("a GRAPH actor config where an LSP binding is missing the server name key") def step_given_lsp_missing_key(context: Context) -> None: context.actor_config = _single_node_with_lsp( [{"languages": ["python"], "auto_detect": True}] ) @given("a GRAPH actor config with two valid LSP bindings on one node") def step_given_two_lsp_bindings(context: Context) -> None: context.actor_config = _single_node_with_lsp( [ {"lsp_server_name": "local/pyright", "languages": ["python"]}, { "lsp_server_name": "local/ts", "languages": ["typescript"], "auto_detect": False, }, ] ) # ──────────────────────────────────────────────────────────── # Given steps — subgraph cycle detection edge cases # ──────────────────────────────────────────────────────────── @given("a GRAPH actor config with a subgraph node that has an empty actor_ref") def step_given_empty_actor_ref(context: Context) -> None: nodes = [ _agent_node("start"), _subgraph_node("sub_node", actor_ref=""), ] edges = [EdgeDefinition(from_node="start", to_node="sub_node")] context.actor_config = _build_graph_config( "workflows/empty-ref", nodes, edges, "start", ["sub_node"] ) context.actor_resolver = lambda name: None @given('a GRAPH actor config with a subgraph referencing actor name "{ref}"') def step_given_subgraph_ref_generic(context: Context, ref: str) -> None: nodes = [ _agent_node("start"), _subgraph_node("sub_node", actor_ref=ref), ] edges = [EdgeDefinition(from_node="start", to_node="sub_node")] context.actor_config = _build_graph_config( "workflows/ref-test", nodes, edges, "start", ["sub_node"] ) @given("a cycle-detection resolver that returns None for all names") def step_given_resolver_none(context: Context) -> None: context.actor_resolver = lambda name: None @given('a cycle-detection resolver where "{ref}" is an LLM type actor') def step_given_resolver_llm_type(context: Context, ref: str) -> None: llm_actor = ActorConfigSchema( name=ref, type=ActorType.LLM, description="An LLM actor", model="gpt-4", ) context.actor_resolver = lambda name, _r=ref, _a=llm_actor: ( _a if name == _r else None ) @given('a cycle-detection resolver where "{ref}" is a GRAPH actor with no route') def step_given_resolver_no_route(context: Context, ref: str) -> None: graph_no_route = ActorConfigSchema.model_construct( name=ref, type=ActorType.GRAPH, description="Graph with no route", model="gpt-4", version="1.0", tools=[], memory=None, context=None, context_view=None, system_prompt=None, route=None, env_vars={}, ) context.actor_resolver = lambda name, _r=ref, _a=graph_no_route: ( _a if name == _r else None ) @given('a GRAPH actor config named "{outer}" with subgraph ref to "{inner}"') def step_given_outer_ref_inner(context: Context, outer: str, inner: str) -> None: nodes = [ _agent_node("main"), _subgraph_node("sub", actor_ref=inner), ] edges = [EdgeDefinition(from_node="main", to_node="sub")] context.actor_config = _build_graph_config(outer, nodes, edges, "main", ["sub"]) context._outer_name = outer context._inner_name = inner @given( 'a cycle-detection resolver where "{mid}" references "{leaf}" ' 'and "{leaf}" references "{back}"' ) def step_given_deep_cycle_resolver( context: Context, mid: str, leaf: str, back: str ) -> None: # mid actor: has subgraph referencing leaf mid_nodes = [_subgraph_node("sub_mid", actor_ref=leaf)] mid_actor = _build_graph_config(mid, mid_nodes, [], "sub_mid", ["sub_mid"]) # leaf actor: has subgraph referencing back (completes the cycle) leaf_nodes = [_subgraph_node("sub_leaf", actor_ref=back)] leaf_actor = _build_graph_config(leaf, leaf_nodes, [], "sub_leaf", ["sub_leaf"]) actors = {mid: mid_actor, leaf: leaf_actor} context.actor_resolver = lambda name, _m=actors: _m.get(name) # ──────────────────────────────────────────────────────────── # Given steps — compile_actor error paths (defensive checks) # ──────────────────────────────────────────────────────────── @given("a GRAPH actor config with a cyclic route that bypasses schema validation") def step_given_intra_cycle(context: Context) -> None: """Create a graph with A→B→A cycle via model_construct to skip pydantic checks.""" nodes = [_agent_node("nodeA"), _agent_node("nodeB")] edges = [ EdgeDefinition(from_node="nodeA", to_node="nodeB"), EdgeDefinition(from_node="nodeB", to_node="nodeA"), ] context.actor_config = _build_graph_config_raw( "workflows/cycle-raw", nodes, edges, "nodeA", ["nodeB"] ) @given( "a GRAPH actor config with an edge whose from_node does not exist in the node list" ) def step_given_missing_from_node(context: Context) -> None: nodes = [_agent_node("real_node")] edges = [EdgeDefinition(from_node="ghost", to_node="real_node")] context.actor_config = _build_graph_config_raw( "workflows/missing-from", nodes, edges, "real_node", ["real_node"] ) context._patch_validate_references = True @given( "a GRAPH actor config with an edge whose to_node does not exist in the node list" ) def step_given_missing_to_node(context: Context) -> None: nodes = [_agent_node("real_node")] edges = [EdgeDefinition(from_node="real_node", to_node="ghost")] context.actor_config = _build_graph_config_raw( "workflows/missing-to", nodes, edges, "real_node", ["real_node"] ) context._patch_validate_references = True @given("a GRAPH actor config whose entry node does not match any compiled node") def step_given_invalid_entry(context: Context) -> None: nodes = [_agent_node("actual")] context.actor_config = _build_graph_config_raw( "workflows/bad-entry", nodes, [], "nonexistent_entry", ["actual"] ) context._patch_validate_references = True @given("a GRAPH actor config whose exit node does not match any compiled node") def step_given_invalid_exit(context: Context) -> None: nodes = [_agent_node("actual")] context.actor_config = _build_graph_config_raw( "workflows/bad-exit", nodes, [], "actual", ["nonexistent_exit"] ) context._patch_validate_references = True # ──────────────────────────────────────────────────────────── # Given steps — model defaults # ──────────────────────────────────────────────────────────── @given("a freshly constructed CompilationMetadata with no arguments") def step_given_default_metadata(context: Context) -> None: context.metadata = CompilationMetadata() @given('a GRAPH actor config named "{name}" with entry "{entry}"') def step_given_named_config(context: Context, name: str, entry: str) -> None: nodes = [_agent_node(entry), _agent_node("end")] edges = [EdgeDefinition(from_node=entry, to_node="end")] context.actor_config = _build_graph_config(name, nodes, edges, entry, ["end"]) # ──────────────────────────────────────────────────────────── # When steps # ──────────────────────────────────────────────────────────── @when("I compile the full-coverage actor config") def step_when_compile(context: Context) -> None: context.compile_error = None needs_patch = getattr(context, "_patch_validate_references", False) try: if needs_patch: with ( patch.object(RouteDefinition, "validate_references", return_value=None), patch.object(RouteDefinition, "detect_cycles", return_value=[]), ): context.compiled = compile_actor(context.actor_config) else: context.compiled = compile_actor(context.actor_config) except Exception as exc: context.compile_error = exc context.compiled = None @when("I compile the full-coverage actor config with a resolver") def step_when_compile_with_resolver(context: Context) -> None: resolver = getattr(context, "actor_resolver", None) context.compile_error = None try: context.compiled = compile_actor(context.actor_config, actor_resolver=resolver) except Exception as exc: context.compile_error = exc context.compiled = None # ──────────────────────────────────────────────────────────── # Then steps — success # ──────────────────────────────────────────────────────────── @then("the compilation should succeed without errors") def step_then_succeed(context: Context) -> None: assert context.compile_error is None, ( f"Expected compilation to succeed but got: {context.compile_error}" ) assert context.compiled is not None # ──────────────────────────────────────────────────────────── # Then steps — node type assertions # ──────────────────────────────────────────────────────────── @then('the compiled node "{node_id}" should have LangGraph type "{type_name}"') def step_then_node_type(context: Context, node_id: str, type_name: str) -> None: assert context.compiled is not None node_cfg = context.compiled.nodes.get(node_id) assert node_cfg is not None, f"Node '{node_id}' not in compiled nodes" expected = lg_nodes.NodeType(type_name) assert node_cfg.type == expected, f"Expected type {expected}, got {node_cfg.type}" @then('the compiled node "{node_id}" should have agent set to "{agent}"') def step_then_node_agent(context: Context, node_id: str, agent: str) -> None: assert context.compiled is not None node_cfg = context.compiled.nodes[node_id] assert node_cfg.agent == agent, f"Expected agent={agent}, got {node_cfg.agent}" @then('the compiled node "{node_id}" should have function set to "{fn}"') def step_then_node_function(context: Context, node_id: str, fn: str) -> None: assert context.compiled is not None node_cfg = context.compiled.nodes[node_id] assert node_cfg.function == fn, f"Expected function={fn}, got {node_cfg.function}" @then('the compiled node "{node_id}" should have tools "{t1}" and "{t2}"') def step_then_node_tools(context: Context, node_id: str, t1: str, t2: str) -> None: assert context.compiled is not None node_cfg = context.compiled.nodes[node_id] assert t1 in node_cfg.tools, f"Expected {t1} in tools, got {node_cfg.tools}" assert t2 in node_cfg.tools, f"Expected {t2} in tools, got {node_cfg.tools}" @then('the compiled node "{node_id}" should have subgraph set to "{ref}"') def step_then_node_subgraph(context: Context, node_id: str, ref: str) -> None: assert context.compiled is not None node_cfg = context.compiled.nodes[node_id] assert node_cfg.subgraph == ref, f"Expected subgraph={ref}, got {node_cfg.subgraph}" # ──────────────────────────────────────────────────────────── # Then steps — edge assertions # ──────────────────────────────────────────────────────────── def _find_edge( compiled: CompiledActor, source: str, target: str ) -> lg_nodes.Edge | None: for e in compiled.edges: if e.source == source and e.target == target: return e return None @then('the compiled edge from "{src}" to "{tgt}" should have a null condition') def step_then_edge_no_condition(context: Context, src: str, tgt: str) -> None: assert context.compiled is not None edge = _find_edge(context.compiled, src, tgt) assert edge is not None, f"Edge {src}->{tgt} not found" assert edge.condition is None, f"Expected None condition, got {edge.condition}" @then( 'the compiled edge from "{src}" to "{tgt}" should have condition expression "{expr}"' ) def step_then_edge_condition(context: Context, src: str, tgt: str, expr: str) -> None: assert context.compiled is not None edge = _find_edge(context.compiled, src, tgt) assert edge is not None, f"Edge {src}->{tgt} not found" assert edge.condition is not None, "Expected condition dict, got None" assert edge.condition.get("expression") == expr, ( f"Expected expression='{expr}', got {edge.condition}" ) @then('the compiled edge from "{src}" to "{tgt}" should have metadata priority {pri:d}') def step_then_edge_priority(context: Context, src: str, tgt: str, pri: int) -> None: assert context.compiled is not None edge = _find_edge(context.compiled, src, tgt) assert edge is not None, f"Edge {src}->{tgt} not found" assert edge.metadata.get("priority") == pri, ( f"Expected priority={pri}, got {edge.metadata}" ) # ──────────────────────────────────────────────────────────── # Then steps — LSP binding assertions # ──────────────────────────────────────────────────────────── @then("the compilation metadata should have {count:d} LSP bindings") def step_then_lsp_count(context: Context, count: int) -> None: assert context.compiled is not None actual = len(context.compiled.metadata.lsp_bindings) assert actual == count, f"Expected {count} LSP bindings, got {actual}" @then('the LSP binding servers should include "{s1}" and "{s2}"') def step_then_lsp_servers(context: Context, s1: str, s2: str) -> None: assert context.compiled is not None servers = [b.lsp_server_name for b in context.compiled.metadata.lsp_bindings] assert s1 in servers, f"{s1} not in {servers}" assert s2 in servers, f"{s2} not in {servers}" # ──────────────────────────────────────────────────────────── # Then steps — error assertions # ──────────────────────────────────────────────────────────── @then("the compilation should fail with a SubgraphCycleError") def step_then_fail_cycle(context: Context) -> None: assert context.compile_error is not None, "Expected SubgraphCycleError, got None" assert isinstance(context.compile_error, SubgraphCycleError), ( f"Expected SubgraphCycleError, got {type(context.compile_error).__name__}: " f"{context.compile_error}" ) @then("the compilation should fail with a MissingNodeError") def step_then_fail_missing(context: Context) -> None: assert context.compile_error is not None, "Expected MissingNodeError, got None" assert isinstance(context.compile_error, MissingNodeError), ( f"Expected MissingNodeError, got {type(context.compile_error).__name__}: " f"{context.compile_error}" ) @then("the compilation should fail with an InvalidEntryExitError") def step_then_fail_entry_exit(context: Context) -> None: assert context.compile_error is not None, "Expected InvalidEntryExitError, got None" assert isinstance(context.compile_error, InvalidEntryExitError), ( f"Expected InvalidEntryExitError, got " f"{type(context.compile_error).__name__}: {context.compile_error}" ) @then('the compilation error message should mention "{text}"') def step_then_error_contains(context: Context, text: str) -> None: assert context.compile_error is not None msg = str(context.compile_error) assert text.lower() in msg.lower(), f"'{text}' not found in '{msg}'" # ──────────────────────────────────────────────────────────── # Then steps — model defaults # ──────────────────────────────────────────────────────────── @then("the metadata node_ids should be an empty list") def step_then_empty_node_ids(context: Context) -> None: assert context.metadata.node_ids == [] @then("the metadata tool_nodes should be an empty list") def step_then_empty_tool_nodes(context: Context) -> None: assert context.metadata.tool_nodes == [] @then("the metadata lsp_bindings should be an empty list") def step_then_empty_lsp_bindings(context: Context) -> None: assert context.metadata.lsp_bindings == [] @then("the metadata subgraph_refs should be an empty dict") def step_then_empty_subgraph_refs(context: Context) -> None: assert context.metadata.subgraph_refs == {} @then("the metadata entry_node should be an empty string") def step_then_empty_entry(context: Context) -> None: assert context.metadata.entry_node == "" @then("the metadata exit_nodes should be an empty list") def step_then_empty_exit_nodes(context: Context) -> None: assert context.metadata.exit_nodes == [] # ──────────────────────────────────────────────────────────── # Then steps — result bundle # ──────────────────────────────────────────────────────────── @then('the compiled actor name should be "{name}"') def step_then_actor_name(context: Context, name: str) -> None: assert context.compiled is not None assert context.compiled.name == name, ( f"Expected name='{name}', got '{context.compiled.name}'" ) @then('the compiled actor entry_point field should be "{entry}"') def step_then_actor_entry(context: Context, entry: str) -> None: assert context.compiled is not None assert context.compiled.entry_point == entry, ( f"Expected entry_point='{entry}', got '{context.compiled.entry_point}'" )