CONTRIBUTORS.md

@@ -75,3 +75,4 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed the `--clone-into` CLI argument for `container-instance`, the `CloneIntoHandler` module, the `devcontainer-instance` snapshot sandbox strategy, and the `ContainerLifecycleState.DISCOVERED` terminology alignment (PR #8304, issue #7555).
 * HAL 9000 has contributed the ACMS Context Tier Hydration documentation (PR #9208 / issue #6175): documented the `context_tier_hydrator` module in the ACMS Architecture section of the specification, covering its public interface, file listing strategy, budget limits, and fragment structure.
 * HAL 9000 has contributed the agent task memory leak fix (#9044): replaced `list.remove` with `set.discard` as the done_callback for asyncio tasks in `Agent._tasks`, preventing unbounded memory growth in long-lived agents and ensuring safe concurrent task removal.
+* HAL 9000 has contributed the plan correction engine (#9562): implemented `CorrectionService` with BFS subtree traversal over both the structural decision tree and the influence DAG, `--mode=revert` (checkpoint restoration, reasoning rollback via `actor_state_ref`, guidance injection, and Strategize phase transition signal) and `--mode=append` (child plan spawning without reverting the existing subtree), `CorrectionAttemptRecord` database persistence for audit trail, cross-plan correction cascading via `CrossPlanCorrectionService`, and the `agents plan correct <id> --mode=revert|append` CLI command with dry-run support.

features/steps/correction_attempt_persistence_steps.py

@@ -11,6 +11,11 @@ from typing import Any
 
 from behave import given, then, when  # type: ignore[import-untyped]
 from behave.runner import Context
+from cleveragents.domain.models.core.decision import (
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+)
 from pydantic import ValidationError
 from sqlalchemy import create_engine, event
 from sqlalchemy.exc import IntegrityError as SAIntegrityError
@@ -24,11 +29,6 @@ from cleveragents.domain.models.core.correction import (
     CorrectionAttemptState,
     CorrectionMode,
 )
-from cleveragents.domain.models.core.decision import (
-    ContextSnapshot,
-    Decision,
-    DecisionType,
-)
 from cleveragents.domain.models.core.plan import (
     NamespacedName,
     Plan,

features/steps/correction_checkpoint_rollback_steps.py

@@ -13,8 +13,8 @@ import tempfile
 from pathlib import Path
 
 from behave import given, then, when
-
 from cleveragents.application.services.checkpoint_service import CheckpointService
+
 from cleveragents.application.services.correction_service import CorrectionService
 from cleveragents.core.exceptions import BusinessRuleViolation, ResourceNotFoundError
 

features/steps/correction_service_coverage_boost_steps.py

@@ -11,14 +11,14 @@ existing step definitions in other feature files.
 """
 
 from behave import given, then, when
+from cleveragents.infrastructure.events.models import DomainEvent
+from cleveragents.infrastructure.events.types import EventType
 
 from cleveragents.application.services.correction_service import CorrectionService
 from cleveragents.domain.models.core.correction import (
     CorrectionMode,
     CorrectionStatus,
 )
-from cleveragents.infrastructure.events.models import DomainEvent
-from cleveragents.infrastructure.events.types import EventType
 
 # ---------------------------------------------------------------------------
 # Mock event buses

features/steps/correction_service_coverage_r3_steps.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 from unittest.mock import patch
 
 from behave import given, then, when
+from cleveragents.infrastructure.events.models import DomainEvent
 
 from cleveragents.application.services.correction_service import CorrectionService
 from cleveragents.core.exceptions import ValidationError
@@ -23,7 +24,6 @@ from cleveragents.domain.models.core.correction import (
     CorrectionImpact,
     CorrectionMode,
 )
-from cleveragents.infrastructure.events.models import DomainEvent
 
 # ---------------------------------------------------------------------------
 # Helper: simple event bus that records events

robot/helper_correction_attempt_persistence.py

@@ -23,6 +23,11 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
 from collections.abc import Callable
 from typing import Any
 
+from cleveragents.domain.models.core.decision import (
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+)
 from sqlalchemy import create_engine, event
 from sqlalchemy.orm import Session, sessionmaker
 
@@ -35,11 +40,6 @@ from cleveragents.domain.models.core.correction import (
     CorrectionAttemptState,
     CorrectionMode,
 )
-from cleveragents.domain.models.core.decision import (
-    ContextSnapshot,
-    Decision,
-    DecisionType,
-)
 from cleveragents.domain.models.core.plan import (
     NamespacedName,
     Plan,

src/cleveragents/application/services/correction_impact_service.py

@@ -0,0 +1,498 @@
+"""Impact analysis service for plan decision corrections.
+
+Encapsulates BFS subtree traversal, risk classification, rollback tier
+computation, and dry-run report generation.  These computations are
+separated from the execution orchestration in ``CorrectionService`` to
+keep each module within the 500-line style limit and to honour the
+Single Responsibility Principle.
+
+Exports:
+    CorrectionImpactService: Service for correction impact analysis.
+"""
+
+from __future__ import annotations
+
+from collections import deque
+from typing import TYPE_CHECKING
+
+import structlog
+
+from cleveragents.core.exceptions import ResourceNotFoundError, ValidationError
+from cleveragents.domain.models.core.correction import (
+    CorrectionDryRunReport,
+    CorrectionImpact,
+    CorrectionMode,
+    CorrectionRequest,
+    CorrectionStatus,
+)
+
+if TYPE_CHECKING:
+    pass
+
+logger = structlog.get_logger(__name__)
+
+# ---------------------------------------------------------------------------
+# Risk-level thresholds
+# ---------------------------------------------------------------------------
+_RISK_LOW_MAX = 3
+_RISK_MEDIUM_MAX = 10
+
+# ---------------------------------------------------------------------------
+# Cost / time estimation constants
+# ---------------------------------------------------------------------------
+_COST_PER_DECISION = 1.5
+_RECOMPUTE_SECONDS_PER_DECISION = 2.0
+
+# ---------------------------------------------------------------------------
+# Terminal correction statuses (immutable after execution)
+# ---------------------------------------------------------------------------
+_TERMINAL_STATUSES: frozenset[CorrectionStatus] = frozenset(
+    {
+        CorrectionStatus.APPLIED,
+        CorrectionStatus.FAILED,
+        CorrectionStatus.CANCELLED,
+        CorrectionStatus.REJECTED,
+    }
+)
+
+# ---------------------------------------------------------------------------
+# Maximum decision tree size (DoS protection)
+# ---------------------------------------------------------------------------
+_MAX_TREE_NODES = 50_000
+
+
+class CorrectionImpactService:
+    """Service for decision correction impact analysis.
+
+    Provides BFS subtree traversal, risk classification, rollback tier
+    computation, subtree isolation validation, and dry-run report
+    generation.  Designed to be used by ``CorrectionService`` for the
+    analytical phase of the correction lifecycle.
+    """
+
+    def analyze_impact(
+        self,
+        correction_id: str,
+        corrections: dict[str, CorrectionRequest],
+        impacts: dict[str, CorrectionImpact],
+        decision_tree: dict[str, list[str]] | None = None,
+        influence_edges: dict[str, list[str]] | None = None,
+    ) -> CorrectionImpact:
+        """Compute the impact of a correction via BFS subtree traversal.
+
+        Traverses **both** the structural tree (parent → children) and
+        the influence DAG (``decision_dependencies`` edges) to compute
+        the full set of transitively affected decisions.
+
+        Args:
+            correction_id: Previously created correction request ID.
+            corrections: Mutable correction request store.
+            impacts: Mutable correction impact store.
+            decision_tree: Adjacency list mapping parent → children.
+            influence_edges: Adjacency list mapping source → targets
+                in the influence DAG (``decision_dependencies``).
+
+        Returns:
+            ``CorrectionImpact`` with affected nodes, excluded nodes,
+            rollback tier depth, and risk level.
+
+        Raises:
+            ResourceNotFoundError: If the correction does not exist.
+            ValidationError: If inputs exceed max size or status
+                is terminal.
+        """
+        request = corrections.get(correction_id)
+        if request is None:
+            raise ResourceNotFoundError(
+                resource_type="correction",
+                resource_id=correction_id,
+            )
+
+        # Guard against pathologically large inputs.
+        tree = decision_tree or {}
+        dag = influence_edges or {}
+        total_keys = len(tree) + len(dag)
+        if total_keys > _MAX_TREE_NODES:
+            raise ValidationError(
+                f"Decision tree + influence DAG too large ({total_keys} keys, "
+                f"max {_MAX_TREE_NODES}). Reduce tree size before analyzing."
+            )
+
+        # Reject analysis on terminal states.
+        if request.status in _TERMINAL_STATUSES:
+            raise ValidationError(
+                f"Cannot analyze correction in terminal '{request.status}' "
+                "status. Impact data is immutable after execution."
+            )
+
+        if request.status == CorrectionStatus.PENDING:
+            request.status = CorrectionStatus.ANALYZING
+
+        affected = self._compute_affected_subtree(request.target_decision_id, tree, dag)
+        risk = self._classify_risk(len(affected))
+
+        all_decisions = self._collect_all_decisions(tree, dag)
+        all_decisions.add(request.target_decision_id)
+        affected_set = set(affected)
+        excluded = sorted(d for d in all_decisions if d not in affected_set)
+
+        tier_depth = self._compute_rollback_tier_depth(request.target_decision_id, tree)
+
+        artifacts = [f"{d}.artifact" for d in affected]
+
+        impact = CorrectionImpact(
+            affected_decisions=affected,
+            excluded_decisions=excluded,
+            affected_files=[f"{d}.py" for d in affected],
+            affected_child_plans=[],
+            estimated_cost=float(len(affected)) * _COST_PER_DECISION,
+            risk_level=risk,
+            rollback_tier="full"
+            if request.mode == CorrectionMode.REVERT
+            else "append_only",
+            rollback_tier_depth=tier_depth,
+            artifacts_to_archive=artifacts,
+        )
+        impacts[correction_id] = impact
+
+        logger.info(
+            "correction.impact_analyzed",
+            correction_id=correction_id,
+            affected_count=len(affected),
+            excluded_count=len(excluded),
+            rollback_tier_depth=tier_depth,
+            risk_level=risk,
+        )
+        return impact
+
+    def generate_dry_run_report(
+        self,
+        correction_id: str,
+        corrections: dict[str, CorrectionRequest],
+        impacts: dict[str, CorrectionImpact],
+        decision_tree: dict[str, list[str]] | None = None,
+        influence_edges: dict[str, list[str]] | None = None,
+    ) -> CorrectionDryRunReport:
+        """Generate a dry-run report without executing the correction.
+
+        Args:
+            correction_id: Correction request ID.
+            corrections: Mutable correction request store.
+            impacts: Mutable correction impact store.
+            decision_tree: Optional decision tree adjacency list.
+            influence_edges: Optional influence DAG adjacency list.
+
+        Returns:
+            ``CorrectionDryRunReport`` describing what *would* happen.
+        """
+        request = corrections.get(correction_id)
+        if request is None:
+            raise ResourceNotFoundError(
+                resource_type="correction",
+                resource_id=correction_id,
+            )
+
+        original_status = request.status
+        original_impact = impacts.get(correction_id)
+        try:
+            impact = self.analyze_impact(
+                correction_id, corrections, impacts, decision_tree, influence_edges
+            )
+        finally:
+            request.status = original_status
+            if original_impact is None:
+                impacts.pop(correction_id, None)
+            else:
+                impacts[correction_id] = original_impact
+
+        warnings: list[str] = []
+        if impact.risk_level == "high":
+            warnings.append(
+                "High risk: more than 10 decisions affected. "
+                "Review carefully before executing."
+            )
+        elif impact.risk_level == "medium":
+            warnings.append("Medium risk: 4-10 decisions affected.")
+        if request.mode == CorrectionMode.REVERT and len(impact.affected_decisions) > 1:
+            warnings.append(
+                f"Revert will invalidate {len(impact.affected_decisions)} decisions "
+                "and archive associated artifacts."
+            )
+        tree = decision_tree or {}
+        actual_root = self._find_root(tree)
+        if (
+            impact.rollback_tier_depth == 0
+            and len(impact.affected_decisions) > 1
+            and actual_root is not None
+            and request.target_decision_id == actual_root
+        ):
+            warnings.append(
+                "Tier 0: root decision targeted — entire decision tree "
+                "will be affected."
+            )
+
+        recompute_seconds = (
+            float(len(impact.affected_decisions)) * _RECOMPUTE_SECONDS_PER_DECISION
+        )
+
+        report = CorrectionDryRunReport(
+            correction_id=correction_id,
+            mode=request.mode,
+            impact=impact,
+            decisions_to_invalidate=impact.affected_decisions
+            if request.mode == CorrectionMode.REVERT
+            else [],
+            estimated_recompute_time_seconds=recompute_seconds,
+            warnings=warnings,
+        )
+
+        logger.info(
+            "correction.dry_run_generated",
+            correction_id=correction_id,
+            warning_count=len(warnings),
+            rollback_tier_depth=impact.rollback_tier_depth,
+            excluded_count=len(impact.excluded_decisions),
+        )
+        return report
+
+    def compute_rollback_tier(
+        self,
+        target_decision_id: str,
+        plan_id: str,
+        decision_tree: dict[str, list[str]] | None = None,
+    ) -> int:
+        """Compute the rollback tier (depth) for a target decision.
+
+        The tier is the number of parent hops from the target to the
+        tree root.
+
+        Args:
+            target_decision_id: Decision to compute the tier for.
+            plan_id: Plan owning the decision tree (used for logging).
+            decision_tree: Adjacency list (parent → children).
+
+        Returns:
+            Non-negative integer representing the tier depth.
+        """
+        tree = decision_tree or {}
+        depth = self._compute_rollback_tier_depth(target_decision_id, tree)
+        logger.info(
+            "correction.rollback_tier_computed",
+            target_decision_id=target_decision_id,
+            plan_id=plan_id,
+            tier_depth=depth,
+        )
+        return depth
+
+    def validate_subtree_isolation(
+        self,
+        target_decision_id: str,
+        decision_tree: dict[str, list[str]],
+        influence_edges: dict[str, list[str]] | None = None,
+    ) -> bool:
+        """Validate that the affected subtree is correctly isolated.
+
+        Confirms two invariants for non-root corrections:
+
+        1. The **root decision** is never in the structural affected set
+           unless it is explicitly the target decision.
+        2. **Sibling decisions** are not in the structural affected set.
+
+        Args:
+            target_decision_id: Decision node that was targeted.
+            decision_tree: Structural tree adjacency list.
+            influence_edges: Accepted for API consistency but not used
+                in isolation checks — structural-only BFS is used per spec.
+
+        Returns:
+            ``True`` if isolation invariants hold, ``False`` otherwise.
+        """
+        tree = decision_tree
+        structural_affected = self._compute_affected_subtree(
+            target_decision_id,
+            tree,
+            influence_edges=None,
+        )
+        structural_set = set(structural_affected)
+
+        root = self._find_root(tree)
+        if root is None:
+            return True
+
+        if root in structural_set and root != target_decision_id:
+            logger.warning(
+                "correction.isolation_violation_root",
+                root=root,
+                target=target_decision_id,
+            )
+            return False
+
+        parent = self._find_parent(target_decision_id, tree)
+        if parent is not None:
+            siblings = [
+                child for child in tree.get(parent, []) if child != target_decision_id
+            ]
+            for sibling in siblings:
+                if sibling in structural_set:
+                    logger.warning(
+                        "correction.isolation_violation_sibling",
+                        sibling=sibling,
+                        target=target_decision_id,
+                    )
+                    return False
+
+        return True
+
+    @staticmethod
+    def _compute_affected_subtree(
+        target_id: str,
+        tree: dict[str, list[str]],
+        influence_edges: dict[str, list[str]] | None = None,
+    ) -> list[str]:
+        """BFS walk from *target_id* through structural tree AND influence DAG.
+
+        Traverses **both** the structural tree (parent → children) and
+        the influence DAG (``decision_dependencies`` edges) to compute
+        the union of all transitively affected decisions.
+
+        Cycle detection is built-in via the ``visited`` set.
+
+        Complexity: O(V + E) where V = decisions, E = tree + DAG edges.
+
+        Args:
+            target_id: Root decision to start BFS from.
+            tree: Structural tree adjacency list (parent → children).
+            influence_edges: Influence DAG adjacency list.
+                Optional; when ``None`` only the structural tree is traversed.
+
+        Returns:
+            All reachable node IDs (inclusive of the target itself),
+            in BFS visit order.
+        """
+        dag = influence_edges or {}
+        affected: list[str] = []
+        visited: set[str] = set()
+        queue: deque[str] = deque([target_id])
+
+        while queue:
+            node = queue.popleft()
+            if node in visited:
+                continue
+            visited.add(node)
+            affected.append(node)
+
+            for neighbor in tree.get(node, []):
+                if neighbor in visited:
+                    logger.warning(
+                        "correction.cycle_detected",
+                        node=neighbor,
+                        source=node,
+                        edge_type="structural",
+                        msg="Back-edge detected during BFS "
+                        "(cycle in decision tree or influence DAG)",
+                    )
+                elif neighbor not in visited:
+                    queue.append(neighbor)
+            for neighbor in dag.get(node, []):
+                if neighbor in visited:
+                    logger.warning(
+                        "correction.cycle_detected",
+                        node=neighbor,
+                        source=node,
+                        edge_type="influence",
+                        msg="Back-edge detected during BFS "
+                        "(cycle in decision tree or influence DAG)",
+                    )
+                elif neighbor not in visited:
+                    queue.append(neighbor)
+
+        influence_count = sum(len(v) for v in dag.values()) if dag else 0
+        if influence_count > 0:
+            logger.info(
+                "correction.influence_traversal",
+                target_id=target_id,
+                total_affected=len(affected),
+                influence_edge_count=influence_count,
+            )
+
+        return affected
+
+    @staticmethod
+    def _classify_risk(affected_count: int) -> str:
+        """Classify risk level based on affected subtree size."""
+        if affected_count <= _RISK_LOW_MAX:
+            return "low"
+        if affected_count <= _RISK_MEDIUM_MAX:
+            return "medium"
+        return "high"
+
+    @staticmethod
+    def _collect_all_decisions(
+        tree: dict[str, list[str]],
+        dag: dict[str, list[str]],
+    ) -> set[str]:
+        """Collect every decision ID from both the tree and DAG edges."""
+        all_ids: set[str] = set()
+        for parent, children in tree.items():
+            all_ids.add(parent)
+            all_ids.update(children)
+        for source, targets in dag.items():
+            all_ids.add(source)
+            all_ids.update(targets)
+        return all_ids
+
+    @staticmethod
+    def _compute_rollback_tier_depth(
+        target_id: str,
+        tree: dict[str, list[str]],
+    ) -> int:
+        """Count parent hops from *target_id* up to the tree root.
+
+        Returns ``0`` when the target is the root or not in the tree.
+        """
+        child_to_parent: dict[str, str] = {}
+        for parent, children in tree.items():
+            for child in children:
+                child_to_parent[child] = parent
+
+        depth = 0
+        current = target_id
+        visited: set[str] = set()
+        while current in child_to_parent and current not in visited:
+            visited.add(current)
+            current = child_to_parent[current]
+            depth += 1
+
+        return depth
+
+    @staticmethod
+    def _find_root(tree: dict[str, list[str]]) -> str | None:
+        """Find the root node of the tree (not a child of any node).
+
+        Returns ``None`` if the tree is empty.
+        """
+        if not tree:
+            return None
+        all_children: set[str] = set()
+        for children in tree.values():
+            all_children.update(children)
+        for parent in tree:
+            if parent not in all_children:
+                return parent
+        return next(iter(tree))
+
+    @staticmethod
+    def _find_parent(
+        target_id: str,
+        tree: dict[str, list[str]],
+    ) -> str | None:
+        """Find the parent of *target_id* in the tree, or ``None``."""
+        for parent, children in tree.items():
+            if target_id in children:
+                return parent
+        return None
+
+
+__all__ = [
+    "CorrectionImpactService",
+]

src/cleveragents/cli/commands/plan_correction_cli.py

@@ -0,0 +1,507 @@
+"""CLI command handlers for plan correction and rollback operations.
+
+Defines the ``correct`` and ``rollback`` command handler functions that are
+registered on the main ``agents plan`` Typer app from ``plan.py``.
+
+Separating these handlers into their own module keeps ``plan.py`` within
+the 500-line style limit while preserving a single ``agents plan`` command
+group.
+"""
+
+from __future__ import annotations
+
+import time
+from datetime import datetime
+from typing import TYPE_CHECKING, Annotated
+
+import typer
+from rich.console import Console
+from rich.panel import Panel
+from rich.table import Table
+
+from cleveragents.application.container import get_container
+from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.core.exceptions import CleverAgentsError, ValidationError
+
+if TYPE_CHECKING:
+    pass
+
+console = Console()
+
+_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
+
+
+def _resolve_active_plan_id_ext() -> str:
+    """Resolve the active plan ID (thin wrapper used by correction CLI).
+
+    Imports ``_resolve_active_plan_id`` lazily from plan to avoid circular
+    imports.
+    """
+    from cleveragents.cli.commands.plan import _resolve_active_plan_id
+
+    return _resolve_active_plan_id()
+
+
+def correct_decision(
+    identifier: Annotated[
+        str,
+        typer.Argument(
+            help="Plan ID (auto-selects root decision) or Decision ID to correct"
+        ),
+    ],
+    mode: Annotated[
+        str,
+        typer.Option(
+            ...,
+            "--mode",
+            "-m",
+            help="Correction mode: revert or append",
+        ),
+    ],
+    guidance: Annotated[
+        str,
+        typer.Option(
+            ...,
+            "--guidance",
+            "-g",
+            help="Guidance text for the correction",
+        ),
+    ],
+    dry_run: Annotated[
+        bool,
+        typer.Option(
+            "--dry-run",
+            help="Only analyze impact, do not execute",
+        ),
+    ] = False,
+    yes: Annotated[
+        bool,
+        typer.Option(
+            "--yes",
+            "-y",
+            help="Skip confirmation prompt",
+        ),
+    ] = False,
+    plan_id: Annotated[
+        str | None,
+        typer.Option(
+            "--plan",
+            "-p",
+            help="Plan ID (uses latest active plan if omitted)",
+        ),
+    ] = None,
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help=_FORMAT_HELP,
+        ),
+    ] = "rich",
+) -> None:
+    """Correct a decision in a plan's decision tree.
+
+    The positional *identifier* can be either a **plan ID** or a
+    **decision ID**.  When a plan ID is given the root decision of that
+    plan is automatically selected as the correction target.
+
+    Supports two modes:
+
+    * **revert** -- undo a decision and recompute affected subtrees
+    * **append** -- add new guidance at a decision point
+
+    Use ``--dry-run`` to preview what would change without executing.
+
+    Examples::
+
+        agents plan correct --mode revert -g "Use FastAPI instead" PLAN-001
+        agents plan correct --mode revert -g "Use FastAPI instead" DEC-001
+        agents plan correct --mode append -g "Add caching layer" --dry-run DEC-002
+    """
+    from cleveragents.core.exceptions import ResourceNotFoundError as RNF
+    from cleveragents.domain.models.core.correction import CorrectionMode
+    from cleveragents.domain.models.core.plan import Plan
+
+    try:
+        try:
+            correction_mode = CorrectionMode(mode)
+        except ValueError as exc:
+            console.print(
+                f"[red]Invalid mode:[/red] {mode}. Must be 'revert' or 'append'."
+            )
+            raise typer.Abort() from exc
+
+        if not guidance.strip():
+            console.print("[red]Error:[/red] --guidance / -g must not be blank.")
+            raise typer.Abort()
+
+        container = get_container()
+        decision_svc = container.decision_service()
+
+        target_decision_id: str
+        resolved_plan_id: str
+        _is_plan = False
+        try:
+            service = container.plan_lifecycle_service()
+            plan_obj = service.get_plan(identifier)
+            if isinstance(plan_obj, Plan):
+                _is_plan = True
+        except RNF:
+            pass
+
+        if _is_plan:
+            resolved_plan_id = identifier
+            decisions = decision_svc.list_decisions(resolved_plan_id)
+            root_decisions = [d for d in decisions if d.parent_decision_id is None]
+            if not root_decisions:
+                console.print(
+                    f"[red]Error:[/red] Plan '{identifier}' has no root decision."
+                )
+                raise typer.Abort()
+            target_decision_id = root_decisions[0].decision_id
+        else:
+            target_decision_id = identifier
+            resolved_plan_id = plan_id or _resolve_active_plan_id_ext()
+
+        decisions = decision_svc.list_decisions(resolved_plan_id)
+        decision_tree: dict[str, list[str]] = {}
+        for d in decisions:
+            if d.parent_decision_id is not None:
+                decision_tree.setdefault(d.parent_decision_id, []).append(d.decision_id)
+
+        influence_edges = decision_svc.get_influence_edges(resolved_plan_id)
+        svc = container.correction_service()
+
+        request = svc.request_correction(
+            plan_id=resolved_plan_id,
+            target_decision_id=target_decision_id,
+            mode=correction_mode,
+            guidance=guidance,
+            dry_run=dry_run,
+        )
+
+        if dry_run:
+            impact = svc.analyze_impact(
+                request.correction_id,
+                decision_tree=decision_tree,
+                influence_edges=influence_edges,
+            )
+            if fmt != OutputFormat.RICH.value:
+                data = {
+                    "correction_id": request.correction_id,
+                    "mode": request.mode.value,
+                    "target_decision": request.target_decision_id,
+                    "affected_decisions": impact.affected_decisions,
+                    "affected_files": impact.affected_files,
+                    "estimated_cost": impact.estimated_cost,
+                    "risk_level": impact.risk_level,
+                }
+                console.print(format_output(data, fmt))
+            else:
+                console.print(
+                    Panel(
+                        f"[bold]Correction ID:[/bold] {request.correction_id}\n"
+                        f"[bold]Mode:[/bold] {request.mode.value}\n"
+                        f"[bold]Target Decision:[/bold] "
+                        f"{request.target_decision_id}\n"
+                        f"[bold]Guidance:[/bold] {request.guidance}\n\n"
+                        f"[bold]Affected Decisions:[/bold] "
+                        f"{', '.join(impact.affected_decisions) or '(none)'}\n"
+                        f"[bold]Affected Files:[/bold] "
+                        f"{', '.join(impact.affected_files) or '(none)'}\n"
+                        f"[bold]Risk Level:[/bold] {impact.risk_level}\n"
+                        f"[bold]Estimated Cost:[/bold] "
+                        f"{impact.estimated_cost or 'N/A'}",
+                        title="Correction Impact (Dry Run)",
+                        expand=False,
+                    )
+                )
+            return
+
+        if not yes:
+            console.print(
+                f"\n[bold]Correction:[/bold] {correction_mode.value} "
+                f"decision {target_decision_id}"
+            )
+            console.print(f"[bold]Guidance:[/bold] {guidance}")
+            confirm = typer.confirm("\nProceed with correction?")
+            if not confirm:
+                console.print("[yellow]Cancelled.[/yellow]")
+                raise typer.Exit(0)
+
+        result = svc.execute_correction(
+            request.correction_id,
+            decision_tree=decision_tree,
+            influence_edges=influence_edges,
+        )
+
+        if fmt != OutputFormat.RICH.value:
+            data = {
+                "correction_id": result.correction_id,
+                "status": result.status.value,
+                "mode": correction_mode.value,
+                "new_decisions": result.new_decisions,
+                "reverted_decisions": result.reverted_decisions,
+            }
+            console.print(format_output(data, fmt))
+        else:
+            console.print(
+                f"[green]\u2713[/green] Correction applied: {result.correction_id}"
+            )
+            if result.reverted_decisions:
+                console.print(f"  Reverted: {', '.join(result.reverted_decisions)}")
+            if result.new_decisions:
+                console.print(f"  New decisions: {', '.join(result.new_decisions)}")
+
+    except RNF as e:
+        console.print(f"[red]Not found:[/red] {e.message}")
+        raise typer.Abort() from e
+    except ValidationError as e:
+        console.print(f"[red]Validation Error:[/red] {e.message}")
+        raise typer.Abort() from e
+    except CleverAgentsError as e:
+        console.print(f"[red]Error:[/red] {e.message}")
+        raise typer.Abort() from e
+
+
+def rollback_plan(
+    plan_id: Annotated[
+        str,
+        typer.Argument(help="Plan ID to rollback"),
+    ],
+    checkpoint_id: Annotated[
+        str | None,
+        typer.Argument(help="Checkpoint ID to restore (positional)"),
+    ] = None,
+    to_checkpoint: Annotated[
+        str | None,
+        typer.Option(
+            "--to-checkpoint",
+            help="Checkpoint ID to restore (named option)",
+        ),
+    ] = None,
+    yes: Annotated[
+        bool,
+        typer.Option(
+            "--yes",
+            "-y",
+            help="Skip confirmation prompt",
+        ),
+    ] = False,
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help=_FORMAT_HELP,
+        ),
+    ] = "rich",
+) -> None:
+    """Restore sandbox state to a named checkpoint.
+
+    Specify checkpoint as positional arg or via ``--to-checkpoint``.
+
+    Examples:
+        agents plan rollback --yes 01ARZ3NDEK... 01BRZ4NFEK...
+        agents plan rollback --to-checkpoint 01BRZ4NFEK... 01ARZ3NDEK...
+    """
+    from datetime import UTC
+
+    from cleveragents.core.exceptions import BusinessRuleViolation
+    from cleveragents.core.exceptions import ResourceNotFoundError as RNF
+    from cleveragents.domain.models.core.decision import DecisionType
+
+    resolved_checkpoint_id = checkpoint_id or to_checkpoint
+    if not resolved_checkpoint_id:
+        console.print(
+            "[red]Error:[/red] checkpoint ID required. "
+            "Provide as positional argument or via --to-checkpoint."
+        )
+        raise typer.Abort()
+
+    container = get_container()
+    svc = container.checkpoint_service()
+
+    if not yes:
+        prompt_text = (
+            f"\nRollback plan {plan_id} to checkpoint {resolved_checkpoint_id}?"
+        )
+        try:
+            checkpoint = svc.get_checkpoint(resolved_checkpoint_id)
+            label = checkpoint.metadata.reason
+            relative_time = _format_relative_time(checkpoint.created_at)
+            label_display = f"'{label}'" if label else resolved_checkpoint_id
+            prompt_text = (
+                f"\nRoll back plan {plan_id} to checkpoint "
+                f"{label_display} (created {relative_time})?"
+            )
+            decisions_count = 0
+            child_plans_count = 0
+            try:
+                decision_svc = container.decision_service()
+                all_decisions = decision_svc.list_decisions(plan_id)
+                cp_time = checkpoint.created_at
+                if cp_time.tzinfo is None:
+                    cp_time = cp_time.replace(tzinfo=UTC)
+                for d in all_decisions:
+                    d_time = d.created_at
+                    if d_time.tzinfo is None:
+                        d_time = d_time.replace(tzinfo=UTC)
+                    if d_time > cp_time:
+                        decisions_count += 1
+                        if d.decision_type in (
+                            DecisionType.SUBPLAN_SPAWN,
+                            DecisionType.SUBPLAN_PARALLEL_SPAWN,
+                        ):
+                            child_plans_count += 1
+            except CleverAgentsError:
+                pass
+
+            side_effects_parts: list[str] = []
+            if decisions_count > 0:
+                side_effects_parts.append(
+                    f"invalidate {decisions_count} "
+                    f"decision{'s' if decisions_count != 1 else ''}"
+                )
+            if child_plans_count > 0:
+                side_effects_parts.append(
+                    f"cancel {child_plans_count} child "
+                    f"plan{'s' if child_plans_count != 1 else ''}"
+                )
+            if side_effects_parts:
+                console.print("This will " + " and ".join(side_effects_parts) + ".")
+        except CleverAgentsError:
+            pass
+
+        confirm = typer.confirm(prompt_text)
+        if not confirm:
+            console.print("[yellow]Rollback cancelled.[/yellow]")
+            raise typer.Abort()
+
+    try:
+        t0 = time.monotonic()
+        result = svc.selective_rollback(plan_id, resolved_checkpoint_id)
+        elapsed = time.monotonic() - t0
+
+        data = {
+            "rollback_summary": {
+                "plan_id": plan_id,
+                "from_checkpoint_id": result.from_checkpoint_id,
+                "restored_files_count": result.restored_files_count,
+            },
+            "changes_reverted": result.changed_paths,
+            "impact": {"files_affected": result.restored_files_count},
+            "post_rollback_state": {
+                "active_checkpoint": result.from_checkpoint_id,
+                "plan_id": plan_id,
+            },
+            "timing": {"elapsed_seconds": round(elapsed, 3)},
+            "messages": ["Rollback completed successfully."],
+        }
+
+        if fmt != OutputFormat.RICH.value:
+            console.print(format_output(data, fmt))
+        else:
+            label = getattr(result, "label", None) or ""
+            label_line = f"\n[bold]Label:[/bold] {label}" if label else ""
+            console.print(
+                Panel(
+                    f"[bold]Plan:[/bold] {plan_id}\n"
+                    f"[bold]Checkpoint:[/bold] {result.from_checkpoint_id}"
+                    f"{label_line}\n"
+                    f"[bold]Files:[/bold] {result.restored_files_count} reverted",
+                    title="Rollback Summary",
+                    expand=False,
+                )
+            )
+
+            changes_table = Table(show_header=True, expand=False)
+            changes_table.add_column("File", style="cyan")
+            changes_table.add_column("Action", style="green")
+            changes_reverted = getattr(result, "changes_reverted", None)
+            if changes_reverted and isinstance(changes_reverted, list):
+                for entry in changes_reverted:
+                    if isinstance(entry, dict):
+                        changes_table.add_row(
+                            str(entry.get("file", "")),
+                            str(entry.get("action", "restored")),
+                        )
+                    else:
+                        changes_table.add_row(str(entry), "restored")
+            else:
+                for path in result.changed_paths:
+                    changes_table.add_row(str(path), "restored")
+            console.print(Panel(changes_table, title="Changes Reverted", expand=False))
+
+            child_plans_invalidated = getattr(result, "child_plans_invalidated", None)
+            sandbox_state = getattr(result, "sandbox", None) or (
+                f"restored to {result.from_checkpoint_id}"
+            )
+            decisions_after_cp = getattr(result, "decisions_after_cp", None)
+            tool_calls_after_cp = getattr(result, "tool_calls_after_cp", None)
+            impact_lines = []
+            if child_plans_invalidated is not None:
+                impact_lines.append(
+                    f"[bold]Child Plans Invalidated:[/bold] {child_plans_invalidated}"
+                )
+            impact_lines.append(f"[bold]Sandbox:[/bold] {sandbox_state}")
+            if decisions_after_cp is not None:
+                impact_lines.append(
+                    f"[bold]Decisions After CP:[/bold] {decisions_after_cp} discarded"
+                )
+            if tool_calls_after_cp is not None:
+                impact_lines.append(
+                    f"[bold]Tool Calls After CP:[/bold] {tool_calls_after_cp} undone"
+                )
+            console.print(Panel("\n".join(impact_lines), title="Impact", expand=False))
+
+            phase = getattr(result, "phase", None) or "execute"
+            state = getattr(result, "state", None) or "queued"
+            checkpoints_remaining = getattr(result, "checkpoints_remaining", None)
+            post_rollback_lines = [
+                f"[bold]Phase:[/bold] {phase}",
+                f"[bold]State:[/bold] {state}",
+            ]
+            if checkpoints_remaining is not None:
+                post_rollback_lines.append(
+                    f"[bold]Checkpoints Remaining:[/bold] {checkpoints_remaining}"
+                )
+            console.print(
+                Panel(
+                    "\n".join(post_rollback_lines),
+                    title="Post-Rollback State",
+                    expand=False,
+                )
+            )
+            console.print("[green]\u2713 OK[/green] Rollback complete")
+
+    except BusinessRuleViolation as e:
+        console.print(f"[red]Rollback blocked:[/red] {e.message}")
+        raise typer.Abort() from e
+    except RNF as e:
+        console.print(f"[red]Not found:[/red] {e.message}")
+        raise typer.Abort() from e
+    except ValidationError as e:
+        console.print(f"[red]Validation Error:[/red] {e.message}")
+        raise typer.Abort() from e
+    except CleverAgentsError as e:
+        console.print(f"[red]Error:[/red] {e.message}")
+        raise typer.Abort() from e
+
+
+
+def _format_relative_time(dt: datetime) -> str:
+    """Format a datetime as a relative time string."""
+    from cleveragents.cli.commands.plan import _format_relative_time as _fmt
+
+    return _fmt(dt)
+
+
+__all__ = ["correct_decision", "rollback_plan"]
+
+
+# Register commands on the plan Typer app.
+from cleveragents.cli.commands.plan import app  # noqa: E402
+
+app.command("correct")(correct_decision)
+app.command("rollback")(rollback_plan)