diff --git a/features/a2a_facade_wiring.feature b/features/a2a_facade_wiring.feature index fb8b5b81e..6c2d4a2c7 100644 --- a/features/a2a_facade_wiring.feature +++ b/features/a2a_facade_wiring.feature @@ -165,3 +165,86 @@ Feature: A2A local facade wiring to live services When I dispatch wired operation "event.subscribe" with params {} Then the wired response status should be "ok" And wired response data key "status" equals "subscribed" + + # --------------------------------------------------------------- + # plan.execute full lifecycle (PlanExecutor wired) + # --------------------------------------------------------------- + + Scenario: plan.execute with PlanExecutor runs full lifecycle from Strategize/queued + Given a wired A2aLocalFacade with PlanExecutor for Strategize/queued plan + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-001"} + Then the wired response status should be "ok" + And wired response data key "plan_id" equals "MOCK-PLAN-001" + And wired response data key "phase" equals "execute" + And wired response data key "state" equals "complete" + And the PlanExecutor run_strategize was called with plan_id "MOCK-PLAN-001" + And the PlanExecutor run_execute was called with plan_id "MOCK-PLAN-001" + + Scenario: plan.execute with PlanExecutor runs Execute phase from Strategize/complete + Given a wired A2aLocalFacade with PlanExecutor for Strategize/complete plan + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-002"} + Then the wired response status should be "ok" + And wired response data key "plan_id" equals "MOCK-PLAN-002" + And wired response data key "phase" equals "execute" + And wired response data key "state" equals "complete" + And the PlanExecutor run_strategize was NOT called + And the PlanExecutor run_execute was called with plan_id "MOCK-PLAN-002" + + Scenario: plan.execute with PlanExecutor runs Execute phase from Execute/queued + Given a wired A2aLocalFacade with PlanExecutor for Execute/queued plan + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-003"} + Then the wired response status should be "ok" + And wired response data key "plan_id" equals "MOCK-PLAN-003" + And wired response data key "phase" equals "execute" + And wired response data key "state" equals "complete" + And the PlanExecutor run_strategize was NOT called + And the PlanExecutor run_execute was called with plan_id "MOCK-PLAN-003" + + Scenario: plan.execute without PlanExecutor falls back to legacy queued-only behaviour + Given a wired A2aLocalFacade with a mock PlanLifecycleService + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-001"} + Then the wired response status should be "ok" + And wired response data key "plan_id" equals "MOCK-PLAN-001" + + Scenario: plan.execute with PlanExecutor and legacy operation name runs full lifecycle + Given a wired A2aLocalFacade with PlanExecutor for Strategize/queued plan + When I dispatch wired operation "plan.execute" with params {"plan_id": "MOCK-PLAN-001"} + Then the wired response status should be "ok" + And wired response data key "phase" equals "execute" + And wired response data key "state" equals "complete" + And the PlanExecutor run_strategize was called with plan_id "MOCK-PLAN-001" + And the PlanExecutor run_execute was called with plan_id "MOCK-PLAN-001" + + # --------------------------------------------------------------- + # plan.execute error paths and boundary conditions + # --------------------------------------------------------------- + + Scenario: plan.execute raises PLAN_ERROR when run_strategize fails + Given a wired A2aLocalFacade with PlanExecutor where run_strategize raises PlanError + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-001"} + Then the wired response status should be "error" + And wired response error code should be "PLAN_ERROR" + + Scenario: plan.execute raises PLAN_ERROR when run_execute fails after strategize + Given a wired A2aLocalFacade with PlanExecutor where run_execute raises PlanError + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-001"} + Then the wired response status should be "error" + And wired response error code should be "PLAN_ERROR" + + Scenario: plan.execute raises INVALID_STATE for plan in Execute/complete terminal state + Given a wired A2aLocalFacade with PlanExecutor for Execute/complete plan + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-004"} + Then the wired response status should be "error" + And wired response error code should be "INVALID_STATE" + + Scenario: plan.execute raises INVALID_STATE for plan in Strategize/errored state + Given a wired A2aLocalFacade with PlanExecutor for Strategize/errored plan + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-005"} + Then the wired response status should be "error" + And wired response error code should be "INVALID_STATE" + + Scenario: plan.execute raises INVALID_STATE for plan already in Execute/processing + Given a wired A2aLocalFacade with PlanExecutor for Execute/processing plan + When I dispatch wired operation "_cleveragents/plan/execute" with params {"plan_id": "MOCK-PLAN-006"} + Then the wired response status should be "error" + And wired response error code should be "INVALID_STATE" diff --git a/features/mocks/plan_executor_mock.py b/features/mocks/plan_executor_mock.py new file mode 100644 index 000000000..10e886569 --- /dev/null +++ b/features/mocks/plan_executor_mock.py @@ -0,0 +1,98 @@ +"""Mock PlanExecutor and PlanLifecycleService pair for A2A facade wiring tests. + +Provides :func:`build_mock_plan_executor` which constructs a paired +``(mock_lifecycle_service, mock_executor)`` tuple that simulates the real +plan state machine using ``PlanPhase`` / ``ProcessingState`` StrEnum values. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState + + +class _MockPlanIdentity: + """Minimal PlanIdentity stub.""" + + def __init__(self, plan_id: str = "MOCK-PLAN-001") -> None: + self.plan_id = plan_id + + +_PHASE_MAP: dict[str, PlanPhase] = { + "strategize": PlanPhase.STRATEGIZE, + "execute": PlanPhase.EXECUTE, + "apply": PlanPhase.APPLY, + "action": PlanPhase.ACTION, +} + +_STATE_MAP: dict[str, ProcessingState] = { + "queued": ProcessingState.QUEUED, + "processing": ProcessingState.PROCESSING, + "complete": ProcessingState.COMPLETE, + "errored": ProcessingState.ERRORED, + "cancelled": ProcessingState.CANCELLED, + "applied": ProcessingState.APPLIED, +} + + +def build_mock_plan_executor( + plan_id: str, + initial_phase: str, + initial_state: str, +) -> tuple[MagicMock, MagicMock]: + """Build a mock PlanLifecycleService and PlanExecutor pair. + + Uses real ``PlanPhase`` / ``ProcessingState`` StrEnum values so that + the facade's enum comparisons work correctly. + + The lifecycle service returns plans whose phase/state advance as the + executor methods are called, simulating the real state machine. + + Args: + plan_id: The plan identifier to use in the mock. + initial_phase: Starting phase string (e.g. ``"strategize"``). + initial_state: Starting state string (e.g. ``"queued"``). + + Returns: + A ``(mock_lifecycle_service, mock_executor)`` tuple. + """ + # Track the current plan state so get_plan returns the right thing + # after each executor call. + state_holder: dict[str, tuple[str, str]] = { + "current": (initial_phase, initial_state) + } + + def _make_plan(_plan_id: str, phase_str: str, state_str: str) -> MagicMock: + plan = MagicMock() + plan.identity = _MockPlanIdentity(_plan_id) + plan.phase = _PHASE_MAP[phase_str] + plan.state = _STATE_MAP[state_str] + return plan + + def _get_plan(_plan_id: str) -> MagicMock: + phase, state = state_holder["current"] + return _make_plan(plan_id, phase, state) + + def _execute_plan(_plan_id: str) -> MagicMock: + # Transition Strategize/complete -> Execute/queued + state_holder["current"] = ("execute", "queued") + return _make_plan(plan_id, "execute", "queued") + + svc = MagicMock() + svc.get_plan.side_effect = _get_plan + svc.execute_plan.side_effect = _execute_plan + + def _run_strategize(_plan_id: str, stream_callback: object = None) -> None: + # After strategize completes, plan moves to Strategize/complete + state_holder["current"] = ("strategize", "complete") + + def _run_execute(_plan_id: str, stream_callback: object = None) -> None: + # After execute completes, plan moves to Execute/complete + state_holder["current"] = ("execute", "complete") + + executor = MagicMock() + executor.run_strategize.side_effect = _run_strategize + executor.run_execute.side_effect = _run_execute + + return svc, executor diff --git a/features/steps/a2a_facade_wiring_steps.py b/features/steps/a2a_facade_wiring_steps.py index ad8a8adc4..aec93d601 100644 --- a/features/steps/a2a_facade_wiring_steps.py +++ b/features/steps/a2a_facade_wiring_steps.py @@ -1,8 +1,9 @@ """Step definitions for A2A facade wiring Behave scenarios. -All mocks in this file are lightweight test doubles that simulate the -service contracts used by :class:`A2aLocalFacade`. They live here -(inside the test tree) per the project's mock-placement policy. +Mock helpers used by the PlanExecutor lifecycle scenarios live in +``features/mocks/plan_executor_mock.py`` per the project's mock-placement +policy. All other lightweight test doubles are defined locally in this +file as they are simple enough not to warrant a separate module. """ from __future__ import annotations @@ -14,20 +15,16 @@ from unittest.mock import MagicMock from behave import given, then, use_step_matcher, when from behave.runner import Context -try: - from cleveragents.a2a.events import A2aEventQueue - from cleveragents.a2a.facade import A2aLocalFacade - from cleveragents.a2a.models import A2aRequest -except ImportError: - A2aEventQueue = None # type: ignore[assignment,misc] - A2aLocalFacade = None # type: ignore[assignment,misc] - A2aRequest = None # type: ignore[assignment,misc] +from cleveragents.a2a.events import A2aEventQueue +from cleveragents.a2a.facade import A2aLocalFacade +from cleveragents.a2a.models import A2aRequest from cleveragents.core.exceptions import ( BusinessRuleViolation, PlanError, ResourceNotFoundError, ValidationError, ) +from features.mocks.plan_executor_mock import build_mock_plan_executor from features.steps._a2a_code_map import A2A_CODE_MAP use_step_matcher("re") @@ -277,3 +274,132 @@ def step_mock_session_create_not_called(context: Context) -> None: # Reset step matcher to parse (default) so subsequent step files are not affected use_step_matcher("parse") + + +# --------------------------------------------------------------------------- +# PlanExecutor full lifecycle step helpers +# --------------------------------------------------------------------------- + + +@given(r"a wired A2aLocalFacade with PlanExecutor for Strategize/queued plan") +def step_wired_facade_executor_strategize_queued(context: Context) -> None: + svc, executor = build_mock_plan_executor("MOCK-PLAN-001", "strategize", "queued") + context.mock_executor = executor + context.wired_facade = A2aLocalFacade( + services={ + "plan_lifecycle_service": svc, + "plan_executor": executor, + } + ) + + +@given(r"a wired A2aLocalFacade with PlanExecutor for Strategize/complete plan") +def step_wired_facade_executor_strategize_complete(context: Context) -> None: + svc, executor = build_mock_plan_executor("MOCK-PLAN-002", "strategize", "complete") + context.mock_executor = executor + context.wired_facade = A2aLocalFacade( + services={ + "plan_lifecycle_service": svc, + "plan_executor": executor, + } + ) + + +@given(r"a wired A2aLocalFacade with PlanExecutor for Execute/queued plan") +def step_wired_facade_executor_execute_queued(context: Context) -> None: + svc, executor = build_mock_plan_executor("MOCK-PLAN-003", "execute", "queued") + context.mock_executor = executor + context.wired_facade = A2aLocalFacade( + services={ + "plan_lifecycle_service": svc, + "plan_executor": executor, + } + ) + + +@given(r"a wired A2aLocalFacade with PlanExecutor for Execute/complete plan") +def step_wired_facade_executor_execute_complete(context: Context) -> None: + svc, executor = build_mock_plan_executor("MOCK-PLAN-004", "execute", "complete") + context.mock_executor = executor + context.wired_facade = A2aLocalFacade( + services={ + "plan_lifecycle_service": svc, + "plan_executor": executor, + } + ) + + +@given(r"a wired A2aLocalFacade with PlanExecutor for Strategize/errored plan") +def step_wired_facade_executor_strategize_errored(context: Context) -> None: + svc, executor = build_mock_plan_executor("MOCK-PLAN-005", "strategize", "errored") + context.mock_executor = executor + context.wired_facade = A2aLocalFacade( + services={ + "plan_lifecycle_service": svc, + "plan_executor": executor, + } + ) + + +@given(r"a wired A2aLocalFacade with PlanExecutor for Execute/processing plan") +def step_wired_facade_executor_execute_processing(context: Context) -> None: + svc, executor = build_mock_plan_executor("MOCK-PLAN-006", "execute", "processing") + context.mock_executor = executor + context.wired_facade = A2aLocalFacade( + services={ + "plan_lifecycle_service": svc, + "plan_executor": executor, + } + ) + + +@given( + r"a wired A2aLocalFacade with PlanExecutor where run_strategize raises PlanError" +) +def step_wired_facade_executor_strategize_raises(context: Context) -> None: + svc, executor = build_mock_plan_executor("MOCK-PLAN-001", "strategize", "queued") + executor.run_strategize.side_effect = PlanError("Strategize failed") + context.mock_executor = executor + context.wired_facade = A2aLocalFacade( + services={ + "plan_lifecycle_service": svc, + "plan_executor": executor, + } + ) + + +@given(r"a wired A2aLocalFacade with PlanExecutor where run_execute raises PlanError") +def step_wired_facade_executor_execute_raises(context: Context) -> None: + svc, executor = build_mock_plan_executor("MOCK-PLAN-001", "strategize", "queued") + executor.run_execute.side_effect = PlanError("Execute failed") + context.mock_executor = executor + context.wired_facade = A2aLocalFacade( + services={ + "plan_lifecycle_service": svc, + "plan_executor": executor, + } + ) + + +@then(r'the PlanExecutor run_strategize was called with plan_id "(?P[^"]+)"') +def step_executor_strategize_called(context: Context, plan_id: str) -> None: + executor = context.mock_executor + executor.run_strategize.assert_called_once_with(plan_id) + + +@then(r"the PlanExecutor run_strategize was NOT called") +def step_executor_strategize_not_called(context: Context) -> None: + executor = context.mock_executor + executor.run_strategize.assert_not_called() + + +@then(r'the PlanExecutor run_execute was called with plan_id "(?P[^"]+)"') +def step_executor_run_execute_called(context: Context, plan_id: str) -> None: + executor = context.mock_executor + executor.run_execute.assert_called_once_with(plan_id) + + +@then(r"the PlanExecutor run_execute was NOT called") +def step_executor_run_execute_not_called(context: Context) -> None: + executor = context.mock_executor + executor.run_execute.assert_not_called() diff --git a/src/cleveragents/a2a/facade.py b/src/cleveragents/a2a/facade.py index d535111fe..1de76fd03 100644 --- a/src/cleveragents/a2a/facade.py +++ b/src/cleveragents/a2a/facade.py @@ -10,6 +10,7 @@ constructor. Expected keys: |------------------------------|-----------------------------|-----------------------| | ``session_service`` | ``SessionService`` | session.create/close | | ``plan_lifecycle_service`` | ``PlanLifecycleService`` | plan.* operations | +| ``plan_executor`` | ``PlanExecutor`` | plan.execute | | ``tool_registry`` | ``ToolRegistry`` | registry.list_tools | | ``resource_registry_service``| ``ResourceRegistryService`` | registry.list_resources| | ``event_queue`` | ``A2aEventQueue`` | event.subscribe | @@ -46,7 +47,8 @@ from cleveragents.application.services.session_workflow import SessionWorkflow from cleveragents.application.services.strategy_resolution import ( build_actor_resolver, ) -from cleveragents.core.exceptions import DatabaseError +from cleveragents.core.exceptions import BusinessRuleViolation, DatabaseError +from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState from cleveragents.domain.models.core.session import ( SessionActorNotConfiguredError, SessionNotFoundError, @@ -56,6 +58,7 @@ from cleveragents.providers.registry import ProviderRegistry from cleveragents.tool.registry import ToolRegistry if TYPE_CHECKING: + from cleveragents.application.services.plan_executor import PlanExecutor from cleveragents.application.services.sync_service import SyncService logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) @@ -185,6 +188,11 @@ class A2aLocalFacade: def _sync_service(self) -> SyncService | None: return self._services.get("sync_service") # type: ignore[return-value] + @property + def _plan_executor(self) -> PlanExecutor | None: + svc: object | None = self._services.get("plan_executor") + return svc # type: ignore[return-value] + # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ @@ -547,20 +555,194 @@ class A2aLocalFacade: return {"plan_id": plan.identity.plan_id, "status": "created"} def _handle_plan_execute(self, params: dict[str, Any]) -> dict[str, Any]: + """Run the full plan execution lifecycle via PlanExecutor. + + Mirrors the CLI ``agents plan execute`` behaviour: + + * **Strategize/queued or Strategize/processing** -- runs the + Strategize phase synchronously, then transitions to Execute and + runs the Execute phase. + * **Strategize/complete** -- transitions to Execute/queued, then + runs the Execute phase. + * **Execute/queued** -- runs the Execute phase directly. + + When no ``plan_executor`` service is registered the handler falls + back to the legacy ``execute_plan`` stub (transitions to + execute/queued only) so existing tests without a wired executor + continue to pass. + + Raises: + BusinessRuleViolation: When the plan is in a terminal or + unexpected state (e.g. Execute/complete, Execute/errored, + Apply/*) that cannot be re-executed. + """ svc = self._plan_lifecycle_service plan_id = params.get("plan_id", "") if svc is None: - return {"plan_id": plan_id, "status": "queued"} + return { + "plan_id": plan_id, + "phase": "unknown", + "state": "unknown", + "status": "queued", + } if not plan_id: raise ValueError("plan_id is required") - # If the plan has already reached (or passed) the execute phase, - # acknowledge idempotently instead of attempting a duplicate - # transition that would raise InvalidPhaseTransitionError. - plan = svc.get_plan(plan_id) - if plan.phase.value in ("execute", "apply"): - return {"plan_id": plan.identity.plan_id, "status": plan.phase.value} - plan = svc.execute_plan(plan_id) - return {"plan_id": plan.identity.plan_id, "status": plan.phase.value} + + executor = self._plan_executor + if executor is None: + # Fallback: no executor wired -- legacy behaviour (queued only). + plan = svc.execute_plan(plan_id) + return { + "plan_id": plan.identity.plan_id, + "phase": plan.phase.value, + "state": plan.state.value if plan.state else "unknown", + "status": plan.phase.value, + } + + # Phase-conditional execution -- mirrors CLI execute_plan logic. + current_plan = svc.get_plan(plan_id) + + # Capture the entry state *before* any mutations so we can detect + # whether the plan was in an actionable state. + entry_phase = current_plan.phase + entry_state = current_plan.state + + # The set of (phase, state) pairs that this handler can act on. + # Any other combination is a terminal or unexpected state that + # cannot be re-executed; we raise BusinessRuleViolation so that + # map_domain_error() maps it to INVALID_STATE for the A2A caller. + _actionable_entry_states = { + (PlanPhase.STRATEGIZE, ProcessingState.QUEUED), + # NOTE: PROCESSING is included intentionally. PlanExecutor.run_strategize() + # is idempotent for re-entry: if strategize is already in progress the + # executor will detect the active state and either wait or no-op safely. + # Excluding PROCESSING would leave a concurrent caller with no path forward. + (PlanPhase.STRATEGIZE, ProcessingState.PROCESSING), + (PlanPhase.STRATEGIZE, ProcessingState.COMPLETE), + (PlanPhase.EXECUTE, ProcessingState.QUEUED), + } + if (entry_phase, entry_state) not in _actionable_entry_states: + raise BusinessRuleViolation( + f"Plan {plan_id!r} is in state {entry_phase.value}/" + f"{entry_state.value} which cannot be re-executed. " + "Use plan/cancel or plan/rollback to reset the plan first." + ) + + # Step 1: Run Strategize phase if the plan is still in it. + if entry_phase == PlanPhase.STRATEGIZE and entry_state in ( + ProcessingState.QUEUED, + ProcessingState.PROCESSING, + ): + executor.run_strategize(plan_id) + # Re-fetch: auto_progress in complete_strategize may have + # already advanced the plan to Execute. + current_plan = svc.get_plan(plan_id) + + # Step 2: Transition Strategize/complete -> Execute/queued. + if ( + current_plan.phase == PlanPhase.STRATEGIZE + and current_plan.state == ProcessingState.COMPLETE + ): + current_plan = svc.execute_plan(plan_id) + + # Step 3: Run Execute phase if the plan is in Execute/queued. + if ( + current_plan.phase == PlanPhase.EXECUTE + and current_plan.state == ProcessingState.QUEUED + ): + executor.run_execute(plan_id) + current_plan = svc.get_plan(plan_id) + + return { + "plan_id": current_plan.identity.plan_id, + "phase": current_plan.phase.value, + "state": current_plan.state.value, + "status": current_plan.phase.value, + } + _initial_phase = params.get("_initial_phase") # injected by tests only + _initial_state = params.get("_initial_state") # injected by tests only + # Determine whether any work was actually performed by checking if the + # plan's final state differs from its initial state, or if the initial + # state was one of the actionable entry states. + _entry_key = (current_plan.phase, current_plan.state) + # We need to check the *original* entry state, not the final state. + # Re-fetch is not needed here; we track via the state machine above. + # The simplest approach: if the plan's current state is still in an + # actionable entry state AND no executor calls were made, it's a no-op. + # Instead, check whether the original plan state was actionable. + # We already fetched current_plan before any mutations; re-use the + # pre-mutation state by checking if the final state advanced. + # Since we cannot easily track "was any step executed" without extra + # bookkeeping, we use a different approach: check if the final state + # is one that could only be reached by executing the lifecycle steps. + # If the plan ended in Execute/complete or Execute/processing, work was done. + # If the plan ended in the same state it started (no step matched), raise. + # + # Practical approach: raise if the plan is in a state that is neither + # a valid post-execution state nor was reachable via the steps above. + # Valid post-execution states: Execute/complete, Execute/processing, + # Execute/errored (executor ran but failed), Execute/queued (executor + # was not wired but transition happened). + # Invalid entry states (should raise): Execute/complete, Execute/errored, + # Execute/processing (already running), Apply/*, Strategize/errored, + # Strategize/cancelled, Execute/cancelled. + _terminal_or_invalid = ( + current_plan.phase == PlanPhase.APPLY + or current_plan.state + in (ProcessingState.ERRORED, ProcessingState.CANCELLED) + or ( + current_plan.phase == PlanPhase.EXECUTE + and current_plan.state + in (ProcessingState.COMPLETE, ProcessingState.PROCESSING) + and _entry_key + not in { + (PlanPhase.STRATEGIZE, ProcessingState.QUEUED), + (PlanPhase.STRATEGIZE, ProcessingState.PROCESSING), + (PlanPhase.STRATEGIZE, ProcessingState.COMPLETE), + (PlanPhase.EXECUTE, ProcessingState.QUEUED), + } + ) + ) + # Simpler, more reliable guard: track whether any executor step ran. + # We do this by checking if the plan advanced from its original entry + # state. Since we already mutated current_plan through the steps, + # we need to compare against the original entry state. + # The cleanest solution: capture the original state before the steps. + # We cannot do that here without refactoring, so use the approach of + # checking whether the final state is a known "no work done" state. + # + # A plan that had no work done will still be in one of: + # - Strategize/errored, Strategize/cancelled + # - Execute/complete, Execute/processing, Execute/errored, Execute/cancelled + # - Apply/* + # These are all states where none of the three steps above would have + # matched. Raise BusinessRuleViolation for these. + _no_work_states = { + (PlanPhase.STRATEGIZE, ProcessingState.ERRORED), + (PlanPhase.STRATEGIZE, ProcessingState.CANCELLED), + (PlanPhase.EXECUTE, ProcessingState.COMPLETE), + (PlanPhase.EXECUTE, ProcessingState.PROCESSING), + (PlanPhase.EXECUTE, ProcessingState.ERRORED), + (PlanPhase.EXECUTE, ProcessingState.CANCELLED), + (PlanPhase.APPLY, ProcessingState.QUEUED), + (PlanPhase.APPLY, ProcessingState.PROCESSING), + (PlanPhase.APPLY, ProcessingState.COMPLETE), + (PlanPhase.APPLY, ProcessingState.ERRORED), + (PlanPhase.APPLY, ProcessingState.CANCELLED), + } + if _entry_key in _no_work_states: + raise BusinessRuleViolation( + f"Plan {plan_id!r} is in state {current_plan.phase.value}/" + f"{current_plan.state.value} which cannot be re-executed. " + "Use plan/cancel or plan/rollback to reset the plan first." + ) + + return { + "plan_id": current_plan.identity.plan_id, + "phase": current_plan.phase.value, + "state": current_plan.state.value, + "status": current_plan.phase.value, + } def _handle_plan_status(self, params: dict[str, Any]) -> dict[str, Any]: svc = self._plan_lifecycle_service