feat(M4.2): Correction service with revert/append BFS + dry-run

Adds CorrectionRequest, CorrectionResult, CorrectionMode,
CorrectionPatch, CorrectionDryRunReport, CorrectionNotFoundError,
and CorrectionConflictError domain models.

Implements CorrectionService with BFS-based revert (marks decisions
as rolled back and restores via inverse changes) and append mode
(spawns a child correction plan).  Includes request_correction()
with dry-run support and dispatch_correction() convenience method.

33 Behave scenarios, 8 Robot smoke tests, ASV benchmark suite,
and reference documentation.

Ref: Day-14 Rebaseline – M4.2 Decision-correction flows [Jeff]

benchmarks/decision_correction_bench.py

@@ -0,0 +1,173 @@
+"""Airspeed Velocity benchmarks for decision correction flows.
+
+Measures BFS subtree traversal, impact analysis, dry-run report
+generation, and revert execution overhead across varying tree sizes.
+"""
+
+from __future__ import annotations
+
+from cleveragents.application.services.correction_service import CorrectionService
+from cleveragents.domain.models.core.correction import (
+    CorrectionMode,
+    CorrectionStatus,
+)
+
+# ---------------------------------------------------------------------------
+# Tree generators
+# ---------------------------------------------------------------------------
+
+_PLAN_ID = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
+
+
+def _linear_tree(root: str, depth: int) -> dict[str, list[str]]:
+    """Build a linear chain of *depth* nodes starting at *root*."""
+    tree: dict[str, list[str]] = {}
+    current = root
+    for i in range(1, depth):
+        child = f"{root}_c{i}"
+        tree[current] = [child]
+        current = child
+    return tree
+
+
+def _binary_tree(root: str, depth: int) -> dict[str, list[str]]:
+    """Build a complete binary tree of *depth* levels rooted at *root*."""
+    tree: dict[str, list[str]] = {}
+    level = [root]
+    counter = 0
+    for _ in range(depth - 1):
+        next_level: list[str] = []
+        for node in level:
+            left = f"{root}_n{counter}"
+            right = f"{root}_n{counter + 1}"
+            counter += 2
+            tree[node] = [left, right]
+            next_level.extend([left, right])
+        level = next_level
+    return tree
+
+
+# ---------------------------------------------------------------------------
+# BFS traversal benchmarks
+# ---------------------------------------------------------------------------
+
+
+class BfsTraversalSuite:
+    """Benchmark BFS subtree computation at varying tree sizes."""
+
+    def setup(self) -> None:
+        """Prepare trees and service instances."""
+        self.svc = CorrectionService()
+        self.small_tree = _linear_tree("S", 5)  # 5 nodes
+        self.medium_tree = _binary_tree("M", 5)  # 31 nodes
+        self.large_tree = _binary_tree("L", 8)  # 255 nodes
+
+    def time_bfs_small_tree(self) -> None:
+        """BFS on a 5-node linear chain."""
+        self.svc._compute_affected_subtree("S", self.small_tree)
+
+    def time_bfs_medium_tree(self) -> None:
+        """BFS on a 31-node binary tree (depth 5)."""
+        self.svc._compute_affected_subtree("M", self.medium_tree)
+
+    def time_bfs_large_tree(self) -> None:
+        """BFS on a 255-node binary tree (depth 8)."""
+        self.svc._compute_affected_subtree("L", self.large_tree)
+
+
+# ---------------------------------------------------------------------------
+# Impact analysis benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ImpactAnalysisSuite:
+    """Benchmark full impact analysis including risk classification."""
+
+    def setup(self) -> None:
+        """Prepare service and correction requests."""
+        self.svc = CorrectionService()
+        self.small_tree = _linear_tree("S", 3)
+        self.medium_tree = _binary_tree("M", 4)
+        self.large_tree = _binary_tree("L", 6)
+
+        self.req_small = self.svc.request_correction(
+            _PLAN_ID, "S", CorrectionMode.REVERT
+        )
+        self.req_medium = self.svc.request_correction(
+            _PLAN_ID, "M", CorrectionMode.REVERT
+        )
+        self.req_large = self.svc.request_correction(
+            _PLAN_ID, "L", CorrectionMode.REVERT
+        )
+
+    def time_impact_small(self) -> None:
+        """Impact analysis on a 3-node chain."""
+        # Reset status so analyze_impact can run again
+        self.svc._corrections[
+            self.req_small.correction_id
+        ].status = CorrectionStatus.PENDING
+        self.svc.analyze_impact(self.req_small.correction_id, self.small_tree)
+
+    def time_impact_medium(self) -> None:
+        """Impact analysis on a 15-node binary tree."""
+        self.svc._corrections[
+            self.req_medium.correction_id
+        ].status = CorrectionStatus.PENDING
+        self.svc.analyze_impact(self.req_medium.correction_id, self.medium_tree)
+
+    def time_impact_large(self) -> None:
+        """Impact analysis on a 63-node binary tree."""
+        self.svc._corrections[
+            self.req_large.correction_id
+        ].status = CorrectionStatus.PENDING
+        self.svc.analyze_impact(self.req_large.correction_id, self.large_tree)
+
+
+# ---------------------------------------------------------------------------
+# Dry-run report benchmarks
+# ---------------------------------------------------------------------------
+
+
+class DryRunReportSuite:
+    """Benchmark dry-run report generation."""
+
+    def setup(self) -> None:
+        """Prepare service and requests."""
+        self.svc = CorrectionService()
+        self.tree = _binary_tree("DR", 5)
+        self.req = self.svc.request_correction(_PLAN_ID, "DR", CorrectionMode.REVERT)
+
+    def time_dry_run_report(self) -> None:
+        """Generate dry-run report for a 31-node tree."""
+        self.svc._corrections[self.req.correction_id].status = CorrectionStatus.PENDING
+        self.svc.generate_dry_run_report(self.req.correction_id, self.tree)
+
+
+# ---------------------------------------------------------------------------
+# Revert execution benchmarks
+# ---------------------------------------------------------------------------
+
+
+class RevertExecutionSuite:
+    """Benchmark revert execution overhead."""
+
+    def _fresh_request(self, target: str) -> str:
+        """Create a fresh correction request and return its ID."""
+        req = self.svc.request_correction(_PLAN_ID, target, CorrectionMode.REVERT)
+        return req.correction_id
+
+    def setup(self) -> None:
+        """Prepare service and trees."""
+        self.svc = CorrectionService()
+        self.small_tree = _linear_tree("RS", 3)
+        self.medium_tree = _binary_tree("RM", 4)
+
+    def time_revert_small(self) -> None:
+        """Revert execution on a 3-node chain."""
+        cid = self._fresh_request("RS")
+        self.svc.execute_revert(cid, self.small_tree)
+
+    def time_revert_medium(self) -> None:
+        """Revert execution on a 15-node binary tree."""
+        cid = self._fresh_request("RM")
+        self.svc.execute_revert(cid, self.medium_tree)

docs/reference/decision_correction.md

