From 4ca4874c4d156f5cc8de0ec9e3a620042cf35ab1 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Wed, 4 Mar 2026 10:49:37 +0000 Subject: [PATCH] feat(correction): implement cross-plan correction cascading with child plan state handling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add CrossPlanCorrectionService that implements the four child-plan-state- dependent behaviours from the specification when a correction's affected subtree includes child plans: - Not yet started → cancel the child plan - In progress → cancel + rollback sandbox to pre-child-plan state - Completed but not applied → cancel + rollback sandbox - Already applied → reject the correction (CorrectionRejection) Key additions: - ChildPlanState enum classifying child plans into 4 states - CorrectionRejection result type with reason and affected applied plan IDs - CascadeAction/CascadeResult models for cascade operation tracking - CorrectionStatus.REJECTED for rejected corrections - Atomic cascade-or-rollback: all child plan actions succeed or the entire cascade is rolled back - Protocol-based dependency injection (ChildPlanLookup, ChildPlanCanceller, SandboxRollbacker) for testability - execute_correction_with_cascade() integrates with CorrectionService flow Testing: - 24 Behave BDD scenarios in cross_plan_correction.feature - 8 Robot Framework end-to-end smoke tests - ASV benchmarks for cascade performance with varying child plan counts ISSUES CLOSED: #547 --- benchmarks/cross_plan_correction_bench.py | 158 ++++++ features/cross_plan_correction.feature | 303 ++++++++++++ .../cross_plan_correction_coverage_steps.py | 208 ++++++++ features/steps/cross_plan_correction_steps.py | 454 ++++++++++++++++++ robot/cross_plan_correction.robot | 66 +++ robot/helper_cross_plan_correction.py | 217 +++++++++ .../application/services/__init__.py | 4 + .../services/cross_plan_correction_service.py | 426 ++++++++++++++++ .../domain/models/core/correction.py | 116 +++++ 9 files changed, 1952 insertions(+) create mode 100644 benchmarks/cross_plan_correction_bench.py create mode 100644 features/cross_plan_correction.feature create mode 100644 features/steps/cross_plan_correction_coverage_steps.py create mode 100644 features/steps/cross_plan_correction_steps.py create mode 100644 robot/cross_plan_correction.robot create mode 100644 robot/helper_cross_plan_correction.py create mode 100644 src/cleveragents/application/services/cross_plan_correction_service.py diff --git a/benchmarks/cross_plan_correction_bench.py b/benchmarks/cross_plan_correction_bench.py new file mode 100644 index 000000000..ec8625706 --- /dev/null +++ b/benchmarks/cross_plan_correction_bench.py @@ -0,0 +1,158 @@ +"""ASV benchmarks for cross-plan correction cascading performance. + +Measures cascade throughput with varying numbers of child plans in +different states: cancel-only, cancel-with-rollback, and mixed states. +""" + +from __future__ import annotations + +import importlib +import sys +from pathlib import Path + +# Ensure the local *source* tree is importable even when ASV has an +# older build of the package installed. +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +import cleveragents # noqa: E402 + +importlib.reload(cleveragents) + +from cleveragents.application.services.cross_plan_correction_service import ( # noqa: E402 + CrossPlanCorrectionService, +) +from cleveragents.domain.models.core.correction import ( # noqa: E402 + ChildPlanState, + CorrectionImpact, + CorrectionMode, +) + + +# --------------------------------------------------------------------------- +# Mock implementations (minimal, for benchmarking) +# --------------------------------------------------------------------------- + + +class _BenchPlanLookup: + def __init__(self, states: dict[str, ChildPlanState]) -> None: + self._states = states + + def get_child_plan_state(self, child_plan_id: str) -> ChildPlanState: + return self._states[child_plan_id] + + +class _BenchPlanCanceller: + def cancel_child_plan(self, child_plan_id: str) -> None: + pass + + +class _BenchSandboxRollbacker: + def rollback_child_plan_sandbox(self, child_plan_id: str) -> None: + pass + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_service( + child_count: int, state: ChildPlanState +) -> tuple[CrossPlanCorrectionService, list[str]]: + plan_ids = [f"CP{i}" for i in range(child_count)] + states = {pid: state for pid in plan_ids} + lookup = _BenchPlanLookup(states) + canceller = _BenchPlanCanceller() + rollbacker = _BenchSandboxRollbacker() + svc = CrossPlanCorrectionService( + plan_lookup=lookup, + plan_canceller=canceller, + sandbox_rollbacker=rollbacker, + ) + return svc, plan_ids + + +# --------------------------------------------------------------------------- +# Benchmark suites +# --------------------------------------------------------------------------- + + +class CascadeCancelSuite: + """Benchmark cascade with cancel-only (not-started) child plans.""" + + params = [1, 5, 10, 50, 100] + param_names = ["child_plan_count"] + + def setup(self, child_plan_count: int) -> None: + self.svc, self.plan_ids = _make_service( + child_plan_count, ChildPlanState.NOT_STARTED + ) + + def time_execute_cascade_cancel(self, child_plan_count: int) -> None: + self.svc.execute_cascade("C1", self.plan_ids) + + def time_evaluate_cascade_cancel(self, child_plan_count: int) -> None: + self.svc.evaluate_cascade("C1", self.plan_ids) + + +class CascadeRollbackSuite: + """Benchmark cascade with cancel+rollback (in-progress) child plans.""" + + params = [1, 5, 10, 50, 100] + param_names = ["child_plan_count"] + + def setup(self, child_plan_count: int) -> None: + self.svc, self.plan_ids = _make_service( + child_plan_count, ChildPlanState.IN_PROGRESS + ) + + def time_execute_cascade_rollback(self, child_plan_count: int) -> None: + self.svc.execute_cascade("C1", self.plan_ids) + + +class CascadeRejectionSuite: + """Benchmark cascade rejection (applied child plans).""" + + params = [1, 5, 10, 50, 100] + param_names = ["child_plan_count"] + + def setup(self, child_plan_count: int) -> None: + self.svc, self.plan_ids = _make_service( + child_plan_count, ChildPlanState.APPLIED + ) + + def time_execute_cascade_rejection(self, child_plan_count: int) -> None: + self.svc.execute_cascade("C1", self.plan_ids) + + +class CorrectionWithCascadeSuite: + """Benchmark full execute_correction_with_cascade flow.""" + + params = [0, 5, 20, 50] + param_names = ["child_plan_count"] + + def setup(self, child_plan_count: int) -> None: + plan_ids = [f"CP{i}" for i in range(child_plan_count)] + states = {pid: ChildPlanState.NOT_STARTED for pid in plan_ids} + lookup = _BenchPlanLookup(states) + canceller = _BenchPlanCanceller() + rollbacker = _BenchSandboxRollbacker() + self.svc = CrossPlanCorrectionService( + plan_lookup=lookup, + plan_canceller=canceller, + sandbox_rollbacker=rollbacker, + ) + self.impact = CorrectionImpact( + affected_decisions=["D1"], + affected_child_plans=plan_ids, + risk_level="low", + rollback_tier="full", + artifacts_to_archive=["D1.artifact"], + ) + + def time_correction_with_cascade(self, child_plan_count: int) -> None: + self.svc.execute_correction_with_cascade( + "C1", self.impact, CorrectionMode.REVERT + ) diff --git a/features/cross_plan_correction.feature b/features/cross_plan_correction.feature new file mode 100644 index 000000000..36f50ac82 --- /dev/null +++ b/features/cross_plan_correction.feature @@ -0,0 +1,303 @@ +Feature: Cross-plan correction cascading + When a correction's affected subtree includes child plans, the child + plan's state determines the cascade behaviour: cancel, cancel with + sandbox rollback, or reject the correction entirely. + + # ───────────────────────────────────────────────────────────────── + # Not-yet-started child plans → cancel only + # ───────────────────────────────────────────────────────────────── + + Scenario: Correction cascades to not-started child plan — cancel + Given a cross-plan correction service + And the child plan "CP1" has state "not_started" + When I execute a cascade correction for correction "C1" with child plans "CP1" + Then the cascade should succeed + And the cancelled child plans should contain "CP1" + And the rolled back child plans should be empty + + # ───────────────────────────────────────────────────────────────── + # In-progress child plans → cancel + sandbox rollback + # ───────────────────────────────────────────────────────────────── + + Scenario: Correction cascades to in-progress child plan — cancel and rollback + Given a cross-plan correction service + And the child plan "CP1" has state "in_progress" + When I execute a cascade correction for correction "C1" with child plans "CP1" + Then the cascade should succeed + And the cancelled child plans should contain "CP1" + And the rolled back child plans should contain "CP1" + + # ───────────────────────────────────────────────────────────────── + # Completed-unapplied child plans → cancel + rollback + # ───────────────────────────────────────────────────────────────── + + Scenario: Correction cascades to completed-unapplied child plan — cancel and rollback + Given a cross-plan correction service + And the child plan "CP1" has state "completed_unapplied" + When I execute a cascade correction for correction "C1" with child plans "CP1" + Then the cascade should succeed + And the cancelled child plans should contain "CP1" + And the rolled back child plans should contain "CP1" + + # ───────────────────────────────────────────────────────────────── + # Already-applied child plans → rejection + # ───────────────────────────────────────────────────────────────── + + Scenario: Correction cascades to already-applied child plan — rejection with CorrectionRejection + Given a cross-plan correction service + And the child plan "CP1" has state "applied" + When I execute a cascade correction for correction "C1" with child plans "CP1" + Then the cascade should be rejected + And the rejection reason should mention "already applied" + And the rejection should list applied child plan "CP1" + + # ───────────────────────────────────────────────────────────────── + # Mixed child plan states + # ───────────────────────────────────────────────────────────────── + + Scenario: Mixed child plan states — reject if any applied even if others are cancellable + Given a cross-plan correction service + And the child plan "CP1" has state "not_started" + And the child plan "CP2" has state "in_progress" + And the child plan "CP3" has state "applied" + When I execute a cascade correction for correction "C1" with child plans "CP1,CP2,CP3" + Then the cascade should be rejected + And the rejection should list applied child plan "CP3" + And the cancelled child plans should be empty + + # ───────────────────────────────────────────────────────────────── + # Atomic cascade: failure during one cancel rolls back all previous + # ───────────────────────────────────────────────────────────────── + + Scenario: Atomic cascade — failure during one cancel rolls back all previous cancels + Given a cross-plan correction service with failing canceller on "CP2" + And the child plan "CP1" has state "not_started" + And the child plan "CP2" has state "not_started" + When I try to execute a cascade correction for correction "C1" with child plans "CP1,CP2" + Then the cascade should raise an error + And the cascade error should mention "CP2" + + # ───────────────────────────────────────────────────────────────── + # Evaluation (dry-run) scenarios + # ───────────────────────────────────────────────────────────────── + + Scenario: Evaluate cascade without executing — no child plans + Given a cross-plan correction service + When I evaluate a cascade for correction "C1" with no child plans + Then the cascade evaluation should succeed + And the cascade actions should be empty + + Scenario: Evaluate cascade — mixed states preview + Given a cross-plan correction service + And the child plan "CP1" has state "not_started" + And the child plan "CP2" has state "in_progress" + When I evaluate a cascade for correction "C1" with child plans "CP1,CP2" + Then the cascade evaluation should succeed + And the cascade action for "CP1" should be "cancel" + And the cascade action for "CP2" should be "cancel_and_rollback" + + # ───────────────────────────────────────────────────────────────── + # Validation + # ───────────────────────────────────────────────────────────────── + + Scenario: Empty correction_id raises validation error + Given a cross-plan correction service + When I try to evaluate a cascade with empty correction_id + Then a cross-plan validation error should be raised + + # ───────────────────────────────────────────────────────────────── + # Multiple not-started child plans — all cancelled + # ───────────────────────────────────────────────────────────────── + + Scenario: Multiple not-started child plans all cancelled + Given a cross-plan correction service + And the child plan "CP1" has state "not_started" + And the child plan "CP2" has state "not_started" + And the child plan "CP3" has state "not_started" + When I execute a cascade correction for correction "C1" with child plans "CP1,CP2,CP3" + Then the cascade should succeed + And the cancelled child plans should contain "CP1" + And the cancelled child plans should contain "CP2" + And the cancelled child plans should contain "CP3" + And the rolled back child plans should be empty + + # ───────────────────────────────────────────────────────────────── + # Integration with CorrectionImpact — execute_correction_with_cascade + # ───────────────────────────────────────────────────────────────── + + Scenario: Execute correction with cascade — no child plans returns result + Given a cross-plan correction service + When I execute a correction with cascade for correction "C1" with no child plans in revert mode + Then the correction with cascade result should be applied + + Scenario: Execute correction with cascade — applied child plan returns rejection + Given a cross-plan correction service + And the child plan "CP1" has state "applied" + When I execute a correction with cascade for correction "C1" with child plan "CP1" in revert mode + Then the correction with cascade result should be a rejection + + Scenario: Execute correction with cascade — cancellable child plan succeeds + Given a cross-plan correction service + And the child plan "CP1" has state "not_started" + When I execute a correction with cascade for correction "C1" with child plan "CP1" in revert mode + Then the correction with cascade result should be applied + + # ───────────────────────────────────────────────────────────────── + # classify_cascade_action unit-level checks + # ───────────────────────────────────────────────────────────────── + + Scenario: classify_cascade_action for not_started returns cancel + When I classify cascade action for state "not_started" + Then the classified action should be "cancel" + + Scenario: classify_cascade_action for in_progress returns cancel_and_rollback + When I classify cascade action for state "in_progress" + Then the classified action should be "cancel_and_rollback" + + Scenario: classify_cascade_action for completed_unapplied returns cancel_and_rollback + When I classify cascade action for state "completed_unapplied" + Then the classified action should be "cancel_and_rollback" + + Scenario: classify_cascade_action for applied returns reject + When I classify cascade action for state "applied" + Then the classified action should be "reject" + + # ───────────────────────────────────────────────────────────────── + # Constructor validation + # ───────────────────────────────────────────────────────────────── + + Scenario: Constructor rejects None plan_lookup + When I try to create a cross-plan service with None plan_lookup + Then a cross-plan validation error should be raised + + Scenario: Constructor rejects None plan_canceller + When I try to create a cross-plan service with None plan_canceller + Then a cross-plan validation error should be raised + + Scenario: Constructor rejects None sandbox_rollbacker + When I try to create a cross-plan service with None sandbox_rollbacker + Then a cross-plan validation error should be raised + + # ───────────────────────────────────────────────────────────────── + # CorrectionRejection model validation + # ───────────────────────────────────────────────────────────────── + + Scenario: CorrectionRejection model has correct fields + When I create a CorrectionRejection with correction_id "C1" and reason "test" and plans "CP1,CP2" + Then the rejection correction_id should be "C1" + And the rejection reason should be "test" + And the rejection affected plans should contain "CP1" + And the rejection affected plans should contain "CP2" + + # ───────────────────────────────────────────────────────────────── + # CascadeAction model validation + # ───────────────────────────────────────────────────────────────── + + Scenario: CascadeAction with invalid action raises error + When I try to create a CascadeAction with invalid action "destroy" + Then a cross-plan pydantic validation error should be raised + + # ───────────────────────────────────────────────────────────────── + # CorrectionStatus REJECTED + # ───────────────────────────────────────────────────────────────── + + Scenario: CorrectionStatus includes REJECTED + Then the CorrectionStatus enum should include "rejected" + + # ───────────────────────────────────────────────────────────────── + # ChildPlanState enum completeness + # ───────────────────────────────────────────────────────────────── + + Scenario: ChildPlanState enum has all four states + Then the ChildPlanState enum should have 4 members + And the ChildPlanState enum should include "not_started" + And the ChildPlanState enum should include "in_progress" + And the ChildPlanState enum should include "completed_unapplied" + And the ChildPlanState enum should include "applied" + + # ───────────────────────────────────────────────────────────────── + # Validation — execute_cascade with empty correction_id + # ───────────────────────────────────────────────────────────────── + + Scenario: execute_cascade with empty correction_id raises validation error + Given a cross-plan correction service + When I try to execute a cascade with empty correction_id + Then a cross-plan validation error should be raised + + # ───────────────────────────────────────────────────────────────── + # Validation — execute_correction_with_cascade with empty id + # ───────────────────────────────────────────────────────────────── + + Scenario: execute_correction_with_cascade with empty correction_id raises validation error + Given a cross-plan correction service + When I try to execute a correction with cascade with empty correction_id + Then a cross-plan validation error should be raised + + # ───────────────────────────────────────────────────────────────── + # Append mode — no child plans + # ───────────────────────────────────────────────────────────────── + + Scenario: Execute correction with cascade in append mode — no child plans returns result with empty lists + Given a cross-plan correction service + When I execute a correction with cascade for correction "C1" with no child plans in append mode + Then the correction with cascade result should be applied + And the correction result reverted decisions should be empty + And the correction result archived artifacts should be empty + + # ───────────────────────────────────────────────────────────────── + # Append mode — cancellable child plan + # ───────────────────────────────────────────────────────────────── + + Scenario: Execute correction with cascade in append mode — cancellable child plan succeeds with empty lists + Given a cross-plan correction service + And the child plan "CP1" has state "not_started" + When I execute a correction with cascade for correction "C1" with child plan "CP1" in append mode + Then the correction with cascade result should be applied + And the correction result reverted decisions should be empty + And the correction result archived artifacts should be empty + + # ───────────────────────────────────────────────────────────────── + # CorrectionImpact model validation — invalid rollback_tier + # ───────────────────────────────────────────────────────────────── + + Scenario: CorrectionImpact rejects invalid rollback_tier + When I try to create a CorrectionImpact with rollback_tier "invalid_tier" + Then a cross-plan pydantic validation error should be raised + + # ───────────────────────────────────────────────────────────────── + # CorrectionImpact model validation — invalid risk_level + # ───────────────────────────────────────────────────────────────── + + Scenario: CorrectionImpact rejects invalid risk_level + When I try to create a CorrectionImpact with risk_level "catastrophic" + Then a cross-plan pydantic validation error should be raised + + # ───────────────────────────────────────────────────────────────── + # Protocol runtime checks + # ───────────────────────────────────────────────────────────────── + + Scenario: Mock plan lookup satisfies ChildPlanLookup protocol + Given a cross-plan correction service + Then the plan lookup should satisfy the ChildPlanLookup protocol + + Scenario: Mock plan canceller satisfies ChildPlanCanceller protocol + Given a cross-plan correction service + Then the plan canceller should satisfy the ChildPlanCanceller protocol + + Scenario: Mock sandbox rollbacker satisfies SandboxRollbacker protocol + Given a cross-plan correction service + Then the sandbox rollbacker should satisfy the SandboxRollbacker protocol + + # ───────────────────────────────────────────────────────────────── + # Whitespace-only correction_id treated as empty + # ───────────────────────────────────────────────────────────────── + + Scenario: execute_cascade with whitespace-only correction_id raises validation error + Given a cross-plan correction service + When I try to execute a cascade with whitespace-only correction_id + Then a cross-plan validation error should be raised + + Scenario: execute_correction_with_cascade with whitespace-only correction_id raises validation error + Given a cross-plan correction service + When I try to execute a correction with cascade with whitespace-only correction_id + Then a cross-plan validation error should be raised diff --git a/features/steps/cross_plan_correction_coverage_steps.py b/features/steps/cross_plan_correction_coverage_steps.py new file mode 100644 index 000000000..1efc326b7 --- /dev/null +++ b/features/steps/cross_plan_correction_coverage_steps.py @@ -0,0 +1,208 @@ +"""Additional step definitions for cross_plan_correction.feature coverage. + +Covers validation edge cases, append-mode flows, protocol isinstance +checks, and model validator error paths that the primary step file +does not exercise. +""" + +from __future__ import annotations + +from behave import then, when +from pydantic import ValidationError as PydanticValidationError + +from cleveragents.application.services.cross_plan_correction_service import ( + ChildPlanCanceller, + ChildPlanLookup, + SandboxRollbacker, +) +from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.core.correction import ( + CorrectionImpact, + CorrectionMode, + CorrectionResult, +) + +# ------------------------------------------------------------------- +# Validation — empty / whitespace correction_id for execute_cascade +# ------------------------------------------------------------------- + + +@when("I try to execute a cascade with empty correction_id") +def step_execute_cascade_empty_cid(context: object) -> None: + try: + context.service.execute_cascade("", []) # type: ignore[attr-defined] + context.error = None # type: ignore[attr-defined] + except ValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +@when("I try to execute a cascade with whitespace-only correction_id") +def step_execute_cascade_whitespace_cid(context: object) -> None: + try: + context.service.execute_cascade(" ", []) # type: ignore[attr-defined] + context.error = None # type: ignore[attr-defined] + except ValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +# ------------------------------------------------------------------- +# Validation — empty / whitespace correction_id for +# execute_correction_with_cascade +# ------------------------------------------------------------------- + + +@when("I try to execute a correction with cascade with empty correction_id") +def step_execute_correction_cascade_empty_cid(context: object) -> None: + impact = CorrectionImpact( + affected_decisions=[], + affected_child_plans=[], + risk_level="low", + rollback_tier="full", + ) + try: + context.service.execute_correction_with_cascade( # type: ignore[attr-defined] + "", impact, CorrectionMode.REVERT + ) + context.error = None # type: ignore[attr-defined] + except ValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +@when("I try to execute a correction with cascade with whitespace-only correction_id") +def step_execute_correction_cascade_whitespace_cid(context: object) -> None: + impact = CorrectionImpact( + affected_decisions=[], + affected_child_plans=[], + risk_level="low", + rollback_tier="full", + ) + try: + context.service.execute_correction_with_cascade( # type: ignore[attr-defined] + " ", impact, CorrectionMode.REVERT + ) + context.error = None # type: ignore[attr-defined] + except ValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +# ------------------------------------------------------------------- +# Append-mode steps +# ------------------------------------------------------------------- + + +@when( + 'I execute a correction with cascade for correction "{cid}" ' + "with no child plans in append mode" +) +def step_execute_correction_cascade_no_children_append( + context: object, cid: str +) -> None: + impact = CorrectionImpact( + affected_decisions=["D1", "D2"], + affected_child_plans=[], + risk_level="low", + rollback_tier="full", + artifacts_to_archive=["D1.artifact", "D2.artifact"], + ) + context.correction_with_cascade_result = ( # type: ignore[attr-defined] + context.service.execute_correction_with_cascade( # type: ignore[attr-defined] + cid, impact, CorrectionMode.APPEND + ) + ) + + +@when( + 'I execute a correction with cascade for correction "{cid}" ' + 'with child plan "{plan_id}" in append mode' +) +def step_execute_correction_cascade_with_child_append( + context: object, cid: str, plan_id: str +) -> None: + impact = CorrectionImpact( + affected_decisions=["D1"], + affected_child_plans=[plan_id], + risk_level="low", + rollback_tier="full", + artifacts_to_archive=["D1.artifact"], + ) + context.correction_with_cascade_result = ( # type: ignore[attr-defined] + context.service.execute_correction_with_cascade( # type: ignore[attr-defined] + cid, impact, CorrectionMode.APPEND + ) + ) + + +# ------------------------------------------------------------------- +# Result assertion steps +# ------------------------------------------------------------------- + + +@then("the correction result reverted decisions should be empty") +def step_correction_result_reverted_empty(context: object) -> None: + result = context.correction_with_cascade_result # type: ignore[attr-defined] + assert isinstance(result, CorrectionResult), ( + f"Expected CorrectionResult, got {type(result)}" + ) + assert len(result.reverted_decisions) == 0, ( + f"Expected empty reverted_decisions, got {result.reverted_decisions}" + ) + + +@then("the correction result archived artifacts should be empty") +def step_correction_result_archived_empty(context: object) -> None: + result = context.correction_with_cascade_result # type: ignore[attr-defined] + assert isinstance(result, CorrectionResult), ( + f"Expected CorrectionResult, got {type(result)}" + ) + assert len(result.archived_artifacts) == 0, ( + f"Expected empty archived_artifacts, got {result.archived_artifacts}" + ) + + +# ------------------------------------------------------------------- +# Model validation steps +# ------------------------------------------------------------------- + + +@when('I try to create a CorrectionImpact with rollback_tier "{tier}"') +def step_create_bad_rollback_tier(context: object, tier: str) -> None: + try: + CorrectionImpact(rollback_tier=tier) + context.error = None # type: ignore[attr-defined] + except PydanticValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +@when('I try to create a CorrectionImpact with risk_level "{level}"') +def step_create_bad_risk_level(context: object, level: str) -> None: + try: + CorrectionImpact(risk_level=level) + context.error = None # type: ignore[attr-defined] + except PydanticValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +# ------------------------------------------------------------------- +# Protocol isinstance checks +# ------------------------------------------------------------------- + + +@then("the plan lookup should satisfy the ChildPlanLookup protocol") +def step_lookup_protocol(context: object) -> None: + assert isinstance(context.lookup, ChildPlanLookup), ( # type: ignore[attr-defined] + "Mock does not satisfy ChildPlanLookup protocol" + ) + + +@then("the plan canceller should satisfy the ChildPlanCanceller protocol") +def step_canceller_protocol(context: object) -> None: + assert isinstance(context.canceller, ChildPlanCanceller), ( # type: ignore[attr-defined] + "Mock does not satisfy ChildPlanCanceller protocol" + ) + + +@then("the sandbox rollbacker should satisfy the SandboxRollbacker protocol") +def step_rollbacker_protocol(context: object) -> None: + assert isinstance(context.rollbacker, SandboxRollbacker), ( # type: ignore[attr-defined] + "Mock does not satisfy SandboxRollbacker protocol" + ) diff --git a/features/steps/cross_plan_correction_steps.py b/features/steps/cross_plan_correction_steps.py new file mode 100644 index 000000000..f68620e1c --- /dev/null +++ b/features/steps/cross_plan_correction_steps.py @@ -0,0 +1,454 @@ +"""Step definitions for cross_plan_correction.feature. + +Exercises CrossPlanCorrectionService cascade flows, rejection logic, +atomic rollback, model validation, and helper function behaviour. +""" + +from __future__ import annotations + +from behave import given, then, when +from pydantic import ValidationError as PydanticValidationError + +from cleveragents.application.services.cross_plan_correction_service import ( + CrossPlanCorrectionService, + classify_cascade_action, +) +from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.core.correction import ( + CascadeAction, + CascadeResult, + ChildPlanState, + CorrectionImpact, + CorrectionMode, + CorrectionRejection, + CorrectionResult, + CorrectionStatus, +) + +# ------------------------------------------------------------------- +# Mock implementations (in-step, minimal — no external mocks needed) +# ------------------------------------------------------------------- + + +class _MockPlanLookup: + """In-memory child plan state lookup.""" + + def __init__(self) -> None: + self._states: dict[str, ChildPlanState] = {} + + def set_state(self, plan_id: str, state: ChildPlanState) -> None: + self._states[plan_id] = state + + def get_child_plan_state(self, child_plan_id: str) -> ChildPlanState: + if child_plan_id not in self._states: + raise KeyError(f"Unknown child plan: {child_plan_id}") + return self._states[child_plan_id] + + +class _MockPlanCanceller: + """In-memory child plan canceller.""" + + def __init__(self, fail_on: str | None = None) -> None: + self.cancelled: list[str] = [] + self._fail_on = fail_on + + def cancel_child_plan(self, child_plan_id: str) -> None: + if self._fail_on and child_plan_id == self._fail_on: + raise RuntimeError(f"Cancel failed for {child_plan_id}") + self.cancelled.append(child_plan_id) + + +class _MockSandboxRollbacker: + """In-memory sandbox rollbacker.""" + + def __init__(self) -> None: + self.rolled_back: list[str] = [] + + def rollback_child_plan_sandbox(self, child_plan_id: str) -> None: + self.rolled_back.append(child_plan_id) + + +# ------------------------------------------------------------------- +# Given steps +# ------------------------------------------------------------------- + + +@given("a cross-plan correction service") +def step_create_cross_plan_service(context: object) -> None: + lookup = _MockPlanLookup() + canceller = _MockPlanCanceller() + rollbacker = _MockSandboxRollbacker() + context.lookup = lookup # type: ignore[attr-defined] + context.canceller = canceller # type: ignore[attr-defined] + context.rollbacker = rollbacker # type: ignore[attr-defined] + context.service = CrossPlanCorrectionService( # type: ignore[attr-defined] + plan_lookup=lookup, + plan_canceller=canceller, + sandbox_rollbacker=rollbacker, + ) + context.cascade_result = None # type: ignore[attr-defined] + context.cascade_error = None # type: ignore[attr-defined] + context.error = None # type: ignore[attr-defined] + context.correction_with_cascade_result = None # type: ignore[attr-defined] + context.classified_action = None # type: ignore[attr-defined] + + +@given('a cross-plan correction service with failing canceller on "{fail_id}"') +def step_create_service_failing_canceller(context: object, fail_id: str) -> None: + lookup = _MockPlanLookup() + canceller = _MockPlanCanceller(fail_on=fail_id) + rollbacker = _MockSandboxRollbacker() + context.lookup = lookup # type: ignore[attr-defined] + context.canceller = canceller # type: ignore[attr-defined] + context.rollbacker = rollbacker # type: ignore[attr-defined] + context.service = CrossPlanCorrectionService( # type: ignore[attr-defined] + plan_lookup=lookup, + plan_canceller=canceller, + sandbox_rollbacker=rollbacker, + ) + context.cascade_result = None # type: ignore[attr-defined] + context.cascade_error = None # type: ignore[attr-defined] + context.error = None # type: ignore[attr-defined] + + +@given('the child plan "{plan_id}" has state "{state}"') +def step_set_child_plan_state(context: object, plan_id: str, state: str) -> None: + context.lookup.set_state(plan_id, ChildPlanState(state)) # type: ignore[attr-defined] + + +# ------------------------------------------------------------------- +# When steps +# ------------------------------------------------------------------- + + +@when( + 'I execute a cascade correction for correction "{cid}" with child plans "{plans}"' +) +def step_execute_cascade(context: object, cid: str, plans: str) -> None: + plan_ids = [p.strip() for p in plans.split(",")] + context.cascade_result = context.service.execute_cascade(cid, plan_ids) # type: ignore[attr-defined] + + +@when( + 'I try to execute a cascade correction for correction "{cid}" with child plans "{plans}"' +) +def step_try_execute_cascade(context: object, cid: str, plans: str) -> None: + plan_ids = [p.strip() for p in plans.split(",")] + try: + context.cascade_result = context.service.execute_cascade(cid, plan_ids) # type: ignore[attr-defined] + context.cascade_error = None # type: ignore[attr-defined] + except Exception as exc: + context.cascade_error = exc # type: ignore[attr-defined] + + +@when('I evaluate a cascade for correction "{cid}" with no child plans') +def step_evaluate_cascade_empty(context: object, cid: str) -> None: + context.cascade_result = context.service.evaluate_cascade(cid, []) # type: ignore[attr-defined] + + +@when('I evaluate a cascade for correction "{cid}" with child plans "{plans}"') +def step_evaluate_cascade(context: object, cid: str, plans: str) -> None: + plan_ids = [p.strip() for p in plans.split(",")] + context.cascade_result = context.service.evaluate_cascade(cid, plan_ids) # type: ignore[attr-defined] + + +@when("I try to evaluate a cascade with empty correction_id") +def step_evaluate_empty_cid(context: object) -> None: + try: + context.service.evaluate_cascade("", []) # type: ignore[attr-defined] + context.error = None # type: ignore[attr-defined] + except ValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +@when('I classify cascade action for state "{state}"') +def step_classify_action(context: object, state: str) -> None: + context.classified_action = classify_cascade_action(ChildPlanState(state)) # type: ignore[attr-defined] + + +@when("I try to create a cross-plan service with None plan_lookup") +def step_create_none_lookup(context: object) -> None: + try: + CrossPlanCorrectionService( + plan_lookup=None, # type: ignore[arg-type] + plan_canceller=_MockPlanCanceller(), + sandbox_rollbacker=_MockSandboxRollbacker(), + ) + context.error = None # type: ignore[attr-defined] + except ValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +@when("I try to create a cross-plan service with None plan_canceller") +def step_create_none_canceller(context: object) -> None: + try: + CrossPlanCorrectionService( + plan_lookup=_MockPlanLookup(), + plan_canceller=None, # type: ignore[arg-type] + sandbox_rollbacker=_MockSandboxRollbacker(), + ) + context.error = None # type: ignore[attr-defined] + except ValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +@when("I try to create a cross-plan service with None sandbox_rollbacker") +def step_create_none_rollbacker(context: object) -> None: + try: + CrossPlanCorrectionService( + plan_lookup=_MockPlanLookup(), + plan_canceller=_MockPlanCanceller(), + sandbox_rollbacker=None, # type: ignore[arg-type] + ) + context.error = None # type: ignore[attr-defined] + except ValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +@when( + 'I create a CorrectionRejection with correction_id "{cid}" ' + 'and reason "{reason}" and plans "{plans}"' +) +def step_create_rejection_model( + context: object, cid: str, reason: str, plans: str +) -> None: + plan_ids = [p.strip() for p in plans.split(",")] + context.rejection = CorrectionRejection( # type: ignore[attr-defined] + correction_id=cid, + reason=reason, + affected_applied_child_plan_ids=plan_ids, + ) + + +@when('I try to create a CascadeAction with invalid action "{action}"') +def step_create_bad_cascade_action(context: object, action: str) -> None: + try: + CascadeAction( + child_plan_id="CP1", + child_plan_state=ChildPlanState.NOT_STARTED, + action=action, + ) + context.error = None # type: ignore[attr-defined] + except PydanticValidationError as exc: + context.error = exc # type: ignore[attr-defined] + + +@when( + 'I execute a correction with cascade for correction "{cid}" ' + "with no child plans in revert mode" +) +def step_execute_correction_cascade_no_children(context: object, cid: str) -> None: + impact = CorrectionImpact( + affected_decisions=["D1", "D2"], + affected_child_plans=[], + risk_level="low", + rollback_tier="full", + artifacts_to_archive=["D1.artifact", "D2.artifact"], + ) + context.correction_with_cascade_result = ( # type: ignore[attr-defined] + context.service.execute_correction_with_cascade( # type: ignore[attr-defined] + cid, impact, CorrectionMode.REVERT + ) + ) + + +@when( + 'I execute a correction with cascade for correction "{cid}" ' + 'with child plan "{plan_id}" in revert mode' +) +def step_execute_correction_cascade_with_child( + context: object, cid: str, plan_id: str +) -> None: + impact = CorrectionImpact( + affected_decisions=["D1"], + affected_child_plans=[plan_id], + risk_level="low", + rollback_tier="full", + artifacts_to_archive=["D1.artifact"], + ) + context.correction_with_cascade_result = ( # type: ignore[attr-defined] + context.service.execute_correction_with_cascade( # type: ignore[attr-defined] + cid, impact, CorrectionMode.REVERT + ) + ) + + +# ------------------------------------------------------------------- +# Then steps +# ------------------------------------------------------------------- + + +@then("the cascade should succeed") +def step_cascade_success(context: object) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert result is not None, "No cascade result" + assert not result.rejected, f"Cascade was rejected: {result.rejection}" + + +@then("the cascade should be rejected") +def step_cascade_rejected(context: object) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert result is not None, "No cascade result" + assert result.rejected, "Cascade was not rejected" + + +@then('the cancelled child plans should contain "{plan_id}"') +def step_cancelled_contains(context: object, plan_id: str) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert plan_id in result.all_cancelled_plan_ids, ( + f"'{plan_id}' not in cancelled: {result.all_cancelled_plan_ids}" + ) + + +@then("the cancelled child plans should be empty") +def step_cancelled_empty(context: object) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert len(result.all_cancelled_plan_ids) == 0, ( + f"Expected empty cancelled list, got {result.all_cancelled_plan_ids}" + ) + + +@then('the rolled back child plans should contain "{plan_id}"') +def step_rolled_back_contains(context: object, plan_id: str) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert plan_id in result.all_rolled_back_plan_ids, ( + f"'{plan_id}' not in rolled back: {result.all_rolled_back_plan_ids}" + ) + + +@then("the rolled back child plans should be empty") +def step_rolled_back_empty(context: object) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert len(result.all_rolled_back_plan_ids) == 0, ( + f"Expected empty rolled back list, got {result.all_rolled_back_plan_ids}" + ) + + +@then('the rejection reason should mention "{fragment}"') +def step_rejection_reason_contains(context: object, fragment: str) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert result.rejection is not None, "No rejection" + assert fragment in result.rejection.reason, ( + f"'{fragment}' not in reason: {result.rejection.reason}" + ) + + +@then('the rejection should list applied child plan "{plan_id}"') +def step_rejection_lists_plan(context: object, plan_id: str) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert result.rejection is not None, "No rejection" + assert plan_id in result.rejection.affected_applied_child_plan_ids, ( + f"'{plan_id}' not in applied IDs: " + f"{result.rejection.affected_applied_child_plan_ids}" + ) + + +@then("the cascade should raise an error") +def step_cascade_error(context: object) -> None: + assert context.cascade_error is not None, "Expected error but none raised" # type: ignore[attr-defined] + + +@then('the cascade error should mention "{fragment}"') +def step_cascade_error_message(context: object, fragment: str) -> None: + assert context.cascade_error is not None, "No error" # type: ignore[attr-defined] + assert fragment in str(context.cascade_error), ( # type: ignore[attr-defined] + f"'{fragment}' not in error: {context.cascade_error}" # type: ignore[attr-defined] + ) + + +@then("the cascade evaluation should succeed") +def step_evaluation_success(context: object) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert result is not None, "No cascade result" + + +@then("the cascade actions should be empty") +def step_actions_empty(context: object) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + assert len(result.cascade_actions) == 0, ( + f"Expected empty actions, got {result.cascade_actions}" + ) + + +@then('the cascade action for "{plan_id}" should be "{action}"') +def step_action_for_plan(context: object, plan_id: str, action: str) -> None: + result: CascadeResult = context.cascade_result # type: ignore[attr-defined] + for ca in result.cascade_actions: + if ca.child_plan_id == plan_id: + assert ca.action == action, ( + f"Expected action '{action}' for {plan_id}, got '{ca.action}'" + ) + return + raise AssertionError(f"No action found for plan '{plan_id}'") + + +@then("a cross-plan validation error should be raised") +def step_cross_plan_validation_error(context: object) -> None: + assert isinstance(context.error, ValidationError), ( # type: ignore[attr-defined] + f"Expected ValidationError, got {type(context.error)}" # type: ignore[attr-defined] + ) + + +@then('the classified action should be "{expected}"') +def step_classified_action(context: object, expected: str) -> None: + assert context.classified_action == expected, ( # type: ignore[attr-defined] + f"Expected '{expected}', got '{context.classified_action}'" # type: ignore[attr-defined] + ) + + +@then('the rejection correction_id should be "{cid}"') +def step_rejection_cid(context: object, cid: str) -> None: + assert context.rejection.correction_id == cid # type: ignore[attr-defined] + + +@then('the rejection reason should be "{reason}"') +def step_rejection_reason(context: object, reason: str) -> None: + assert context.rejection.reason == reason # type: ignore[attr-defined] + + +@then('the rejection affected plans should contain "{plan_id}"') +def step_rejection_plans_contain(context: object, plan_id: str) -> None: + assert plan_id in context.rejection.affected_applied_child_plan_ids # type: ignore[attr-defined] + + +@then("a cross-plan pydantic validation error should be raised") +def step_pydantic_error(context: object) -> None: + assert isinstance(context.error, PydanticValidationError), ( # type: ignore[attr-defined] + f"Expected PydanticValidationError, got {type(context.error)}" # type: ignore[attr-defined] + ) + + +@then('the CorrectionStatus enum should include "{value}"') +def step_status_includes(context: object, value: str) -> None: + values = [s.value for s in CorrectionStatus] + assert value in values, f"'{value}' not in CorrectionStatus: {values}" + + +@then("the ChildPlanState enum should have {count:d} members") +def step_child_state_count(context: object, count: int) -> None: + assert len(ChildPlanState) == count, ( + f"Expected {count} members, got {len(ChildPlanState)}" + ) + + +@then('the ChildPlanState enum should include "{value}"') +def step_child_state_includes(context: object, value: str) -> None: + values = [s.value for s in ChildPlanState] + assert value in values, f"'{value}' not in ChildPlanState: {values}" + + +@then("the correction with cascade result should be applied") +def step_correction_cascade_applied(context: object) -> None: + result = context.correction_with_cascade_result # type: ignore[attr-defined] + assert isinstance(result, CorrectionResult), ( + f"Expected CorrectionResult, got {type(result)}" + ) + assert result.status == CorrectionStatus.APPLIED + + +@then("the correction with cascade result should be a rejection") +def step_correction_cascade_rejection(context: object) -> None: + result = context.correction_with_cascade_result # type: ignore[attr-defined] + assert isinstance(result, CorrectionRejection), ( + f"Expected CorrectionRejection, got {type(result)}" + ) diff --git a/robot/cross_plan_correction.robot b/robot/cross_plan_correction.robot new file mode 100644 index 000000000..1c055dddc --- /dev/null +++ b/robot/cross_plan_correction.robot @@ -0,0 +1,66 @@ +*** Settings *** +Documentation End-to-end tests for cross-plan correction cascading: +... cancel, cancel+rollback, rejection, and atomic cascade. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} robot/helper_cross_plan_correction.py + +*** Test Cases *** +Cancel Not-Started Child Plan + [Documentation] Cascade cancels a not-yet-started child plan + [Tags] phase2 correction cross-plan cancel + ${result}= Run Process ${PYTHON} ${HELPER} cancel-not-started cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} cancel-not-started-ok + +Cancel And Rollback In-Progress Child Plan + [Documentation] Cascade cancels and rolls back an in-progress child plan + [Tags] phase2 correction cross-plan rollback + ${result}= Run Process ${PYTHON} ${HELPER} cancel-rollback-in-progress cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} cancel-rollback-in-progress-ok + +Cancel And Rollback Completed Unapplied Child Plan + [Documentation] Cascade cancels and rolls back a completed-unapplied child plan + [Tags] phase2 correction cross-plan rollback + ${result}= Run Process ${PYTHON} ${HELPER} cancel-rollback-completed cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} cancel-rollback-completed-ok + +Reject Applied Child Plan + [Documentation] Cascade rejects correction when child plan is already applied + [Tags] phase2 correction cross-plan rejection + ${result}= Run Process ${PYTHON} ${HELPER} reject-applied cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} reject-applied-ok + +Mixed States Rejection + [Documentation] Reject correction when any child plan is applied, even if others are cancellable + [Tags] phase2 correction cross-plan rejection mixed + ${result}= Run Process ${PYTHON} ${HELPER} mixed-states-reject cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} mixed-states-reject-ok + +Atomic Cascade Failure Rollback + [Documentation] Failure during cascade rolls back all completed actions + [Tags] phase2 correction cross-plan atomic + ${result}= Run Process ${PYTHON} ${HELPER} atomic-failure cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} atomic-failure-ok + +Execute Correction With Cascade No Children + [Documentation] Correction with cascade and no child plans succeeds normally + [Tags] phase2 correction cross-plan integration + ${result}= Run Process ${PYTHON} ${HELPER} cascade-no-children cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} cascade-no-children-ok + +Execute Correction With Cascade Rejection + [Documentation] Correction with cascade rejects when child plan is applied + [Tags] phase2 correction cross-plan integration rejection + ${result}= Run Process ${PYTHON} ${HELPER} cascade-rejection cwd=${WORKSPACE} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} cascade-rejection-ok diff --git a/robot/helper_cross_plan_correction.py b/robot/helper_cross_plan_correction.py new file mode 100644 index 000000000..7218a77f8 --- /dev/null +++ b/robot/helper_cross_plan_correction.py @@ -0,0 +1,217 @@ +"""Helper script for cross-plan correction Robot Framework smoke tests. + +Exercises CrossPlanCorrectionService cascade flows, rejection logic, +and atomic rollback without requiring persistence infrastructure. +""" + +from __future__ import annotations + +import sys + +from cleveragents.application.services.cross_plan_correction_service import ( + CrossPlanCorrectionService, +) +from cleveragents.domain.models.core.correction import ( + ChildPlanState, + CorrectionImpact, + CorrectionMode, + CorrectionRejection, + CorrectionResult, +) + +__all__: list[str] = [] + + +# --------------------------------------------------------------------------- +# Mock implementations +# --------------------------------------------------------------------------- + + +class MockPlanLookup: + """In-memory child plan state lookup.""" + + def __init__(self) -> None: + self._states: dict[str, ChildPlanState] = {} + + def set_state(self, plan_id: str, state: ChildPlanState) -> None: + self._states[plan_id] = state + + def get_child_plan_state(self, child_plan_id: str) -> ChildPlanState: + return self._states[child_plan_id] + + +class MockPlanCanceller: + """In-memory child plan canceller.""" + + def __init__(self, fail_on: str | None = None) -> None: + self.cancelled: list[str] = [] + self._fail_on = fail_on + + def cancel_child_plan(self, child_plan_id: str) -> None: + if self._fail_on and child_plan_id == self._fail_on: + raise RuntimeError(f"Cancel failed for {child_plan_id}") + self.cancelled.append(child_plan_id) + + +class MockSandboxRollbacker: + """In-memory sandbox rollbacker.""" + + def __init__(self) -> None: + self.rolled_back: list[str] = [] + + def rollback_child_plan_sandbox(self, child_plan_id: str) -> None: + self.rolled_back.append(child_plan_id) + + +def _make_service( + states: dict[str, ChildPlanState], + fail_on: str | None = None, +) -> tuple[ + CrossPlanCorrectionService, MockPlanLookup, MockPlanCanceller, MockSandboxRollbacker +]: + lookup = MockPlanLookup() + for pid, s in states.items(): + lookup.set_state(pid, s) + canceller = MockPlanCanceller(fail_on=fail_on) + rollbacker = MockSandboxRollbacker() + svc = CrossPlanCorrectionService( + plan_lookup=lookup, + plan_canceller=canceller, + sandbox_rollbacker=rollbacker, + ) + return svc, lookup, canceller, rollbacker + + +# --------------------------------------------------------------------------- +# Command handlers +# --------------------------------------------------------------------------- + + +def _cancel_not_started() -> None: + svc, _, _canceller, _rollbacker = _make_service({"CP1": ChildPlanState.NOT_STARTED}) + result = svc.execute_cascade("C1", ["CP1"]) + assert not result.rejected + assert "CP1" in result.all_cancelled_plan_ids + assert len(result.all_rolled_back_plan_ids) == 0 + print("cancel-not-started-ok") + + +def _cancel_rollback_in_progress() -> None: + svc, _, _canceller, _rollbacker = _make_service({"CP1": ChildPlanState.IN_PROGRESS}) + result = svc.execute_cascade("C1", ["CP1"]) + assert not result.rejected + assert "CP1" in result.all_cancelled_plan_ids + assert "CP1" in result.all_rolled_back_plan_ids + print("cancel-rollback-in-progress-ok") + + +def _cancel_rollback_completed() -> None: + svc, _, _, _ = _make_service({"CP1": ChildPlanState.COMPLETED_UNAPPLIED}) + result = svc.execute_cascade("C1", ["CP1"]) + assert not result.rejected + assert "CP1" in result.all_cancelled_plan_ids + assert "CP1" in result.all_rolled_back_plan_ids + print("cancel-rollback-completed-ok") + + +def _reject_applied() -> None: + svc, _, _, _ = _make_service({"CP1": ChildPlanState.APPLIED}) + result = svc.execute_cascade("C1", ["CP1"]) + assert result.rejected + assert result.rejection is not None + assert "CP1" in result.rejection.affected_applied_child_plan_ids + print("reject-applied-ok") + + +def _mixed_states_reject() -> None: + svc, _, _, _ = _make_service( + { + "CP1": ChildPlanState.NOT_STARTED, + "CP2": ChildPlanState.IN_PROGRESS, + "CP3": ChildPlanState.APPLIED, + } + ) + result = svc.execute_cascade("C1", ["CP1", "CP2", "CP3"]) + assert result.rejected + assert result.rejection is not None + assert "CP3" in result.rejection.affected_applied_child_plan_ids + assert len(result.all_cancelled_plan_ids) == 0 + print("mixed-states-reject-ok") + + +def _atomic_failure() -> None: + svc, _, _, _ = _make_service( + {"CP1": ChildPlanState.NOT_STARTED, "CP2": ChildPlanState.NOT_STARTED}, + fail_on="CP2", + ) + try: + svc.execute_cascade("C1", ["CP1", "CP2"]) + raise AssertionError("Expected RuntimeError") + except RuntimeError as exc: + assert "CP2" in str(exc) + print("atomic-failure-ok") + + +def _cascade_no_children() -> None: + svc, _, _, _ = _make_service({}) + impact = CorrectionImpact( + affected_decisions=["D1"], + affected_child_plans=[], + risk_level="low", + rollback_tier="full", + artifacts_to_archive=["D1.artifact"], + ) + result = svc.execute_correction_with_cascade("C1", impact, CorrectionMode.REVERT) + assert isinstance(result, CorrectionResult) + assert result.status == "applied" + print("cascade-no-children-ok") + + +def _cascade_rejection() -> None: + svc, _, _, _ = _make_service({"CP1": ChildPlanState.APPLIED}) + impact = CorrectionImpact( + affected_decisions=["D1"], + affected_child_plans=["CP1"], + risk_level="low", + rollback_tier="full", + artifacts_to_archive=["D1.artifact"], + ) + result = svc.execute_correction_with_cascade("C1", impact, CorrectionMode.REVERT) + assert isinstance(result, CorrectionRejection) + print("cascade-rejection-ok") + + +# --------------------------------------------------------------------------- +# Dispatcher +# --------------------------------------------------------------------------- + +_COMMANDS = { + "cancel-not-started": _cancel_not_started, + "cancel-rollback-in-progress": _cancel_rollback_in_progress, + "cancel-rollback-completed": _cancel_rollback_completed, + "reject-applied": _reject_applied, + "mixed-states-reject": _mixed_states_reject, + "atomic-failure": _atomic_failure, + "cascade-no-children": _cascade_no_children, + "cascade-rejection": _cascade_rejection, +} + + +def main() -> None: + """Dispatch to the requested command.""" + if len(sys.argv) < 2: + print(f"Usage: {sys.argv[0]} ", file=sys.stderr) + sys.exit(1) + + command = sys.argv[1] + handler = _COMMANDS.get(command) + if handler is None: + print(f"Unknown command: {command}", file=sys.stderr) + print(f"Available: {', '.join(sorted(_COMMANDS))}", file=sys.stderr) + sys.exit(1) + + handler() + + +if __name__ == "__main__": + main() diff --git a/src/cleveragents/application/services/__init__.py b/src/cleveragents/application/services/__init__.py index 5eb60d0a5..156ef37b7 100644 --- a/src/cleveragents/application/services/__init__.py +++ b/src/cleveragents/application/services/__init__.py @@ -18,6 +18,9 @@ from cleveragents.application.services.config_service import ( from cleveragents.application.services.correction_service import ( CorrectionService, ) +from cleveragents.application.services.cross_plan_correction_service import ( + CrossPlanCorrectionService, +) from cleveragents.application.services.decision_service import ( DecisionNotFoundError, DecisionService, @@ -156,6 +159,7 @@ __all__ = [ "ContainerUnavailableError", "ContextFragment", "CorrectionService", + "CrossPlanCorrectionService", "DecisionNotFoundError", "DecisionService", "DecompositionConfig", diff --git a/src/cleveragents/application/services/cross_plan_correction_service.py b/src/cleveragents/application/services/cross_plan_correction_service.py new file mode 100644 index 000000000..c9a90ebf2 --- /dev/null +++ b/src/cleveragents/application/services/cross_plan_correction_service.py @@ -0,0 +1,426 @@ +"""Cross-plan correction cascading service. + +Implements the four child-plan-state-dependent behaviours defined in the +specification when a correction's affected subtree includes child plans: + +| Child Plan State | Action | +|---------------------------|-------------------------------------| +| Not yet started | Cancel the child plan | +| In progress | Cancel + rollback sandbox | +| Completed but not applied | Cancel + rollback sandbox | +| Already applied | Reject the correction | + +All cascading actions are **atomic**: either every child plan action +succeeds, or the entire cascade is rolled back to the pre-correction +state. +""" + +from __future__ import annotations + +from typing import Protocol, runtime_checkable + +import structlog + +from cleveragents.core.exceptions import ValidationError +from cleveragents.domain.models.core.correction import ( + CascadeAction, + CascadeResult, + ChildPlanState, + CorrectionImpact, + CorrectionMode, + CorrectionRejection, + CorrectionResult, + CorrectionStatus, +) + +logger = structlog.get_logger(__name__) + + +# --------------------------------------------------------------------------- +# Protocols for dependency injection +# --------------------------------------------------------------------------- + + +@runtime_checkable +class ChildPlanLookup(Protocol): + """Protocol for resolving child plan states.""" + + def get_child_plan_state(self, child_plan_id: str) -> ChildPlanState: + """Return the current state of the given child plan. + + Args: + child_plan_id: The child plan to inspect. + + Returns: + The classified state of the child plan. + + Raises: + KeyError: If the child plan cannot be found. + """ + ... + + +@runtime_checkable +class ChildPlanCanceller(Protocol): + """Protocol for cancelling child plans.""" + + def cancel_child_plan(self, child_plan_id: str) -> None: + """Cancel the specified child plan. + + Args: + child_plan_id: The child plan to cancel. + + Raises: + RuntimeError: If cancellation fails. + """ + ... + + +@runtime_checkable +class SandboxRollbacker(Protocol): + """Protocol for rolling back a child plan's sandbox.""" + + def rollback_child_plan_sandbox(self, child_plan_id: str) -> None: + """Roll back the sandbox associated with a child plan. + + Args: + child_plan_id: The child plan whose sandbox to roll back. + + Raises: + RuntimeError: If rollback fails. + """ + ... + + +# --------------------------------------------------------------------------- +# State classification helper +# --------------------------------------------------------------------------- + +_CANCEL_ONLY_STATES = frozenset({ChildPlanState.NOT_STARTED}) +_CANCEL_AND_ROLLBACK_STATES = frozenset( + { + ChildPlanState.IN_PROGRESS, + ChildPlanState.COMPLETED_UNAPPLIED, + } +) +_BLOCKING_STATES = frozenset({ChildPlanState.APPLIED}) + + +def classify_cascade_action(state: ChildPlanState) -> str: + """Determine what cascade action to take for a given child plan state. + + Args: + state: The observed child plan state. + + Returns: + One of ``'cancel'``, ``'cancel_and_rollback'``, or ``'reject'``. + + Raises: + ValidationError: If the state is unrecognised. + """ + if state in _CANCEL_ONLY_STATES: + return "cancel" + if state in _CANCEL_AND_ROLLBACK_STATES: + return "cancel_and_rollback" + if state in _BLOCKING_STATES: + return "reject" + raise ValidationError(f"Unrecognised child plan state: {state}") + + +# --------------------------------------------------------------------------- +# Service +# --------------------------------------------------------------------------- + + +class CrossPlanCorrectionService: + """Orchestrates cross-plan correction cascading. + + Works alongside ``CorrectionService`` to handle the case where + a correction's affected subtree includes child plans. The caller + first uses ``CorrectionService.analyze_impact`` to obtain an + ``CorrectionImpact`` with ``affected_child_plans``, then delegates + to this service to resolve the child-plan-dependent cascading. + + All cascade operations are atomic: if any action fails mid-way, + previously completed actions are undone before raising. + """ + + def __init__( + self, + plan_lookup: ChildPlanLookup, + plan_canceller: ChildPlanCanceller, + sandbox_rollbacker: SandboxRollbacker, + ) -> None: + if plan_lookup is None: + raise ValidationError("plan_lookup must not be None") + if plan_canceller is None: + raise ValidationError("plan_canceller must not be None") + if sandbox_rollbacker is None: + raise ValidationError("sandbox_rollbacker must not be None") + + self._plan_lookup = plan_lookup + self._plan_canceller = plan_canceller + self._sandbox_rollbacker = sandbox_rollbacker + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def evaluate_cascade( + self, + correction_id: str, + affected_child_plan_ids: list[str], + ) -> CascadeResult: + """Evaluate what cascade actions would be taken, without executing. + + Inspects each child plan's state and determines the action. + If any child plan is already applied, the cascade is marked + as rejected. + + Args: + correction_id: The parent correction request ID. + affected_child_plan_ids: Child plan IDs in the affected subtree. + + Returns: + ``CascadeResult`` describing the planned actions. + + Raises: + ValidationError: If correction_id is empty. + """ + if not correction_id or not correction_id.strip(): + raise ValidationError("correction_id must not be empty") + + actions: list[CascadeAction] = [] + applied_ids: list[str] = [] + + for plan_id in affected_child_plan_ids: + state = self._plan_lookup.get_child_plan_state(plan_id) + action_type = classify_cascade_action(state) + + if state == ChildPlanState.APPLIED: + applied_ids.append(plan_id) + + actions.append( + CascadeAction( + child_plan_id=plan_id, + child_plan_state=state, + action=action_type, + sandbox_rolled_back=action_type == "cancel_and_rollback", + ) + ) + + rejected = len(applied_ids) > 0 + rejection = None + if rejected: + rejection = CorrectionRejection( + correction_id=correction_id, + reason=( + "Correction rejected: child plan(s) already applied. " + "Applied changes cannot be unilaterally reverted. " + "Correct the child plan independently or use " + "--mode=append on the parent." + ), + affected_applied_child_plan_ids=applied_ids, + ) + + logger.info( + "cross_plan_correction.cascade_evaluated", + correction_id=correction_id, + child_plan_count=len(affected_child_plan_ids), + rejected=rejected, + applied_count=len(applied_ids), + ) + + return CascadeResult( + correction_id=correction_id, + cascade_actions=actions, + rejected=rejected, + rejection=rejection, + all_cancelled_plan_ids=[], + all_rolled_back_plan_ids=[], + ) + + def execute_cascade( + self, + correction_id: str, + affected_child_plan_ids: list[str], + ) -> CascadeResult: + """Execute cross-plan correction cascade atomically. + + Evaluates child plan states, then executes the required actions. + If any child plan is already applied, the correction is rejected + without performing any mutations. If a failure occurs mid-cascade, + all completed actions are rolled back. + + Args: + correction_id: The parent correction request ID. + affected_child_plan_ids: Child plan IDs in the affected subtree. + + Returns: + ``CascadeResult`` with executed actions. + + Raises: + ValidationError: If correction_id is empty. + """ + if not correction_id or not correction_id.strip(): + raise ValidationError("correction_id must not be empty") + + evaluation = self.evaluate_cascade(correction_id, affected_child_plan_ids) + + if evaluation.rejected: + logger.warning( + "cross_plan_correction.cascade_rejected", + correction_id=correction_id, + applied_ids=evaluation.rejection.affected_applied_child_plan_ids + if evaluation.rejection + else [], + ) + return evaluation + + cancelled_ids: list[str] = [] + rolled_back_ids: list[str] = [] + executed_actions: list[CascadeAction] = [] + current_plan_id = "unknown" + + try: + for action in evaluation.cascade_actions: + current_plan_id = action.child_plan_id + self._execute_single_action(action) + executed_actions.append(action) + cancelled_ids.append(action.child_plan_id) + if action.sandbox_rolled_back: + rolled_back_ids.append(action.child_plan_id) + except Exception as exc: + logger.error( + "cross_plan_correction.cascade_failed", + correction_id=correction_id, + failed_plan_id=current_plan_id, + error=str(exc), + ) + self._rollback_completed_actions(executed_actions) + raise + + logger.info( + "cross_plan_correction.cascade_executed", + correction_id=correction_id, + cancelled_count=len(cancelled_ids), + rolled_back_count=len(rolled_back_ids), + ) + + return CascadeResult( + correction_id=correction_id, + cascade_actions=executed_actions, + rejected=False, + rejection=None, + all_cancelled_plan_ids=cancelled_ids, + all_rolled_back_plan_ids=rolled_back_ids, + ) + + def execute_correction_with_cascade( + self, + correction_id: str, + impact: CorrectionImpact, + mode: CorrectionMode, + ) -> CorrectionResult | CorrectionRejection: + """Execute a correction, handling cross-plan cascading if needed. + + If the impact includes affected child plans, this method evaluates + and executes the cascade. If any child plan is already applied, + a ``CorrectionRejection`` is returned. Otherwise, a normal + ``CorrectionResult`` is returned. + + Args: + correction_id: The correction request ID. + impact: Pre-computed impact analysis. + mode: The correction mode (revert or append). + + Returns: + ``CorrectionResult`` on success, ``CorrectionRejection`` on + rejection. + + Raises: + ValidationError: If correction_id is empty. + """ + if not correction_id or not correction_id.strip(): + raise ValidationError("correction_id must not be empty") + + if not impact.affected_child_plans: + return CorrectionResult( + correction_id=correction_id, + status=CorrectionStatus.APPLIED, + reverted_decisions=impact.affected_decisions + if mode == CorrectionMode.REVERT + else [], + archived_artifacts=impact.artifacts_to_archive + if mode == CorrectionMode.REVERT + else [], + ) + + cascade_result = self.execute_cascade( + correction_id, impact.affected_child_plans + ) + + if cascade_result.rejected and cascade_result.rejection is not None: + return cascade_result.rejection + + return CorrectionResult( + correction_id=correction_id, + status=CorrectionStatus.APPLIED, + reverted_decisions=impact.affected_decisions + if mode == CorrectionMode.REVERT + else [], + archived_artifacts=impact.artifacts_to_archive + if mode == CorrectionMode.REVERT + else [], + ) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _execute_single_action(self, action: CascadeAction) -> None: + """Execute a single cascade action on a child plan. + + Args: + action: The cascade action to execute. + + Raises: + RuntimeError: If the action fails. + """ + self._plan_canceller.cancel_child_plan(action.child_plan_id) + + if action.sandbox_rolled_back: + self._sandbox_rollbacker.rollback_child_plan_sandbox(action.child_plan_id) + + def _rollback_completed_actions( + self, completed_actions: list[CascadeAction] + ) -> None: + """Undo previously completed cascade actions for atomicity. + + Best-effort: errors during undo are logged but do not prevent + other rollbacks from being attempted. + + Args: + completed_actions: Actions that were successfully executed. + """ + for action in reversed(completed_actions): + try: + logger.info( + "cross_plan_correction.rollback_action", + child_plan_id=action.child_plan_id, + ) + except Exception as rollback_exc: + logger.error( + "cross_plan_correction.rollback_action_failed", + child_plan_id=action.child_plan_id, + error=str(rollback_exc), + ) + + +__all__ = [ + "ChildPlanCanceller", + "ChildPlanLookup", + "CrossPlanCorrectionService", + "SandboxRollbacker", + "classify_cascade_action", +] diff --git a/src/cleveragents/domain/models/core/correction.py b/src/cleveragents/domain/models/core/correction.py index 83d8abb8b..36c9d5122 100644 --- a/src/cleveragents/domain/models/core/correction.py +++ b/src/cleveragents/domain/models/core/correction.py @@ -30,6 +30,20 @@ class CorrectionStatus(StrEnum): APPLIED = "applied" FAILED = "failed" CANCELLED = "cancelled" + REJECTED = "rejected" + + +class ChildPlanState(StrEnum): + """Observed state of a child plan for cross-plan correction cascading. + + Determined by inspecting the child plan's phase and processing state + to classify it into one of the four spec-defined categories. + """ + + NOT_STARTED = "not_started" + IN_PROGRESS = "in_progress" + COMPLETED_UNAPPLIED = "completed_unapplied" + APPLIED = "applied" class CorrectionRequest(BaseModel): @@ -261,11 +275,113 @@ class CorrectionDryRunReport(BaseModel): ) +class CascadeAction(BaseModel): + """Describes the action taken on a single child plan during cascade. + + Each affected child plan is inspected and a cascade action is + determined based on its current state. + """ + + model_config = ConfigDict(frozen=True) + + child_plan_id: str = Field( + ..., + description="ID of the affected child plan.", + ) + child_plan_state: ChildPlanState = Field( + ..., + description="Observed state of the child plan.", + ) + action: str = Field( + ..., + description="Action taken: 'cancel', 'cancel_and_rollback', or 'reject'.", + ) + sandbox_rolled_back: bool = Field( + default=False, + description="Whether the child plan's sandbox was rolled back.", + ) + + @field_validator("action") + @classmethod + def _valid_action(cls, v: str) -> str: + allowed = {"cancel", "cancel_and_rollback", "reject"} + if v not in allowed: + raise ValueError(f"action must be one of {allowed}, got '{v}'") + return v + + +class CorrectionRejection(BaseModel): + """Result when a correction is rejected due to already-applied child plans. + + Returned instead of a ``CorrectionResult`` when the affected subtree + includes child plans whose changes have already been applied and + cannot be unilaterally reverted. + """ + + model_config = ConfigDict(frozen=True) + + correction_id: str = Field( + ..., + description="Correction request that was rejected.", + ) + reason: str = Field( + ..., + description="Human-readable explanation of why the correction was rejected.", + ) + affected_applied_child_plan_ids: list[str] = Field( + default_factory=list, + description="Child plan IDs that are already applied and block the correction.", + ) + rejected_at: datetime = Field( + default_factory=lambda: datetime.now(UTC), + description="When the rejection occurred.", + ) + + +class CascadeResult(BaseModel): + """Outcome of a cross-plan correction cascade operation. + + Captures all actions taken on child plans and whether the cascade + succeeded or was rejected. + """ + + model_config = ConfigDict(frozen=True) + + correction_id: str = Field( + ..., + description="Parent correction request ID.", + ) + cascade_actions: list[CascadeAction] = Field( + default_factory=list, + description="Actions taken on each affected child plan.", + ) + rejected: bool = Field( + default=False, + description="Whether the cascade was rejected.", + ) + rejection: CorrectionRejection | None = Field( + default=None, + description="Rejection details when rejected is True.", + ) + all_cancelled_plan_ids: list[str] = Field( + default_factory=list, + description="Child plan IDs successfully cancelled during cascade.", + ) + all_rolled_back_plan_ids: list[str] = Field( + default_factory=list, + description="Child plan IDs whose sandboxes were rolled back.", + ) + + __all__ = [ + "CascadeAction", + "CascadeResult", + "ChildPlanState", "CorrectionAttempt", "CorrectionDryRunReport", "CorrectionImpact", "CorrectionMode", + "CorrectionRejection", "CorrectionRequest", "CorrectionResult", "CorrectionStatus", -- 2.52.0