Files
placeholder/features/steps/actor_compiler_coverage_steps.py
freemo c47e6445d0 feat(actor): compile hierarchical actor configs to LangGraph
Add ActorCompiler module that translates GRAPH-type ActorConfigSchema
definitions into LangGraph NodeConfig/Edge structures with LSP binding
metadata. Includes subgraph resolution with cross-actor cycle detection,
entry/exit validation, and CompilationMetadata for diagnostics.

New files:
- src/cleveragents/actor/compiler.py: Core compiler with compile_actor()
- features/actor_compiler.feature: 13 Behave scenarios
- features/steps/actor_compiler_steps.py: Step definitions
- robot/actor_compiler.robot: 4 Robot smoke tests
- benchmarks/actor_compiler_bench.py: ASV performance benchmarks
- docs/reference/actor_compiler.md: Compilation pipeline reference

Modified:
- src/cleveragents/actor/__init__.py: Export compiler types
- vulture_whitelist.py: Whitelist new public API

ISSUES CLOSED: #158
2026-02-24 17:57:18 +00:00

747 lines
30 KiB
Python

"""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}'"
)