"""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 typing import Any 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() _ENVELOPE_KEYS = {"command", "status", "exit_code", "data", "timing", "messages"} def _unwrap_envelope(parsed: Any) -> Any: """Return the ``data`` field if *parsed* is a spec envelope, else *parsed* as-is.""" if isinstance(parsed, dict) and _ENVELOPE_KEYS.issubset(parsed.keys()): return parsed["data"] return parsed _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()) data = _unwrap_envelope(parsed) assert isinstance(data, 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()) data = _unwrap_envelope(parsed) assert isinstance(data, 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