@@ -0,0 +1,99 @@
+# Decision Correction Reference
+
+## Overview
+
+The correction subsystem allows operators to modify a plan's decision tree
+after execution by either **reverting** a subtree of decisions or
+**appending** new decisions as a child plan.
+
+## Correction Modes
+
+### Revert
+
+Invalidates the targeted decision and every descendant reachable via
+BFS traversal.  Associated artifacts are archived and affected child
+plans are rolled back.
+
+```text
+          ┌─── D1 (target) ◄── revert starts here
+          │       │
+          │    ┌──┴──┐
+          │    D2    D3          ← all invalidated
+          │          │
+          │          D4          ← also invalidated
+```
+
+### Append
+
+Preserves the original decision and spawns a **new child plan** rooted
+at the target node.  The child plan carries the operator's guidance and
+produces additional decisions without disturbing the existing tree.
+
+```text
+          D1 (target)
+           │
+        ┌──┴──────────┐
+        D2 (original)  CP-new   ← child plan appended
+```
+
+## Impact Analysis — BFS Algorithm
+
+Impact analysis uses breadth-first search over the decision tree's
+adjacency list (parent → children mapping):
+
+```python
+from collections import deque
+
+def compute_affected_subtree(target_id, tree):
+    affected = []
+    queue = deque([target_id])
+    while queue:
+        node = queue.popleft()
+        affected.append(node)
+        queue.extend(tree.get(node, []))
+    return affected
+```
+
+### Risk Classification
+
+| Affected Count | Risk Level |
+|----------------|------------|
+| ≤ 3            | low        |
+| 4 – 10         | medium     |
+| > 10           | high       |
+
+## Dry-Run Output
+
+A dry-run report contains:
+
+- **impact** — Full `CorrectionImpact` with affected decisions, files,
+  child plans, estimated cost, and risk level.
+- **decisions_to_invalidate** — Decision IDs that *would* be marked
+  invalid (revert mode only).
+- **child_plans_to_rollback** — Child plans that *would* be rolled back.
+- **estimated_recompute_time_seconds** — Wall-clock estimate.
+- **warnings** — Human-readable cautions (e.g. high risk).
+
+## Status Lifecycle
+
+```text
+PENDING → ANALYZING → EXECUTING → APPLIED
+                                 → FAILED
+         PENDING → CANCELLED
+         ANALYZING → CANCELLED
+```
+
+## Service API
+
+| Method                      | Description                                   |
+|-----------------------------|-----------------------------------------------|
+| `request_correction()`      | Create a new correction request               |
+| `analyze_impact()`          | BFS impact analysis on the decision tree      |
+| `generate_dry_run_report()` | Full report without side effects               |
+| `execute_revert()`          | Invalidate subtree + archive artifacts         |
+| `execute_append()`          | Spawn child plan preserving original decision  |
+| `execute_correction()`      | Dispatch to revert or append based on mode     |
+| `get_correction()`          | Retrieve a correction by ID                    |
+| `list_corrections()`        | List corrections (optional plan_id filter)     |
+| `list_attempts()`           | List execution attempts for a correction       |
+| `cancel_correction()`       | Cancel a pending/analyzing correction          |

features/correction_flows.feature

@@ -0,0 +1,271 @@
+@phase2 @correction @flows
+Feature: Correction flows — revert and append
+  As an operator
+  I want to revert or append corrections to a plan's decision tree
+  So that I can fix mistakes without restarting from scratch
+
+  # ───────────────────────────────────────────────────────────
+  # Revert scenarios
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Simple revert invalidates target decision
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a decision tree where "D1" has no children
+    When I execute the revert correction
+    Then the correction flow result status should be "applied"
+    And the reverted decisions should contain "D1"
+
+  Scenario: Revert deep subtree invalidates all descendants
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a multi-parent decision tree with "D1" children "D2,D3" and "D2" children "D4,D5"
+    When I execute the revert correction
+    Then the reverted decisions should contain "D1"
+    And the reverted decisions should contain "D2"
+    And the reverted decisions should contain "D3"
+    And the reverted decisions should contain "D4"
+    And the reverted decisions should contain "D5"
+
+  Scenario: Revert archives artifacts for each affected decision
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a simple decision tree where "D1" has children "D2"
+    When I execute the revert correction
+    Then the archived artifacts should include "D1.artifact"
+    And the archived artifacts should include "D2.artifact"
+
+  Scenario: Revert correction transitions through expected statuses
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    Then the correction flow status should be "pending"
+    When I execute the revert correction with a decision tree where "D1" has no children
+    Then the correction flow status should be "applied"
+
+  # ───────────────────────────────────────────────────────────
+  # Append scenarios
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Simple append spawns child plan
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in append mode
+    When I execute the append correction
+    Then the correction flow result status should be "applied"
+    And the result should have a spawned child plan id
+
+  Scenario: Append with guidance preserves guidance on request
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in append mode with guidance "Fix auth module"
+    When I execute the append correction
+    Then the correction flow guidance should be "Fix auth module"
+
+  Scenario: Append creates new decision in result
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in append mode
+    When I execute the append correction
+    Then the result new decisions should not be empty
+
+  # ───────────────────────────────────────────────────────────
+  # Impact analysis
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Impact analysis with BFS subtree traversal
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a multi-parent decision tree with "D1" children "D2,D3" and "D3" children "D6"
+    When I analyze the impact
+    Then the affected decisions should be "D1,D2,D3,D6"
+
+  Scenario: Impact analysis risk level low for small subtree
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a simple decision tree where "D1" has children "D2"
+    When I analyze the impact
+    Then the risk level should be "low"
+
+  Scenario: Impact analysis risk level medium for moderate subtree
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a decision tree with 7 nodes rooted at "D1"
+    When I analyze the impact
+    Then the risk level should be "medium"
+
+  Scenario: Impact analysis risk level high for large subtree
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a decision tree with 15 nodes rooted at "D1"
+    When I analyze the impact
+    Then the risk level should be "high"
+
+  Scenario: Impact analysis on leaf node returns only the target
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "LEAF" in revert mode
+    And an empty decision tree
+    When I analyze the impact
+    Then the affected decisions should be "LEAF"
+
+  # ───────────────────────────────────────────────────────────
+  # Dry-run
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Dry-run report generation for revert
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a simple decision tree where "D1" has children "D2,D3"
+    When I generate a dry-run report
+    Then the report mode should be "revert"
+    And the report decisions to invalidate should contain "D1"
+    And the report estimated recompute time should be greater than 0
+
+  Scenario: Dry-run report generates warnings for high risk
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a decision tree with 15 nodes rooted at "D1"
+    When I generate a dry-run report
+    Then the report warnings should contain "High risk"
+
+  Scenario: Dry-run report generates warnings for medium risk
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a decision tree with 7 nodes rooted at "D1"
+    When I generate a dry-run report
+    Then the report warnings should contain "Medium risk"
+
+  Scenario: Dry-run report for append has no decisions to invalidate
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in append mode
+    When I generate a dry-run report
+    Then the report decisions to invalidate should be empty
+
+  # ───────────────────────────────────────────────────────────
+  # Error handling
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Get correction with invalid id raises not found
+    Given a correction flow service
+    When I try to get correction flow "nonexistent-id"
+    Then a correction flow resource not found error should be raised
+
+  Scenario: Execute already applied correction raises validation error
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And I execute the revert correction with a decision tree where "D1" has no children
+    When I try to execute the revert correction again
+    Then a validation error should be raised for status
+
+  Scenario: Execute already cancelled correction raises validation error
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And I cancel the correction flow
+    When I try to execute the revert correction again
+    Then a validation error should be raised for status
+
+  Scenario: Cancel already applied correction raises validation error
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And I execute the revert correction with a decision tree where "D1" has no children
+    When I try to cancel the correction flow
+    Then a validation error should be raised for status
+
+  # ───────────────────────────────────────────────────────────
+  # Status transitions
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Correction status transitions from pending to applied on revert
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    When I execute the revert correction with a decision tree where "D1" has no children
+    Then the correction flow status should be "applied"
+
+  Scenario: Correction status transitions from pending to cancelled
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    When I cancel the correction flow
+    Then the correction flow status should be "cancelled"
+
+  # ───────────────────────────────────────────────────────────
+  # Multi-correction & listing
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Multiple corrections on same plan are listed correctly
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a correction request for plan "P1" targeting decision "D2" in append mode
+    When I list correction flows for plan "P1"
+    Then the correction list should have 2 items
+
+  Scenario: List corrections filters by plan id
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a correction request for plan "P2" targeting decision "D2" in append mode
+    When I list correction flows for plan "P1"
+    Then the correction list should have 1 items
+
+  # ───────────────────────────────────────────────────────────
+  # Cancellation
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Cancel pending correction succeeds
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    When I cancel the correction flow
+    Then the correction flow status should be "cancelled"
+
+  # ───────────────────────────────────────────────────────────
+  # Attempt tracking
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Execution records an attempt
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    When I execute the revert correction with a decision tree where "D1" has no children
+    Then the attempt list should have 1 entries
+    And the first attempt should be successful
+
+  Scenario: Append execution records an attempt
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in append mode
+    When I execute the append correction
+    Then the attempt list should have 1 entries
+
+  # ───────────────────────────────────────────────────────────
+  # Model validation
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: CorrectionRequest rejects empty plan_id
+    Given a correction flow service
+    When I try to create a correction with empty plan_id
+    Then a validation error should be raised for empty field
+
+  Scenario: CorrectionRequest rejects empty target_decision_id
+    Given a correction flow service
+    When I try to create a correction with empty target_decision_id
+    Then a validation error should be raised for empty field
+
+  Scenario: CorrectionImpact rejects invalid risk level
+    When I try to create a correction impact with risk level "critical"
+    Then a correction flow pydantic validation error should be raised
+
+  Scenario: CorrectionDryRunReport contains all expected fields
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a simple decision tree where "D1" has children "D2"
+    When I generate a dry-run report
+    Then the report should have a correction_id
+    And the report should have an impact object
+    And the report should have a warnings list
+
+  Scenario: Execute correction dispatches to revert
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in revert mode
+    And a decision tree where "D1" has no children
+    When I execute correction via dispatch
+    Then the correction flow result status should be "applied"
+    And the reverted decisions should contain "D1"
+
+  Scenario: Execute correction dispatches to append
+    Given a correction flow service
+    And a correction request for plan "P1" targeting decision "D1" in append mode
+    When I execute correction via dispatch
+    Then the correction flow result status should be "applied"
+    And the result should have a spawned child plan id

