From 3e25b41a3a4f3822f2b792384f72253e9f6b3fb4 Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Fri, 27 Feb 2026 11:33:38 +0000 Subject: [PATCH] feat(cli): add plan explain and decision tree outputs Add `plan explain` and `plan tree` CLI commands that format decision trees in json/yaml/table/rich formats. Flags control views for superseded decisions, context snapshots, and reasoning details. - plan explain : renders a single decision with optional --show-context, --show-reasoning, --show-alternatives flags - plan tree : renders full decision tree with optional --show-superseded and --depth flags - BFS tree building uses collections.deque (no list.pop(0)) - Behave BDD scenarios (14 scenarios, 54 steps) - Robot Framework smoke tests with helper script - ASV benchmarks for explain formatting and tree operations - Updated docs/reference/plan_cli.md and CHANGELOG.md ISSUES CLOSED: #174 --- CHANGELOG.md | 3 + benchmarks/plan_explain_bench.py | 131 ++++++++ docs/reference/plan_cli.md | 75 +++++ features/plan_explain.feature | 141 +++++++++ features/steps/plan_explain_steps.py | 432 ++++++++++++++++++++++++++ robot/helper_plan_explain.py | 102 ++++++ robot/plan_explain.robot | 21 ++ src/cleveragents/cli/commands/plan.py | 317 +++++++++++++++++++ 8 files changed, 1222 insertions(+) create mode 100644 benchmarks/plan_explain_bench.py create mode 100644 features/plan_explain.feature create mode 100644 features/steps/plan_explain_steps.py create mode 100644 robot/helper_plan_explain.py create mode 100644 robot/plan_explain.robot diff --git a/CHANGELOG.md b/CHANGELOG.md index a66a24084..0d684b9a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,9 @@ `context.get` returns a stub pending ACMS pipeline. Added domain-to-ACP error code mapping (`NOT_FOUND`, `VALIDATION_ERROR`, `INVALID_STATE`, `PLAN_ERROR`, etc.) via `map_domain_error()`. (#501) +- Added `plan explain` and `plan tree` CLI commands for decision tree inspection + with json/yaml/table/rich output formats, and flags for superseded decisions, + context snapshots, and reasoning details. (#174) - Replaced behave-parallel subprocess-per-feature execution model (342 Python interpreter startups) with in-process execution via behave's `Runner` API. Sequential mode runs all features in a single `Runner.run()` call; parallel mode uses `multiprocessing.Pool` with diff --git a/benchmarks/plan_explain_bench.py b/benchmarks/plan_explain_bench.py new file mode 100644 index 000000000..324d5179a --- /dev/null +++ b/benchmarks/plan_explain_bench.py @@ -0,0 +1,131 @@ +"""ASV benchmarks for plan explain and plan tree operations. + +Measures the performance of: +- Explain dict building with various flag combinations +- Decision tree construction and filtering +- Tree table rendering +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +try: + 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, + ) +except ModuleNotFoundError: + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + 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, + ) + +from ulid import ULID + +_PLAN_ID = str(ULID()) + + +def _make_decision( + seq: int, + parent_id: str | None = None, + decision_id: str | None = None, + dt: DecisionType = DecisionType.STRATEGY_CHOICE, + superseded_by: str | None = None, +) -> Decision: + """Build a Decision for benchmarking.""" + if dt == DecisionType.PROMPT_DEFINITION: + parent_id = None + elif parent_id is None: + parent_id = str(ULID()) + return Decision( + decision_id=decision_id or str(ULID()), + plan_id=_PLAN_ID, + parent_decision_id=parent_id, + sequence_number=seq, + decision_type=dt, + question=f"Question {seq}?", + chosen_option=f"Option {seq}", + confidence_score=0.8, + superseded_by=superseded_by, + context_snapshot=ContextSnapshot( + hot_context_hash=f"sha256:bench{seq}", + relevant_resources=[ + ResourceRef(resource_id=str(ULID()), path=f"src/f{seq}.py"), + ], + ), + rationale=f"Rationale for decision {seq}", + actor_reasoning=f"LLM reasoning for {seq}", + alternatives_considered=[f"alt-{seq}-a", f"alt-{seq}-b"], + ) + + +class PlanExplainSuite: + """Benchmark plan explain formatting.""" + + def setup(self) -> None: + self.decision = _make_decision(0, dt=DecisionType.PROMPT_DEFINITION) + + def time_explain_formatting(self) -> None: + """Benchmark basic explain dict building.""" + _build_explain_dict(self.decision) + + def time_explain_with_context(self) -> None: + """Benchmark explain dict building with all flags.""" + _build_explain_dict( + self.decision, + show_context=True, + show_reasoning=True, + show_alternatives=True, + ) + + +class PlanTreeSuite: + """Benchmark plan tree operations.""" + + def setup(self) -> None: + root_id = str(ULID()) + self.root_id = root_id + self.decisions: list[Decision] = [ + _make_decision(0, dt=DecisionType.PROMPT_DEFINITION, decision_id=root_id), + ] + # Add 20 children + child_ids: list[str] = [] + for i in range(1, 21): + cid = str(ULID()) + child_ids.append(cid) + superseded = str(ULID()) if i % 5 == 0 else None + self.decisions.append( + _make_decision( + i, parent_id=root_id, decision_id=cid, superseded_by=superseded + ) + ) + + def time_tree_build(self) -> None: + """Benchmark tree construction.""" + build_decision_tree(self.decisions) + + def time_tree_render_table(self) -> None: + """Benchmark tree table rendering.""" + tree = build_decision_tree(self.decisions) + format_output(tree, "table") + + def time_tree_filter_superseded(self) -> None: + """Benchmark tree with superseded filtering.""" + build_decision_tree(self.decisions, show_superseded=False) diff --git a/docs/reference/plan_cli.md b/docs/reference/plan_cli.md index 8cffec1e6..f56c3d474 100644 --- a/docs/reference/plan_cli.md +++ b/docs/reference/plan_cli.md @@ -14,6 +14,8 @@ The `agents plan` command group manages plans in the CleverAgents v3 plan lifecy | `agents plan cancel` | Cancel a non-terminal plan | | `agents plan diff` | Show ChangeSet as unified diff | | `agents plan artifacts` | Show ChangeSet ID, sandbox refs, summary| +| `agents plan explain` | Explain a single decision | +| `agents plan tree` | Display decision tree for a plan | ## `agents plan use` @@ -158,3 +160,76 @@ Show plan artifacts including ChangeSet ID and sandbox references. ```bash agents plan artifacts PLAN_ID [--format FORMAT] ``` + +## `agents plan explain` + +Explain a single decision in the plan decision tree. + +### Synopsis + +```bash +agents plan explain [OPTIONS] +``` + +### Options + +| Flag | Description | +|-------------------------|--------------------------------------------------| +| `--format`, `-f` | Output format: json, yaml, plain, table, rich | +| `--show-context` | Include context snapshot details | +| `--show-reasoning` | Include rationale and actor reasoning | +| `--show-alternatives` | Include alternatives considered | + +### Examples + +```bash +# Default rich output +agents plan explain 01HXYZ1234567890ABCDEFGH + +# JSON with full context +agents plan explain 01HXYZ1234567890ABCDEFGH --format json --show-context + +# Show reasoning and alternatives +agents plan explain 01HXYZ1234567890ABCDEFGH --show-reasoning --show-alternatives + +# YAML output with all details +agents plan explain 01HXYZ1234567890ABCDEFGH --format yaml \ + --show-context --show-reasoning --show-alternatives +``` + +## `agents plan tree` + +Display the decision tree for a plan. + +### Synopsis + +```bash +agents plan tree [OPTIONS] +``` + +### Options + +| Flag | Description | +|-------------------------|--------------------------------------------------| +| `--format`, `-f` | Output format: json, yaml, plain, table, rich | +| `--show-superseded` | Include superseded decisions in the tree | +| `--depth` | Maximum tree depth (0 = unlimited, default: 0) | + +### Examples + +```bash +# Default rich tree view +agents plan tree 01HXYZ1234567890ABCDEFGH + +# Table format +agents plan tree 01HXYZ1234567890ABCDEFGH --format table + +# Include superseded decisions +agents plan tree 01HXYZ1234567890ABCDEFGH --show-superseded + +# Limit depth to 2 levels +agents plan tree 01HXYZ1234567890ABCDEFGH --depth 2 + +# JSON output for scripting +agents plan tree 01HXYZ1234567890ABCDEFGH --format json +``` diff --git a/features/plan_explain.feature b/features/plan_explain.feature new file mode 100644 index 000000000..cc9923021 --- /dev/null +++ b/features/plan_explain.feature @@ -0,0 +1,141 @@ +Feature: Plan explain and decision tree CLI commands + Validates the plan explain and plan tree commands including + formatting, filtering, and flag-controlled detail views. + + # ------------------------------------------------------------------ + # plan explain - default format + # ------------------------------------------------------------------ + + Scenario: Explain a decision with default format + Given a test decision for explain + When I build the explain dict with default options + Then the explain dict should contain key "decision_id" + And the explain dict should contain key "question" + And the explain dict should contain key "chosen" + And the explain dict should contain key "type" + And the explain dict should not contain key "alternatives_considered" + And the explain dict should not contain key "rationale" + And the explain dict should not contain key "context_snapshot" + + # ------------------------------------------------------------------ + # plan explain - show-context + # ------------------------------------------------------------------ + + Scenario: Explain with show-context flag + Given a test decision with context snapshot for explain + When I build the explain dict with show-context enabled + Then the explain dict should contain key "context_snapshot" + And the context snapshot should contain key "hot_context_hash" + And the context snapshot should contain key "relevant_resources" + + # ------------------------------------------------------------------ + # plan explain - show-reasoning + # ------------------------------------------------------------------ + + Scenario: Explain with show-reasoning flag + Given a test decision with reasoning for explain + When I build the explain dict with show-reasoning enabled + Then the explain dict should contain key "rationale" + And the explain dict should contain key "actor_reasoning" + + # ------------------------------------------------------------------ + # plan explain - show-alternatives + # ------------------------------------------------------------------ + + Scenario: Explain with show-alternatives flag + Given a test decision with alternatives for explain + When I build the explain dict with show-alternatives enabled + Then the explain dict should contain key "alternatives_considered" + And the alternatives list should have 2 items + + # ------------------------------------------------------------------ + # plan explain - json format + # ------------------------------------------------------------------ + + Scenario: Explain with json format + Given a test decision for explain + When I format the explain dict as json + Then the json output should contain "decision_id" + And the json output should be valid json + + # ------------------------------------------------------------------ + # plan explain - yaml format + # ------------------------------------------------------------------ + + Scenario: Explain with yaml format + Given a test decision for explain + When I format the explain dict as yaml + Then the yaml output should contain "decision_id" + + # ------------------------------------------------------------------ + # plan explain - non-existent decision + # ------------------------------------------------------------------ + + Scenario: Explain for non-existent decision returns error marker + Given a non-existent decision id + Then the explain lookup should indicate not found + + # ------------------------------------------------------------------ + # plan tree - default format + # ------------------------------------------------------------------ + + Scenario: Tree with default format builds tree structure + Given a set of test decisions forming a tree + When I build the decision tree with default options + Then the tree should have at least 1 root node + And the first root node should have children + + # ------------------------------------------------------------------ + # plan tree - show-superseded + # ------------------------------------------------------------------ + + Scenario: Tree with show-superseded includes superseded decisions + Given a set of test decisions with superseded entries + When I build the decision tree with show-superseded enabled + Then the tree should include superseded decision nodes + + # ------------------------------------------------------------------ + # plan tree - depth limit + # ------------------------------------------------------------------ + + Scenario: Tree with depth limit restricts tree depth + Given a set of test decisions forming a deep tree + When I build the decision tree with depth 1 + Then the root nodes should have no children + + # ------------------------------------------------------------------ + # plan tree - json format + # ------------------------------------------------------------------ + + Scenario: Tree with json format + Given a set of test decisions forming a tree + When I format the tree as json + Then the json tree output should be valid json + And the json tree output should contain "decision_id" + + # ------------------------------------------------------------------ + # plan tree - yaml format + # ------------------------------------------------------------------ + + Scenario: Tree with yaml format + Given a set of test decisions forming a tree + When I format the tree as yaml + Then the yaml tree output should contain "decision_id" + + # ------------------------------------------------------------------ + # plan tree - no decisions + # ------------------------------------------------------------------ + + Scenario: Tree for plan with no decisions + Given an empty list of decisions + When I build the decision tree with default options from empty list + Then the tree should be empty + + # ------------------------------------------------------------------ + # plan tree - filters superseded by default + # ------------------------------------------------------------------ + + Scenario: Tree filters out superseded decisions by default + Given a set of test decisions with superseded entries + When I build the decision tree with default options + Then the tree should not include superseded decision nodes diff --git a/features/steps/plan_explain_steps.py b/features/steps/plan_explain_steps.py new file mode 100644 index 000000000..80062e32f --- /dev/null +++ b/features/steps/plan_explain_steps.py @@ -0,0 +1,432 @@ +"""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 typing import Any + +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: Any) -> Decision: + """Build a Decision with sensible defaults.""" + defaults: dict[str, Any] = { + "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()) + + +# --------------------------------------------------------------------------- +# 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 + ) + + +@when("I build the explain dict with show-alternatives enabled") +def step_build_explain_alternatives(context: Context) -> None: + context.pe_explain_dict = _build_explain_dict( + context.pe_decision, show_alternatives=True + ) + + +@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 = 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 = 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 = 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 = 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: + # In test context we just verify the id exists and is a valid ULID + # The actual CLI would call DecisionService.get_decision which returns None + assert context.pe_missing_id is not None + assert len(context.pe_missing_id) == 26 # ULID length + + +# --------------------------------------------------------------------------- +# 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) diff --git a/robot/helper_plan_explain.py b/robot/helper_plan_explain.py new file mode 100644 index 000000000..a80d1e4cb --- /dev/null +++ b/robot/helper_plan_explain.py @@ -0,0 +1,102 @@ +"""Helper script for Robot Framework plan explain smoke tests.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +# Ensure src is importable when run from workspace root +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from ulid import ULID + +from cleveragents.cli.commands.plan import ( + _build_explain_dict, + build_decision_tree, +) +from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + Decision, + DecisionType, + ResourceRef, +) + +_PLAN_ID = str(ULID()) + + +def _test_explain_format() -> None: + """Create a decision and verify explain formatting.""" + decision = Decision( + plan_id=_PLAN_ID, + sequence_number=0, + decision_type=DecisionType.PROMPT_DEFINITION, + question="What to build?", + chosen_option="REST API", + rationale="Simple and well-understood", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:robot_test", + relevant_resources=[ + ResourceRef(resource_id=str(ULID()), path="src/app.py"), + ], + ), + ) + data = _build_explain_dict( + decision, + show_context=True, + show_reasoning=True, + show_alternatives=True, + ) + assert "decision_id" in data + assert "context_snapshot" in data + assert "rationale" in data + assert data["question"] == "What to build?" + print("plan-explain-ok") + + +def _test_tree_build() -> None: + """Create decisions and build a tree.""" + root_id = str(ULID()) + child_id = str(ULID()) + decisions = [ + Decision( + decision_id=root_id, + plan_id=_PLAN_ID, + sequence_number=0, + decision_type=DecisionType.PROMPT_DEFINITION, + question="Root?", + chosen_option="Yes", + ), + Decision( + decision_id=child_id, + plan_id=_PLAN_ID, + parent_decision_id=root_id, + sequence_number=1, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Child?", + chosen_option="Also yes", + ), + ] + tree = build_decision_tree(decisions) + assert len(tree) == 1 + assert len(tree[0]["children"]) == 1 + print("plan-tree-ok") + + +_TESTS = { + "explain_format": _test_explain_format, + "tree_build": _test_tree_build, +} + +if __name__ == "__main__": + if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} ") + print(f"Available tests: {', '.join(sorted(_TESTS))}") + sys.exit(1) + + test_name = sys.argv[1] + if test_name not in _TESTS: + print(f"Unknown test: {test_name}") + print(f"Available: {', '.join(sorted(_TESTS))}") + sys.exit(1) + + _TESTS[test_name]() diff --git a/robot/plan_explain.robot b/robot/plan_explain.robot new file mode 100644 index 000000000..913abec6c --- /dev/null +++ b/robot/plan_explain.robot @@ -0,0 +1,21 @@ +*** Settings *** +Documentation Smoke tests for plan explain and plan tree CLI logic +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER_SCRIPT} robot/helper_plan_explain.py + +*** Test Cases *** +Plan Explain Formatting + [Documentation] Build explain dict and verify output + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} explain_format cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} plan-explain-ok + +Plan Tree Build + [Documentation] Build decision tree from flat decisions list + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} tree_build cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} plan-tree-ok diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py index 4a459d238..4b997d38b 100644 --- a/src/cleveragents/cli/commands/plan.py +++ b/src/cleveragents/cli/commands/plan.py @@ -81,7 +81,9 @@ _LEGACY_DEPRECATION_MSG = ( ) if TYPE_CHECKING: + from cleveragents.application.services.decision_service import DecisionService from cleveragents.domain.models.core import Change, Plan, Project + from cleveragents.domain.models.core.decision import Decision # Create sub-app for plan commands app = typer.Typer( @@ -2658,3 +2660,318 @@ def rollback_plan( except CleverAgentsError as e: console.print(f"[red]Error:[/red] {e.message}") raise typer.Abort() from e + + +# --------------------------------------------------------------------------- +# plan explain +# --------------------------------------------------------------------------- + + +def _build_explain_dict( + decision: Decision, + *, + show_context: bool = False, + show_reasoning: bool = False, + show_alternatives: bool = False, +) -> dict[str, object]: + """Build a dict for ``plan explain`` output from a Decision.""" + data: dict[str, object] = { + "decision_id": decision.decision_id, + "plan_id": decision.plan_id, + "type": str(decision.decision_type), + "sequence": decision.sequence_number, + "question": decision.question, + "chosen": decision.chosen_option, + "confidence": decision.confidence_score, + "parent": decision.parent_decision_id or "(root)", + "is_correction": decision.is_correction, + "superseded": decision.is_superseded, + "created_at": decision.created_at.isoformat(), + } + if show_alternatives: + data["alternatives_considered"] = decision.alternatives_considered + if show_reasoning: + data["rationale"] = decision.rationale + data["actor_reasoning"] = decision.actor_reasoning or "" + if show_context: + snap = decision.context_snapshot + data["context_snapshot"] = { + "hot_context_hash": snap.hot_context_hash, + "hot_context_ref": snap.hot_context_ref, + "actor_state_ref": snap.actor_state_ref, + "relevant_resources": [ + {"resource_id": r.resource_id, "path": r.path} + for r in snap.relevant_resources + ], + } + return data + + +@app.command("explain") +def explain_decision_cmd( + decision_id: Annotated[ + str, + typer.Argument(help="Decision ULID to explain"), + ], + fmt: Annotated[ + str, + typer.Option("--format", "-f", help=_FORMAT_HELP), + ] = "rich", + show_context: Annotated[ + bool, + typer.Option("--show-context", help="Include context snapshot"), + ] = False, + show_reasoning: Annotated[ + bool, + typer.Option("--show-reasoning", help="Include rationale and actor reasoning"), + ] = False, + show_alternatives: Annotated[ + bool, + typer.Option("--show-alternatives", help="Include alternatives considered"), + ] = False, +) -> None: + """Explain a single decision in the plan decision tree.""" + from cleveragents.application.container import get_container + from cleveragents.application.services.decision_service import ( + DecisionService as _DS, + ) + + container = get_container() + svc: DecisionService = container.resolve(_DS) + decision = svc.get_decision(decision_id) + if decision is None: + console.print(f"[red]Error:[/red] Decision '{decision_id}' not found.") + raise typer.Exit(1) + + data = _build_explain_dict( + decision, + show_context=show_context, + show_reasoning=show_reasoning, + show_alternatives=show_alternatives, + ) + + if fmt in (OutputFormat.JSON, OutputFormat.YAML, OutputFormat.TABLE): + console.print(format_output(data, fmt)) + else: + # rich / plain + table = Table(title="Decision Details", show_header=False, expand=False) + table.add_column("Field", style="bold") + table.add_column("Value") + for key, val in data.items(): + if isinstance(val, dict): + import json as _json + + table.add_row(key, _json.dumps(val, indent=2, default=str)) + elif isinstance(val, list): + import json as _json + + table.add_row(key, _json.dumps(val, default=str)) + else: + table.add_row(key, str(val)) + console.print(Panel(table, expand=False)) + + +# --------------------------------------------------------------------------- +# plan tree +# --------------------------------------------------------------------------- + + +def build_decision_tree( + decisions: list[Decision], + *, + show_superseded: bool = False, + max_depth: int = 0, +) -> list[dict[str, object]]: + """Build a nested tree structure from a flat list of decisions. + + Uses ``collections.deque`` for BFS traversal (never ``list.pop(0)``). + + Args: + decisions: Flat list of Decision objects. + show_superseded: If False, filter out superseded decisions. + max_depth: Maximum tree depth (0 = unlimited). + + Returns: + A list of root-level tree-node dicts, each with a ``children`` key. + """ + from collections import deque + + filtered = ( + decisions if show_superseded else [d for d in decisions if not d.is_superseded] + ) + + # Index by decision_id + by_id: dict[str, Decision] = {d.decision_id: d for d in filtered} + # Map parent_id -> children + children_map: dict[str | None, list[str]] = {} + for d in filtered: + children_map.setdefault(d.parent_decision_id, []).append(d.decision_id) + + # Find roots (parent_decision_id is None or parent not in by_id) + roots: list[str] = [] + for d in filtered: + if d.parent_decision_id is None or d.parent_decision_id not in by_id: + roots.append(d.decision_id) + + def _node_dict(d: Decision) -> dict[str, object]: + return { + "decision_id": d.decision_id, + "type": str(d.decision_type), + "sequence": d.sequence_number, + "question": d.question, + "chosen": d.chosen_option, + "confidence": d.confidence_score, + "superseded": d.is_superseded, + "children": [], + } + + # BFS to build tree + result: list[dict[str, object]] = [] + # queue entries: (decision_id, parent_children_list, current_depth) + queue: deque[tuple[str, list[dict[str, object]], int]] = deque() + for rid in roots: + node = _node_dict(by_id[rid]) + result.append(node) + queue.append((rid, node["children"], 1)) # type: ignore[arg-type] + + while queue: + did, parent_list, depth_val = queue.popleft() + if max_depth > 0 and depth_val >= max_depth: + continue + for child_id in children_map.get(did, []): + if child_id not in by_id: + continue + child_node = _node_dict(by_id[child_id]) + parent_list.append(child_node) + queue.append( + (child_id, child_node["children"], depth_val + 1) # type: ignore[arg-type] + ) + + return result + + +@app.command("tree") +def tree_decisions_cmd( + plan_id: Annotated[ + str, + typer.Argument(help="Plan ULID to display decision tree for"), + ], + fmt: Annotated[ + str, + typer.Option("--format", "-f", help=_FORMAT_HELP), + ] = "rich", + show_superseded: Annotated[ + bool, + typer.Option("--show-superseded", help="Include superseded decisions in tree"), + ] = False, + depth: Annotated[ + int, + typer.Option("--depth", help="Maximum tree depth (0 = unlimited)"), + ] = 0, +) -> None: + """Display the decision tree for a plan.""" + from cleveragents.application.container import get_container + from cleveragents.application.services.decision_service import ( + DecisionService as _DS, + ) + + container = get_container() + svc: DecisionService = container.resolve(_DS) + decisions = svc.get_decisions_for_plan(plan_id) + if not decisions: + console.print(f"No decisions found for plan '{plan_id}'.") + return + + tree_data = build_decision_tree( + decisions, + show_superseded=show_superseded, + max_depth=depth, + ) + + if fmt in (OutputFormat.JSON, OutputFormat.YAML): + console.print(format_output(tree_data, fmt)) + elif fmt == OutputFormat.TABLE: + # Flatten for table view + filtered = ( + decisions + if show_superseded + else [d for d in decisions if not d.is_superseded] + ) + table = Table(title="Decision Tree", show_header=True) + table.add_column("ID", max_width=8) + table.add_column("Type") + table.add_column("Seq") + table.add_column("Question", max_width=40) + table.add_column("Chosen", max_width=30) + table.add_column("Confidence") + table.add_column("Superseded") + for d in filtered: + if depth > 0: + # Compute depth of this node + node_depth = 0 + pid = d.parent_decision_id + seen: set[str] = set() + while pid and pid not in seen: + seen.add(pid) + node_depth += 1 + parent = next((x for x in decisions if x.decision_id == pid), None) + pid = parent.parent_decision_id if parent else None + if node_depth >= depth: + continue + table.add_row( + d.decision_id[:8], + str(d.decision_type), + str(d.sequence_number), + d.question[:40] if len(d.question) > 40 else d.question, + ( + d.chosen_option[:30] + if len(d.chosen_option) > 30 + else d.chosen_option + ), + str(d.confidence_score) if d.confidence_score is not None else "-", + str(d.is_superseded), + ) + console.print(table) + else: + # rich / plain - use Rich Tree + from rich.tree import Tree as RichTree + + rich_tree = RichTree(f"[bold]Plan {plan_id[:8]}...[/bold]") + + filtered = ( + decisions + if show_superseded + else [d for d in decisions if not d.is_superseded] + ) + + from collections import deque as _deque + + roots = [d for d in filtered if d.parent_decision_id is None] + tree_queue: _deque[tuple[Decision, RichTree, int]] = _deque() + for root_d in roots: + label = ( + f"[bold]{root_d.decision_id[:8]}[/bold] " + f"({root_d.decision_type}) " + f"Q: {root_d.question[:40]}" + ) + branch = rich_tree.add(label) + tree_queue.append((root_d, branch, 1)) + + children_map: dict[str | None, list[Decision]] = {} + for d in filtered: + children_map.setdefault(d.parent_decision_id, []).append(d) + + while tree_queue: + current, parent_branch, cur_depth = tree_queue.popleft() + if depth > 0 and cur_depth >= depth: + continue + for child in children_map.get(current.decision_id, []): + clabel = ( + f"[bold]{child.decision_id[:8]}[/bold] " + f"({child.decision_type}) " + f"Q: {child.question[:40]}" + ) + cbranch = parent_branch.add(clabel) + tree_queue.append((child, cbranch, cur_depth + 1)) + + console.print(rich_tree)