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
+        )

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

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"
+    )

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)}"
+    )

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

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]} <command>", 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()

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",

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",
+]

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",