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 <decision_id>: renders a single decision with optional
  --show-context, --show-reasoning, --show-alternatives flags
- plan tree <plan_id>: 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
This commit is contained in:
khyari hamza
2026-02-27 11:33:38 +00:00
parent 4b9df961f0
commit 3e25b41a3a
8 changed files with 1222 additions and 0 deletions
+3
View File
@@ -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
+131
View File
@@ -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)
+75
View File
@@ -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 <DECISION_ID> [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 <PLAN_ID> [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
```
+141
View File
@@ -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
+432
View File
@@ -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)
+102
View File
@@ -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]} <test_name>")
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]()
+21
View File
@@ -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
+317
View File
@@ -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)