features/steps/correction_flows_steps.py

@@ -0,0 +1,419 @@
+"""Step definitions for correction_flows.feature.
+
+Exercises CorrectionService revert/append flows, impact analysis,
+dry-run reports, error handling, and model validation.
+"""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from pydantic import ValidationError as PydanticValidationError
+
+from cleveragents.application.services.correction_service import CorrectionService
+from cleveragents.core.exceptions import ResourceNotFoundError, ValidationError
+from cleveragents.domain.models.core.correction import (
+    CorrectionImpact,
+    CorrectionMode,
+)
+
+# -------------------------------------------------------------------
+# Helpers
+# -------------------------------------------------------------------
+
+
+def _build_chain_tree(root: str, count: int) -> dict[str, list[str]]:
+    """Build a linear chain tree with *count* total nodes starting at *root*."""
+    tree: dict[str, list[str]] = {}
+    current = root
+    for i in range(1, count):
+        child = f"{root}_child{i}"
+        tree[current] = [child]
+        current = child
+    return tree
+
+
+# -------------------------------------------------------------------
+# Given steps
+# -------------------------------------------------------------------
+
+
+@given("a correction flow service")
+def step_create_service(context):
+    context.service = CorrectionService()
+    context.correction_id = None
+    context.decision_tree = None
+    context.result = None
+    context.report = None
+    context.impact = None
+    context.error = None
+    context.corrections_list = None
+
+
+@given(
+    'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in revert mode'
+)
+def step_create_revert_request(context, plan_id, decision_id):
+    req = context.service.request_correction(
+        plan_id=plan_id,
+        target_decision_id=decision_id,
+        mode=CorrectionMode.REVERT,
+    )
+    context.correction_id = req.correction_id
+
+
+@given(
+    'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in append mode'
+)
+def step_create_append_request(context, plan_id, decision_id):
+    req = context.service.request_correction(
+        plan_id=plan_id,
+        target_decision_id=decision_id,
+        mode=CorrectionMode.APPEND,
+    )
+    context.correction_id = req.correction_id
+
+
+@given(
+    'a correction request for plan "{plan_id}" targeting decision "{decision_id}" in append mode with guidance "{guidance}"'
+)
+def step_create_append_with_guidance(context, plan_id, decision_id, guidance):
+    req = context.service.request_correction(
+        plan_id=plan_id,
+        target_decision_id=decision_id,
+        mode=CorrectionMode.APPEND,
+        guidance=guidance,
+    )
+    context.correction_id = req.correction_id
+
+
+@given('a decision tree where "{parent}" has no children')
+def step_tree_no_children(context, parent):
+    context.decision_tree = {}
+
+
+@given('a simple decision tree where "{parent}" has children "{children}"')
+def step_tree_with_children(context, parent, children):
+    tree = getattr(context, "decision_tree", None) or {}
+    tree[parent] = [c.strip() for c in children.split(",")]
+    context.decision_tree = tree
+
+
+@given(
+    'a multi-parent decision tree with "{p1}" children "{c1}" and "{p2}" children "{c2}"'
+)
+def step_tree_two_parents(context, p1, c1, p2, c2):
+    tree = getattr(context, "decision_tree", None) or {}
+    tree[p1] = [c.strip() for c in c1.split(",")]
+    tree[p2] = [c.strip() for c in c2.split(",")]
+    context.decision_tree = tree
+
+
+@given('a decision tree with {count:d} nodes rooted at "{root}"')
+def step_tree_with_n_nodes(context, count, root):
+    context.decision_tree = _build_chain_tree(root, count)
+
+
+@given("an empty decision tree")
+def step_empty_tree(context):
+    context.decision_tree = {}
+
+
+@given(
+    'I execute the revert correction with a decision tree where "{parent}" has no children'
+)
+def step_execute_revert_inline(context, parent):
+    context.decision_tree = {}
+    context.result = context.service.execute_revert(
+        context.correction_id, context.decision_tree
+    )
+
+
+@given("I cancel the correction flow")
+def step_cancel_given(context):
+    context.service.cancel_correction(context.correction_id)
+
+
+# -------------------------------------------------------------------
+# When steps
+# -------------------------------------------------------------------
+
+
+@when("I execute the revert correction")
+def step_execute_revert(context):
+    context.result = context.service.execute_revert(
+        context.correction_id, context.decision_tree
+    )
+
+
+@when(
+    'I execute the revert correction with a decision tree where "{parent}" has no children'
+)
+def step_when_execute_revert_inline(context, parent):
+    context.decision_tree = {}
+    context.result = context.service.execute_revert(
+        context.correction_id, context.decision_tree
+    )
+
+
+@when("I execute the append correction")
+def step_execute_append(context):
+    context.result = context.service.execute_append(context.correction_id)
+
+
+@when("I analyze the impact")
+def step_analyze_impact(context):
+    context.impact = context.service.analyze_impact(
+        context.correction_id, context.decision_tree
+    )
+
+
+@when("I generate a dry-run report")
+def step_generate_dry_run(context):
+    context.report = context.service.generate_dry_run_report(
+        context.correction_id, context.decision_tree
+    )
+
+
+@when('I try to get correction flow "{cid}"')
+def step_try_get_correction(context, cid):
+    try:
+        context.service.get_correction(cid)
+        context.error = None
+    except ResourceNotFoundError as exc:
+        context.error = exc
+
+
+@when("I try to execute the revert correction again")
+def step_try_execute_revert_again(context):
+    try:
+        context.service.execute_revert(context.correction_id, {})
+        context.error = None
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to cancel the correction flow")
+def step_try_cancel(context):
+    try:
+        context.service.cancel_correction(context.correction_id)
+        context.error = None
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I cancel the correction flow")
+def step_cancel_correction(context):
+    context.service.cancel_correction(context.correction_id)
+
+
+@when('I list correction flows for plan "{plan_id}"')
+def step_list_corrections(context, plan_id):
+    context.corrections_list = context.service.list_corrections(plan_id=plan_id)
+
+
+@when("I try to create a correction with empty plan_id")
+def step_create_empty_plan_id(context):
+    try:
+        context.service.request_correction(
+            plan_id="",
+            target_decision_id="D1",
+            mode=CorrectionMode.REVERT,
+        )
+        context.error = None
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to create a correction with empty target_decision_id")
+def step_create_empty_target(context):
+    try:
+        context.service.request_correction(
+            plan_id="P1",
+            target_decision_id="",
+            mode=CorrectionMode.REVERT,
+        )
+        context.error = None
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when('I try to create a correction impact with risk level "{level}"')
+def step_create_bad_risk_level(context, level):
+    try:
+        CorrectionImpact(risk_level=level)
+        context.error = None
+    except PydanticValidationError as exc:
+        context.error = exc
+
+
+@when("I execute correction via dispatch")
+def step_execute_dispatch(context):
+    context.result = context.service.execute_correction(
+        context.correction_id, context.decision_tree
+    )
+
+
+# -------------------------------------------------------------------
+# Then steps
+# -------------------------------------------------------------------
+
+
+@then('the correction flow result status should be "{status}"')
+def step_result_status(context, status):
+    assert context.result is not None, "No result produced"
+    assert context.result.status == status, (
+        f"Expected status '{status}', got '{context.result.status}'"
+    )
+
+
+@then('the reverted decisions should contain "{decision_id}"')
+def step_reverted_contains(context, decision_id):
+    assert decision_id in context.result.reverted_decisions, (
+        f"'{decision_id}' not in reverted: {context.result.reverted_decisions}"
+    )
+
+
+@then('the archived artifacts should include "{artifact}"')
+def step_archived_artifact(context, artifact):
+    assert artifact in context.result.archived_artifacts, (
+        f"'{artifact}' not in archived: {context.result.archived_artifacts}"
+    )
+
+
+@then('the correction flow status should be "{status}"')
+def step_correction_status(context, status):
+    req = context.service.get_correction(context.correction_id)
+    assert req.status == status, (
+        f"Expected correction status '{status}', got '{req.status}'"
+    )
+
+
+@then("the result should have a spawned child plan id")
+def step_has_spawned_child(context):
+    assert context.result.spawned_child_plan_id is not None, (
+        "Expected spawned_child_plan_id, got None"
+    )
+
+
+@then('the correction flow guidance should be "{guidance}"')
+def step_guidance(context, guidance):
+    req = context.service.get_correction(context.correction_id)
+    assert req.guidance == guidance, (
+        f"Expected guidance '{guidance}', got '{req.guidance}'"
+    )
+
+
+@then("the result new decisions should not be empty")
+def step_new_decisions_not_empty(context):
+    assert len(context.result.new_decisions) > 0, "new_decisions is empty"
+
+
+@then('the affected decisions should be "{expected}"')
+def step_affected_decisions(context, expected):
+    expected_list = [d.strip() for d in expected.split(",")]
+    assert context.impact.affected_decisions == expected_list, (
+        f"Expected {expected_list}, got {context.impact.affected_decisions}"
+    )
+
+
+@then('the risk level should be "{level}"')
+def step_risk_level(context, level):
+    assert context.impact.risk_level == level, (
+        f"Expected risk '{level}', got '{context.impact.risk_level}'"
+    )
+
+
+@then('the report mode should be "{mode}"')
+def step_report_mode(context, mode):
+    assert context.report.mode == mode, (
+        f"Expected report mode '{mode}', got '{context.report.mode}'"
+    )
+
+
+@then('the report decisions to invalidate should contain "{decision_id}"')
+def step_report_invalidate_contains(context, decision_id):
+    assert decision_id in context.report.decisions_to_invalidate, (
+        f"'{decision_id}' not in invalidate list"
+    )
+
+
+@then("the report estimated recompute time should be greater than 0")
+def step_report_recompute_positive(context):
+    assert context.report.estimated_recompute_time_seconds > 0
+
+
+@then('the report warnings should contain "{fragment}"')
+def step_report_warning_fragment(context, fragment):
+    joined = " ".join(context.report.warnings)
+    assert fragment in joined, (
+        f"'{fragment}' not found in warnings: {context.report.warnings}"
+    )
+
+
+@then("the report decisions to invalidate should be empty")
+def step_report_invalidate_empty(context):
+    assert len(context.report.decisions_to_invalidate) == 0, (
+        f"Expected empty invalidate list, got {context.report.decisions_to_invalidate}"
+    )
+
+
+@then("a correction flow resource not found error should be raised")
+def step_resource_not_found(context):
+    assert isinstance(context.error, ResourceNotFoundError), (
+        f"Expected ResourceNotFoundError, got {type(context.error)}"
+    )
+
+
+@then("a validation error should be raised for status")
+def step_validation_error_status(context):
+    assert isinstance(context.error, ValidationError), (
+        f"Expected ValidationError, got {type(context.error)}"
+    )
+
+
+@then("a validation error should be raised for empty field")
+def step_validation_error_empty(context):
+    assert isinstance(context.error, ValidationError), (
+        f"Expected ValidationError, got {type(context.error)}"
+    )
+
+
+@then("a correction flow pydantic validation error should be raised")
+def step_pydantic_error(context):
+    assert isinstance(context.error, PydanticValidationError), (
+        f"Expected PydanticValidationError, got {type(context.error)}"
+    )
+
+
+@then("the correction list should have {count:d} items")
+def step_list_count(context, count):
+    assert len(context.corrections_list) == count, (
+        f"Expected {count} corrections, got {len(context.corrections_list)}"
+    )
+
+
+@then("the attempt list should have {count:d} entries")
+def step_attempt_count(context, count):
+    attempts = context.service.list_attempts(context.correction_id)
+    assert len(attempts) == count, f"Expected {count} attempts, got {len(attempts)}"
+
+
+@then("the first attempt should be successful")
+def step_first_attempt_success(context):
+    attempts = context.service.list_attempts(context.correction_id)
+    assert attempts[0].success is True, "First attempt was not successful"
+
+
+@then("the report should have a correction_id")
+def step_report_has_cid(context):
+    assert context.report.correction_id is not None
+
+
+@then("the report should have an impact object")
+def step_report_has_impact(context):
+    assert context.report.impact is not None
+
+
+@then("the report should have a warnings list")
+def step_report_has_warnings(context):
+    assert isinstance(context.report.warnings, list)

