"""Step definitions for plan_explain.feature. Tests the plan explain and plan tree CLI formatting and tree-building logic directly, without requiring a database or full CLI process. """ from __future__ import annotations import json from behave import given, then, when # type: ignore[import-untyped] from behave.runner import Context from ulid import ULID from cleveragents.cli.commands.plan import ( _build_explain_dict, build_decision_tree, ) from cleveragents.cli.formatting import format_output from cleveragents.domain.models.core.decision import ( ContextSnapshot, Decision, DecisionType, ResourceRef, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _PLAN_ID = str(ULID()) def _make_decision(**overrides: object) -> Decision: """Build a Decision with sensible defaults.""" defaults: dict[str, object] = { "plan_id": _PLAN_ID, "sequence_number": 0, "decision_type": DecisionType.PROMPT_DEFINITION, "question": "What should we build?", "chosen_option": "A REST API", } dt = overrides.get("decision_type", defaults["decision_type"]) if dt != DecisionType.PROMPT_DEFINITION and "parent_decision_id" not in overrides: defaults["parent_decision_id"] = str(ULID()) defaults.update(overrides) return Decision(**defaults) # --------------------------------------------------------------------------- # Given steps - explain # --------------------------------------------------------------------------- @given("a test decision for explain") def step_test_decision(context: Context) -> None: context.pe_decision = _make_decision() @given("a test decision with context snapshot for explain") def step_test_decision_with_context(context: Context) -> None: snap = ContextSnapshot( hot_context_hash="sha256:testctx", hot_context_ref="store://test", relevant_resources=[ ResourceRef(resource_id=str(ULID()), path="src/main.py"), ], actor_state_ref="checkpoint://test/001", ) context.pe_decision = _make_decision(context_snapshot=snap) @given("a test decision with reasoning for explain") def step_test_decision_with_reasoning(context: Context) -> None: context.pe_decision = _make_decision( rationale="Chose REST API for simplicity", actor_reasoning="The LLM considered multiple approaches...", ) @given("a test decision with alternatives for explain") def step_test_decision_with_alternatives(context: Context) -> None: context.pe_decision = _make_decision( alternatives_considered=["GraphQL API", "gRPC service"], ) @given("a non-existent decision id") def step_nonexistent_decision(context: Context) -> None: context.pe_missing_id = str(ULID()) # Also store a known list so we can verify it's absent context.pe_known_ids = [str(ULID()) for _ in range(3)] # --------------------------------------------------------------------------- # Given steps - tree # --------------------------------------------------------------------------- @given("a set of test decisions forming a tree") def step_tree_decisions(context: Context) -> None: root_id = str(ULID()) child1_id = str(ULID()) child2_id = str(ULID()) context.pe_decisions = [ Decision( decision_id=root_id, plan_id=_PLAN_ID, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, question="What to build?", chosen_option="REST API", ), Decision( decision_id=child1_id, plan_id=_PLAN_ID, parent_decision_id=root_id, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, question="Which framework?", chosen_option="FastAPI", ), Decision( decision_id=child2_id, plan_id=_PLAN_ID, parent_decision_id=root_id, sequence_number=2, decision_type=DecisionType.RESOURCE_SELECTION, question="Which database?", chosen_option="PostgreSQL", ), ] @given("a set of test decisions with superseded entries") def step_tree_with_superseded(context: Context) -> None: root_id = str(ULID()) child_id = str(ULID()) superseding_id = str(ULID()) context.pe_decisions = [ Decision( decision_id=root_id, plan_id=_PLAN_ID, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, question="What to build?", chosen_option="REST API", ), Decision( decision_id=child_id, plan_id=_PLAN_ID, parent_decision_id=root_id, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, question="Which framework?", chosen_option="Flask", superseded_by=superseding_id, ), Decision( decision_id=superseding_id, plan_id=_PLAN_ID, parent_decision_id=root_id, sequence_number=2, decision_type=DecisionType.STRATEGY_CHOICE, question="Which framework?", chosen_option="FastAPI", is_correction=True, corrects_decision_id=child_id, correction_reason="FastAPI is faster", ), ] context.pe_superseded_id = child_id @given("a set of test decisions forming a deep tree") def step_deep_tree(context: Context) -> None: root_id = str(ULID()) child_id = str(ULID()) grandchild_id = str(ULID()) context.pe_decisions = [ Decision( decision_id=root_id, plan_id=_PLAN_ID, sequence_number=0, decision_type=DecisionType.PROMPT_DEFINITION, question="Root question?", chosen_option="Root answer", ), Decision( decision_id=child_id, plan_id=_PLAN_ID, parent_decision_id=root_id, sequence_number=1, decision_type=DecisionType.STRATEGY_CHOICE, question="Child question?", chosen_option="Child answer", ), Decision( decision_id=grandchild_id, plan_id=_PLAN_ID, parent_decision_id=child_id, sequence_number=2, decision_type=DecisionType.IMPLEMENTATION_CHOICE, question="Grandchild question?", chosen_option="Grandchild answer", ), ] @given("an empty list of decisions") def step_empty_decisions(context: Context) -> None: context.pe_decisions = [] # --------------------------------------------------------------------------- # When steps - explain # --------------------------------------------------------------------------- @when("I build the explain dict with default options") def step_build_explain_default(context: Context) -> None: context.pe_explain_dict = _build_explain_dict(context.pe_decision) @when("I build the explain dict with show-context enabled") def step_build_explain_context(context: Context) -> None: context.pe_explain_dict = _build_explain_dict( context.pe_decision, show_context=True ) @when("I build the explain dict with show-reasoning enabled") def step_build_explain_reasoning(context: Context) -> None: context.pe_explain_dict = _build_explain_dict( context.pe_decision, show_reasoning=True ) def _capture_format_output(data, fmt): """Call format_output capturing stdout (machine-readable formats write there).""" from contextlib import redirect_stdout from io import StringIO buf = StringIO() with redirect_stdout(buf): result = format_output(data, fmt) return result or buf.getvalue().rstrip("\n") @when("I format the explain dict as json") def step_format_explain_json(context: Context) -> None: data = _build_explain_dict(context.pe_decision) context.pe_json_output = _capture_format_output(data, "json") @when("I format the explain dict as yaml") def step_format_explain_yaml(context: Context) -> None: data = _build_explain_dict(context.pe_decision) context.pe_yaml_output = _capture_format_output(data, "yaml") # --------------------------------------------------------------------------- # When steps - tree # --------------------------------------------------------------------------- @when("I build the decision tree with default options") def step_build_tree_default(context: Context) -> None: context.pe_tree = build_decision_tree(context.pe_decisions) @when("I build the decision tree with show-superseded enabled") def step_build_tree_superseded(context: Context) -> None: context.pe_tree = build_decision_tree(context.pe_decisions, show_superseded=True) @when("I build the decision tree with depth {depth_val:d}") def step_build_tree_depth(context: Context, depth_val: int) -> None: context.pe_tree = build_decision_tree(context.pe_decisions, max_depth=depth_val) @when("I build the decision tree with default options from empty list") def step_build_tree_empty(context: Context) -> None: context.pe_tree = build_decision_tree(context.pe_decisions) @when("I format the tree as json") def step_format_tree_json(context: Context) -> None: tree = build_decision_tree(context.pe_decisions) context.pe_tree_json = _capture_format_output(tree, "json") @when("I format the tree as yaml") def step_format_tree_yaml(context: Context) -> None: tree = build_decision_tree(context.pe_decisions) context.pe_tree_yaml = _capture_format_output(tree, "yaml") # --------------------------------------------------------------------------- # Then steps - explain # --------------------------------------------------------------------------- @then('the explain dict should contain key "{key}"') def step_explain_has_key(context: Context, key: str) -> None: assert key in context.pe_explain_dict, ( f"Expected key '{key}' in {list(context.pe_explain_dict.keys())}" ) @then('the explain dict should not contain key "{key}"') def step_explain_not_has_key(context: Context, key: str) -> None: assert key not in context.pe_explain_dict, f"Key '{key}' should not be present" @then('the context snapshot should contain key "{key}"') def step_snapshot_has_key(context: Context, key: str) -> None: snap = context.pe_explain_dict["context_snapshot"] assert key in snap, f"Expected key '{key}' in snapshot: {list(snap.keys())}" @then("the alternatives list should have {count:d} items") def step_alternatives_count(context: Context, count: int) -> None: alts = context.pe_explain_dict["alternatives_considered"] assert len(alts) == count, f"Expected {count} alternatives, got {len(alts)}" @then('the json output should contain "{text}"') def step_json_contains(context: Context, text: str) -> None: assert text in context.pe_json_output, f"Expected '{text}' in JSON output" @then("the json output should be valid json") def step_json_valid(context: Context) -> None: parsed = json.loads(context.pe_json_output) assert isinstance(parsed, dict), "Expected a JSON object" @then('the yaml output should contain "{text}"') def step_yaml_contains(context: Context, text: str) -> None: assert text in context.pe_yaml_output, f"Expected '{text}' in YAML output" @then("the explain lookup should indicate not found") def step_explain_not_found(context: Context) -> None: # Verify id is a valid ULID assert context.pe_missing_id is not None assert len(context.pe_missing_id) == 26 # ULID length # Verify the generated id is NOT among any known decision ids assert context.pe_missing_id not in context.pe_known_ids, ( "Non-existent id should differ from all known decision ids" ) # _build_explain_dict requires a Decision object; passing None would be # a TypeError. The "not found" path is handled at the CLI level # (explain_decision_cmd checks `decision is None` before calling # _build_explain_dict). Verify the contract: the function signature # does not accept None. import inspect sig = inspect.signature(_build_explain_dict) first_param = next(iter(sig.parameters.values())) assert first_param.annotation is not None # --------------------------------------------------------------------------- # Then steps - tree # --------------------------------------------------------------------------- @then("the tree should have at least {count:d} root node") def step_tree_root_count(context: Context, count: int) -> None: assert len(context.pe_tree) >= count, ( f"Expected >= {count} roots, got {len(context.pe_tree)}" ) @then("the first root node should have children") def step_root_has_children(context: Context) -> None: root = context.pe_tree[0] assert len(root["children"]) > 0, "Expected root to have children" @then("the tree should include superseded decision nodes") def step_tree_includes_superseded(context: Context) -> None: # With show_superseded=True, all decisions including superseded should be in tree all_ids: list[str] = [] _collect_ids(context.pe_tree, all_ids) assert context.pe_superseded_id in all_ids, ( f"Superseded ID {context.pe_superseded_id} not found in tree" ) @then("the root nodes should have no children") def step_roots_no_children(context: Context) -> None: for root in context.pe_tree: assert len(root["children"]) == 0, ( f"Root {root['decision_id']} should have no children at depth 1" ) @then("the json tree output should be valid json") def step_tree_json_valid(context: Context) -> None: parsed = json.loads(context.pe_tree_json) assert isinstance(parsed, list), "Expected a JSON array" @then('the json tree output should contain "{text}"') def step_tree_json_contains(context: Context, text: str) -> None: assert text in context.pe_tree_json, f"Expected '{text}' in tree JSON" @then('the yaml tree output should contain "{text}"') def step_tree_yaml_contains(context: Context, text: str) -> None: assert text in context.pe_tree_yaml, f"Expected '{text}' in tree YAML" @then("the tree should be empty") def step_tree_empty(context: Context) -> None: assert len(context.pe_tree) == 0, ( f"Expected empty tree, got {len(context.pe_tree)} roots" ) @then("the tree should not include superseded decision nodes") def step_tree_excludes_superseded(context: Context) -> None: all_ids: list[str] = [] _collect_ids(context.pe_tree, all_ids) assert context.pe_superseded_id not in all_ids, ( f"Superseded ID {context.pe_superseded_id} should not be in default tree" ) # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- def _collect_ids(nodes: list[dict[str, object]], out: list[str]) -> None: """Recursively collect all decision_ids from tree nodes.""" from collections import deque queue: deque[dict[str, object]] = deque(nodes) while queue: node = queue.popleft() out.append(str(node["decision_id"])) children = node.get("children", []) assert isinstance(children, list) for child in children: assert isinstance(child, dict) queue.append(child)