"""Step definitions for plan_explain_cli_coverage.feature. Exercises the CLI-level code paths for plan explain, plan tree, plan correct, plan resume, plan revert, and the read-only plan guards, using the Typer CliRunner with mocked services. """ from __future__ import annotations import json from unittest.mock import MagicMock, patch from behave import given, then, when from behave.runner import Context from typer.testing import CliRunner from ulid import ULID from cleveragents.application.services.plan_lifecycle_service import ( InvalidPhaseTransitionError, ) from cleveragents.cli.commands.plan import app as plan_app from cleveragents.core.exceptions import ( CleverAgentsError, PlanError, ValidationError, ) from cleveragents.domain.models.core.decision import ( ContextSnapshot, Decision, DecisionType, ResourceRef, ) runner = CliRunner() _PATCH_CONTAINER = "cleveragents.application.container.get_container" _PATCH_LIFECYCLE = "cleveragents.cli.commands.plan._get_lifecycle_service" _PATCH_RESUME_SVC_MOD = ( "cleveragents.application.services.plan_resume_service.PlanResumeService" ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_decision( decision_id: str | None = None, plan_id: str | None = None, sequence: int = 0, parent_id: str | None = None, dtype: DecisionType = DecisionType.PROMPT_DEFINITION, question: str = "What should we build?", chosen: str = "A REST API", superseded_by: str | None = None, is_correction: bool = False, corrects_decision_id: str | None = None, correction_reason: str | None = None, confidence_score: float | None = 0.85, rationale: str = "", actor_reasoning: str | None = None, alternatives: list[str] | None = None, context_snapshot: ContextSnapshot | None = None, ) -> Decision: did = decision_id or str(ULID()) pid = plan_id or str(ULID()) kwargs: dict = { "decision_id": did, "plan_id": pid, "sequence_number": sequence, "decision_type": dtype, "question": question, "chosen_option": chosen, "confidence_score": confidence_score, "rationale": rationale, "superseded_by": superseded_by, "is_correction": is_correction, } if parent_id is not None: kwargs["parent_decision_id"] = parent_id if corrects_decision_id is not None: kwargs["corrects_decision_id"] = corrects_decision_id if correction_reason is not None: kwargs["correction_reason"] = correction_reason if actor_reasoning is not None: kwargs["actor_reasoning"] = actor_reasoning if alternatives is not None: kwargs["alternatives_considered"] = alternatives if context_snapshot is not None: kwargs["context_snapshot"] = context_snapshot return Decision(**kwargs) def _mock_container_with_decision_svc(svc_mock: MagicMock) -> MagicMock: container = MagicMock() container.decision_service.return_value = svc_mock return container def _make_tree_decisions() -> list[Decision]: root_id = str(ULID()) child_id = str(ULID()) return [ _make_decision(decision_id=root_id, sequence=0), _make_decision( decision_id=child_id, parent_id=root_id, sequence=1, dtype=DecisionType.STRATEGY_CHOICE, question="Which framework?", chosen="FastAPI", ), ] def _make_deep_decisions() -> list[Decision]: root_id = str(ULID()) child_id = str(ULID()) grandchild_id = str(ULID()) return [ _make_decision(decision_id=root_id, sequence=0), _make_decision( decision_id=child_id, parent_id=root_id, sequence=1, dtype=DecisionType.STRATEGY_CHOICE, question="Child question?", chosen="Child answer", ), _make_decision( decision_id=grandchild_id, parent_id=child_id, sequence=2, dtype=DecisionType.IMPLEMENTATION_CHOICE, question="Grandchild question?", chosen="Grandchild answer", ), ] # --------------------------------------------------------------------------- # GIVEN steps - explain # --------------------------------------------------------------------------- @given("pec a mock DecisionService returning a valid decision") def step_pec_mock_decision_svc(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_decision = _make_decision(decision_id=context.pec_decision_id) svc = MagicMock() svc.get_decision.return_value = context.pec_decision context.pec_container = _mock_container_with_decision_svc(svc) @given("pec a mock DecisionService returning None for get_decision") def step_pec_mock_decision_none(context: Context) -> None: context.pec_decision_id = str(ULID()) svc = MagicMock() svc.get_decision.return_value = None svc.list_decisions.return_value = [] context.pec_container = _mock_container_with_decision_svc(svc) @given("pec a mock DecisionService returning a decision with context snapshot") def step_pec_mock_decision_ctx(context: Context) -> None: context.pec_decision_id = str(ULID()) snap = ContextSnapshot( hot_context_hash="sha256:abc123", hot_context_ref="store://ctx/1", relevant_resources=[ ResourceRef(resource_id=str(ULID()), path="src/main.py"), ], actor_state_ref="checkpoint://state/1", ) context.pec_decision = _make_decision( decision_id=context.pec_decision_id, context_snapshot=snap, ) svc = MagicMock() svc.get_decision.return_value = context.pec_decision context.pec_container = _mock_container_with_decision_svc(svc) @given("pec a mock DecisionService returning a decision with reasoning") def step_pec_mock_decision_reasoning(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_decision = _make_decision( decision_id=context.pec_decision_id, rationale="Chose REST API for simplicity", actor_reasoning="The LLM considered approaches...", ) svc = MagicMock() svc.get_decision.return_value = context.pec_decision context.pec_container = _mock_container_with_decision_svc(svc) @given("pec a mock DecisionService returning a decision with alternatives") def step_pec_mock_decision_alts(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_decision = _make_decision( decision_id=context.pec_decision_id, alternatives=["GraphQL API", "gRPC service"], ) svc = MagicMock() svc.get_decision.return_value = context.pec_decision context.pec_container = _mock_container_with_decision_svc(svc) # --------------------------------------------------------------------------- # GIVEN steps - tree # --------------------------------------------------------------------------- @given("pec a mock DecisionService returning a list of decisions") def step_pec_mock_tree_decisions(context: Context) -> None: context.pec_plan_id = str(ULID()) decisions = _make_tree_decisions() svc = MagicMock() svc.list_decisions.return_value = decisions context.pec_container = _mock_container_with_decision_svc(svc) @given("pec a mock DecisionService returning a deep decision list") def step_pec_mock_deep_decisions(context: Context) -> None: context.pec_plan_id = str(ULID()) decisions = _make_deep_decisions() svc = MagicMock() svc.list_decisions.return_value = decisions context.pec_container = _mock_container_with_decision_svc(svc) @given("pec a mock DecisionService returning decisions with superseded") def step_pec_mock_superseded_decisions(context: Context) -> None: context.pec_plan_id = str(ULID()) root_id = str(ULID()) old_id = str(ULID()) new_id = str(ULID()) decisions = [ _make_decision(decision_id=root_id, sequence=0), _make_decision( decision_id=old_id, parent_id=root_id, sequence=1, dtype=DecisionType.STRATEGY_CHOICE, question="Which framework?", chosen="Flask", superseded_by=new_id, ), _make_decision( decision_id=new_id, parent_id=root_id, sequence=2, dtype=DecisionType.STRATEGY_CHOICE, question="Which framework?", chosen="FastAPI", is_correction=True, corrects_decision_id=old_id, correction_reason="Better performance", ), ] svc = MagicMock() svc.list_decisions.return_value = decisions context.pec_container = _mock_container_with_decision_svc(svc) @given("pec a mock DecisionService returning an empty list") def step_pec_mock_empty_decisions(context: Context) -> None: context.pec_plan_id = str(ULID()) svc = MagicMock() svc.list_decisions.return_value = [] context.pec_container = _mock_container_with_decision_svc(svc) # --------------------------------------------------------------------------- # GIVEN steps - orphan edge case # --------------------------------------------------------------------------- @given("pec a decision list where a child references a missing parent") def step_pec_orphan_decisions(context: Context) -> None: """Build three decisions where the grandchild is superseded. With ``show_superseded=False`` the grandchild is excluded from ``by_id`` but its entry in ``children_map`` (built from ALL decisions) still references its ``decision_id`` under its parent. This forces the BFS orphan guard (``child_id not in by_id``) to fire and skip the missing node. """ root_id = str(ULID()) child_id = str(ULID()) grandchild_id = str(ULID()) context.pec_orphan_root_id = root_id context.pec_orphan_decisions = [ _make_decision(decision_id=root_id, sequence=0), _make_decision( decision_id=child_id, parent_id=root_id, sequence=1, dtype=DecisionType.STRATEGY_CHOICE, question="Child question?", chosen="Child answer", ), _make_decision( decision_id=grandchild_id, parent_id=child_id, sequence=2, dtype=DecisionType.IMPLEMENTATION_CHOICE, question="Grandchild question?", chosen="Grandchild answer", superseded_by=str(ULID()), ), ] # --------------------------------------------------------------------------- # GIVEN steps - resolve active plan # --------------------------------------------------------------------------- @given("pec a lifecycle service with one active plan") def step_pec_lifecycle_active(context: Context) -> None: context.pec_expected_plan_id = str(ULID()) plan = MagicMock() plan.is_terminal = False plan.identity.plan_id = context.pec_expected_plan_id svc = MagicMock() svc.list_plans.return_value = [plan] context.pec_lifecycle_svc = svc # --------------------------------------------------------------------------- # GIVEN steps - revert error handlers # --------------------------------------------------------------------------- @given("pec a lifecycle service that raises InvalidPhaseTransitionError on revert") def step_pec_revert_invalid_phase(context: Context) -> None: from cleveragents.domain.models.core.plan import PlanPhase context.pec_plan_id = str(ULID()) svc = MagicMock() svc.revert_plan.side_effect = InvalidPhaseTransitionError( from_phase=PlanPhase.APPLY, to_phase=PlanPhase.STRATEGIZE, message="Cannot revert from apply phase", ) context.pec_lifecycle_svc = svc @given("pec a lifecycle service that raises PlanError on revert") def step_pec_revert_plan_error(context: Context) -> None: context.pec_plan_id = str(ULID()) svc = MagicMock() svc.revert_plan.side_effect = PlanError("Plan is terminal") context.pec_lifecycle_svc = svc # --------------------------------------------------------------------------- # GIVEN steps - correct # --------------------------------------------------------------------------- @given("pec a mock CorrectionService with dry-run impact") def step_pec_correct_dry_run(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_plan_id = str(ULID()) request = MagicMock() request.correction_id = str(ULID()) request.mode.value = "revert" request.target_decision_id = context.pec_decision_id request.guidance = "Use FastAPI instead" impact = MagicMock() impact.affected_decisions = ["DEC-A", "DEC-B"] impact.affected_files = ["src/api.py"] impact.estimated_cost = "low" impact.risk_level = "medium" svc = MagicMock() svc.request_correction.return_value = request svc.analyze_impact.return_value = impact context.pec_correction_svc = svc @given("pec a mock CorrectionService with execute result") def step_pec_correct_execute(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_plan_id = str(ULID()) request = MagicMock() request.correction_id = str(ULID()) request.mode.value = "revert" request.target_decision_id = context.pec_decision_id request.guidance = "Use FastAPI instead" result = MagicMock() result.correction_id = request.correction_id result.status.value = "applied" result.new_decisions = [] result.reverted_decisions = [] svc = MagicMock() svc.request_correction.return_value = request svc.execute_correction.return_value = result context.pec_correction_svc = svc @given("pec a mock CorrectionService with reverted and new decisions") def step_pec_correct_with_changes(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_plan_id = str(ULID()) request = MagicMock() request.correction_id = str(ULID()) request.mode.value = "revert" request.target_decision_id = context.pec_decision_id request.guidance = "Use FastAPI instead" result = MagicMock() result.correction_id = request.correction_id result.status.value = "applied" result.new_decisions = ["DEC-NEW-1"] result.reverted_decisions = ["DEC-OLD-1"] svc = MagicMock() svc.request_correction.return_value = request svc.execute_correction.return_value = result context.pec_correction_svc = svc @given("pec a mock CorrectionService that raises ResourceNotFoundError") def step_pec_correct_rnf(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_plan_id = str(ULID()) from cleveragents.core.exceptions import ResourceNotFoundError svc = MagicMock() svc.request_correction.side_effect = ResourceNotFoundError("Decision not found") context.pec_correction_svc = svc @given("pec a mock CorrectionService that raises ValidationError") def step_pec_correct_validation(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_plan_id = str(ULID()) svc = MagicMock() svc.request_correction.side_effect = ValidationError("Invalid correction mode") context.pec_correction_svc = svc @given("pec a mock CorrectionService that raises CleverAgentsError") def step_pec_correct_ca_error(context: Context) -> None: context.pec_decision_id = str(ULID()) context.pec_plan_id = str(ULID()) svc = MagicMock() svc.request_correction.side_effect = CleverAgentsError("Service unavailable") context.pec_correction_svc = svc # --------------------------------------------------------------------------- # GIVEN steps - resume # --------------------------------------------------------------------------- @given("pec a mock PlanResumeService returning a summary") def step_pec_resume_svc(context: Context) -> None: context.pec_plan_id = str(ULID()) summary = MagicMock() summary.plan_id = context.pec_plan_id summary.phase = "execute" summary.processing_state = "in_progress" summary.last_completed_step = 3 summary.next_step_index = 4 summary.total_steps = 10 summary.decision_id = str(ULID()) summary.last_checkpoint_id = str(ULID()) summary.sandbox_ref = "/tmp/sandbox/plan-001" summary.as_cli_dict.return_value = { "plan_id": summary.plan_id, "phase": "execute", "next_step": 4, } context.pec_resume_summary = summary svc = MagicMock() svc.resume_plan.return_value = summary context.pec_resume_svc = svc @given("pec a mock PlanResumeService that raises PlanError") def step_pec_resume_error(context: Context) -> None: context.pec_plan_id = str(ULID()) svc = MagicMock() svc.resume_plan.side_effect = PlanError("Plan is terminal, cannot resume") context.pec_resume_svc = svc # --------------------------------------------------------------------------- # GIVEN steps - read-only plan # --------------------------------------------------------------------------- @given("pec a lifecycle service returning a read-only plan") def step_pec_readonly_plan(context: Context) -> None: context.pec_plan_id = str(ULID()) plan = MagicMock() plan.read_only = True svc = MagicMock() svc.get_plan.return_value = plan context.pec_lifecycle_svc = svc # --------------------------------------------------------------------------- # WHEN steps - explain # --------------------------------------------------------------------------- @when('pec I invoke "explain" with the decision id') def step_pec_invoke_explain(context: Context) -> None: with patch(_PATCH_CONTAINER, return_value=context.pec_container): context.pec_result = runner.invoke( plan_app, ["explain", context.pec_decision_id] ) @when('pec I invoke "explain" with format "{fmt}"') def step_pec_invoke_explain_fmt(context: Context, fmt: str) -> None: with patch(_PATCH_CONTAINER, return_value=context.pec_container): context.pec_result = runner.invoke( plan_app, ["explain", context.pec_decision_id, "--format", fmt] ) @when('pec I invoke "explain" with flags "{flags}"') def step_pec_invoke_explain_flags(context: Context, flags: str) -> None: with patch(_PATCH_CONTAINER, return_value=context.pec_container): context.pec_result = runner.invoke( plan_app, ["explain", context.pec_decision_id, *flags.split()], ) # --------------------------------------------------------------------------- # WHEN steps - tree # --------------------------------------------------------------------------- @when('pec I invoke "tree" with a plan id') def step_pec_invoke_tree(context: Context) -> None: with patch(_PATCH_CONTAINER, return_value=context.pec_container): context.pec_result = runner.invoke(plan_app, ["tree", context.pec_plan_id]) @when('pec I invoke "tree" with format "{fmt}"') def step_pec_invoke_tree_fmt(context: Context, fmt: str) -> None: with patch(_PATCH_CONTAINER, return_value=context.pec_container): context.pec_result = runner.invoke( plan_app, ["tree", context.pec_plan_id, "--format", fmt] ) @when('pec I invoke "tree" with format "{fmt}" and depth {depth:d}') def step_pec_invoke_tree_fmt_depth(context: Context, fmt: str, depth: int) -> None: with patch(_PATCH_CONTAINER, return_value=context.pec_container): context.pec_result = runner.invoke( plan_app, ["tree", context.pec_plan_id, "--format", fmt, "--depth", str(depth)], ) @when('pec I invoke "tree" with depth {depth:d}') def step_pec_invoke_tree_depth(context: Context, depth: int) -> None: with patch(_PATCH_CONTAINER, return_value=context.pec_container): context.pec_result = runner.invoke( plan_app, ["tree", context.pec_plan_id, "--depth", str(depth)] ) @when('pec I invoke "tree" with flags "{flags}"') def step_pec_invoke_tree_flags(context: Context, flags: str) -> None: with patch(_PATCH_CONTAINER, return_value=context.pec_container): context.pec_result = runner.invoke( plan_app, ["tree", context.pec_plan_id, *flags.split()] ) # --------------------------------------------------------------------------- # WHEN steps - orphan edge case # --------------------------------------------------------------------------- @when("pec I build the tree from orphan decisions") def step_pec_build_orphan_tree(context: Context) -> None: from cleveragents.cli.commands.plan import build_decision_tree context.pec_tree = build_decision_tree(context.pec_orphan_decisions) # --------------------------------------------------------------------------- # WHEN steps - resolve active plan # --------------------------------------------------------------------------- @when("pec I call resolve active plan id") def step_pec_resolve_active(context: Context) -> None: from cleveragents.cli.commands.plan import _resolve_active_plan_id with patch(_PATCH_LIFECYCLE, return_value=context.pec_lifecycle_svc): context.pec_resolved_id = _resolve_active_plan_id() # --------------------------------------------------------------------------- # WHEN steps - revert # --------------------------------------------------------------------------- @when('pec I invoke "revert" with plan id and target "{target}"') def step_pec_invoke_revert(context: Context, target: str) -> None: with patch(_PATCH_LIFECYCLE, return_value=context.pec_lifecycle_svc): context.pec_result = runner.invoke( plan_app, ["revert", context.pec_plan_id, "--to-phase", target] ) # --------------------------------------------------------------------------- # WHEN steps - correct # --------------------------------------------------------------------------- def _invoke_correct( context: Context, extra_args: list[str] | None = None, input_text: str | None = None ) -> None: args = [ "correct", context.pec_decision_id, "--mode", "revert", "--guidance", "Use FastAPI instead", "--plan", context.pec_plan_id, ] if extra_args: args.extend(extra_args) # Mock DecisionService resolved via DI container (issue #606 fix) mock_decision_svc = MagicMock() mock_decision_svc.list_decisions.return_value = [] mock_decision_svc.get_influence_edges.return_value = {} mock_container = MagicMock() mock_container.decision_service.return_value = mock_decision_svc mock_container.correction_service.return_value = context.pec_correction_svc with patch(_PATCH_CONTAINER, return_value=mock_container): context.pec_result = runner.invoke(plan_app, args, input=input_text) @when('pec I invoke "correct" in dry-run mode with rich format') def step_pec_correct_dryrun(context: Context) -> None: _invoke_correct(context, extra_args=["--dry-run"]) @when('pec I invoke "correct" with yes flag') def step_pec_correct_yes(context: Context) -> None: _invoke_correct(context, extra_args=["--yes"]) @when('pec I invoke "correct" without yes and decline') def step_pec_correct_decline(context: Context) -> None: _invoke_correct(context, input_text="n\n") # --------------------------------------------------------------------------- # WHEN steps - resume # --------------------------------------------------------------------------- @when('pec I invoke "resume" with a plan id') def step_pec_invoke_resume(context: Context) -> None: with ( patch(_PATCH_LIFECYCLE, return_value=MagicMock()), patch( _PATCH_RESUME_SVC_MOD, return_value=context.pec_resume_svc, ), ): context.pec_result = runner.invoke(plan_app, ["resume", context.pec_plan_id]) @when('pec I invoke "resume" with dry-run flag') def step_pec_invoke_resume_dryrun(context: Context) -> None: with ( patch(_PATCH_LIFECYCLE, return_value=MagicMock()), patch( _PATCH_RESUME_SVC_MOD, return_value=context.pec_resume_svc, ), ): context.pec_result = runner.invoke( plan_app, ["resume", context.pec_plan_id, "--dry-run"] ) @when('pec I invoke "resume" with format "{fmt}"') def step_pec_invoke_resume_fmt(context: Context, fmt: str) -> None: with ( patch(_PATCH_LIFECYCLE, return_value=MagicMock()), patch( _PATCH_RESUME_SVC_MOD, return_value=context.pec_resume_svc, ), ): context.pec_result = runner.invoke( plan_app, ["resume", context.pec_plan_id, "--format", fmt] ) # --------------------------------------------------------------------------- # WHEN steps - read-only guards # --------------------------------------------------------------------------- @when('pec I invoke "execute" with the read-only plan id') def step_pec_invoke_execute_readonly(context: Context) -> None: with patch(_PATCH_LIFECYCLE, return_value=context.pec_lifecycle_svc): context.pec_result = runner.invoke(plan_app, ["execute", context.pec_plan_id]) @when('pec I invoke "apply" with the read-only plan id') def step_pec_invoke_apply_readonly(context: Context) -> None: with patch(_PATCH_LIFECYCLE, return_value=context.pec_lifecycle_svc): context.pec_result = runner.invoke( plan_app, ["apply", "--yes", context.pec_plan_id] ) # --------------------------------------------------------------------------- # THEN steps # --------------------------------------------------------------------------- @then("pec the exit code should be 0") def step_pec_exit_0(context: Context) -> None: assert context.pec_result.exit_code == 0, ( f"Expected exit 0, got {context.pec_result.exit_code}.\n" f"Output: {context.pec_result.output}" ) @then("pec the exit code should be {code:d}") def step_pec_exit_code(context: Context, code: int) -> None: assert context.pec_result.exit_code == code, ( f"Expected exit {code}, got {context.pec_result.exit_code}.\n" f"Output: {context.pec_result.output}" ) @then("pec the exit code should be nonzero") def step_pec_exit_nonzero(context: Context) -> None: assert context.pec_result.exit_code != 0, ( f"Expected nonzero exit, got {context.pec_result.exit_code}.\n" f"Output: {context.pec_result.output}" ) @then('pec the output should contain "{text}"') def step_pec_output_contains(context: Context, text: str) -> None: assert text in context.pec_result.output, ( f"Expected '{text}' in output.\nOutput: {context.pec_result.output}" ) @then('pec the output should not contain "{text}"') def step_pec_output_not_contains(context: Context, text: str) -> None: assert text not in context.pec_result.output, ( f"Did not expect '{text}' in output.\nOutput: {context.pec_result.output}" ) @then("pec the output should be valid json") def step_pec_output_valid_json(context: Context) -> None: parsed = json.loads(context.pec_result.output.strip()) assert isinstance(parsed, dict), "Expected JSON object" @then("pec the output should be valid json list") def step_pec_output_valid_json_list(context: Context) -> None: parsed = json.loads(context.pec_result.output.strip()) assert isinstance(parsed, list), "Expected JSON array" @then("pec the tree should exclude the superseded grandchild") def step_pec_tree_excludes_orphan(context: Context) -> None: # Tree should have exactly one root with one child and zero grandchildren. assert len(context.pec_tree) == 1, f"Expected 1 root, got {len(context.pec_tree)}" root = context.pec_tree[0] assert root["decision_id"] == context.pec_orphan_root_id children = root["children"] assert len(children) == 1, ( # type: ignore[arg-type] f"Expected 1 child, got {len(children)}" # type: ignore[arg-type] ) # The grandchild was superseded and filtered from by_id; the BFS # orphan guard must have skipped it, leaving no grandchildren. grandchildren = children[0]["children"] # type: ignore[index] assert len(grandchildren) == 0, ( # type: ignore[arg-type] f"Expected 0 grandchildren, got {len(grandchildren)}" # type: ignore[arg-type] ) @then("pec the resolved plan id should match the active plan") def step_pec_resolved_matches(context: Context) -> None: assert context.pec_resolved_id == context.pec_expected_plan_id