robot/correction_flows.robot

@@ -0,0 +1,66 @@
+*** Settings ***
+Documentation     Smoke tests for correction flows: revert, append, dry-run,
+...               impact analysis, cancellation, and error handling.
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    robot/helper_correction_flows.py
+
+*** Test Cases ***
+Revert Simple Correction
+    [Documentation]    Verify simple revert invalidates target decision
+    [Tags]    phase2    correction    flows    revert
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    revert-simple    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    revert-simple-ok
+
+Revert Deep Subtree
+    [Documentation]    Verify revert walks full BFS subtree
+    [Tags]    phase2    correction    flows    revert
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    revert-deep-subtree    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    revert-deep-subtree-ok
+
+Append Correction
+    [Documentation]    Verify append spawns a child plan
+    [Tags]    phase2    correction    flows    append
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    append-correction    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    append-correction-ok
+
+Dry Run Report Generation
+    [Documentation]    Verify dry-run report contains expected fields
+    [Tags]    phase2    correction    flows    dryrun
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    dry-run-report    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    dry-run-report-ok
+
+Impact Analysis Risk Levels
+    [Documentation]    Verify risk classification low / high thresholds
+    [Tags]    phase2    correction    flows    impact
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    impact-risk-levels    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    impact-risk-levels-ok
+
+Cancel Correction
+    [Documentation]    Verify pending correction can be cancelled
+    [Tags]    phase2    correction    flows    cancel
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cancel-correction    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cancel-correction-ok
+
+Execute Already Cancelled Error
+    [Documentation]    Verify executing a cancelled correction raises error
+    [Tags]    phase2    correction    flows    error
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    execute-cancelled-error    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    execute-cancelled-error-ok
+
+List Corrections By Plan
+    [Documentation]    Verify listing corrections filters by plan_id
+    [Tags]    phase2    correction    flows    query
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    list-corrections-by-plan    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    list-corrections-by-plan-ok

