diff --git a/features/plan_tree_decision_rendering.feature b/features/plan_tree_decision_rendering.feature new file mode 100644 index 000000000..0a0c7ead0 --- /dev/null +++ b/features/plan_tree_decision_rendering.feature @@ -0,0 +1,79 @@ +Feature: Plan Tree Decision Rendering + As a user + I want to view the decision tree of a plan + So that I can audit the decision history and identify correction points + + Background: + Given a plan with ID "plan-test-001" + And the plan has the following decisions: + | decision_id | type | parent_id | question | chosen_option | status | + | d-001 | prompt_definition | None | What is the task? | Analyze code | completed | + | d-002 | strategy_choice | d-001 | Which strategy to use? | Iterative | completed | + | d-003 | tool_selection | d-002 | Which tool for linting? | ruff | completed | + | d-004 | parameter | d-002 | Set max retries? | 3 | reverted | + | d-005 | branch | d-001 | Await confirmation? | pending | pending | + + Scenario: Display plan tree in rich format + When I run "agents plan tree plan-test-001" + Then the output should contain "Plan Decision Tree: plan-test-001" + And the output should contain "d-001" + And the output should contain "prompt_definition" + And the output should contain "completed" + And the output should contain "d-002" + And the output should contain "strategy_choice" + And the output should contain "d-003" + And the output should contain "tool_selection" + And the output should contain "d-004" + And the output should contain "parameter" + And the output should contain "reverted" + And the output should contain "d-005" + And the output should contain "branch" + And the output should contain "pending" + + Scenario: Display plan tree in plain text format + When I run "agents plan tree plan-test-001 --format plain" + Then the output should contain "Plan Decision Tree: plan-test-001" + And the output should contain "`--" or "|--" + And the output should contain "[completed]" + And the output should contain "[reverted]" + And the output should contain "[pending]" + + Scenario: Display plan tree in JSON format + When I run "agents plan tree plan-test-001 --format json" + Then the output should be valid JSON + And the JSON should contain "plan_id": "plan-test-001" + And the JSON should contain "root" + And the JSON root should have "decision_id": "d-001" + And the JSON root should have "children" array with 2 items + + Scenario: Limit tree depth to 1 + When I run "agents plan tree plan-test-001 --depth 1" + Then the output should contain "d-001" + And the output should contain "d-002" + And the output should NOT contain "d-003" + And the output should NOT contain "d-004" + + Scenario: Limit tree depth to 2 + When I run "agents plan tree plan-test-001 --depth 2" + Then the output should contain "d-001" + And the output should contain "d-002" + And the output should contain "d-003" + And the output should contain "d-004" + And the output should NOT contain "d-005" + + Scenario: Handle non-existent plan ID + When I run "agents plan tree plan-nonexistent" + Then the exit code should be non-zero + And the output should contain "not found" + + Scenario: Handle plan with no decisions + Given a plan with ID "plan-empty" + When I run "agents plan tree plan-empty" + Then the output should contain "not found" or "no decisions" + + Scenario: Tree structure is hierarchical + When I run "agents plan tree plan-test-001 --format plain" + Then the output should show d-002 as a child of d-001 + And the output should show d-003 as a child of d-002 + And the output should show d-004 as a child of d-002 + And the output should show d-005 as a child of d-001 diff --git a/features/steps/plan_tree_decision_rendering_steps.py b/features/steps/plan_tree_decision_rendering_steps.py new file mode 100644 index 000000000..abc4dca9c --- /dev/null +++ b/features/steps/plan_tree_decision_rendering_steps.py @@ -0,0 +1,253 @@ +"""Step definitions for plan tree decision rendering feature.""" + +from __future__ import annotations + +import json +import subprocess +from datetime import datetime + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.domain.models.core.decision import Decision, DecisionType +from cleveragents.domain.models.core.plan import Plan + + +@given('a plan with ID "{plan_id}"') +def step_create_plan(context: Context, plan_id: str) -> None: + """Create a test plan with the given ID.""" + if not hasattr(context, "plans"): + context.plans = {} + + plan = Plan( + plan_id=plan_id, + name=f"Test Plan {plan_id}", + description="Test plan for decision tree rendering", + ) + context.plans[plan_id] = plan + + +@given("the plan has the following decisions") +def step_add_decisions(context: Context) -> None: + """Add decisions to the current plan.""" + if not hasattr(context, "decisions"): + context.decisions = {} + + # Get the last created plan + plan_id = list(context.plans.keys())[-1] + + for row in context.table: + decision_id = row["decision_id"] + decision_type = DecisionType(row["type"]) + parent_id = row["parent_id"] if row["parent_id"] != "None" else None + question = row["question"] + chosen_option = row["chosen_option"] + status = row["status"] + + decision = Decision( + decision_id=decision_id, + plan_id=plan_id, + parent_decision_id=parent_id, + sequence_number=len(context.decisions), + decision_type=decision_type, + question=question, + chosen_option=chosen_option, + rationale=f"Rationale for {chosen_option}", + created_at=datetime.now(datetime.UTC), + ) + + # Add status attribute (not in base model, but needed for rendering) + decision.status = status + + context.decisions[decision_id] = decision + + +@when('I run "agents plan tree {plan_id}"') +def step_run_plan_tree_command(context: Context, plan_id: str) -> None: + """Run the plan tree command.""" + context.last_command = ["agents", "plan", "tree", plan_id] + step_run_command(context) + + +@when('I run "agents plan tree {plan_id} --format {format_type}"') +def step_run_plan_tree_with_format( + context: Context, plan_id: str, format_type: str +) -> None: + """Run the plan tree command with format option.""" + context.last_command = [ + "agents", + "plan", + "tree", + plan_id, + "--format", + format_type, + ] + step_run_command(context) + + +@when('I run "agents plan tree {plan_id} --depth {depth}"') +def step_run_plan_tree_with_depth( + context: Context, plan_id: str, depth: str +) -> None: + """Run the plan tree command with depth option.""" + context.last_command = [ + "agents", + "plan", + "tree", + plan_id, + "--depth", + depth, + ] + step_run_command(context) + + +@when('I run "agents plan tree {plan_id} --format {format_type} --depth {depth}"') +def step_run_plan_tree_with_format_and_depth( + context: Context, plan_id: str, format_type: str, depth: str +) -> None: + """Run the plan tree command with both format and depth options.""" + context.last_command = [ + "agents", + "plan", + "tree", + plan_id, + "--format", + format_type, + "--depth", + depth, + ] + step_run_command(context) + + +def step_run_command(context: Context) -> None: + """Execute the command and capture output.""" + try: + result = subprocess.run( + context.last_command, + capture_output=True, + text=True, + timeout=10, + ) + context.last_exit_code = result.returncode + context.last_output = result.stdout + result.stderr + except subprocess.TimeoutExpired: + context.last_exit_code = 124 + context.last_output = "Command timed out" + except Exception as e: + context.last_exit_code = 1 + context.last_output = str(e) + + +@then("the output should contain {text}") +def step_output_contains(context: Context, text: str) -> None: + """Check that output contains the given text.""" + # Remove quotes if present + text = text.strip('"') + assert text in context.last_output, ( + f"Expected '{text}' in output, but got:\n{context.last_output}" + ) + + +@then("the output should NOT contain {text}") +def step_output_not_contains(context: Context, text: str) -> None: + """Check that output does not contain the given text.""" + # Remove quotes if present + text = text.strip('"') + assert text not in context.last_output, ( + f"Expected '{text}' NOT in output, but got:\n{context.last_output}" + ) + + +@then("the exit code should be non-zero") +def step_exit_code_nonzero(context: Context) -> None: + """Check that exit code is non-zero.""" + assert context.last_exit_code != 0, ( + f"Expected non-zero exit code, but got {context.last_exit_code}" + ) + + +@then("the output should be valid JSON") +def step_output_is_valid_json(context: Context) -> None: + """Check that output is valid JSON.""" + try: + context.last_json = json.loads(context.last_output) + except json.JSONDecodeError as e: + raise AssertionError( + f"Output is not valid JSON: {e}\nOutput: {context.last_output}" + ) from e + + +@then('the JSON should contain "{key}": "{value}"') +def step_json_contains_key_value( + context: Context, key: str, value: str +) -> None: + """Check that JSON contains a specific key-value pair.""" + assert key in context.last_json, ( + f"Expected key '{key}' in JSON, but got: {context.last_json}" + ) + assert context.last_json[key] == value, ( + f"Expected {key}={value}, but got {key}={context.last_json[key]}" + ) + + +@then("the JSON root should have {key}: {value}") +def step_json_root_has_key_value( + context: Context, key: str, value: str +) -> None: + """Check that JSON root has a specific key-value pair.""" + root = context.last_json.get("root", {}) + assert key in root, ( + f"Expected key '{key}' in JSON root, but got: {root}" + ) + assert root[key] == value, ( + f"Expected {key}={value}, but got {key}={root[key]}" + ) + + +@then("the JSON root should have {key} array with {count} items") +def step_json_root_has_array( + context: Context, key: str, count: str +) -> None: + """Check that JSON root has an array with specific count.""" + root = context.last_json.get("root", {}) + assert key in root, ( + f"Expected key '{key}' in JSON root, but got: {root}" + ) + assert isinstance(root[key], list), ( + f"Expected {key} to be an array, but got {type(root[key])}" + ) + expected_count = int(count) + assert len(root[key]) == expected_count, ( + f"Expected {key} to have {expected_count} items, " + f"but got {len(root[key])}" + ) + + +@then("the output should show {child_id} as a child of {parent_id}") +def step_output_shows_hierarchy( + context: Context, child_id: str, parent_id: str +) -> None: + """Check that output shows correct hierarchy.""" + # This is a simplified check - in a real scenario, we'd parse the tree structure + lines = context.last_output.split("\n") + parent_line_idx = None + child_line_idx = None + + for i, line in enumerate(lines): + if parent_id in line: + parent_line_idx = i + if child_id in line: + child_line_idx = i + + assert parent_line_idx is not None, ( + f"Parent {parent_id} not found in output" + ) + assert child_line_idx is not None, ( + f"Child {child_id} not found in output" + ) + assert parent_line_idx < child_line_idx, ( + f"Parent {parent_id} should appear before child {child_id}" + ) + + +__all__ = [] diff --git a/features/test_plan_tree_renderers.py b/features/test_plan_tree_renderers.py new file mode 100644 index 000000000..f4103f435 --- /dev/null +++ b/features/test_plan_tree_renderers.py @@ -0,0 +1,260 @@ +"""Unit tests for plan tree renderers.""" + +from __future__ import annotations + +from datetime import datetime + +import pytest + +from cleveragents.application.services.plan_tree_service import DecisionTreeNode +from cleveragents.cli.output.plan_tree_renderers import ( + JsonPlanTreeRenderer, + PlainPlanTreeRenderer, + RichPlanTreeRenderer, + get_renderer, +) +from cleveragents.domain.models.core.decision import Decision, DecisionType + + +@pytest.fixture +def sample_decision_tree() -> DecisionTreeNode: + """Create a sample decision tree for testing.""" + now = datetime.now(datetime.UTC) + + root_decision = Decision( + decision_id="d-001", + plan_id="plan-001", + parent_decision_id=None, + sequence_number=0, + decision_type=DecisionType.PROMPT_DEFINITION, + question="What is the task?", + chosen_option="Analyze code", + rationale="User requested analysis", + created_at=now, + ) + root_decision.status = "completed" + + child_decision = Decision( + decision_id="d-002", + plan_id="plan-001", + parent_decision_id="d-001", + sequence_number=1, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which strategy?", + chosen_option="Iterative", + rationale="Best approach", + created_at=now, + ) + child_decision.status = "completed" + + root_node = DecisionTreeNode(decision=root_decision, depth=0) + child_node = DecisionTreeNode(decision=child_decision, depth=1) + root_node.children.append(child_node) + + return root_node + + +class TestRichPlanTreeRenderer: + """Tests for RichPlanTreeRenderer.""" + + def test_render_includes_plan_id( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render includes plan ID.""" + renderer = RichPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + assert "Plan Decision Tree: plan-001" in output + + def test_render_includes_decision_ids( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render includes decision IDs.""" + renderer = RichPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + assert "d-001" in output + assert "d-002" in output + + def test_render_includes_decision_types( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render includes decision types.""" + renderer = RichPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + assert "prompt_definition" in output + assert "strategy_choice" in output + + def test_render_includes_status( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render includes status.""" + renderer = RichPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + assert "[completed]" in output + + def test_render_includes_timestamp( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render includes timestamp.""" + renderer = RichPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + # Check for ISO format timestamp + assert "T" in output # ISO format includes T + + def test_render_includes_summary( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render includes summary.""" + renderer = RichPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + assert "User requested analysis" in output or "Analyze code" in output + + +class TestPlainPlanTreeRenderer: + """Tests for PlainPlanTreeRenderer.""" + + def test_render_includes_plan_id( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render includes plan ID.""" + renderer = PlainPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + assert "Plan Decision Tree: plan-001" in output + + def test_render_uses_ascii_connectors( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render uses ASCII connectors.""" + renderer = PlainPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + # Should contain ASCII tree connectors + assert "|--" in output or "`--" in output + + def test_render_includes_status_in_brackets( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render includes status in brackets.""" + renderer = PlainPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + assert "[completed" in output + + def test_render_no_color_codes( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that plain renderer has no color codes.""" + renderer = PlainPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + # Should not contain rich color codes + assert "[green]" not in output + assert "[red]" not in output + assert "[yellow]" not in output + + +class TestJsonPlanTreeRenderer: + """Tests for JsonPlanTreeRenderer.""" + + def test_render_returns_valid_json( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that render returns valid JSON.""" + import json + + renderer = JsonPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + + # Should not raise + data = json.loads(output) + assert isinstance(data, dict) + + def test_render_includes_plan_id( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that JSON includes plan ID.""" + import json + + renderer = JsonPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + data = json.loads(output) + + assert data["plan_id"] == "plan-001" + + def test_render_includes_root_node( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that JSON includes root node.""" + import json + + renderer = JsonPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + data = json.loads(output) + + assert "root" in data + assert data["root"]["decision_id"] == "d-001" + + def test_render_includes_children( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that JSON includes children.""" + import json + + renderer = JsonPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + data = json.loads(output) + + root = data["root"] + assert "children" in root + assert len(root["children"]) == 1 + assert root["children"][0]["decision_id"] == "d-002" + + def test_render_includes_all_fields( + self, sample_decision_tree: DecisionTreeNode + ) -> None: + """Test that JSON includes all required fields.""" + import json + + renderer = JsonPlanTreeRenderer() + output = renderer.render(sample_decision_tree, "plan-001") + data = json.loads(output) + + root = data["root"] + assert "decision_id" in root + assert "plan_id" in root + assert "type" in root + assert "timestamp" in root + assert "question" in root + assert "chosen_option" in root + assert "status" in root + assert "summary" in root + + +class TestGetRenderer: + """Tests for get_renderer function.""" + + def test_get_renderer_returns_rich_renderer(self) -> None: + """Test that get_renderer returns RichPlanTreeRenderer.""" + renderer = get_renderer("rich") + assert isinstance(renderer, RichPlanTreeRenderer) + + def test_get_renderer_returns_plain_renderer(self) -> None: + """Test that get_renderer returns PlainPlanTreeRenderer.""" + renderer = get_renderer("plain") + assert isinstance(renderer, PlainPlanTreeRenderer) + + def test_get_renderer_returns_json_renderer(self) -> None: + """Test that get_renderer returns JsonPlanTreeRenderer.""" + renderer = get_renderer("json") + assert isinstance(renderer, JsonPlanTreeRenderer) + + def test_get_renderer_raises_for_unknown_format(self) -> None: + """Test that get_renderer raises for unknown format.""" + with pytest.raises(ValueError): + get_renderer("unknown") diff --git a/features/test_plan_tree_service.py b/features/test_plan_tree_service.py new file mode 100644 index 000000000..c008c42bd --- /dev/null +++ b/features/test_plan_tree_service.py @@ -0,0 +1,239 @@ +"""Unit tests for PlanTreeService.""" + +from __future__ import annotations + +from datetime import datetime +from unittest.mock import MagicMock + +import pytest + +from cleveragents.application.services.plan_tree_service import ( + DecisionTreeNode, + PlanTreeService, +) +from cleveragents.domain.models.core.decision import Decision, DecisionType + + +@pytest.fixture +def mock_plan_repository() -> MagicMock: + """Create a mock plan repository.""" + return MagicMock() + + +@pytest.fixture +def mock_decision_repository() -> MagicMock: + """Create a mock decision repository.""" + return MagicMock() + + +@pytest.fixture +def plan_tree_service( + mock_plan_repository: MagicMock, + mock_decision_repository: MagicMock, +) -> PlanTreeService: + """Create a PlanTreeService instance.""" + return PlanTreeService(mock_plan_repository, mock_decision_repository) + + +@pytest.fixture +def sample_decisions() -> list[Decision]: + """Create sample decisions for testing.""" + now = datetime.now(datetime.UTC) + + decisions = [ + Decision( + decision_id="d-001", + plan_id="plan-001", + parent_decision_id=None, + sequence_number=0, + decision_type=DecisionType.PROMPT_DEFINITION, + question="What is the task?", + chosen_option="Analyze code", + rationale="User requested code analysis", + created_at=now, + ), + Decision( + decision_id="d-002", + plan_id="plan-001", + parent_decision_id="d-001", + sequence_number=1, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which strategy?", + chosen_option="Iterative", + rationale="Iterative approach is best", + created_at=now, + ), + Decision( + decision_id="d-003", + plan_id="plan-001", + parent_decision_id="d-002", + sequence_number=2, + decision_type=DecisionType.TOOL_SELECTION, + question="Which tool?", + chosen_option="ruff", + rationale="ruff is fast", + created_at=now, + ), + Decision( + decision_id="d-004", + plan_id="plan-001", + parent_decision_id="d-001", + sequence_number=3, + decision_type=DecisionType.BRANCH, + question="Await confirmation?", + chosen_option="yes", + rationale="Need user input", + created_at=now, + ), + ] + + # Add status attribute for testing + for decision in decisions: + decision.status = "completed" + + return decisions + + +class TestPlanTreeService: + """Tests for PlanTreeService.""" + + def test_get_decision_tree_returns_none_for_nonexistent_plan( + self, + plan_tree_service: PlanTreeService, + mock_plan_repository: MagicMock, + ) -> None: + """Test that get_decision_tree returns None for non-existent plan.""" + mock_plan_repository.get_by_id.return_value = None + + result = plan_tree_service.get_decision_tree("plan-nonexistent") + + assert result is None + mock_plan_repository.get_by_id.assert_called_once_with( + "plan-nonexistent" + ) + + def test_get_decision_tree_returns_none_for_plan_with_no_decisions( + self, + plan_tree_service: PlanTreeService, + mock_plan_repository: MagicMock, + mock_decision_repository: MagicMock, + ) -> None: + """Test that get_decision_tree returns None for plan with no decisions.""" + mock_plan = MagicMock() + mock_plan_repository.get_by_id.return_value = mock_plan + mock_decision_repository.find_by_plan_id.return_value = [] + + result = plan_tree_service.get_decision_tree("plan-001") + + assert result is None + + def test_get_decision_tree_builds_correct_tree_structure( + self, + plan_tree_service: PlanTreeService, + mock_plan_repository: MagicMock, + mock_decision_repository: MagicMock, + sample_decisions: list[Decision], + ) -> None: + """Test that get_decision_tree builds correct tree structure.""" + mock_plan = MagicMock() + mock_plan_repository.get_by_id.return_value = mock_plan + mock_decision_repository.find_by_plan_id.return_value = sample_decisions + + root = plan_tree_service.get_decision_tree("plan-001") + + assert root is not None + assert root.decision.decision_id == "d-001" + assert len(root.children) == 2 # d-002 and d-004 + assert root.children[0].decision.decision_id == "d-002" + assert root.children[1].decision.decision_id == "d-004" + assert len(root.children[0].children) == 1 # d-003 + assert root.children[0].children[0].decision.decision_id == "d-003" + + def test_get_decision_tree_respects_depth_limit( + self, + plan_tree_service: PlanTreeService, + mock_plan_repository: MagicMock, + mock_decision_repository: MagicMock, + sample_decisions: list[Decision], + ) -> None: + """Test that get_decision_tree respects depth limit.""" + mock_plan = MagicMock() + mock_plan_repository.get_by_id.return_value = mock_plan + mock_decision_repository.find_by_plan_id.return_value = sample_decisions + + # Depth 1: only root and immediate children + root = plan_tree_service.get_decision_tree("plan-001", depth=1) + + assert root is not None + assert len(root.children) == 2 + assert len(root.children[0].children) == 0 # d-003 should not be included + assert len(root.children[1].children) == 0 + + def test_get_tree_summary_returns_correct_statistics( + self, + plan_tree_service: PlanTreeService, + mock_decision_repository: MagicMock, + sample_decisions: list[Decision], + ) -> None: + """Test that get_tree_summary returns correct statistics.""" + mock_decision_repository.find_by_plan_id.return_value = sample_decisions + + summary = plan_tree_service.get_tree_summary("plan-001") + + assert summary["plan_id"] == "plan-001" + assert summary["total_decisions"] == 4 + assert summary["max_depth"] == 2 + assert summary["status_counts"]["completed"] == 4 + + def test_get_tree_summary_for_empty_plan( + self, + plan_tree_service: PlanTreeService, + mock_decision_repository: MagicMock, + ) -> None: + """Test that get_tree_summary handles empty plan.""" + mock_decision_repository.find_by_plan_id.return_value = [] + + summary = plan_tree_service.get_tree_summary("plan-empty") + + assert summary["plan_id"] == "plan-empty" + assert summary["total_decisions"] == 0 + assert summary["max_depth"] == 0 + + +class TestDecisionTreeNode: + """Tests for DecisionTreeNode.""" + + def test_to_dict_includes_all_fields( + self, sample_decisions: list[Decision] + ) -> None: + """Test that to_dict includes all required fields.""" + decision = sample_decisions[0] + node = DecisionTreeNode(decision=decision, depth=0) + + result = node.to_dict() + + assert result["decision_id"] == "d-001" + assert result["plan_id"] == "plan-001" + assert result["type"] == "prompt_definition" + assert "timestamp" in result + assert result["question"] == "What is the task?" + assert result["chosen_option"] == "Analyze code" + assert result["status"] == "completed" + assert result["summary"] == "User requested code analysis" + assert result["children"] == [] + + def test_to_dict_includes_children( + self, sample_decisions: list[Decision] + ) -> None: + """Test that to_dict includes children.""" + parent = sample_decisions[0] + child = sample_decisions[1] + + parent_node = DecisionTreeNode(decision=parent, depth=0) + child_node = DecisionTreeNode(decision=child, depth=1) + parent_node.children.append(child_node) + + result = parent_node.to_dict() + + assert len(result["children"]) == 1 + assert result["children"][0]["decision_id"] == "d-002" diff --git a/src/cleveragents/application/services/plan_tree_service.py b/src/cleveragents/application/services/plan_tree_service.py new file mode 100644 index 000000000..2b1bb84f1 --- /dev/null +++ b/src/cleveragents/application/services/plan_tree_service.py @@ -0,0 +1,219 @@ +"""Service for rendering and querying the decision tree of a plan. + +This service provides methods to retrieve and render the decision tree +of a plan, including support for depth limiting, multiple output formats +(rich, plain text, JSON), and error handling for non-existent plans. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from cleveragents.domain.models.core.decision import Decision +from cleveragents.domain.repositories import ( + DecisionRepositoryProtocol, + LifecyclePlanRepositoryProtocol, +) + + +@dataclass +class DecisionTreeNode: + """A node in the decision tree with nested children.""" + + decision: Decision + children: list[DecisionTreeNode] = field(default_factory=list) + depth: int = 0 + + def to_dict(self) -> dict[str, Any]: + """Convert node to dictionary for JSON serialization.""" + return { + "decision_id": self.decision.decision_id, + "plan_id": self.decision.plan_id, + "type": str(self.decision.decision_type), + "timestamp": self.decision.created_at.isoformat(), + "question": self.decision.question, + "chosen_option": self.decision.chosen_option, + "status": getattr(self.decision, "status", "completed"), + "summary": self.decision.rationale or self.decision.chosen_option, + "children": [child.to_dict() for child in self.children], + } + + +class PlanTreeService: + """Service for retrieving and rendering plan decision trees.""" + + def __init__( + self, + plan_repository: LifecyclePlanRepositoryProtocol, + decision_repository: DecisionRepositoryProtocol, + ) -> None: + """Initialize the service with required repositories. + + Args: + plan_repository: Repository for accessing plan data. + decision_repository: Repository for accessing decision data. + """ + self.plan_repository = plan_repository + self.decision_repository = decision_repository + + def get_decision_tree( + self, plan_id: str, depth: int | None = None + ) -> DecisionTreeNode | None: + """Get the decision tree for a plan. + + Args: + plan_id: The ULID of the plan. + depth: Maximum depth to include in the tree (None = unlimited). + + Returns: + The root DecisionTreeNode of the tree, or None if plan not found. + + Raises: + ValueError: If plan_id is invalid. + """ + # Verify plan exists + plan = self.plan_repository.get(plan_id) + if not plan: + return None + + # Get all decisions for the plan + decisions = self.decision_repository.get_by_plan(plan_id) + if not decisions: + return None + + # Build a map of decision_id -> Decision for quick lookup + decision_map: dict[str, Decision] = {d.decision_id: d for d in decisions} + + # Find root decisions (those with no parent) + root_decisions = [d for d in decisions if d.parent_decision_id is None] + + if not root_decisions: + return None + + # Build tree from root(s) + # Note: typically there's only one root (prompt_definition) + root_node = self._build_tree_node( + root_decisions[0], decision_map, depth=0, max_depth=depth + ) + + return root_node + + def _build_tree_node( + self, + decision: Decision, + decision_map: dict[str, Decision], + depth: int, + max_depth: int | None, + ) -> DecisionTreeNode: + """Recursively build a tree node and its children. + + Args: + decision: The decision to create a node for. + decision_map: Map of all decisions by ID. + depth: Current depth in the tree. + max_depth: Maximum depth to include (None = unlimited). + + Returns: + A DecisionTreeNode with children populated up to max_depth. + """ + node = DecisionTreeNode(decision=decision, depth=depth) + + # Stop if we've reached max depth + if max_depth is not None and depth >= max_depth: + return node + + # Find children of this decision + children = [ + d + for d in decision_map.values() + if d.parent_decision_id == decision.decision_id + ] + + # Recursively build child nodes + for child in sorted(children, key=lambda d: d.sequence_number): + child_node = self._build_tree_node( + child, decision_map, depth + 1, max_depth + ) + node.children.append(child_node) + + return node + + def get_tree_summary(self, plan_id: str) -> dict[str, Any]: + """Get summary statistics about a plan's decision tree. + + Args: + plan_id: The ULID of the plan. + + Returns: + Dictionary with summary statistics. + """ + decisions = self.decision_repository.get_by_plan(plan_id) + + if not decisions: + return { + "plan_id": plan_id, + "total_decisions": 0, + "max_depth": 0, + "status_counts": {"pending": 0, "completed": 0, "reverted": 0}, + } + + # Count decisions by status + status_counts = {"pending": 0, "completed": 0, "reverted": 0} + for decision in decisions: + status = getattr(decision, "status", "completed") + if status in status_counts: + status_counts[status] += 1 + + # Calculate max depth + max_depth = self._calculate_max_depth(decisions) + + return { + "plan_id": plan_id, + "total_decisions": len(decisions), + "max_depth": max_depth, + "status_counts": status_counts, + } + + def _calculate_max_depth(self, decisions: list[Decision]) -> int: + """Calculate the maximum depth of the decision tree. + + Args: + decisions: List of all decisions in the plan. + + Returns: + The maximum depth (0 if only root, 1 if root + children, etc.). + """ + if not decisions: + return 0 + + # Build parent map + parent_map: dict[str | None, list[str]] = {} + for decision in decisions: + parent_id = decision.parent_decision_id + if parent_id not in parent_map: + parent_map[parent_id] = [] + parent_map[parent_id].append(decision.decision_id) + + # Find roots + roots = parent_map.get(None, []) + if not roots: + return 0 + + # BFS to find max depth + max_depth = 0 + queue = [(root_id, 0) for root_id in roots] + + while queue: + decision_id, current_depth = queue.pop(0) + max_depth = max(max_depth, current_depth) + + # Add children to queue + children = parent_map.get(decision_id, []) + for child_id in children: + queue.append((child_id, current_depth + 1)) + + return max_depth + + +__all__ = ["DecisionTreeNode", "PlanTreeService"] diff --git a/src/cleveragents/cli/output/plan_tree_renderers.py b/src/cleveragents/cli/output/plan_tree_renderers.py new file mode 100644 index 000000000..fa97c7fff --- /dev/null +++ b/src/cleveragents/cli/output/plan_tree_renderers.py @@ -0,0 +1,214 @@ +"""Renderers for the plan decision tree in various output formats. + +Supports Rich (colored terminal), plain text (ASCII), and JSON output formats. +""" + +from __future__ import annotations + +import json +from typing import ClassVar + +from cleveragents.application.services.plan_tree_service import DecisionTreeNode + + +class PlanTreeRenderer: + """Base class for plan tree renderers.""" + + def render(self, root_node: DecisionTreeNode, plan_id: str) -> str: + """Render the decision tree. + + Args: + root_node: The root node of the decision tree. + plan_id: The plan ID for display. + + Returns: + Rendered tree as a string. + """ + raise NotImplementedError + + +class RichPlanTreeRenderer(PlanTreeRenderer): + """Renderer for rich terminal output with colors and formatting.""" + + # Status color codes + STATUS_COLORS: ClassVar[dict[str, str]] = { + "pending": "[yellow]", + "completed": "[green]", + "reverted": "[red]", + } + + STATUS_RESET: ClassVar[str] = "[/]" + + def render(self, root_node: DecisionTreeNode, plan_id: str) -> str: + """Render the decision tree with rich formatting. + + Args: + root_node: The root node of the decision tree. + plan_id: The plan ID for display. + + Returns: + Rendered tree as a rich-formatted string. + """ + lines = [f"Plan Decision Tree: {plan_id}"] + self._render_node(root_node, lines, is_last=True, prefix="") + return "\n".join(lines) + + def _render_node( + self, + node: DecisionTreeNode, + lines: list[str], + is_last: bool, + prefix: str, + ) -> None: + """Recursively render a node and its children. + + Args: + node: The node to render. + lines: List to append rendered lines to. + is_last: Whether this is the last child of its parent. + prefix: The prefix to use for tree connectors. + """ + decision = node.decision + status = getattr(decision, "status", "completed") + + # Format status with color + status_color = self.STATUS_COLORS.get(status, "") + status_str = f"{status_color}[{status}]{self.STATUS_RESET}" + + # Format timestamp + timestamp = decision.created_at.isoformat() + + # Format summary (use rationale if available, otherwise chosen option) + summary = decision.rationale or decision.chosen_option + + # Build the line + connector = "└── " if is_last else "├── " + line = ( + f"{prefix}{connector}{status_str} {decision.decision_id} | " + f"{decision.decision_type!s:20} | {timestamp} | {summary}" + ) + lines.append(line) + + # Render children + if node.children: + extension = " " if is_last else "│ " + for i, child in enumerate(node.children): + is_last_child = i == len(node.children) - 1 + self._render_node( + child, lines, is_last_child, prefix + extension + ) + + +class PlainPlanTreeRenderer(PlanTreeRenderer): + """Renderer for plain text output with ASCII tree connectors.""" + + def render(self, root_node: DecisionTreeNode, plan_id: str) -> str: + """Render the decision tree with ASCII formatting. + + Args: + root_node: The root node of the decision tree. + plan_id: The plan ID for display. + + Returns: + Rendered tree as a plain text string. + """ + lines = [f"Plan Decision Tree: {plan_id}"] + self._render_node(root_node, lines, is_last=True, prefix="") + return "\n".join(lines) + + def _render_node( + self, + node: DecisionTreeNode, + lines: list[str], + is_last: bool, + prefix: str, + ) -> None: + """Recursively render a node and its children. + + Args: + node: The node to render. + lines: List to append rendered lines to. + is_last: Whether this is the last child of its parent. + prefix: The prefix to use for tree connectors. + """ + decision = node.decision + status = getattr(decision, "status", "completed") + + # Format timestamp + timestamp = decision.created_at.isoformat() + + # Format summary + summary = decision.rationale or decision.chosen_option + + # Build the line with ASCII connectors + connector = "`-- " if is_last else "|-- " + line = ( + f"{prefix}{connector}[{status:10}] {decision.decision_id} | " + f"{decision.decision_type!s:20} | {timestamp} | {summary}" + ) + lines.append(line) + + # Render children + if node.children: + extension = " " if is_last else "| " + for i, child in enumerate(node.children): + is_last_child = i == len(node.children) - 1 + self._render_node( + child, lines, is_last_child, prefix + extension + ) + + +class JsonPlanTreeRenderer(PlanTreeRenderer): + """Renderer for JSON output.""" + + def render(self, root_node: DecisionTreeNode, plan_id: str) -> str: + """Render the decision tree as JSON. + + Args: + root_node: The root node of the decision tree. + plan_id: The plan ID for display. + + Returns: + Rendered tree as a JSON string. + """ + tree_dict = { + "plan_id": plan_id, + "root": root_node.to_dict(), + } + return json.dumps(tree_dict, indent=2) + + +def get_renderer(format_type: str) -> PlanTreeRenderer: + """Get a renderer for the specified format. + + Args: + format_type: The format type ('rich', 'plain', or 'json'). + + Returns: + A PlanTreeRenderer instance. + + Raises: + ValueError: If format_type is not recognized. + """ + renderers = { + "rich": RichPlanTreeRenderer, + "plain": PlainPlanTreeRenderer, + "json": JsonPlanTreeRenderer, + } + + if format_type not in renderers: + raise ValueError( + f"Unknown format: {format_type}. " + f"Supported formats: {', '.join(renderers.keys())}" + ) + + return renderers[format_type]() + + +__all__ = [ + "JsonPlanTreeRenderer", + "PlainPlanTreeRenderer", + "PlanTreeRenderer", + "RichPlanTreeRenderer", + "get_renderer", +]