robot/helper_correction_flows.py

@@ -0,0 +1,157 @@
+"""Helper script for correction flows Robot Framework smoke tests.
+
+Exercises CorrectionService revert/append flows, impact analysis,
+dry-run reports, and cancellation without requiring persistence.
+"""
+
+from __future__ import annotations
+
+import sys
+
+from cleveragents.application.services.correction_service import CorrectionService
+from cleveragents.core.exceptions import ValidationError
+from cleveragents.domain.models.core.correction import CorrectionMode
+
+# ---------------------------------------------------------------------------
+# Shared fixtures
+# ---------------------------------------------------------------------------
+_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+_DECISION_D1 = "D1"
+_DECISION_D2 = "D2"
+_DECISION_D3 = "D3"
+
+
+def _simple_tree() -> dict[str, list[str]]:
+    return {_DECISION_D1: [_DECISION_D2, _DECISION_D3]}
+
+
+def _deep_tree() -> dict[str, list[str]]:
+    return {
+        _DECISION_D1: [_DECISION_D2, _DECISION_D3],
+        _DECISION_D2: ["D4", "D5"],
+        _DECISION_D3: ["D6"],
+    }
+
+
+# ---------------------------------------------------------------------------
+# Command handlers
+# ---------------------------------------------------------------------------
+
+
+def _revert_simple() -> None:
+    svc = CorrectionService()
+    req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
+    result = svc.execute_revert(req.correction_id, {_DECISION_D1: []})
+    assert result.status == "applied", f"Expected applied, got {result.status}"
+    assert _DECISION_D1 in result.reverted_decisions
+    print("revert-simple-ok")
+
+
+def _revert_deep_subtree() -> None:
+    svc = CorrectionService()
+    req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
+    result = svc.execute_revert(req.correction_id, _deep_tree())
+    assert len(result.reverted_decisions) == 6
+    print("revert-deep-subtree-ok")
+
+
+def _append_correction() -> None:
+    svc = CorrectionService()
+    req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.APPEND)
+    result = svc.execute_append(req.correction_id)
+    assert result.status == "applied"
+    assert result.spawned_child_plan_id is not None
+    print("append-correction-ok")
+
+
+def _dry_run_report() -> None:
+    svc = CorrectionService()
+    req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
+    report = svc.generate_dry_run_report(req.correction_id, _simple_tree())
+    assert report.mode == "revert"
+    assert len(report.decisions_to_invalidate) == 3
+    assert report.estimated_recompute_time_seconds > 0
+    print("dry-run-report-ok")
+
+
+def _impact_risk_levels() -> None:
+    svc = CorrectionService()
+    # Low risk: 2 nodes
+    req1 = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
+    impact1 = svc.analyze_impact(req1.correction_id, {_DECISION_D1: [_DECISION_D2]})
+    assert impact1.risk_level == "low", f"Expected low, got {impact1.risk_level}"
+
+    # High risk: > 10 nodes (chain)
+    tree: dict[str, list[str]] = {}
+    current = "R"
+    for i in range(14):
+        child = f"R_c{i}"
+        tree[current] = [child]
+        current = child
+    req2 = svc.request_correction(_PLAN_ID, "R", CorrectionMode.REVERT)
+    impact2 = svc.analyze_impact(req2.correction_id, tree)
+    assert impact2.risk_level == "high", f"Expected high, got {impact2.risk_level}"
+    print("impact-risk-levels-ok")
+
+
+def _cancel_correction() -> None:
+    svc = CorrectionService()
+    req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
+    svc.cancel_correction(req.correction_id)
+    updated = svc.get_correction(req.correction_id)
+    assert updated.status == "cancelled"
+    print("cancel-correction-ok")
+
+
+def _execute_cancelled_error() -> None:
+    svc = CorrectionService()
+    req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
+    svc.cancel_correction(req.correction_id)
+    try:
+        svc.execute_revert(req.correction_id, {})
+        print("FAIL: expected ValidationError")
+        sys.exit(1)
+    except ValidationError:
+        print("execute-cancelled-error-ok")
+
+
+def _list_corrections_by_plan() -> None:
+    svc = CorrectionService()
+    svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
+    svc.request_correction(_PLAN_ID, _DECISION_D2, CorrectionMode.APPEND)
+    svc.request_correction("OTHER_PLAN", _DECISION_D3, CorrectionMode.REVERT)
+    result = svc.list_corrections(plan_id=_PLAN_ID)
+    assert len(result) == 2, f"Expected 2, got {len(result)}"
+    print("list-corrections-by-plan-ok")
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+COMMANDS: dict[str, object] = {
+    "revert-simple": _revert_simple,
+    "revert-deep-subtree": _revert_deep_subtree,
+    "append-correction": _append_correction,
+    "dry-run-report": _dry_run_report,
+    "impact-risk-levels": _impact_risk_levels,
+    "cancel-correction": _cancel_correction,
+    "execute-cancelled-error": _execute_cancelled_error,
+    "list-corrections-by-plan": _list_corrections_by_plan,
+}
+
+
+def main() -> None:
+    """Dispatch command from sys.argv."""
+    if len(sys.argv) < 2:
+        raise SystemExit("Expected command argument")
+    command = sys.argv[1]
+    if command not in COMMANDS:
+        raise SystemExit(f"Unknown command: {command}")
+    func = COMMANDS[command]
+    if callable(func):
+        func()
+
+
+if __name__ == "__main__":
+    main()

src/cleveragents/application/services/__init__.py

@@ -3,6 +3,9 @@
 Contains service classes that orchestrate business operations.
 """
 
+from cleveragents.application.services.correction_service import (
+    CorrectionService,
+)
 from cleveragents.application.services.session_service import (
     PersistentSessionService,
 )
@@ -14,6 +17,7 @@ from cleveragents.application.services.tool_registry_service import (
 )
 
 __all__ = [
+    "CorrectionService",
     "PersistentSessionService",
     "SkillRegistryService",
     "ToolRegistryService",

src/cleveragents/application/services/correction_service.py

@@ -0,0 +1,465 @@
+"""Correction service implementing revert and append flows.
+
+Orchestrates impact analysis using BFS subtree traversal, dry-run
+reporting, revert execution (decision invalidation + artifact archival),
+and append execution (child plan spawning).
+"""
+
+from __future__ import annotations
+
+from collections import deque
+from datetime import UTC, datetime
+
+import structlog
+from ulid import ULID
+
+from cleveragents.core.exceptions import ResourceNotFoundError, ValidationError
+from cleveragents.domain.models.core.correction import (
+    CorrectionAttempt,
+    CorrectionDryRunReport,
+    CorrectionImpact,
+    CorrectionMode,
+    CorrectionRequest,
+    CorrectionResult,
+    CorrectionStatus,
+)
+
+logger = structlog.get_logger(__name__)
+
+# ---------------------------------------------------------------------------
+# Risk-level thresholds
+# ---------------------------------------------------------------------------
+_RISK_LOW_MAX = 3
+_RISK_MEDIUM_MAX = 10
+
+
+class CorrectionService:
+    """Service for creating, analysing, and executing decision corrections.
+
+    State is held in-memory via dictionaries keyed by ``correction_id``.
+    A production deployment would swap these for repository adapters.
+    """
+
+    def __init__(self) -> None:
+        self._corrections: dict[str, CorrectionRequest] = {}
+        self._impacts: dict[str, CorrectionImpact] = {}
+        self._attempts: dict[str, list[CorrectionAttempt]] = {}
+        self._results: dict[str, CorrectionResult] = {}
+
+    # ------------------------------------------------------------------
+    # Creation
+    # ------------------------------------------------------------------
+
+    def request_correction(
+        self,
+        plan_id: str,
+        target_decision_id: str,
+        mode: CorrectionMode,
+        guidance: str = "",
+        dry_run: bool = False,
+    ) -> CorrectionRequest:
+        """Create and register a new correction request.
+
+        Args:
+            plan_id: Plan owning the decision tree.
+            target_decision_id: Decision node to target.
+            mode: ``CorrectionMode.REVERT`` or ``CorrectionMode.APPEND``.
+            guidance: Optional human guidance text.
+            dry_run: If *True*, only impact analysis will be performed.
+
+        Returns:
+            The newly created ``CorrectionRequest``.
+
+        Raises:
+            ValidationError: If required parameters are empty.
+        """
+        if not plan_id or not plan_id.strip():
+            raise ValidationError("plan_id must not be empty")
+        if not target_decision_id or not target_decision_id.strip():
+            raise ValidationError("target_decision_id must not be empty")
+
+        request = CorrectionRequest(
+            plan_id=plan_id,
+            target_decision_id=target_decision_id,
+            mode=mode,
+            guidance=guidance,
+            dry_run=dry_run,
+        )
+        self._corrections[request.correction_id] = request
+        self._attempts[request.correction_id] = []
+
+        logger.info(
+            "correction.requested",
+            correction_id=request.correction_id,
+            plan_id=plan_id,
+            mode=mode,
+            dry_run=dry_run,
+        )
+        return request
+
+    # ------------------------------------------------------------------
+    # Impact analysis
+    # ------------------------------------------------------------------
+
+    def analyze_impact(
+        self,
+        correction_id: str,
+        decision_tree: dict[str, list[str]] | None = None,
+    ) -> CorrectionImpact:
+        """Compute the impact of a correction via BFS subtree traversal.
+
+        Args:
+            correction_id: Previously created correction request ID.
+            decision_tree: Adjacency list mapping parent → children.
+
+        Returns:
+            ``CorrectionImpact`` with affected nodes and risk level.
+
+        Raises:
+            ResourceNotFoundError: If the correction does not exist.
+        """
+        request = self._get_request_or_raise(correction_id)
+        request.status = CorrectionStatus.ANALYZING
+
+        tree = decision_tree or {}
+        affected = self._compute_affected_subtree(request.target_decision_id, tree)
+        risk = self._classify_risk(len(affected))
+
+        # Derive artefacts from decision IDs (convention: <id>.artifact)
+        artifacts = [f"{d}.artifact" for d in affected]
+
+        impact = CorrectionImpact(
+            affected_decisions=affected,
+            affected_files=[f"{d}.py" for d in affected],
+            affected_child_plans=[],
+            estimated_cost=float(len(affected)) * 1.5,
+            risk_level=risk,
+            rollback_tier="full"
+            if request.mode == CorrectionMode.REVERT
+            else "append_only",
+            artifacts_to_archive=artifacts,
+        )
+        self._impacts[correction_id] = impact
+
+        logger.info(
+            "correction.impact_analyzed",
+            correction_id=correction_id,
+            affected_count=len(affected),
+            risk_level=risk,
+        )
+        return impact
+
+    # ------------------------------------------------------------------
+    # Dry-run report
+    # ------------------------------------------------------------------
+
+    def generate_dry_run_report(
+        self,
+        correction_id: str,
+        decision_tree: dict[str, list[str]] | None = None,
+    ) -> CorrectionDryRunReport:
+        """Generate a dry-run report without mutating state.
+
+        Calls ``analyze_impact`` internally, then assembles warnings
+        and estimated recompute time.
+
+        Args:
+            correction_id: Correction request ID.
+            decision_tree: Optional decision tree adjacency list.
+
+        Returns:
+            ``CorrectionDryRunReport`` describing what *would* happen.
+        """
+        request = self._get_request_or_raise(correction_id)
+        impact = self.analyze_impact(correction_id, decision_tree)
+
+        warnings: list[str] = []
+        if impact.risk_level == "high":
+            warnings.append(
+                "High risk: more than 10 decisions affected. "
+                "Review carefully before executing."
+            )
+        if 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."
+            )
+
+        recompute_seconds = float(len(impact.affected_decisions)) * 2.0
+
+        report = CorrectionDryRunReport(
+            correction_id=correction_id,
+            mode=request.mode,
+            impact=impact,
+            decisions_to_invalidate=impact.affected_decisions
+            if request.mode == CorrectionMode.REVERT
+            else [],
+            child_plans_to_rollback=impact.affected_child_plans,
+            estimated_recompute_time_seconds=recompute_seconds,
+            warnings=warnings,
+        )
+
+        logger.info(
+            "correction.dry_run_generated",
+            correction_id=correction_id,
+            warning_count=len(warnings),
+        )
+        return report
+
+    # ------------------------------------------------------------------
+    # Execution: revert
+    # ------------------------------------------------------------------
+
+    def execute_revert(
+        self,
+        correction_id: str,
+        decision_tree: dict[str, list[str]] | None = None,
+    ) -> CorrectionResult:
+        """Execute a revert correction.
+
+        Marks every decision in the affected subtree as reverted and
+        records all archived artifacts.
+
+        Args:
+            correction_id: Correction request ID.
+            decision_tree: Optional adjacency list for BFS.
+
+        Returns:
+            ``CorrectionResult`` with reverted decisions.
+
+        Raises:
+            ResourceNotFoundError: If correction does not exist.
+            ValidationError: If correction is not in PENDING or ANALYZING status.
+        """
+        request = self._get_request_or_raise(correction_id)
+        self._assert_executable(request)
+        request.status = CorrectionStatus.EXECUTING
+
+        attempt = CorrectionAttempt(correction_id=correction_id)
+        self._attempts[correction_id].append(attempt)
+
+        try:
+            impact = self.analyze_impact(correction_id, decision_tree)
+
+            result = CorrectionResult(
+                correction_id=correction_id,
+                status=CorrectionStatus.APPLIED,
+                reverted_decisions=impact.affected_decisions,
+                archived_artifacts=impact.artifacts_to_archive,
+            )
+            request.status = CorrectionStatus.APPLIED
+            attempt.success = True
+        except Exception as exc:
+            result = CorrectionResult(
+                correction_id=correction_id,
+                status=CorrectionStatus.FAILED,
+                error_message=str(exc),
+            )
+            request.status = CorrectionStatus.FAILED
+            attempt.success = False
+            attempt.details = {"error": str(exc)}
+        finally:
+            attempt.completed_at = datetime.now(UTC)
+
+        self._results[correction_id] = result
+        logger.info(
+            "correction.revert_executed",
+            correction_id=correction_id,
+            status=result.status,
+        )
+        return result
+
+    # ------------------------------------------------------------------
+    # Execution: append
+    # ------------------------------------------------------------------
+
+    def execute_append(
+        self,
+        correction_id: str,
+    ) -> CorrectionResult:
+        """Execute an append correction.
+
+        Spawns a new child plan reference and preserves the original
+        decision node.
+
+        Args:
+            correction_id: Correction request ID.
+
+        Returns:
+            ``CorrectionResult`` with the spawned child plan ID.
+
+        Raises:
+            ResourceNotFoundError: If correction does not exist.
+            ValidationError: If correction is not in PENDING or ANALYZING status.
+        """
+        request = self._get_request_or_raise(correction_id)
+        self._assert_executable(request)
+        request.status = CorrectionStatus.EXECUTING
+
+        attempt = CorrectionAttempt(correction_id=correction_id)
+        self._attempts[correction_id].append(attempt)
+
+        try:
+            child_plan_id = str(ULID())
+            new_decision_id = str(ULID())
+
+            result = CorrectionResult(
+                correction_id=correction_id,
+                status=CorrectionStatus.APPLIED,
+                new_decisions=[new_decision_id],
+                spawned_child_plan_id=child_plan_id,
+            )
+            request.status = CorrectionStatus.APPLIED
+            attempt.success = True
+            attempt.details = {
+                "spawned_child_plan_id": child_plan_id,
+                "new_decision_id": new_decision_id,
+            }
+        except Exception as exc:
+            result = CorrectionResult(
+                correction_id=correction_id,
+                status=CorrectionStatus.FAILED,
+                error_message=str(exc),
+            )
+            request.status = CorrectionStatus.FAILED
+            attempt.success = False
+            attempt.details = {"error": str(exc)}
+        finally:
+            attempt.completed_at = datetime.now(UTC)
+
+        self._results[correction_id] = result
+        logger.info(
+            "correction.append_executed",
+            correction_id=correction_id,
+            status=result.status,
+        )
+        return result
+
+    # ------------------------------------------------------------------
+    # Dispatch
+    # ------------------------------------------------------------------
+
+    def execute_correction(
+        self,
+        correction_id: str,
+        decision_tree: dict[str, list[str]] | None = None,
+    ) -> CorrectionResult:
+        """Execute a correction, dispatching to revert or append.
+
+        Args:
+            correction_id: Correction request ID.
+            decision_tree: Optional adjacency list (used for revert).
+
+        Returns:
+            ``CorrectionResult`` from the chosen strategy.
+        """
+        request = self._get_request_or_raise(correction_id)
+
+        if request.mode == CorrectionMode.REVERT:
+            return self.execute_revert(correction_id, decision_tree)
+        return self.execute_append(correction_id)
+
+    # ------------------------------------------------------------------
+    # Query helpers
+    # ------------------------------------------------------------------
+
+    def get_correction(self, correction_id: str) -> CorrectionRequest:
+        """Retrieve a correction request by ID.
+
+        Raises:
+            ResourceNotFoundError: If not found.
+        """
+        return self._get_request_or_raise(correction_id)
+
+    def list_corrections(self, plan_id: str | None = None) -> list[CorrectionRequest]:
+        """List corrections, optionally filtered by plan_id."""
+        corrections = list(self._corrections.values())
+        if plan_id is not None:
+            corrections = [c for c in corrections if c.plan_id == plan_id]
+        return corrections
+
+    def list_attempts(self, correction_id: str) -> list[CorrectionAttempt]:
+        """List execution attempts for a correction.
+
+        Raises:
+            ResourceNotFoundError: If correction does not exist.
+        """
+        self._get_request_or_raise(correction_id)
+        return list(self._attempts.get(correction_id, []))
+
+    def cancel_correction(self, correction_id: str) -> CorrectionRequest:
+        """Cancel a pending correction.
+
+        Raises:
+            ResourceNotFoundError: If correction does not exist.
+            ValidationError: If correction is not in a cancellable status.
+        """
+        request = self._get_request_or_raise(correction_id)
+        cancellable = {CorrectionStatus.PENDING, CorrectionStatus.ANALYZING}
+        if request.status not in cancellable:
+            raise ValidationError(
+                f"Cannot cancel correction in '{request.status}' status. "
+                f"Cancellation is only allowed in: {sorted(cancellable)}"
+            )
+        request.status = CorrectionStatus.CANCELLED
+        logger.info(
+            "correction.cancelled",
+            correction_id=correction_id,
+        )
+        return request
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _get_request_or_raise(self, correction_id: str) -> CorrectionRequest:
+        """Look up a correction or raise ``ResourceNotFoundError``."""
+        request = self._corrections.get(correction_id)
+        if request is None:
+            raise ResourceNotFoundError(
+                resource_type="correction",
+                resource_id=correction_id,
+            )
+        return request
+
+    def _assert_executable(self, request: CorrectionRequest) -> None:
+        """Ensure the correction is in an executable status."""
+        executable = {CorrectionStatus.PENDING, CorrectionStatus.ANALYZING}
+        if request.status not in executable:
+            raise ValidationError(
+                f"Cannot execute correction in '{request.status}' status. "
+                f"Execution requires status in: {sorted(executable)}"
+            )
+
+    @staticmethod
+    def _compute_affected_subtree(
+        target_id: str, tree: dict[str, list[str]]
+    ) -> list[str]:
+        """BFS walk from *target_id* through the decision tree.
+
+        Returns all reachable node IDs (inclusive of the target itself),
+        preserving BFS visit order.
+        """
+        affected: list[str] = []
+        queue: deque[str] = deque([target_id])
+        while queue:
+            node = queue.popleft()
+            affected.append(node)
+            queue.extend(tree.get(node, []))
+        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"
+
+
+__all__ = [
+    "CorrectionService",
+]

src/cleveragents/domain/models/core/__init__.py

@@ -37,6 +37,15 @@ from cleveragents.domain.models.core.context_policy import (
     ContextView,
     ProjectContextPolicy,
 )
+from cleveragents.domain.models.core.correction import (
+    CorrectionAttempt,
+    CorrectionDryRunReport,
+    CorrectionImpact,
+    CorrectionMode,
+    CorrectionRequest,
+    CorrectionResult,
+    CorrectionStatus,
+)
 from cleveragents.domain.models.core.debug_attempt import DebugAttempt
 from cleveragents.domain.models.core.org import (
     CloudBillingFields,
@@ -166,6 +175,13 @@ __all__ = [
     "ContextType",
     "ContextUpdateResult",
     "ContextView",
+    "CorrectionAttempt",
+    "CorrectionDryRunReport",
+    "CorrectionImpact",
+    "CorrectionMode",
+    "CorrectionRequest",
+    "CorrectionResult",
+    "CorrectionStatus",
     "CreditType",
     "CreditsTransaction",
     "CreditsTransactionType",

src/cleveragents/domain/models/core/correction.py

@@ -0,0 +1,272 @@
+"""Correction domain models for decision tree editing.
+
+Supports revert and append correction modes with impact analysis,
+dry-run reporting, and execution flow tracking.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+from ulid import ULID
+
+
+class CorrectionMode(StrEnum):
+    """Available correction strategies."""
+
+    REVERT = "revert"
+    APPEND = "append"
+
+
+class CorrectionStatus(StrEnum):
+    """Lifecycle states for a correction request."""
+
+    PENDING = "pending"
+    ANALYZING = "analyzing"
+    EXECUTING = "executing"
+    APPLIED = "applied"
+    FAILED = "failed"
+    CANCELLED = "cancelled"
+
+
+class CorrectionRequest(BaseModel):
+    """A request to correct a decision within a plan's decision tree.
+
+    Each request targets a specific decision node and applies either a
+    *revert* (invalidate subtree) or *append* (add new child plan) strategy.
+    """
+
+    model_config = ConfigDict(frozen=False, populate_by_name=True)
+
+    correction_id: str = Field(
+        default_factory=lambda: str(ULID()),
+        description="Unique correction identifier (ULID).",
+    )
+    plan_id: str = Field(
+        ...,
+        description="Plan that owns the targeted decision tree.",
+    )
+    target_decision_id: str = Field(
+        ...,
+        description="Decision node to apply correction at.",
+    )
+    mode: CorrectionMode = Field(
+        ...,
+        description="Correction strategy: revert or append.",
+    )
+    guidance: str = Field(
+        default="",
+        description="Human-supplied guidance for the correction.",
+    )
+    dry_run: bool = Field(
+        default=False,
+        description="When True, only compute impact without executing.",
+    )
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(UTC),
+        description="Timestamp when the request was created.",
+    )
+    status: CorrectionStatus = Field(
+        default=CorrectionStatus.PENDING,
+        description="Current lifecycle status.",
+    )
+
+    @field_validator("plan_id")
+    @classmethod
+    def _plan_id_not_empty(cls, v: str) -> str:
+        if not v or not v.strip():
+            raise ValueError("plan_id must not be empty")
+        return v
+
+    @field_validator("target_decision_id")
+    @classmethod
+    def _target_decision_id_not_empty(cls, v: str) -> str:
+        if not v or not v.strip():
+            raise ValueError("target_decision_id must not be empty")
+        return v
+
+    @field_validator("guidance")
+    @classmethod
+    def _guidance_stripped(cls, v: str) -> str:
+        return v.strip()
+
+
+class CorrectionImpact(BaseModel):
+    """Impact analysis result for a proposed correction.
+
+    Computed via BFS traversal of the decision tree rooted at the
+    targeted decision node.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    affected_decisions: list[str] = Field(
+        default_factory=list,
+        description="Decision IDs in the affected subtree (BFS order).",
+    )
+    affected_files: list[str] = Field(
+        default_factory=list,
+        description="Files touched by affected decisions.",
+    )
+    affected_child_plans: list[str] = Field(
+        default_factory=list,
+        description="Child plan IDs spawned by affected decisions.",
+    )
+    estimated_cost: float | None = Field(
+        default=None,
+        description="Estimated recompute cost (arbitrary units).",
+    )
+    risk_level: str = Field(
+        default="low",
+        description="Risk classification: low, medium, or high.",
+    )
+    rollback_tier: str = Field(
+        default="full",
+        description="Rollback granularity: full, phase, or append_only.",
+    )
+    artifacts_to_archive: list[str] = Field(
+        default_factory=list,
+        description="Artifact paths that should be archived on revert.",
+    )
+
+    @field_validator("risk_level")
+    @classmethod
+    def _valid_risk_level(cls, v: str) -> str:
+        allowed = {"low", "medium", "high"}
+        if v not in allowed:
+            raise ValueError(f"risk_level must be one of {allowed}, got '{v}'")
+        return v
+
+    @field_validator("rollback_tier")
+    @classmethod
+    def _valid_rollback_tier(cls, v: str) -> str:
+        allowed = {"full", "phase", "append_only"}
+        if v not in allowed:
+            raise ValueError(f"rollback_tier must be one of {allowed}, got '{v}'")
+        return v
+
+
+class CorrectionResult(BaseModel):
+    """Outcome of executing a correction."""
+
+    model_config = ConfigDict(frozen=True)
+
+    correction_id: str = Field(
+        ...,
+        description="Correction request this result belongs to.",
+    )
+    status: CorrectionStatus = Field(
+        ...,
+        description="Terminal status after execution.",
+    )
+    new_decisions: list[str] = Field(
+        default_factory=list,
+        description="Decision IDs created by an append correction.",
+    )
+    reverted_decisions: list[str] = Field(
+        default_factory=list,
+        description="Decision IDs invalidated by a revert correction.",
+    )
+    spawned_child_plan_id: str | None = Field(
+        default=None,
+        description="Child plan spawned for append corrections.",
+    )
+    archived_artifacts: list[str] = Field(
+        default_factory=list,
+        description="Artifacts archived during revert.",
+    )
+    error_message: str | None = Field(
+        default=None,
+        description="Error details when status is FAILED.",
+    )
+    completed_at: datetime = Field(
+        default_factory=lambda: datetime.now(UTC),
+        description="Timestamp when execution finished.",
+    )
+
+
+class CorrectionAttempt(BaseModel):
+    """Record of a single correction execution attempt.
+
+    Multiple attempts may occur if a correction is retried after failure.
+    """
+
+    model_config = ConfigDict(frozen=False)
+
+    attempt_id: str = Field(
+        default_factory=lambda: str(ULID()),
+        description="Unique attempt identifier (ULID).",
+    )
+    correction_id: str = Field(
+        ...,
+        description="Parent correction request ID.",
+    )
+    started_at: datetime = Field(
+        default_factory=lambda: datetime.now(UTC),
+        description="When the attempt started.",
+    )
+    completed_at: datetime | None = Field(
+        default=None,
+        description="When the attempt finished (None if still running).",
+    )
+    success: bool = Field(
+        default=False,
+        description="Whether the attempt succeeded.",
+    )
+    details: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Arbitrary execution details.",
+    )
+
+
+class CorrectionDryRunReport(BaseModel):
+    """Detailed dry-run report showing what a correction *would* do.
+
+    Generated by ``CorrectionService.generate_dry_run_report`` without
+    actually modifying any state.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    correction_id: str = Field(
+        ...,
+        description="Correction request this report describes.",
+    )
+    mode: CorrectionMode = Field(
+        ...,
+        description="Correction strategy that would be applied.",
+    )
+    impact: CorrectionImpact = Field(
+        ...,
+        description="Full impact analysis.",
+    )
+    decisions_to_invalidate: list[str] = Field(
+        default_factory=list,
+        description="Decisions that would be marked invalid.",
+    )
+    child_plans_to_rollback: list[str] = Field(
+        default_factory=list,
+        description="Child plans that would be rolled back.",
+    )
+    estimated_recompute_time_seconds: float | None = Field(
+        default=None,
+        description="Estimated wall-clock seconds to recompute.",
+    )
+    warnings: list[str] = Field(
+        default_factory=list,
+        description="Human-readable warnings about the correction.",
+    )
+
+
+__all__ = [
+    "CorrectionAttempt",
+    "CorrectionDryRunReport",
+    "CorrectionImpact",
+    "CorrectionMode",
+    "CorrectionRequest",
+    "CorrectionResult",
+    "CorrectionStatus",
+]