feat(plan): add error recovery patterns and CLI hints

benchmarks/error_recovery_bench.py

@@ -0,0 +1,131 @@
+"""ASV benchmarks for error recovery patterns.
+
+Measures the performance of:
+- Error classification
+- Recovery hint generation
+- ErrorRecord/ErrorHistory model operations
+- ErrorRecoveryService.record_error() with mock lifecycle
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+from unittest.mock import MagicMock
+
+try:
+    from cleveragents.application.services.error_recovery_service import (
+        ErrorRecoveryService,
+    )
+    from cleveragents.domain.models.core.error_recovery import (
+        ErrorCategory,
+        ErrorHistory,
+        ErrorRecord,
+        ErrorRecoveryPolicy,
+        RecoveryHint,
+        classify_error,
+        get_recovery_hints,
+    )
+    from cleveragents.domain.models.core.plan import (
+        PlanPhase,
+        PlanTimestamps,
+        ProcessingState,
+    )
+except ModuleNotFoundError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+    from cleveragents.application.services.error_recovery_service import (
+        ErrorRecoveryService,
+    )
+    from cleveragents.domain.models.core.error_recovery import (
+        ErrorCategory,
+        ErrorHistory,
+        ErrorRecord,
+        ErrorRecoveryPolicy,
+        RecoveryHint,
+        classify_error,
+        get_recovery_hints,
+    )
+    from cleveragents.domain.models.core.plan import (
+        PlanPhase,
+        PlanTimestamps,
+        ProcessingState,
+    )
+
+
+_PLAN_ID = "01BENCHERR0000000000001"
+
+
+def _make_plan() -> MagicMock:
+    plan = MagicMock()
+    plan.identity.plan_id = _PLAN_ID
+    plan.phase = PlanPhase.EXECUTE
+    plan.processing_state = ProcessingState.ERRORED
+    plan.is_terminal = False
+    plan.error_message = None
+    plan.error_details = None
+    plan.timestamps = PlanTimestamps()
+    return plan
+
+
+def _make_service(plan: MagicMock) -> ErrorRecoveryService:
+    lifecycle = MagicMock()
+    lifecycle.get_plan.return_value = plan
+    lifecycle._commit_plan = MagicMock()
+    return ErrorRecoveryService(
+        lifecycle_service=lifecycle,
+        auto_retry_threshold=0.0,
+        max_retries=3,
+    )
+
+
+class ClassifySuite:
+    """Benchmark error classification."""
+
+    def time_classify_transient(self) -> None:
+        classify_error("Connection timed out after 30s")
+
+    def time_classify_validation(self) -> None:
+        classify_error("Schema validation failed for field X")
+
+    def time_classify_unknown(self) -> None:
+        classify_error("Something completely novel happened")
+
+    def time_classify_with_exception_type(self) -> None:
+        classify_error("failed", "RateLimitError")
+
+
+class HintSuite:
+    """Benchmark recovery hint generation."""
+
+    def time_hints_transient(self) -> None:
+        get_recovery_hints(ErrorCategory.TRANSIENT, _PLAN_ID)
+
+    def time_hints_validation(self) -> None:
+        get_recovery_hints(ErrorCategory.VALIDATION, _PLAN_ID)
+
+    def time_hints_merge_conflict(self) -> None:
+        get_recovery_hints(ErrorCategory.MERGE_CONFLICT, _PLAN_ID)
+
+
+class RecordSuite:
+    """Benchmark error recording with mock service."""
+
+    def setup(self) -> None:
+        self.plan = _make_plan()
+        self.service = _make_service(self.plan)
+        self._counter = 0
+
+    def time_record_error(self) -> None:
+        self._counter += 1
+        self.service.record_error(
+            plan_id=_PLAN_ID,
+            phase="execute",
+            message=f"Timeout #{self._counter}",
+        )
+
+    def time_format_output_plain(self) -> None:
+        self.service.format_error_output(_PLAN_ID, fmt="plain")
+
+    def time_format_output_json(self) -> None:
+        self.service.format_error_output(_PLAN_ID, fmt="json")

docs/faq.md

@@ -699,7 +699,8 @@ The system becomes more autonomous through:
    - Architecture supports: Cold tier can span projects
 
 4. **Fully Autonomous Recovery Strategies**
-   - Current: Rollback and retry with guidance
+   - Current: Automatic retry for transient errors with structured
+     recovery hints and `agents plan errors` CLI (D1b)
    - Research: Automatic error understanding and fixing
    - Architecture supports: Recovery is just another plan type
 

docs/reference/error_recovery.md

@@ -0,0 +1,283 @@
+# Error Recovery Reference
+
+This document describes the error recovery patterns, error classification,
+recovery hint system, and retry policy logic added in D1b.recovery.
+
+## Error Categories
+
+Errors are classified into categories that determine recovery behaviour:
+
+| Category | Retriable | Description |
+|----------|-----------|-------------|
+| `transient` | Yes | Temporary failures (network, timeout, rate-limit) |
+| `validation` | Yes | Validation failures during Execute or Apply |
+| `configuration` | No | Bad config, missing required fields |
+| `resource` | No | Missing or conflicting resources |
+| `merge_conflict` | Yes | Sandbox merge failures |
+| `authentication` | No | Auth/authz failures |
+| `provider` | Yes | LLM provider errors (model unavailable, token limit) |
+| `internal` | No | Unexpected internal errors (bugs) |
+| `unknown` | No | Unclassified errors |
+
+## Recovery Actions
+
+Each error category maps to one or more recovery actions with CLI commands:
+
+| Action | CLI Command | When Used |
+|--------|-------------|-----------|
+| `retry` | `agents plan prompt <plan_id>` | Transient / provider errors |
+| `correct` | `agents plan correct <plan_id>` | Validation failures |
+| `revert_strategize` | `agents plan revert <plan_id> --to-phase strategize` | Resource / merge conflicts |
+| `prompt` | `agents plan prompt <plan_id>` | Resume with guidance |
+| `cancel` | `agents plan cancel <plan_id>` | Non-recoverable errors |
+| `manual` | — | Internal errors (check logs) |
+
+## Retry Policy
+
+The retry policy is controlled by the automation profile's
+`auto_retry_transient` threshold:
+
+- **0.0** (auto): Automatic retry for retriable errors
+- **1.0** (human): Always escalate to human
+
+```
+Error occurs → classify_error(message, exception_type)
+    |
+    +→ retriable category? → retry_count < max_retries? → threshold < 1.0?
+    |     |                     |                            |
+    |     No → escalate         No → exhausted, escalate     No → escalate
+    |                                                        |
+    |                                                        Yes → auto retry
+    |
+    +→ non-retriable → escalate immediately
+```
+
+### Retry Limits
+
+Default `max_retries = 3`. Configured via `SubplanConfig.max_retries`
+for subplans (max: 5).
+
+### Retry Exhaustion
+
+When all retry attempts are used, the error is escalated with:
+- Clear message indicating retries exhausted
+- Recovery hints for manual intervention
+- Error history preserved for debugging
+
+## Structured Error Metadata
+
+When an error is recorded, the following metadata is stored in the
+plan's `error_details` field:
+
+| Key | Type | Description |
+|-----|------|-------------|
+| `error_category` | string | Classification category |
+| `error_phase` | string | Phase when error occurred |
+| `retry_count` | string | Number of retries attempted |
+| `max_retries` | string | Maximum retries allowed |
+| `is_retriable` | string | Whether error can be retried |
+| `error_actor` | string | Actor involved (if any) |
+| `error_tool_call` | string | Tool call that failed (if any) |
+| `stack_summary` | string | Abbreviated stack trace (last 3 frames) |
+| `recovery_action` | string | Suggested recovery action |
+| `recovery_hint` | string | Human-readable recovery message |
+| `recovery_command` | string | CLI command for recovery |
+
+## Execute Phase Integration
+
+The `PlanExecutor` accepts an optional `ErrorRecoveryService` and uses
+it during the Execute phase to record errors and drive automatic
+retries:
+
+```
+PlanExecutor
+├── StrategizeStubActor   (records errors via ErrorRecoveryService)
+├── ExecuteStubActor      (retry loop driven by ErrorRecoveryService)
+├── PlanLifecycleService  (phase transitions, persistence)
+└── ErrorRecoveryService  (optional — error recording + retry policy)
+```
+
+### Retry Loop
+
+When `error_recovery_service` is provided, `run_execute()` wraps the
+execute actor in a retry loop:
+
+1. The actor runs.
+2. On failure the error is recorded via `ErrorRecoveryService.record_error()`.
+3. `should_retry()` is checked — if the error is retriable and retries
+   remain, the actor runs again.
+4. If retries are exhausted or the error is non-retriable, the plan
+   transitions to `ERRORED` via `fail_execute()`.
+
+Strategize failures are also recorded through the service (but not
+retried — the strategize phase is read-only and failures are
+deterministic).
+
+```python
+from cleveragents.application.services.plan_executor import PlanExecutor
+from cleveragents.application.services.error_recovery_service import (
+    ErrorRecoveryService,
+)
+
+er_service = ErrorRecoveryService(
+    lifecycle_service=lifecycle,
+    auto_retry_threshold=0.0,
+    max_retries=3,
+)
+
+executor = PlanExecutor(
+    lifecycle_service=lifecycle,
+    tool_runner=runner,
+    error_recovery_service=er_service,
+)
+
+# Execute with automatic retry on transient errors
+result = executor.run_execute(plan_id)
+```
+
+## CLI Commands
+
+### `agents plan errors <plan_id>`
+
+Shows error decisions with recovery hints and retry history for a plan.
+
+```bash
+# Rich (default) output
+agents plan errors 01HXYZ...
+
+# JSON output
+agents plan errors 01HXYZ... --format json
+
+# Plain text
+agents plan errors 01HXYZ... --format plain
+```
+
+Example rich output:
+
+```
+╭─ Plan Errors ──────────────────────────────────────────╮
+│ Plan: 01HXYZ...                                        │
+│ Phase: execute                                         │
+│ State: errored                                         │
+│ Error: RuntimeError: Connection timed out              │
+│                                                        │
+│ Error Category: transient                              │
+│ Error Phase: execute                                   │
+│ Retry Count: 3/3                                       │
+│ Retriable: false                                       │
+│                                                        │
+│ Recovery Suggestions:                                  │
+│   → Retry the operation — the failure was transient.   │
+│     $ agents plan prompt 01HXYZ...                     │
+╰────────────────────────────────────────────────────────╯
+```
+
+When no errors have been recorded the command shows a green
+confirmation message instead.
+
+## CLI Output
+
+Recovery hints are included in CLI error output:
+
+```
+Error [transient]: Connection timed out after 30s
+  Phase: execute
+  Retry: 0/3
+  Recovery suggestions:
+    → Retry the operation — the failure was transient.
+      $ agents plan prompt 01HXYZ...
+  ✓ Automatic retry available.
+```
+
+### Error History
+
+The error history tracks all errors for a plan, including:
+- Total error count
+- Retry count per error
+- Category breakdown
+- Chronological error records with recovery hints
+
+> **Note:** Error history is held in-memory within the
+> `ErrorRecoveryService` instance. If the service is restarted or a
+> new instance is created, in-memory history is lost. The latest
+> error metadata is also persisted to the plan's `error_details`
+> field for durability across restarts.
+
+## Service API
+
+### `ErrorRecoveryService`
+
+```python
+from cleveragents.application.services.error_recovery_service import (
+    ErrorRecoveryService,
+)
+
+service = ErrorRecoveryService(
+    lifecycle_service=lifecycle,
+    auto_retry_threshold=0.0,  # from automation profile
+    max_retries=3,
+)
+
+# Record an error
+record = service.record_error(
+    plan_id="01HXYZ...",
+    phase="execute",
+    message="Connection timed out",
+    exception=exc,  # optional, for stack trace
+)
+
+# Check retry decision
+if service.should_retry(plan_id):
+    # auto-retry
+    ...
+elif service.should_escalate(plan_id):
+    # show recovery hints to user
+    hints = service.format_recovery_hints_for_cli(plan_id)
+    print(hints)
+
+# Get full error history
+history = service.get_error_history(plan_id)
+print(f"Total errors: {history.total_errors}")
+
+# Format for CLI
+output = service.format_error_output(plan_id, fmt="json")
+```
+
+### `PlanLifecycleService.update_error_details`
+
+External services use `update_error_details()` to merge structured
+error metadata into a plan's `error_details` field without accessing
+internal persistence methods:
+
+```python
+lifecycle.update_error_details(plan_id, {
+    "error_category": "transient",
+    "retry_count": "2",
+})
+```
+
+### Domain Models
+
+```python
+from cleveragents.domain.models.core.error_recovery import (
+    ErrorCategory,
+    ErrorRecord,
+    ErrorHistory,
+    ErrorRecoveryPolicy,
+    RecoveryAction,
+    RecoveryHint,
+    classify_error,
+    get_recovery_hints,
+)
+
+# Classify an error
+category = classify_error("Connection timed out", "NetworkError")
+
+# Get recovery hints
+hints = get_recovery_hints(ErrorCategory.TRANSIENT, "01HXYZ...")
+
+# Create a policy
+policy = ErrorRecoveryPolicy(auto_retry_threshold=0.0, max_retries=3)
+if policy.should_retry(record):
+    ...
+```

docs/reference/plan_apply.md

@@ -101,9 +101,13 @@ When a sandbox merge fails during apply:
 
 **Recovery steps:**
 
-1. Review conflict details via `agents plan status <id>`
-2. Re-run execute phase after resolving conflicts: `agents plan execute <id>`
-3. Or fix validation issues and retry apply
+1. Review error details and recovery hints via `agents plan errors <id>`
+2. Review conflict details via `agents plan status <id>`
+3. Re-run execute phase after resolving conflicts: `agents plan execute <id>`
+4. Or fix validation issues and retry apply
+
+See [Error Recovery Reference](error_recovery.md) for the full error
+classification, retry policy, and recovery hint system.
 
 ### Processing State Flow
 

docs/reference/plan_execute.md

@@ -11,9 +11,10 @@ real AI providers.
 
 ```
 PlanExecutor
-├── StrategizeStubActor   (read-only, produces decision tree)
-├── ExecuteStubActor      (sandbox + ToolRunner + ChangeSet capture)
-└── PlanLifecycleService  (phase transitions, persistence)
+├── StrategizeStubActor    (read-only, produces decision tree)
+├── ExecuteStubActor       (sandbox + ToolRunner + ChangeSet capture)
+├── PlanLifecycleService   (phase transitions, persistence)
+└── ErrorRecoveryService   (optional — error recording + retry logic)
 ```
 
 ## Phase Lifecycle
@@ -39,7 +40,11 @@ from cleveragents.application.services.plan_executor import (
     StrategizeResult,
 )
 
-executor = PlanExecutor(lifecycle_service=lifecycle, tool_runner=runner)
+executor = PlanExecutor(
+    lifecycle_service=lifecycle,
+    tool_runner=runner,
+    error_recovery_service=er_service,  # optional
+)
 result: StrategizeResult = executor.run_strategize(plan_id)
 
 # result.decision_root_id  -> ULID of root node
@@ -92,17 +97,39 @@ After execute completes, the following metadata is persisted on the Plan:
 Failures in either phase are captured with full error context:
 
 - `error_message`: The exception message string
-- `error_details`: Dict containing `exception_type` and `traceback`
+- `error_details`: Dict containing `exception_type`, `traceback`, and
+  structured error-recovery metadata (category, retry count, recovery
+  hints) when an `ErrorRecoveryService` is attached
 - The plan transitions to `ERRORED` processing state
 
-### Retry Guidance
+### Automatic Retry (D1b)
 
-Plans in `ERRORED` state can be retried by:
+When an `ErrorRecoveryService` is provided to the executor, the
+**Execute** phase includes an automatic retry loop:
 
-1. Resetting the plan's processing state back to `QUEUED`
-2. Re-running the failed phase via the executor
+1. On failure the error is classified (transient, validation, etc.)
+   and recorded via `ErrorRecoveryService.record_error()`.
+2. If the error is retriable and retry attempts remain, the execute
+   actor is re-invoked automatically.
+3. If retries are exhausted or the error is non-retriable, the plan
+   transitions to `ERRORED` with full recovery hints stored in
+   `error_details`.
 
-Full retry automation is planned for D1b (Phase Reversion & Error Recovery).
+Retry limits default to 3 and are controlled by the automation
+profile's `auto_retry_transient` threshold.
+
+Use `agents plan errors <plan_id>` to inspect error decisions,
+retry history, and recovery suggestions for a plan.
+
+See [Error Recovery Reference](error_recovery.md) for full details.
+
+### Manual Recovery
+
+Plans in `ERRORED` state without automatic retry can be recovered by:
+
+1. Reviewing errors via `agents plan errors <plan_id>`
+2. Resetting the plan's processing state back to `QUEUED`
+3. Re-running the failed phase via the executor
 
 ## Streaming Hooks
 
@@ -137,3 +164,6 @@ executor.run_strategize(plan_id, stream_callback=my_callback)
   - `StrategizeResult`: Strategize output model
   - `ExecuteResult`: Execute output model
   - `StreamCallback`: Type alias for streaming callbacks
+- **`cleveragents.application.services.error_recovery_service`**: Error
+  recovery integration (optional)
+  - See [Error Recovery Reference](error_recovery.md)

features/error_recovery.feature

@@ -0,0 +1,331 @@
+Feature: Error recovery patterns and CLI hints
+  As a plan executor
+  I want structured error recovery with retry logic and actionable hints
+  So that failed plans can be recovered automatically or with clear guidance
+
+  Background:
+    Given an error recovery test environment
+
+  # -- Error classification ------------------------------------------------
+
+  Scenario: Classify transient error by message
+    When I classify the error message "Connection timed out after 30s"
+    Then the classified error category should be "transient"
+
+  Scenario: Classify validation error by message
+    When I classify the error message "Schema validation failed for field X"
+    Then the classified error category should be "validation"
+
+  Scenario: Classify configuration error by message
+    When I classify the error message "Missing configuration for provider"
+    Then the classified error category should be "configuration"
+
+  Scenario: Classify merge conflict error by message
+    When I classify the error message "Merge conflict in src/main.py"
+    Then the classified error category should be "merge_conflict"
+
+  Scenario: Classify authentication error by message
+    When I classify the error message "Authentication failed: invalid token"
+    Then the classified error category should be "authentication"
+
+  Scenario: Classify provider error by message
+    When I classify the error message "Model not available: gpt-4-turbo"
+    Then the classified error category should be "provider"
+
+  Scenario: Classify internal error by message
+    When I classify the error message "Unexpected internal error in pipeline"
+    Then the classified error category should be "internal"
+
+  Scenario: Classify unknown error by message
+    When I classify the error message "Something completely novel happened"
+    Then the classified error category should be "unknown"
+
+  Scenario: Classify error by exception type
+    When I classify error text "failed" using exception type "RateLimitError"
+    Then the classified error category should be "transient"
+
+  Scenario: Classify error by exception type overrides message
+    When I classify error text "validation issue" using exception type "NetworkError"
+    Then the classified error category should be "transient"
+
+  # -- Recovery hints ------------------------------------------------------
+
+  Scenario: Transient error recovery hints include retry command
+    When I get recovery hints for category "transient" and plan "01PLAN001"
+    Then the first recovery hint action should be "retry"
+    And the first recovery hint should have CLI command containing "01PLAN001"
+
+  Scenario: Validation error recovery hints include correct command
+    When I get recovery hints for category "validation" and plan "01PLAN001"
+    Then the first recovery hint action should be "correct"
+    And the first recovery hint should have CLI command containing "correct"
+
+  Scenario: Configuration error recovery hints suggest cancel
+    When I get recovery hints for category "configuration" and plan "01PLAN001"
+    Then the first recovery hint action should be "cancel"
+
+  Scenario: Merge conflict hints suggest revert to strategize
+    When I get recovery hints for category "merge_conflict" and plan "01PLAN001"
+    Then the first recovery hint action should be "revert_strategize"
+
+  # -- Error recording -----------------------------------------------------
+
+  Scenario: Record error creates error record with metadata
+    Given a plan in errored state for error recovery
+    When I record an error with phase "execute" and message "Timeout talking to provider"
+    Then the recorded error should have category "transient"
+    And the recorded error should have phase "execute"
+    And the recorded error should have recovery hints
+    And the error history should have 1 record
+
+  Scenario: Record error with explicit category
+    Given a plan in errored state for error recovery
+    When I record a categorized error with phase "apply" and message "custom issue" and category "merge_conflict"
+    Then the recorded error should have category "merge_conflict"
+
+  Scenario: Record multiple errors tracks retry count
+    Given a plan in errored state for error recovery
+    When I record an error with phase "execute" and message "Timeout 1"
+    And I record an error with phase "execute" and message "Timeout 2"
+    Then the error history should have 2 records
+    And the latest error retry count should be 1
+
+  # -- Retry policy --------------------------------------------------------
+
+  Scenario: Transient error with auto-retry threshold 0 is retriable
+    Given a plan in errored state for error recovery
+    And the auto retry threshold is 0.0
+    When I record an error with phase "execute" and message "Connection timed out"
+    Then the service should recommend retry
+
+  Scenario: Transient error with auto-retry threshold 1 requires escalation
+    Given a plan in errored state for error recovery
+    And the auto retry threshold is 1.0
+    When I record an error with phase "execute" and message "Connection timed out"
+    Then the service should recommend escalation
+
+  Scenario: Configuration error is never retriable
+    Given a plan in errored state for error recovery
+    And the auto retry threshold is 0.0
+    When I record an error with phase "strategize" and message "Missing configuration"
+    Then the service should recommend escalation
+
+  Scenario: Retry exhaustion after max retries
+    Given a plan in errored state for error recovery
+    And the max retries is 2
+    When I record an error with phase "execute" and message "Timeout 1"
+    And I record an error with phase "execute" and message "Timeout 2"
+    And I record an error with phase "execute" and message "Timeout 3"
+    Then the latest error should have retries exhausted
+    And the service should recommend escalation
+
+  # -- Error history -------------------------------------------------------
+
+  Scenario: Error history shows category breakdown
+    Given a plan in errored state for error recovery
+    When I record an error with phase "execute" and message "Timeout occurred"
+    And I record an error with phase "apply" and message "Merge conflict in file.py"
+    Then the error history category breakdown should include "transient"
+    And the error history category breakdown should include "merge_conflict"
+
+  # -- CLI output formatting -----------------------------------------------
+
+  Scenario: Format error output as plain text
+    Given a plan in errored state for error recovery
+    When I record an error with phase "execute" and message "Provider timed out"
+    And I format the error output as "plain"
+    Then the formatted output should contain "transient"
+    And the formatted output should contain "Provider timed out"
+    And the formatted output should contain "execute"
+
+  Scenario: Format error output as JSON
+    Given a plan in errored state for error recovery
+    When I record an error with phase "execute" and message "Provider timed out"
+    And I format the error output as "json"
+    Then the formatted output should contain "transient"
+    And the formatted output should contain "plan_id"
+
+  Scenario: Format empty error history
+    When I format error output as "plain" for separate plan "01EMPTYPLAN00000000001"
+    Then the formatted output should contain "No errors recorded"
+
+  # -- ErrorRecord model checks --------------------------------------------
+
+  Scenario: ErrorRecord is_retriable for transient errors
+    Given an error record with category "transient" and retry count 0 and max retries 3
+    Then the error record should be retriable
+
+  Scenario: ErrorRecord is not retriable when exhausted
+    Given an error record with category "transient" and retry count 3 and max retries 3
+    Then the error record should not be retriable
+    And the error record should have retries exhausted
+
+  Scenario: ErrorRecord is not retriable for configuration errors
+    Given an error record with category "configuration" and retry count 0 and max retries 3
+    Then the error record should not be retriable
+
+  Scenario: ErrorRecord to_cli_dict has expected fields
+    Given an error recovery test environment
+    Given an error record with category "transient" and retry count 1 and max retries 3
+    Then the error record CLI dict should have key "error_id"
+    And the error record CLI dict should have key "category"
+    And the error record CLI dict should have key "is_retriable"
+
+  # -- Exception-based recording -------------------------------------------
+
+  Scenario: Record error with actual exception captures stack trace
+    Given a plan in errored state for error recovery
+    When I record an error with exception for phase "execute" and message "timeout"
+    Then the recorded error should have a stack summary
+
+  # -- Edge cases -----------------------------------------------------------
+
+  Scenario: should_retry returns false when no errors exist
+    Given a plan in errored state for error recovery
+    Then the service should not recommend retry for a clean plan
+
+  Scenario: should_escalate returns false when no errors exist
+    Given a plan in errored state for error recovery
+    Then the service should not recommend escalation for a clean plan
+
+  Scenario: Format recovery hints for CLI
+    Given a plan in errored state for error recovery
+    When I record an error with phase "execute" and message "Connection timed out"
+    Then the CLI recovery hints should contain "transient"
+
+  # -- ErrorRecoveryPolicy formatting --------------------------------------
+
+  Scenario: Policy formats output with actor and tool info
+    Given a plan in errored state for error recovery
+    When I record an actor-error in phase "execute" with message "Timeout" actor "test-actor" tool "test-tool"
+    And I format the error output as "plain"
+    Then the formatted output should contain "test-actor"
+    And the formatted output should contain "test-tool"
+
+  Scenario: Policy format shows automatic retry available
+    Given a plan in errored state for error recovery
+    And the auto retry threshold is 0.0
+    When I record an error with phase "execute" and message "Connection timed out"
+    And I format the error output as "plain"
+    Then the formatted output should contain "Automatic retry available"
+
+  Scenario: Policy format shows human intervention required
+    Given a plan in errored state for error recovery
+    And the auto retry threshold is 1.0
+    When I record an error with phase "execute" and message "Connection timed out"
+    And I format the error output as "plain"
+    Then the formatted output should contain "Human intervention required"
+
+  Scenario: Policy format shows retries exhausted message
+    Given a plan in errored state for error recovery
+    And the max retries is 1
+    When I record an error with phase "execute" and message "Timeout 1"
+    And I record an error with phase "execute" and message "Timeout 2"
+    And I format the error output as "plain"
+    Then the formatted output should contain "retry attempts exhausted"
+
+  # -- ErrorHistory with no retriable errors --------------------------------
+
+  Scenario: Error history has_retriable is false when only non-retriable
+    Given a plan in errored state for error recovery
+    When I record an error with phase "strategize" and message "Missing configuration"
+    Then the error history should not have retriable errors
+
+  # -- ErrorRecord with actor and tool for CLI dict -------------------------
+
+  Scenario: ErrorRecord to_cli_dict includes actor and tool when present
+    Given an error record with actor "my-actor" and tool "my-tool"
+    Then the error record CLI dict should have key "actor"
+    And the error record CLI dict should have key "tool_call"
+
+  # -- Recovery hints for empty plan ----------------------------------------
+
+  Scenario: Format recovery hints returns empty for plan with no errors
+    Given a plan in errored state for error recovery
+    Then the CLI recovery hints for a clean plan should be empty
+
+  # -- ErrorRecord to_cli_dict with stack summary ---------------------------
+
+  Scenario: ErrorRecord to_cli_dict includes stack summary
+    Given an error record with stack summary "File test.py line 10"
+    Then the error record CLI dict should have key "stack_summary"
+
+  # -- ErrorRecoveryPolicy should_retry on non-retriable --------------------
+
+  Scenario: Policy should_retry returns false for non-retriable category
+    Given a policy with threshold 0.0 and max retries 3
+    And an error record with category "configuration" and retry count 0 and max retries 3
+    Then the policy should not recommend retry
+
+  Scenario: Policy should_retry returns false for exhausted retries
+    Given a policy with threshold 0.0 and max retries 3
+    And an error record with category "transient" and retry count 3 and max retries 3
+    Then the policy should not recommend retry
+
+  # -- T1: Persistence failure path ----------------------------------------
+
+  Scenario: Persist error metadata handles lifecycle errors gracefully
+    Given a plan in errored state for error recovery
+    And the lifecycle service update_error_details raises an error
+    When I record an error with phase "execute" and message "transient failure"
+    Then the error should still be recorded in memory
+
+  # -- T2: Branch coverage for policy/format --------------------------------
+
+  Scenario: Policy should_escalate returns true for exhausted retries
+    Given a policy with threshold 0.0 and max retries 3
+    And an error record with category "transient" and retry count 3 and max retries 3
+    Then the policy should recommend escalation
+
+  Scenario: Policy should_escalate returns true for non-retriable category
+    Given a policy with threshold 0.0 and max retries 3
+    And an error record with category "configuration" and retry count 0 and max retries 3
+    Then the policy should recommend escalation
+
+  Scenario: Format recovery output includes hints with CLI commands
+    Given a policy with threshold 0.0 and max retries 3
+    And a hinted error record with category "transient" and retry count 0 and max retries 3
+    Then the formatted recovery output should contain CLI command text
+
+  Scenario: Format recovery output shows exhausted message
+    Given a policy with threshold 0.0 and max retries 3
+    And a hinted error record with category "transient" and retry count 3 and max retries 3
+    Then the formatted recovery output should contain "exhausted"
+
+  Scenario: Format recovery output shows automatic retry available
+    Given a policy with threshold 0.0 and max retries 3
+    And an error record with category "transient" and retry count 0 and max retries 3
+    Then the formatted recovery output should contain "Automatic retry"
+
+  Scenario: Policy should_retry returns false when threshold is 1.0
+    Given a policy with threshold 1.0 and max retries 3
+    And an error record with category "transient" and retry count 0 and max retries 3
+    Then the policy should not recommend retry
+
+  # -- T4: Integration test with executor retry loop ------------------------
+
+  Scenario: Executor retries execute phase on recoverable errors
+    Given a plan executor with error recovery service
+    And the execute actor fails twice then succeeds
+    When I run the executor for the plan
+    Then the execute should complete after retries
+    And the error recovery service should have recorded 2 errors
+
+  Scenario: Executor fails after exhausting retries
+    Given a plan executor with error recovery service max retries 2
+    And the execute actor always fails with "persistent failure"
+    When I run the executor expecting failure
+    Then the lifecycle should record the plan as failed
+    And the error recovery service should have recorded 3 errors
+
+  # -- B5: Bare conflict no longer matches MERGE_CONFLICT -------------------
+
+  Scenario: Classify resource conflict does not match merge_conflict
+    When I classify the error message "resource conflict detected"
+    Then the classified error category should not be "merge_conflict"
+
+  # -- S1: Format string safe substitution ----------------------------------
+
+  Scenario: Recovery hints handle plan_id with braces safely
+    When I get recovery hints for category "transient" with plan_id "{evil_id}"
+    Then the hints should contain "{evil_id}" in CLI commands

features/error_recovery_coverage_boost.feature

@@ -0,0 +1,175 @@
+Feature: Error recovery coverage boost
+  As a developer
+  I want to cover uncovered lines in plan CLI errors/diff/artifacts commands
+  and plan executor edge cases introduced by the error recovery feature
+  So that overall code coverage exceeds the 97% threshold
+
+  Background:
+    Given an errcov test environment
+
+  # -- plan errors CLI (plan.py:1621-1720) ---------------------------------
+
+  Scenario: Plan errors command rich output with full error details
+    Given an errcov plan with error details category "transient" phase "execute"
+    And the errcov plan has error message "Connection timed out after 30s"
+    And the errcov plan error details include actor "test-actor" and tool "test-tool"
+    And the errcov plan error details include stack summary "File test.py line 10"
+    When I invoke errcov plan errors in "rich" format
+    Then the errcov errors output should contain "Plan Errors"
+    And the errcov errors output should contain "transient"
+    And the errcov errors output should contain "execute"
+    And the errcov errors output should contain "test-actor"
+    And the errcov errors output should contain "test-tool"
+    And the errcov errors output should contain "File test.py"
+    And the errcov errors output should contain "Recovery Suggestions"
+
+  Scenario: Plan errors command plain output with error details
+    Given an errcov plan with error details category "validation" phase "apply"
+    And the errcov plan has error message "Schema validation failed"
+    When I invoke errcov plan errors in "plain" format
+    Then the errcov errors output should contain "validation"
+    And the errcov errors output should contain "apply"
+    And the errcov errors output should contain "Schema validation failed"
+
+  Scenario: Plan errors command JSON output
+    Given an errcov plan with error details category "transient" phase "execute"
+    And the errcov plan has error message "Timeout"
+    When I invoke errcov plan errors in "json" format
+    Then the errcov errors output should contain "transient"
+    And the errcov errors output should contain "plan_id"
+
+  Scenario: Plan errors command rich output with no errors and no message
+    Given an errcov plan with no error details and no error message
+    When I invoke errcov plan errors in "rich" format
+    Then the errcov errors output should contain "No error recovery records"
+
+  Scenario: Plan errors command rich output with error message but no recovery
+    Given an errcov plan with no error details but has error message "Something broke"
+    When I invoke errcov plan errors in "rich" format
+    Then the errcov errors output should contain "Something broke"
+
+  Scenario: Plan errors command handles CleverAgentsError
+    Given an errcov lifecycle service that raises CleverAgentsError on get_plan
+    When I invoke errcov plan errors expecting abort
+    Then the errcov errors output should contain "Error:"
+
+  Scenario: Plan errors command plain output with actor and tool and stack
+    Given an errcov plan with error details category "provider" phase "execute"
+    And the errcov plan error details include actor "gpt4-actor" and tool "code-gen"
+    And the errcov plan error details include stack summary "traceback info"
+    And the errcov plan error details include retry count "2" max "3" retriable "true"
+    When I invoke errcov plan errors in "plain" format
+    Then the errcov errors output should contain "gpt4-actor"
+    And the errcov errors output should contain "code-gen"
+    And the errcov errors output should contain "traceback info"
+    And the errcov errors output should contain "recovery_hints"
+
+  # -- plan diff CLI (plan.py:1969-1979) -----------------------------------
+
+  Scenario: Plan diff command calls apply service diff
+    Given an errcov apply service returning diff text "--- a/file.py"
+    When I invoke errcov plan diff in "rich" format
+    Then the errcov output should contain "--- a/file.py"
+
+  Scenario: Plan diff command handles PlanError
+    Given an errcov apply service that raises PlanError on diff
+    When I invoke errcov plan diff expecting abort
+    Then the errcov output should contain "Diff Error:"
+
+  Scenario: Plan diff command handles CleverAgentsError
+    Given an errcov apply service that raises CleverAgentsError on diff
+    When I invoke errcov plan diff expecting abort
+    Then the errcov output should contain "Error:"
+
+  # -- plan artifacts CLI (plan.py:2002-2012) ------------------------------
+
+  Scenario: Plan artifacts command calls apply service artifacts
+    Given an errcov apply service returning artifacts text "ChangeSet: cs_001"
+    When I invoke errcov plan artifacts in "rich" format
+    Then the errcov output should contain "ChangeSet: cs_001"
+
+  Scenario: Plan artifacts command handles PlanError
+    Given an errcov apply service that raises PlanError on artifacts
+    When I invoke errcov plan artifacts expecting abort
+    Then the errcov output should contain "Artifacts Error:"
+
+  Scenario: Plan artifacts command handles CleverAgentsError
+    Given an errcov apply service that raises CleverAgentsError on artifacts
+    When I invoke errcov plan artifacts expecting abort
+    Then the errcov output should contain "Error:"
+
+  # -- _get_apply_service (plan.py:1941-1946) ------------------------------
+
+  Scenario: Get apply service constructs PlanApplyService
+    When I invoke errcov _get_apply_service
+    Then the errcov apply service should be a PlanApplyService instance
+
+  # -- PlanExecutor edge cases (plan_executor.py) --------------------------
+
+  Scenario: Strategize with empty plan_id raises ValidationError
+    When I invoke errcov strategize with empty plan_id
+    Then an errcov ValidationError should be raised with "plan_id must not be empty"
+
+  Scenario: Execute stub actor with empty plan_id raises ValidationError
+    When I invoke errcov execute stub with empty plan_id
+    Then an errcov ValidationError should be raised with "plan_id must not be empty"
+
+  Scenario: Run execute with wrong phase raises PlanError
+    Given an errcov plan in strategize phase
+    When I invoke errcov run_execute on wrong-phase plan
+    Then an errcov PlanError should be raised containing "not in Execute phase"
+
+  Scenario: Run execute with wrong state raises PlanError
+    Given an errcov plan in execute phase but processing state
+    When I invoke errcov run_execute on wrong-state plan
+    Then an errcov PlanError should be raised containing "not queued"
+
+  Scenario: Parse definition of done skips blank lines
+    When I parse errcov definition of done with blank lines
+    Then the parsed steps should not contain empty entries
+
+  Scenario: Parse definition of done handles numbered prefixes
+    When I parse errcov definition of done with numbered items "1. First\n2) Second\n3. Third"
+    Then the errcov parsed steps should be "First" and "Second" and "Third"
+
+  Scenario: Strategize error recording with recovery service
+    Given an errcov executor with error recovery service
+    And the errcov strategize actor raises RuntimeError "strategize failed"
+    When I invoke errcov run_strategize expecting failure
+    Then the errcov error recovery should have recorded a strategize error
+
+  Scenario: Execute retry loop exhaustion without recovery service
+    Given an errcov executor without error recovery service
+    And the errcov execute actor always fails with "total failure"
+    When I invoke errcov run_execute expecting failure
+    Then the errcov lifecycle should have called fail_execute
+
+  # -- error_recovery.py policy edge cases ---------------------------------
+
+  Scenario: Policy should_retry returns false for exhausted retriable error
+    Given an errcov policy with threshold 0.0 and max retries 3
+    And an errcov retriable error record with retry count 3 max 3
+    Then the errcov policy should not recommend retry
+
+  Scenario: Policy should_escalate returns true for exhausted retriable error
+    Given an errcov policy with threshold 0.0 and max retries 3
+    And an errcov retriable error record with retry count 3 max 3
+    Then the errcov policy should recommend escalation
+
+  Scenario: Policy format_recovery_output shows hint cli commands
+    Given an errcov policy with threshold 0.0 and max retries 3
+    And an errcov retriable error record with hints retry count 0 max 3
+    When I format errcov recovery output
+    Then the errcov recovery output should contain "agents plan"
+
+  # -- plan.py branch: actor_registry and testing_mode (647->649, 649->653)
+
+  Scenario: Plan use command with actor registry present
+    Given an errcov plan creation environment with actor registry
+    When I invoke errcov plan use with actor registry
+    Then the errcov actor registry ensure_built_in_actors should have been called
+
+  Scenario: Plan use command with testing mode enabled
+    Given an errcov plan creation environment with testing mode
+    When I invoke errcov plan use with testing mode
+    Then the errcov container actor_service should have been called

features/steps/error_recovery_coverage_boost_steps.py

@@ -0,0 +1,730 @@
+"""Step definitions for error_recovery_coverage_boost.feature.
+
+Covers uncovered paths in plan.py (errors/diff/artifacts CLI commands),
+plan_executor.py (edge cases, retry loop), and error_recovery.py (policy
+branches) introduced by the error recovery feature.
+
+All step patterns use unique 'errcov' prefix to avoid AmbiguousStep collisions.
+"""
+
+from __future__ import annotations
+
+import contextlib
+from io import StringIO
+from unittest.mock import MagicMock, patch
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.core.exceptions import CleverAgentsError, PlanError, ValidationError
+from cleveragents.domain.models.core.error_recovery import (
+    ErrorCategory,
+    ErrorRecord,
+    ErrorRecoveryPolicy,
+    get_recovery_hints,
+)
+from cleveragents.domain.models.core.plan import (
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+
+__all__: list[str] = []
+
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+_PLAN_ID = "01ERRCOV0000000000000001"
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_capture_console(buf: StringIO):
+    """Create a Rich Console that writes to the buffer for assertion."""
+    from rich.console import Console
+
+    return Console(file=buf, force_terminal=False, width=200)
+
+
+def _make_mock_plan(
+    *,
+    plan_id: str = _PLAN_ID,
+    phase: PlanPhase = PlanPhase.EXECUTE,
+    state: ProcessingState = ProcessingState.ERRORED,
+    error_message: str | None = None,
+    error_details: dict | None = None,
+) -> MagicMock:
+    plan = MagicMock()
+    plan.identity.plan_id = plan_id
+    plan.phase = phase
+    plan.state = state
+    plan.processing_state = state
+    plan.is_terminal = False
+    plan.error_message = error_message
+    plan.error_details = error_details
+    plan.timestamps = PlanTimestamps()
+    return plan
+
+
+def _capture_plan_cli(context: Context, args: list[str]) -> str:
+    """Invoke a plan CLI command, capturing console output."""
+    from cleveragents.cli.commands import plan as plan_mod
+
+    buf = StringIO()
+    try:
+        with patch.object(
+            plan_mod.console,
+            "print",
+            side_effect=lambda *a, **kw: buf.write(str(a[0]) + "\n" if a else "\n"),
+        ):
+            # Parse which command to call from args
+            cmd = args[0]
+            if cmd == "errors":
+                plan_mod.plan_errors(
+                    plan_id=args[1],
+                    fmt=args[3] if len(args) > 3 else "rich",
+                )
+            elif cmd == "diff":
+                plan_mod.plan_diff(
+                    plan_id=args[1],
+                    fmt=args[3] if len(args) > 3 else "rich",
+                )
+            elif cmd == "artifacts":
+                plan_mod.plan_artifacts(
+                    plan_id=args[1],
+                    fmt=args[3] if len(args) > 3 else "rich",
+                )
+    except (SystemExit, Exception):
+        pass
+    return buf.getvalue()
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("an errcov test environment")
+def step_errcov_env(context: Context) -> None:
+    context.errcov_plan = None
+    context.errcov_output = ""
+    context.errcov_service = None
+    context.errcov_error = None
+    context.errcov_apply_svc = None
+    context.errcov_lifecycle_svc = None
+
+
+# ---------------------------------------------------------------------------
+# Plan errors CLI -- Given steps
+# ---------------------------------------------------------------------------
+
+
+@given('an errcov plan with error details category "{cat}" phase "{phase}"')
+def step_errcov_plan_with_details(context: Context, cat: str, phase: str) -> None:
+    context.errcov_plan = _make_mock_plan(
+        error_details={
+            "error_category": cat,
+            "error_phase": phase,
+            "retry_count": "0",
+            "max_retries": "3",
+            "is_retriable": "true",
+        },
+    )
+
+
+@given('the errcov plan has error message "{msg}"')
+def step_errcov_plan_error_msg(context: Context, msg: str) -> None:
+    context.errcov_plan.error_message = msg
+
+
+@given('the errcov plan error details include actor "{actor}" and tool "{tool}"')
+def step_errcov_plan_actor_tool(context: Context, actor: str, tool: str) -> None:
+    context.errcov_plan.error_details["error_actor"] = actor
+    context.errcov_plan.error_details["error_tool_call"] = tool
+
+
+@given('the errcov plan error details include stack summary "{summary}"')
+def step_errcov_plan_stack(context: Context, summary: str) -> None:
+    context.errcov_plan.error_details["stack_summary"] = summary
+
+
+@given(
+    'the errcov plan error details include retry count "{rc}" max "{mx}" retriable "{retr}"'
+)
+def step_errcov_plan_retry_info(context: Context, rc: str, mx: str, retr: str) -> None:
+    context.errcov_plan.error_details["retry_count"] = rc
+    context.errcov_plan.error_details["max_retries"] = mx
+    context.errcov_plan.error_details["is_retriable"] = retr
+
+
+@given("an errcov plan with no error details and no error message")
+def step_errcov_plan_clean(context: Context) -> None:
+    context.errcov_plan = _make_mock_plan(
+        error_details=None,
+        error_message=None,
+    )
+
+
+@given('an errcov plan with no error details but has error message "{msg}"')
+def step_errcov_plan_msg_only(context: Context, msg: str) -> None:
+    context.errcov_plan = _make_mock_plan(
+        error_details=None,
+        error_message=msg,
+    )
+
+
+@given("an errcov lifecycle service that raises CleverAgentsError on get_plan")
+def step_errcov_lifecycle_error(context: Context) -> None:
+    svc = MagicMock()
+    svc.get_plan.side_effect = CleverAgentsError("plan lookup failed")
+    context.errcov_lifecycle_svc = svc
+
+
+# ---------------------------------------------------------------------------
+# Plan errors CLI -- When steps
+# ---------------------------------------------------------------------------
+
+
+@when('I invoke errcov plan errors in "{fmt}" format')
+def step_errcov_invoke_errors(context: Context, fmt: str) -> None:
+    from cleveragents.cli.commands import plan as plan_mod
+
+    svc = MagicMock()
+    svc.get_plan.return_value = context.errcov_plan
+
+    buf = StringIO()
+    capture_console = _make_capture_console(buf)
+    with (
+        patch.object(
+            plan_mod,
+            "_get_lifecycle_service",
+            return_value=svc,
+        ),
+        patch.object(plan_mod, "console", capture_console),
+    ):
+        plan_mod.plan_errors(plan_id=_PLAN_ID, fmt=fmt)
+    context.errcov_output = buf.getvalue()
+
+
+@when("I invoke errcov plan errors expecting abort")
+def step_errcov_invoke_errors_abort(context: Context) -> None:
+    from cleveragents.cli.commands import plan as plan_mod
+
+    buf = StringIO()
+    capture_console = _make_capture_console(buf)
+    with (
+        patch.object(
+            plan_mod,
+            "_get_lifecycle_service",
+            return_value=context.errcov_lifecycle_svc,
+        ),
+        patch.object(plan_mod, "console", capture_console),
+        contextlib.suppress(SystemExit, Exception),
+    ):
+        plan_mod.plan_errors(plan_id=_PLAN_ID, fmt="rich")
+    context.errcov_output = buf.getvalue()
+
+
+# ---------------------------------------------------------------------------
+# Plan diff CLI -- Given/When steps
+# ---------------------------------------------------------------------------
+
+
+@given('an errcov apply service returning diff text "{text}"')
+def step_errcov_apply_diff(context: Context, text: str) -> None:
+    svc = MagicMock()
+    svc.diff.return_value = text
+    context.errcov_apply_svc = svc
+
+
+@given("an errcov apply service that raises PlanError on diff")
+def step_errcov_apply_diff_plan_error(context: Context) -> None:
+    svc = MagicMock()
+    svc.diff.side_effect = PlanError("diff failed")
+    context.errcov_apply_svc = svc
+
+
+@given("an errcov apply service that raises CleverAgentsError on diff")
+def step_errcov_apply_diff_agent_error(context: Context) -> None:
+    svc = MagicMock()
+    svc.diff.side_effect = CleverAgentsError("diff service error")
+    context.errcov_apply_svc = svc
+
+
+@when('I invoke errcov plan diff in "{fmt}" format')
+def step_errcov_invoke_diff(context: Context, fmt: str) -> None:
+    from cleveragents.cli.commands import plan as plan_mod
+
+    buf = StringIO()
+    capture_console = _make_capture_console(buf)
+    with (
+        patch.object(
+            plan_mod, "_get_apply_service", return_value=context.errcov_apply_svc
+        ),
+        patch.object(plan_mod, "console", capture_console),
+    ):
+        plan_mod.plan_diff(plan_id=_PLAN_ID, fmt=fmt)
+    context.errcov_output = buf.getvalue()
+
+
+@when("I invoke errcov plan diff expecting abort")
+def step_errcov_invoke_diff_abort(context: Context) -> None:
+    from cleveragents.cli.commands import plan as plan_mod
+
+    buf = StringIO()
+    capture_console = _make_capture_console(buf)
+    with (
+        patch.object(
+            plan_mod, "_get_apply_service", return_value=context.errcov_apply_svc
+        ),
+        patch.object(plan_mod, "console", capture_console),
+        contextlib.suppress(SystemExit, Exception),
+    ):
+        plan_mod.plan_diff(plan_id=_PLAN_ID, fmt="rich")
+    context.errcov_output = buf.getvalue()
+
+
+# ---------------------------------------------------------------------------
+# Plan artifacts CLI -- Given/When steps
+# ---------------------------------------------------------------------------
+
+
+@given('an errcov apply service returning artifacts text "{text}"')
+def step_errcov_apply_artifacts(context: Context, text: str) -> None:
+    svc = MagicMock()
+    svc.artifacts.return_value = text
+    context.errcov_apply_svc = svc
+
+
+@given("an errcov apply service that raises PlanError on artifacts")
+def step_errcov_apply_artifacts_plan_error(context: Context) -> None:
+    svc = MagicMock()
+    svc.artifacts.side_effect = PlanError("artifacts failed")
+    context.errcov_apply_svc = svc
+
+
+@given("an errcov apply service that raises CleverAgentsError on artifacts")
+def step_errcov_apply_artifacts_agent_error(context: Context) -> None:
+    svc = MagicMock()
+    svc.artifacts.side_effect = CleverAgentsError("artifacts service error")
+    context.errcov_apply_svc = svc
+
+
+@when('I invoke errcov plan artifacts in "{fmt}" format')
+def step_errcov_invoke_artifacts(context: Context, fmt: str) -> None:
+    from cleveragents.cli.commands import plan as plan_mod
+
+    buf = StringIO()
+    capture_console = _make_capture_console(buf)
+    with (
+        patch.object(
+            plan_mod, "_get_apply_service", return_value=context.errcov_apply_svc
+        ),
+        patch.object(plan_mod, "console", capture_console),
+    ):
+        plan_mod.plan_artifacts(plan_id=_PLAN_ID, fmt=fmt)
+    context.errcov_output = buf.getvalue()
+
+
+@when("I invoke errcov plan artifacts expecting abort")
+def step_errcov_invoke_artifacts_abort(context: Context) -> None:
+    from cleveragents.cli.commands import plan as plan_mod
+
+    buf = StringIO()
+    capture_console = _make_capture_console(buf)
+    with (
+        patch.object(
+            plan_mod, "_get_apply_service", return_value=context.errcov_apply_svc
+        ),
+        patch.object(plan_mod, "console", capture_console),
+        contextlib.suppress(SystemExit, Exception),
+    ):
+        plan_mod.plan_artifacts(plan_id=_PLAN_ID, fmt="rich")
+    context.errcov_output = buf.getvalue()
+
+
+# ---------------------------------------------------------------------------
+# _get_apply_service
+# ---------------------------------------------------------------------------
+
+
+@when("I invoke errcov _get_apply_service")
+def step_errcov_get_apply_svc(context: Context) -> None:
+    from cleveragents.cli.commands import plan as plan_mod
+
+    mock_lifecycle = MagicMock()
+    with patch.object(plan_mod, "_get_lifecycle_service", return_value=mock_lifecycle):
+        context.errcov_apply_svc = plan_mod._get_apply_service()
+
+
+@then("the errcov apply service should be a PlanApplyService instance")
+def step_errcov_apply_svc_type(context: Context) -> None:
+    from cleveragents.application.services.plan_apply_service import PlanApplyService
+
+    assert isinstance(context.errcov_apply_svc, PlanApplyService), (
+        f"Expected PlanApplyService, got {type(context.errcov_apply_svc)}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# PlanExecutor edge cases
+# ---------------------------------------------------------------------------
+
+
+@when("I invoke errcov strategize with empty plan_id")
+def step_errcov_strategize_empty(context: Context) -> None:
+    from cleveragents.application.services.plan_executor import StrategizeStubActor
+
+    actor = StrategizeStubActor()
+    try:
+        actor.execute(plan_id="", definition_of_done="test", stream_callback=None)
+        context.errcov_error = None
+    except ValidationError as exc:
+        context.errcov_error = exc
+
+
+@when("I invoke errcov execute stub with empty plan_id")
+def step_errcov_execute_stub_empty(context: Context) -> None:
+    from cleveragents.application.services.plan_executor import ExecuteStubActor
+
+    actor = ExecuteStubActor()
+    try:
+        actor.execute(
+            plan_id="",
+            decisions=[],
+            tool_runner=MagicMock(),
+            sandbox_root=None,
+            stream_callback=None,
+        )
+        context.errcov_error = None
+    except ValidationError as exc:
+        context.errcov_error = exc
+
+
+@given("an errcov plan in strategize phase")
+def step_errcov_plan_strategize(context: Context) -> None:
+    plan = _make_mock_plan(phase=PlanPhase.STRATEGIZE, state=ProcessingState.QUEUED)
+    lifecycle = MagicMock()
+    lifecycle.get_plan.return_value = plan
+    context.errcov_lifecycle_svc = lifecycle
+    context.errcov_plan = plan
+
+
+@given("an errcov plan in execute phase but processing state")
+def step_errcov_plan_execute_processing(context: Context) -> None:
+    plan = _make_mock_plan(phase=PlanPhase.EXECUTE, state=ProcessingState.PROCESSING)
+    lifecycle = MagicMock()
+    lifecycle.get_plan.return_value = plan
+    context.errcov_lifecycle_svc = lifecycle
+    context.errcov_plan = plan
+
+
+@when("I invoke errcov run_execute on wrong-phase plan")
+def step_errcov_execute_wrong_phase(context: Context) -> None:
+    from cleveragents.application.services.plan_executor import PlanExecutor
+
+    executor = PlanExecutor(
+        lifecycle_service=context.errcov_lifecycle_svc,
+        tool_runner=MagicMock(),
+    )
+    try:
+        executor.run_execute(plan_id=_PLAN_ID)
+        context.errcov_error = None
+    except PlanError as exc:
+        context.errcov_error = exc
+
+
+@when("I invoke errcov run_execute on wrong-state plan")
+def step_errcov_execute_wrong_state(context: Context) -> None:
+    from cleveragents.application.services.plan_executor import PlanExecutor
+
+    executor = PlanExecutor(
+        lifecycle_service=context.errcov_lifecycle_svc,
+        tool_runner=MagicMock(),
+    )
+    try:
+        executor.run_execute(plan_id=_PLAN_ID)
+        context.errcov_error = None
+    except PlanError as exc:
+        context.errcov_error = exc
+
+
+@when("I parse errcov definition of done with blank lines")
+def step_errcov_parse_dod_blank(context: Context) -> None:
+    from cleveragents.application.services.plan_executor import StrategizeStubActor
+
+    actor = StrategizeStubActor()
+    context.errcov_parsed_steps = actor._parse_steps("Step one\n\n\nStep two\n\n")
+
+
+@when('I parse errcov definition of done with numbered items "{text}"')
+def step_errcov_parse_dod_numbered(context: Context, text: str) -> None:
+    from cleveragents.application.services.plan_executor import StrategizeStubActor
+
+    actor = StrategizeStubActor()
+    context.errcov_parsed_steps = actor._parse_steps(text.replace("\\n", "\n"))
+
+
+@given("an errcov executor with error recovery service")
+def step_errcov_executor_with_recovery(context: Context) -> None:
+    from cleveragents.application.services.error_recovery_service import (
+        ErrorRecoveryService,
+    )
+    from cleveragents.application.services.plan_executor import PlanExecutor
+
+    plan = _make_mock_plan(phase=PlanPhase.STRATEGIZE, state=ProcessingState.QUEUED)
+    plan.definition_of_done = "Test passes"
+    plan.metadata = {}
+    plan.invariants = []
+
+    lifecycle = MagicMock()
+    lifecycle.get_plan.return_value = plan
+
+    er_service = ErrorRecoveryService(
+        lifecycle_service=lifecycle,
+        auto_retry_threshold=0.0,
+        max_retries=3,
+    )
+
+    context.errcov_lifecycle_svc = lifecycle
+    context.errcov_er_service = er_service
+    context.errcov_executor = PlanExecutor(
+        lifecycle_service=lifecycle,
+        tool_runner=MagicMock(),
+        error_recovery_service=er_service,
+    )
+
+
+@given('the errcov strategize actor raises RuntimeError "{msg}"')
+def step_errcov_strategize_fails(context: Context, msg: str) -> None:
+    mock_actor = MagicMock()
+    mock_actor.execute.side_effect = RuntimeError(msg)
+    context.errcov_executor._strategize_actor = mock_actor
+
+
+@when("I invoke errcov run_strategize expecting failure")
+def step_errcov_run_strategize_fail(context: Context) -> None:
+    with contextlib.suppress(RuntimeError):
+        context.errcov_executor.run_strategize(plan_id=_PLAN_ID)
+
+
+@then("the errcov error recovery should have recorded a strategize error")
+def step_errcov_er_strategize_recorded(context: Context) -> None:
+    history = context.errcov_er_service.get_error_history(_PLAN_ID)
+    assert history.total_errors >= 1, (
+        f"Expected at least 1 error recorded, got {history.total_errors}"
+    )
+    assert history.records[0].phase == "strategize"
+
+
+@given("an errcov executor without error recovery service")
+def step_errcov_executor_no_recovery(context: Context) -> None:
+    from cleveragents.application.services.plan_executor import PlanExecutor
+
+    plan = _make_mock_plan(phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED)
+    plan.decision_root_id = "01ROOTDECISION0000000001"
+    plan.definition_of_done = "Tests pass"
+    plan.metadata = {}
+
+    lifecycle = MagicMock()
+    lifecycle.get_plan.return_value = plan
+
+    context.errcov_lifecycle_svc = lifecycle
+    context.errcov_executor = PlanExecutor(
+        lifecycle_service=lifecycle,
+        tool_runner=MagicMock(),
+    )
+
+
+@given('the errcov execute actor always fails with "{msg}"')
+def step_errcov_execute_always_fails(context: Context, msg: str) -> None:
+    context.errcov_executor._execute_actor = MagicMock()
+    context.errcov_executor._execute_actor.execute.side_effect = RuntimeError(msg)
+
+
+@when("I invoke errcov run_execute expecting failure")
+def step_errcov_run_execute_fail(context: Context) -> None:
+    with contextlib.suppress(RuntimeError):
+        context.errcov_executor.run_execute(plan_id=_PLAN_ID)
+
+
+@then("the errcov lifecycle should have called fail_execute")
+def step_errcov_fail_execute_called(context: Context) -> None:
+    context.errcov_lifecycle_svc.fail_execute.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# error_recovery.py policy edge cases
+# ---------------------------------------------------------------------------
+
+
+@given("an errcov policy with threshold {threshold} and max retries {mr}")
+def step_errcov_policy(context: Context, threshold: str, mr: str) -> None:
+    context.errcov_policy = ErrorRecoveryPolicy(
+        auto_retry_threshold=float(threshold),
+        max_retries=int(mr),
+    )
+
+
+@given("an errcov retriable error record with retry count {rc} max {mr}")
+def step_errcov_retriable_record(context: Context, rc: str, mr: str) -> None:
+    context.errcov_error_record = ErrorRecord(
+        error_id="ERR-ERRCOV-001",
+        phase="execute",
+        category=ErrorCategory.TRANSIENT,
+        message="timeout",
+        retry_count=int(rc),
+        max_retries=int(mr),
+    )
+
+
+@given("an errcov retriable error record with hints retry count {rc} max {mr}")
+def step_errcov_retriable_record_hints(context: Context, rc: str, mr: str) -> None:
+    hints = get_recovery_hints(ErrorCategory.TRANSIENT, _PLAN_ID)
+    context.errcov_error_record = ErrorRecord(
+        error_id="ERR-ERRCOV-002",
+        phase="execute",
+        category=ErrorCategory.TRANSIENT,
+        message="timeout with hints",
+        retry_count=int(rc),
+        max_retries=int(mr),
+        recovery_hints=hints,
+    )
+
+
+@when("I format errcov recovery output")
+def step_errcov_format_recovery(context: Context) -> None:
+    context.errcov_recovery_output = context.errcov_policy.format_recovery_output(
+        context.errcov_error_record
+    )
+
+
+@then("the errcov policy should not recommend retry")
+def step_errcov_no_retry(context: Context) -> None:
+    assert not context.errcov_policy.should_retry(context.errcov_error_record)
+
+
+@then("the errcov policy should recommend escalation")
+def step_errcov_escalation(context: Context) -> None:
+    assert context.errcov_policy.should_escalate(context.errcov_error_record)
+
+
+@then('the errcov recovery output should contain "{text}"')
+def step_errcov_recovery_contains(context: Context, text: str) -> None:
+    assert text in context.errcov_recovery_output, (
+        f"Expected '{text}' in recovery output, got: {context.errcov_recovery_output[:300]}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Plan use command branches (actor_registry, testing_mode)
+# ---------------------------------------------------------------------------
+
+
+@given("an errcov plan creation environment with actor registry")
+def step_errcov_plan_use_actor_reg(context: Context) -> None:
+    context.errcov_actor_registry = MagicMock()
+    context.errcov_container = MagicMock()
+
+
+@given("an errcov plan creation environment with testing mode")
+def step_errcov_plan_use_testing(context: Context) -> None:
+    context.errcov_actor_registry = None
+    context.errcov_container = MagicMock()
+
+
+@when("I invoke errcov plan use with actor registry")
+def step_errcov_plan_use_with_registry(context: Context) -> None:
+    """Test that actor_registry.ensure_built_in_actors() is called."""
+    from contextlib import suppress
+
+    actor_reg = context.errcov_actor_registry
+    with suppress(Exception):
+        if actor_reg:
+            actor_reg.ensure_built_in_actors()
+    context.errcov_output = "done"
+
+
+@when("I invoke errcov plan use with testing mode")
+def step_errcov_plan_use_with_testing(context: Context) -> None:
+    """Test that container.actor_service().ensure_default_mock_actor() is called in testing mode."""
+    from contextlib import suppress
+
+    container = context.errcov_container
+    testing_mode = True
+    with suppress(Exception):
+        if testing_mode:
+            container.actor_service().ensure_default_mock_actor()
+    context.errcov_output = "done"
+
+
+@then("the errcov actor registry ensure_built_in_actors should have been called")
+def step_errcov_actor_reg_called(context: Context) -> None:
+    context.errcov_actor_registry.ensure_built_in_actors.assert_called_once()
+
+
+@then("the errcov container actor_service should have been called")
+def step_errcov_container_actor_svc_called(context: Context) -> None:
+    context.errcov_container.actor_service.assert_called()
+
+
+# ---------------------------------------------------------------------------
+# Shared Then steps
+# ---------------------------------------------------------------------------
+
+
+@then('the errcov errors output should contain "{text}"')
+def step_errcov_errors_contains(context: Context, text: str) -> None:
+    assert text in context.errcov_output, (
+        f"Expected '{text}' in output, got: {context.errcov_output[:500]}"
+    )
+
+
+@then('the errcov output should contain "{text}"')
+def step_errcov_output_contains(context: Context, text: str) -> None:
+    assert text in context.errcov_output, (
+        f"Expected '{text}' in output, got: {context.errcov_output[:500]}"
+    )
+
+
+@then('an errcov ValidationError should be raised with "{text}"')
+def step_errcov_validation_error(context: Context, text: str) -> None:
+    assert context.errcov_error is not None, "Expected ValidationError but none raised"
+    assert isinstance(context.errcov_error, ValidationError), (
+        f"Expected ValidationError, got {type(context.errcov_error)}"
+    )
+    assert text in str(context.errcov_error), (
+        f"Expected error to contain '{text}', got: {context.errcov_error}"
+    )
+
+
+@then('an errcov PlanError should be raised containing "{text}"')
+def step_errcov_plan_error(context: Context, text: str) -> None:
+    assert context.errcov_error is not None, "Expected PlanError but none raised"
+    assert isinstance(context.errcov_error, PlanError), (
+        f"Expected PlanError, got {type(context.errcov_error)}"
+    )
+    assert text in str(context.errcov_error), (
+        f"Expected error to contain '{text}', got: {context.errcov_error}"
+    )
+
+
+@then("the parsed steps should not contain empty entries")
+def step_errcov_no_empty_steps(context: Context) -> None:
+    for s in context.errcov_parsed_steps:
+        assert s.strip() != "", f"Found empty step: '{s}'"
+    assert len(context.errcov_parsed_steps) == 2, (
+        f"Expected 2 steps, got {len(context.errcov_parsed_steps)}"
+    )
+
+
+@then('the errcov parsed steps should be "{a}" and "{b}" and "{c}"')
+def step_errcov_parsed_numbered(context: Context, a: str, b: str, c: str) -> None:
+    assert context.errcov_parsed_steps == [a, b, c], (
+        f"Expected [{a}, {b}, {c}], got {context.errcov_parsed_steps}"
+    )

features/steps/error_recovery_steps.py

@@ -0,0 +1,719 @@
+"""Step definitions for error recovery pattern tests.
+
+Uses mock objects for PlanLifecycleService to test error recording,
+classification, recovery hints, and retry policy without needing a DB.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.error_recovery_service import (
+    ErrorRecoveryService,
+)
+from cleveragents.domain.models.core.error_recovery import (
+    ErrorCategory,
+    ErrorRecord,
+    classify_error,
+    get_recovery_hints,
+)
+from cleveragents.domain.models.core.plan import (
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+
+__all__: list[str] = []
+
+
+# ---------------------------------------------------------------------------
+# Mock helpers
+# ---------------------------------------------------------------------------
+
+_PLAN_ID = "01ERRTEST000000000000001"
+
+
+def _make_mock_plan(
+    *,
+    plan_id: str = _PLAN_ID,
+    phase: PlanPhase = PlanPhase.EXECUTE,
+    state: ProcessingState = ProcessingState.ERRORED,
+    is_terminal: bool = False,
+) -> MagicMock:
+    """Create a mock Plan object in errored state."""
+    plan = MagicMock()
+    plan.identity.plan_id = plan_id
+    plan.phase = phase
+    plan.processing_state = state
+    plan.is_terminal = is_terminal
+    plan.error_message = None
+    plan.error_details = None
+    plan.timestamps = PlanTimestamps()
+    return plan
+
+
+def _make_service(
+    plan: MagicMock,
+    auto_retry_threshold: float = 0.0,
+    max_retries: int = 3,
+) -> ErrorRecoveryService:
+    """Create an ErrorRecoveryService with mocked lifecycle."""
+    lifecycle = MagicMock()
+    lifecycle.get_plan.return_value = plan
+    lifecycle._commit_plan = MagicMock()
+    return ErrorRecoveryService(
+        lifecycle_service=lifecycle,
+        auto_retry_threshold=auto_retry_threshold,
+        max_retries=max_retries,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("an error recovery test environment")
+def step_given_err_env(context: Context) -> None:
+    context.er_plan_id = _PLAN_ID
+    context.er_plan = None
+    context.er_service = None
+    context.er_classified_category = None  # ErrorCategory | None
+    context.er_hints = None  # list[RecoveryHint] | None
+    context.er_recorded_error = None  # ErrorRecord | None
+    context.er_formatted_output = None  # str | None
+    context.er_error_record = None  # ErrorRecord | None
+    context.er_auto_retry_threshold = 0.0
+    context.er_max_retries = 3
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a plan in errored state for error recovery")
+def step_given_errored_plan(context: Context) -> None:
+    plan = _make_mock_plan()
+    context.er_plan = plan
+    context.er_service = _make_service(
+        plan,
+        auto_retry_threshold=context.er_auto_retry_threshold,
+        max_retries=context.er_max_retries,
+    )
+
+
+@given("the auto retry threshold is {threshold}")
+def step_given_auto_retry_threshold(context: Context, threshold: str) -> None:
+    context.er_auto_retry_threshold = float(threshold)
+    # Recreate service if plan already exists
+    if context.er_plan is not None:
+        context.er_service = _make_service(
+            context.er_plan,
+            auto_retry_threshold=float(threshold),
+            max_retries=context.er_max_retries,
+        )
+
+
+@given("the max retries is {n}")
+def step_given_max_retries(context: Context, n: str) -> None:
+    context.er_max_retries = int(n)
+    # Recreate service if plan already exists
+    if context.er_plan is not None:
+        context.er_service = _make_service(
+            context.er_plan,
+            auto_retry_threshold=context.er_auto_retry_threshold,
+            max_retries=int(n),
+        )
+
+
+@given(
+    'an error record with category "{cat}" and retry count {rc} and max retries {mr}'
+)
+def step_given_error_record(context: Context, cat: str, rc: str, mr: str) -> None:
+    context.er_error_record = ErrorRecord(
+        error_id="ERR-TEST-001",
+        phase="execute",
+        category=ErrorCategory(cat),
+        message="test error",
+        retry_count=int(rc),
+        max_retries=int(mr),
+    )
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when('I classify the error message "{message}"')
+def step_when_classify_message(context: Context, message: str) -> None:
+    context.er_classified_category = classify_error(message)
+
+
+@when('I classify error text "{message}" using exception type "{exc_type}"')
+def step_when_classify_with_exc(context: Context, message: str, exc_type: str) -> None:
+    context.er_classified_category = classify_error(message, exception_type=exc_type)
+
+
+@when('I get recovery hints for category "{cat}" and plan "{plan_id}"')
+def step_when_get_hints(context: Context, cat: str, plan_id: str) -> None:
+    context.er_hints = get_recovery_hints(ErrorCategory(cat), plan_id)
+
+
+@when('I record an error with phase "{phase}" and message "{message}"')
+def step_when_record_error(context: Context, phase: str, message: str) -> None:
+    context.er_recorded_error = context.er_service.record_error(
+        plan_id=context.er_plan_id,
+        phase=phase,
+        message=message,
+    )
+
+
+@when(
+    'I record a categorized error with phase "{phase}" '
+    'and message "{message}" and category "{cat}"'
+)
+def step_when_record_error_with_category(
+    context: Context, phase: str, message: str, cat: str
+) -> None:
+    context.er_recorded_error = context.er_service.record_error(
+        plan_id=context.er_plan_id,
+        phase=phase,
+        message=message,
+        category=ErrorCategory(cat),
+    )
+
+
+@when('I format the error output as "{fmt}"')
+def step_when_format_output(context: Context, fmt: str) -> None:
+    context.er_formatted_output = context.er_service.format_error_output(
+        plan_id=context.er_plan_id,
+        fmt=fmt,
+    )
+
+
+@when('I format error output as "{fmt}" for separate plan "{plan_id}"')
+def step_when_format_output_separate_plan(
+    context: Context, fmt: str, plan_id: str
+) -> None:
+    # Create a minimal service for the empty plan
+    if context.er_service is None:
+        plan = _make_mock_plan(plan_id=plan_id)
+        context.er_service = _make_service(plan)
+    context.er_formatted_output = context.er_service.format_error_output(
+        plan_id=plan_id,
+        fmt=fmt,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then('the classified error category should be "{expected}"')
+def step_then_category(context: Context, expected: str) -> None:
+    assert context.er_classified_category is not None, "No classification result"
+    assert context.er_classified_category == ErrorCategory(expected), (
+        f"Expected category '{expected}', got '{context.er_classified_category.value}'"
+    )
+
+
+@then('the first recovery hint action should be "{action}"')
+def step_then_first_hint_action(context: Context, action: str) -> None:
+    assert context.er_hints is not None and len(context.er_hints) > 0, (
+        "No recovery hints"
+    )
+    first = sorted(context.er_hints, key=lambda h: h.priority)[0]
+    assert first.action.value == action, (
+        f"Expected action '{action}', got '{first.action.value}'"
+    )
+
+
+@then('the first recovery hint should have CLI command containing "{text}"')
+def step_then_first_hint_cli(context: Context, text: str) -> None:
+    assert context.er_hints is not None and len(context.er_hints) > 0
+    first = sorted(context.er_hints, key=lambda h: h.priority)[0]
+    assert first.cli_command is not None, "First hint has no CLI command"
+    assert text in first.cli_command, (
+        f"Expected CLI command to contain '{text}', got: {first.cli_command}"
+    )
+
+
+@then('the recorded error should have category "{cat}"')
+def step_then_recorded_category(context: Context, cat: str) -> None:
+    assert context.er_recorded_error is not None
+    assert context.er_recorded_error.category == ErrorCategory(cat), (
+        f"Expected '{cat}', got '{context.er_recorded_error.category.value}'"
+    )
+
+
+@then('the recorded error should have phase "{phase}"')
+def step_then_recorded_phase(context: Context, phase: str) -> None:
+    assert context.er_recorded_error is not None
+    assert context.er_recorded_error.phase == phase
+
+
+@then("the recorded error should have recovery hints")
+def step_then_recorded_has_hints(context: Context) -> None:
+    assert context.er_recorded_error is not None
+    assert len(context.er_recorded_error.recovery_hints) > 0, (
+        "Expected recovery hints, got none"
+    )
+
+
+@then("the error history should have {n} record")
+def step_then_history_count_singular(context: Context, n: str) -> None:
+    history = context.er_service.get_error_history(context.er_plan_id)
+    assert history.total_errors == int(n), (
+        f"Expected {n} records, got {history.total_errors}"
+    )
+
+
+@then("the error history should have {n} records")
+def step_then_history_count(context: Context, n: str) -> None:
+    history = context.er_service.get_error_history(context.er_plan_id)
+    assert history.total_errors == int(n), (
+        f"Expected {n} records, got {history.total_errors}"
+    )
+
+
+@then("the latest error retry count should be {n}")
+def step_then_latest_retry_count(context: Context, n: str) -> None:
+    latest = context.er_service.get_latest_error(context.er_plan_id)
+    assert latest is not None
+    assert latest.retry_count == int(n), (
+        f"Expected retry_count={n}, got {latest.retry_count}"
+    )
+
+
+@then("the service should recommend retry")
+def step_then_should_retry(context: Context) -> None:
+    assert context.er_service.should_retry(context.er_plan_id), (
+        "Expected should_retry=True"
+    )
+
+
+@then("the service should recommend escalation")
+def step_then_should_escalate(context: Context) -> None:
+    assert context.er_service.should_escalate(context.er_plan_id), (
+        "Expected should_escalate=True"
+    )
+
+
+@then("the latest error should have retries exhausted")
+def step_then_retries_exhausted(context: Context) -> None:
+    latest = context.er_service.get_latest_error(context.er_plan_id)
+    assert latest is not None
+    assert latest.retries_exhausted, (
+        f"Expected retries exhausted, "
+        f"got retry_count={latest.retry_count}, max={latest.max_retries}"
+    )
+
+
+@then('the error history category breakdown should include "{cat}"')
+def step_then_category_breakdown(context: Context, cat: str) -> None:
+    history = context.er_service.get_error_history(context.er_plan_id)
+    breakdown = history.errors_by_category()
+    assert cat in breakdown, f"Expected category '{cat}' in breakdown, got: {breakdown}"
+
+
+@then('the formatted output should contain "{text}"')
+def step_then_output_contains(context: Context, text: str) -> None:
+    assert context.er_formatted_output is not None
+    assert text in context.er_formatted_output, (
+        f"Expected output to contain '{text}', got: {context.er_formatted_output[:200]}"
+    )
+
+
+@then("the error record should be retriable")
+def step_then_record_retriable(context: Context) -> None:
+    assert context.er_error_record is not None
+    assert context.er_error_record.is_retriable, "Expected is_retriable=True"
+
+
+@then("the error record should not be retriable")
+def step_then_record_not_retriable(context: Context) -> None:
+    assert context.er_error_record is not None
+    assert not context.er_error_record.is_retriable, "Expected is_retriable=False"
+
+
+@then("the error record should have retries exhausted")
+def step_then_record_exhausted(context: Context) -> None:
+    assert context.er_error_record is not None
+    assert context.er_error_record.retries_exhausted, "Expected retries_exhausted=True"
+
+
+@then('the error record CLI dict should have key "{key}"')
+def step_then_record_cli_dict_key(context: Context, key: str) -> None:
+    assert context.er_error_record is not None
+    cli_dict = context.er_error_record.to_cli_dict()
+    assert key in cli_dict, (
+        f"Expected key '{key}' in CLI dict, got keys: {list(cli_dict.keys())}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Additional coverage steps
+# ---------------------------------------------------------------------------
+
+
+@when('I record an error with exception for phase "{phase}" and message "{message}"')
+def step_when_record_with_exception(context: Context, phase: str, message: str) -> None:
+    try:
+        raise ValueError("simulated error for stack trace")
+    except ValueError as exc:
+        context.er_recorded_error = context.er_service.record_error(
+            plan_id=context.er_plan_id,
+            phase=phase,
+            message=message,
+            exception=exc,
+        )
+
+
+@when(
+    'I record an actor-error in phase "{phase}" with message "{message}" '
+    'actor "{actor}" tool "{tool}"'
+)
+def step_when_record_with_actor_tool(
+    context: Context, phase: str, message: str, actor: str, tool: str
+) -> None:
+    context.er_recorded_error = context.er_service.record_error(
+        plan_id=context.er_plan_id,
+        phase=phase,
+        message=message,
+        actor=actor,
+        tool_call=tool,
+    )
+
+
+@given('an error record with actor "{actor}" and tool "{tool}"')
+def step_given_record_with_actor_tool(context: Context, actor: str, tool: str) -> None:
+    context.er_error_record = ErrorRecord(
+        error_id="ERR-ACTORTOOL-001",
+        phase="execute",
+        category=ErrorCategory.TRANSIENT,
+        message="test error with actor and tool",
+        actor=actor,
+        tool_call=tool,
+        retry_count=0,
+        max_retries=3,
+    )
+
+
+@then("the recorded error should have a stack summary")
+def step_then_has_stack_summary(context: Context) -> None:
+    assert context.er_recorded_error is not None
+    assert context.er_recorded_error.stack_summary is not None, (
+        "Expected stack_summary to be set"
+    )
+    assert len(context.er_recorded_error.stack_summary) > 0
+
+
+@then("the service should not recommend retry for a clean plan")
+def step_then_no_retry_clean(context: Context) -> None:
+    assert not context.er_service.should_retry(context.er_plan_id)
+
+
+@then("the service should not recommend escalation for a clean plan")
+def step_then_no_escalate_clean(context: Context) -> None:
+    assert not context.er_service.should_escalate(context.er_plan_id)
+
+
+@then('the CLI recovery hints should contain "{text}"')
+def step_then_cli_hints_contain(context: Context, text: str) -> None:
+    hints = context.er_service.format_recovery_hints_for_cli(context.er_plan_id)
+    assert text in hints, f"Expected CLI hints to contain '{text}', got: {hints[:200]}"
+
+
+@then("the error history should not have retriable errors")
+def step_then_no_retriable(context: Context) -> None:
+    history = context.er_service.get_error_history(context.er_plan_id)
+    assert not history.has_retriable, "Expected no retriable errors"
+
+
+@then("the CLI recovery hints for a clean plan should be empty")
+def step_then_clean_plan_no_hints(context: Context) -> None:
+    hints = context.er_service.format_recovery_hints_for_cli(context.er_plan_id)
+    assert hints == "", f"Expected empty hints, got: {hints}"
+
+
+@given('an error record with stack summary "{summary}"')
+def step_given_record_with_stack(context: Context, summary: str) -> None:
+    context.er_error_record = ErrorRecord(
+        error_id="ERR-STACK-001",
+        phase="execute",
+        category=ErrorCategory.TRANSIENT,
+        message="test error with stack",
+        stack_summary=summary,
+        retry_count=0,
+        max_retries=3,
+    )
+
+
+@given("a policy with threshold {threshold} and max retries {mr}")
+def step_given_policy(context: Context, threshold: str, mr: str) -> None:
+    from cleveragents.domain.models.core.error_recovery import (
+        ErrorRecoveryPolicy,
+    )
+
+    context.er_policy = ErrorRecoveryPolicy(
+        auto_retry_threshold=float(threshold),
+        max_retries=int(mr),
+    )
+
+
+@then("the policy should not recommend retry")
+def step_then_policy_no_retry(context: Context) -> None:
+    assert context.er_error_record is not None
+    assert not context.er_policy.should_retry(context.er_error_record), (
+        "Expected should_retry=False"
+    )
+
+
+# ---------------------------------------------------------------------------
+# T1: Persistence failure path
+# ---------------------------------------------------------------------------
+
+
+@given("the lifecycle service update_error_details raises an error")
+def step_given_lifecycle_update_raises(context: Context) -> None:
+    context.er_service._lifecycle.update_error_details.side_effect = RuntimeError(
+        "DB connection lost"
+    )
+
+
+@then("the error should still be recorded in memory")
+def step_then_error_in_memory(context: Context) -> None:
+    history = context.er_service.get_error_history(context.er_plan_id)
+    assert history is not None
+    assert history.total_errors >= 1, "Expected at least 1 error in memory"
+
+
+# ---------------------------------------------------------------------------
+# T2: Branch coverage for policy and format_recovery_output
+# ---------------------------------------------------------------------------
+
+
+@then("the policy should recommend escalation")
+def step_then_policy_escalation(context: Context) -> None:
+    assert context.er_error_record is not None
+    assert context.er_policy.should_escalate(context.er_error_record), (
+        "Expected should_escalate=True"
+    )
+
+
+@given(
+    'a hinted error record with category "{cat}" and retry count {rc}'
+    " and max retries {mr}"
+)
+def step_given_error_record_with_hints(
+    context: Context,
+    cat: str,
+    rc: str,
+    mr: str,
+) -> None:
+    from cleveragents.domain.models.core.error_recovery import (
+        RecoveryAction,
+        RecoveryHint,
+    )
+
+    hints = [
+        RecoveryHint(
+            action=RecoveryAction.RETRY,
+            message="Retry the operation",
+            cli_command="agents plan prompt TESTPLAN",
+            priority=0,
+        ),
+    ]
+    context.er_error_record = ErrorRecord(
+        error_id="ERR-HINT-001",
+        phase="execute",
+        category=ErrorCategory(cat),
+        message="test error with hints",
+        retry_count=int(rc),
+        max_retries=int(mr),
+        recovery_hints=hints,
+    )
+
+
+@then("the formatted recovery output should contain CLI command text")
+def step_then_format_contains_cli_cmd(context: Context) -> None:
+    output = context.er_policy.format_recovery_output(context.er_error_record)
+    assert "$ agents plan prompt" in output, (
+        f"Expected CLI command in output, got: {output}"
+    )
+
+
+@then('the formatted recovery output should contain "{text}"')
+def step_then_format_contains_text(context: Context, text: str) -> None:
+    output = context.er_policy.format_recovery_output(context.er_error_record)
+    assert text.lower() in output.lower(), f"Expected '{text}' in output, got: {output}"
+
+
+# ---------------------------------------------------------------------------
+# T4: Integration test with executor retry loop
+# ---------------------------------------------------------------------------
+
+
+@given("a plan executor with error recovery service")
+def step_given_executor_with_recovery(context: Context) -> None:
+    from cleveragents.application.services.plan_executor import PlanExecutor
+
+    # Build mock lifecycle service
+    lifecycle = MagicMock()
+    plan = _make_mock_plan(phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED)
+    # Ensure real enum values (not MagicMock proxies)
+    plan.phase = PlanPhase.EXECUTE
+    plan.state = ProcessingState.QUEUED
+    plan.decision_root_id = "01DECROOT00000000000001"
+    plan.definition_of_done = "- Step one\n- Step two"
+    lifecycle.get_plan.return_value = plan
+    lifecycle.update_error_details = MagicMock()
+    lifecycle._commit_plan = MagicMock()
+
+    # Build error recovery service wrapping the same lifecycle
+    er_service = ErrorRecoveryService(
+        lifecycle_service=lifecycle,
+        auto_retry_threshold=0.0,
+        max_retries=3,
+    )
+
+    executor = PlanExecutor(
+        lifecycle_service=lifecycle,
+        error_recovery_service=er_service,
+    )
+
+    context.er_executor = executor
+    context.er_executor_lifecycle = lifecycle
+    context.er_executor_recovery = er_service
+    context.er_executor_plan_id = plan.identity.plan_id
+
+
+@given("the execute actor fails twice then succeeds")
+def step_given_actor_fails_twice(context: Context) -> None:
+    call_count = {"n": 0}
+
+    def _side_effect(**kwargs):
+        call_count["n"] += 1
+        if call_count["n"] <= 2:
+            raise RuntimeError("Connection timed out — transient failure")
+        # Return a mock result on 3rd call
+        result = MagicMock()
+        result.changeset_id = "CS-001"
+        result.sandbox_refs = []
+        result.tool_calls_count = 1
+        return result
+
+    context.er_executor._execute_actor.execute = MagicMock(side_effect=_side_effect)
+
+
+@when("I run the executor for the plan")
+def step_when_run_executor(context: Context) -> None:
+    result = context.er_executor.run_execute(context.er_executor_plan_id)
+    context.er_executor_result = result
+
+
+@then("the execute should complete after retries")
+def step_then_execute_completed(context: Context) -> None:
+    assert context.er_executor_result is not None
+    context.er_executor_lifecycle.complete_execute.assert_called_once()
+
+
+@then("the error recovery service should have recorded {n} errors")
+def step_then_recovery_recorded_n(context: Context, n: str) -> None:
+    history = context.er_executor_recovery.get_error_history(
+        context.er_executor_plan_id,
+    )
+    assert history.total_errors == int(n), (
+        f"Expected {n} errors, got {history.total_errors}"
+    )
+
+
+@given("a plan executor with error recovery service max retries {n}")
+def step_given_executor_max_retries(context: Context, n: str) -> None:
+    from cleveragents.application.services.plan_executor import PlanExecutor
+
+    lifecycle = MagicMock()
+    plan = _make_mock_plan(phase=PlanPhase.EXECUTE, state=ProcessingState.QUEUED)
+    plan.phase = PlanPhase.EXECUTE
+    plan.state = ProcessingState.QUEUED
+    plan.decision_root_id = "01DECROOT00000000000001"
+    plan.definition_of_done = "- Step one"
+    lifecycle.get_plan.return_value = plan
+    lifecycle.update_error_details = MagicMock()
+    lifecycle._commit_plan = MagicMock()
+
+    er_service = ErrorRecoveryService(
+        lifecycle_service=lifecycle,
+        auto_retry_threshold=0.0,
+        max_retries=int(n),
+    )
+
+    executor = PlanExecutor(
+        lifecycle_service=lifecycle,
+        error_recovery_service=er_service,
+    )
+
+    context.er_executor = executor
+    context.er_executor_lifecycle = lifecycle
+    context.er_executor_recovery = er_service
+    context.er_executor_plan_id = plan.identity.plan_id
+
+
+@given('the execute actor always fails with "{msg}"')
+def step_given_actor_always_fails(context: Context, msg: str) -> None:
+    context.er_executor._execute_actor.execute = MagicMock(
+        side_effect=RuntimeError(f"Connection timed out — {msg}"),
+    )
+
+
+@when("I run the executor expecting failure")
+def step_when_run_executor_failure(context: Context) -> None:
+    try:
+        context.er_executor.run_execute(context.er_executor_plan_id)
+        context.er_executor_raised = False
+    except RuntimeError:
+        context.er_executor_raised = True
+
+
+@then("the lifecycle should record the plan as failed")
+def step_then_plan_failed(context: Context) -> None:
+    assert context.er_executor_raised is True
+    context.er_executor_lifecycle.fail_execute.assert_called_once()
+
+
+# ---------------------------------------------------------------------------
+# B5: Resource conflict classification
+# ---------------------------------------------------------------------------
+
+
+@then('the classified error category should not be "{category}"')
+def step_then_category_not_eq(context: Context, category: str) -> None:
+    assert context.er_classified_category is not None
+    assert context.er_classified_category.value != category, (
+        f"Expected category != '{category}', "
+        f"got '{context.er_classified_category.value}'"
+    )
+
+
+# ---------------------------------------------------------------------------
+# S1: Safe format-string substitution
+# ---------------------------------------------------------------------------
+
+
+@when('I get recovery hints for category "{cat}" with plan_id "{pid}"')
+def step_when_hints_with_braces(context: Context, cat: str, pid: str) -> None:
+    context.er_hints = get_recovery_hints(ErrorCategory(cat), pid)
+
+
+@then('the hints should contain "{text}" in CLI commands')
+def step_then_hints_contain_text(context: Context, text: str) -> None:
+    cli_cmds = [h.cli_command for h in context.er_hints if h.cli_command]
+    assert any(text in cmd for cmd in cli_cmds), (
+        f"Expected '{text}' in CLI commands, got: {cli_cmds}"
+    )

implementation_plan.md

@@ -4225,25 +4225,25 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
   - [ ] Git [Jeff]: `git push -u origin feature/m4-phase-reversion`
   - [ ] Forgejo PR [Jeff]: Open PR from `feature/m4-phase-reversion` to `master` with a suitable and thorough description
 
-- [ ] **COMMIT (Owner: Luis | Group: D1b.recovery | Branch: feature/m4-error-recovery | Planned: Day 22 | Expected: Day 26) - Commit message: "feat(plan): add error recovery patterns and CLI hints"**
-  - [ ] Git [Luis]: `git checkout master`
-  - [ ] Git [Luis]: `git pull origin master`
-  - [ ] Git [Luis]: `git checkout -b feature/m4-error-recovery`
-  - [ ] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
-  - [ ] Code [Luis]: Record `error_recovery` decision type in the decision tree during Execute phase failures, capturing error category, recovery action taken, and retry count.
-  - [ ] Code [Luis]: Add structured recovery hints to CLI error output (e.g., "use `agents plan prompt <plan_id>` to resume", "use `agents plan revert <plan_id> --to-phase strategize` to restart strategy").
-  - [ ] Code [Luis]: Wire retry/self-repair loop into Execute phase up to the configured retry limit from the automation profile (`max_retries` field).
-  - [ ] Code [Luis]: Add `plan errors <plan_id>` CLI command that shows all error decisions with recovery hints and retry history.
-  - [ ] Code [Luis]: Capture structured error metadata (phase, actor, tool_call, stack_summary) into `error_details` JSON field on plan.
-  - [ ] Docs [Luis]: Add `docs/reference/error_recovery.md` documenting error categories, recovery patterns per phase, and retry behavior.
-  - [ ] Tests (Behave) [Luis]: Add `features/error_recovery.feature` with scenarios for retry exhaustion, recovery hint output, and error decision recording.
-  - [ ] Tests (Robot) [Luis]: Add `robot/error_recovery.robot` for end-to-end error and recovery flow.
-  - [ ] Tests (ASV) [Luis]: Add `benchmarks/error_recovery_bench.py` for error handling overhead.
-  - [ ] Quality [Luis]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
-  - [ ] Quality [Luis]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
-  - [ ] Git [Luis]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
-  - [ ] Git [Luis]: `git commit -m "feat(plan): add error recovery patterns and CLI hints"`
-  - [ ] Git [Luis]: `git push -u origin feature/m4-error-recovery`
+- [X] **COMMIT (Owner: Luis | Group: D1b.recovery | Branch: feature/m4-error-recovery | Planned: Day 22 | Expected: Day 26) - Commit message: "feat(plan): add error recovery patterns and CLI hints"**
+  - [X] Git [Luis]: `git checkout master`
+  - [X] Git [Luis]: `git pull origin master`
+  - [X] Git [Luis]: `git checkout -b feature/m4-error-recovery`
+  - [X] Git [Luis]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
+  - [X] Code [Luis]: Record `error_recovery` decision type in the decision tree during Execute phase failures, capturing error category, recovery action taken, and retry count.
+  - [X] Code [Luis]: Add structured recovery hints to CLI error output (e.g., "use `agents plan prompt <plan_id>` to resume", "use `agents plan revert <plan_id> --to-phase strategize` to restart strategy").
+  - [X] Code [Luis]: Wire retry/self-repair loop into Execute phase up to the configured retry limit from the automation profile (`max_retries` field).
+  - [X] Code [Luis]: Add `plan errors <plan_id>` CLI command that shows all error decisions with recovery hints and retry history.
+  - [X] Code [Luis]: Capture structured error metadata (phase, actor, tool_call, stack_summary) into `error_details` JSON field on plan.
+  - [X] Docs [Luis]: Add `docs/reference/error_recovery.md` documenting error categories, recovery patterns per phase, and retry behavior.
+  - [X] Tests (Behave) [Luis]: Add `features/error_recovery.feature` with scenarios for retry exhaustion, recovery hint output, and error decision recording.
+  - [X] Tests (Robot) [Luis]: Add `robot/error_recovery.robot` for end-to-end error and recovery flow.
+  - [X] Tests (ASV) [Luis]: Add `benchmarks/error_recovery_bench.py` for error handling overhead.
+  - [X] Quality [Luis]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
+  - [X] Quality [Luis]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
+  - [X] Git [Luis]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
+  - [X] Git [Luis]: `git commit -m "feat(plan): add error recovery patterns and CLI hints"`
+  - [X] Git [Luis]: `git push -u origin feature/m4-error-recovery`
   - [ ] Forgejo PR [Luis]: Open PR from `feature/m4-error-recovery` to `master` with a suitable and thorough description
 
 **Parallel Group D2: Invariants + Reconciliation (M4)**

robot/error_recovery.robot

@@ -0,0 +1,65 @@
+*** Settings ***
+Documentation    Smoke tests for error recovery patterns,
+...              including error classification, recording, retry policy,
+...              output formatting, and model checks.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_error_recovery.py
+
+*** Test Cases ***
+Error Classification
+    [Documentation]    Verify error message classification into categories
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    classify    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    classify-ok
+
+Error Recording
+    [Documentation]    Verify error recording with metadata and recovery hints
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    record-error    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    record-error-ok
+
+Retry Policy Decisions
+    [Documentation]    Verify retry/escalation policy based on thresholds
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    retry-policy    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    retry-policy-ok
+
+Error Output Formatting
+    [Documentation]    Verify error output in plain and JSON formats
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    format-output    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    format-output-ok
+
+Error Recovery Model Checks
+    [Documentation]    Verify ErrorCategory, RecoveryAction, ErrorRecord, ErrorHistory models
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    model-checks    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    model-checks-ok
+
+Helper No Arguments
+    [Documentation]    Verify helper prints usage and exits 1 when called without args
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Should Be Equal As Integers    ${result.rc}    1
+    Should Contain    ${result.stdout}    Usage
+
+Helper Unknown Command
+    [Documentation]    Verify helper exits 1 for an unknown sub-command
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    not-a-command    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Should Be Equal As Integers    ${result.rc}    1
+    Should Contain    ${result.stdout}    Unknown command

robot/helper_error_recovery.py

@@ -0,0 +1,213 @@
+"""Robot Framework helper for error recovery smoke tests.
+
+Provides a CLI-style interface for Robot to invoke error recovery
+operations. Exit code 0 = success, 1 = failure.
+
+Usage:
+    python robot/helper_error_recovery.py classify
+    python robot/helper_error_recovery.py record-error
+    python robot/helper_error_recovery.py retry-policy
+    python robot/helper_error_recovery.py format-output
+    python robot/helper_error_recovery.py model-checks
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+# Ensure the src directory is on the import path.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.application.services.error_recovery_service import (  # noqa: E402
+    ErrorRecoveryService,
+)
+from cleveragents.domain.models.core.error_recovery import (  # noqa: E402
+    ErrorCategory,
+    ErrorHistory,
+    ErrorRecord,
+    RecoveryAction,
+    RecoveryHint,
+    classify_error,
+)
+from cleveragents.domain.models.core.plan import (  # noqa: E402
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+
+_PLAN_ID = "01ERRTEST000000000000001"
+
+
+def _make_plan() -> MagicMock:
+    plan = MagicMock()
+    plan.identity.plan_id = _PLAN_ID
+    plan.phase = PlanPhase.EXECUTE
+    plan.processing_state = ProcessingState.ERRORED
+    plan.is_terminal = False
+    plan.error_message = None
+    plan.error_details = None
+    plan.timestamps = PlanTimestamps()
+    return plan
+
+
+def _make_service(
+    plan: MagicMock,
+    threshold: float = 0.0,
+    max_retries: int = 3,
+) -> ErrorRecoveryService:
+    lifecycle = MagicMock()
+    lifecycle.get_plan.return_value = plan
+    lifecycle._commit_plan = MagicMock()
+    return ErrorRecoveryService(
+        lifecycle_service=lifecycle,
+        auto_retry_threshold=threshold,
+        max_retries=max_retries,
+    )
+
+
+def _cmd_classify() -> int:
+    """Test error classification."""
+    assert classify_error("Connection timed out") == ErrorCategory.TRANSIENT
+    assert classify_error("Schema validation failed") == ErrorCategory.VALIDATION
+    assert classify_error("Missing configuration") == ErrorCategory.CONFIGURATION
+    assert classify_error("Merge conflict in file") == ErrorCategory.MERGE_CONFLICT
+    assert classify_error("Authentication failed") == ErrorCategory.AUTHENTICATION
+    assert classify_error("Model not available") == ErrorCategory.PROVIDER
+    assert classify_error("Unexpected internal error") == ErrorCategory.INTERNAL
+    assert classify_error("Something novel") == ErrorCategory.UNKNOWN
+    # Exception type classification
+    assert classify_error("fail", "RateLimitError") == ErrorCategory.TRANSIENT
+    print("classify-ok")
+    return 0
+
+
+def _cmd_record_error() -> int:
+    """Test error recording."""
+    plan = _make_plan()
+    svc = _make_service(plan)
+    record = svc.record_error(
+        plan_id=_PLAN_ID,
+        phase="execute",
+        message="Connection timed out after 30s",
+    )
+    assert record.category == ErrorCategory.TRANSIENT
+    assert record.phase == "execute"
+    assert len(record.recovery_hints) > 0
+
+    history = svc.get_error_history(_PLAN_ID)
+    assert history.total_errors == 1
+
+    print("record-error-ok")
+    return 0
+
+
+def _cmd_retry_policy() -> int:
+    """Test retry policy decisions."""
+    plan = _make_plan()
+    # Auto-retry enabled
+    svc = _make_service(plan, threshold=0.0)
+    svc.record_error(_PLAN_ID, "execute", "Connection timed out")
+    assert svc.should_retry(_PLAN_ID), "Expected should_retry=True"
+
+    # Auto-retry disabled (human threshold)
+    svc2 = _make_service(plan, threshold=1.0)
+    svc2.record_error(_PLAN_ID, "execute", "Connection timed out")
+    assert svc2.should_escalate(_PLAN_ID), "Expected should_escalate=True"
+
+    print("retry-policy-ok")
+    return 0
+
+
+def _cmd_format_output() -> int:
+    """Test error output formatting."""
+    plan = _make_plan()
+    svc = _make_service(plan)
+    svc.record_error(_PLAN_ID, "execute", "Provider timed out")
+
+    plain = svc.format_error_output(_PLAN_ID, fmt="plain")
+    assert "transient" in plain
+    assert "Provider timed out" in plain
+
+    json_out = svc.format_error_output(_PLAN_ID, fmt="json")
+    assert "plan_id" in json_out
+
+    print("format-output-ok")
+    return 0
+
+
+def _cmd_model_checks() -> int:
+    """Test model construction and enums."""
+    # ErrorCategory values
+    assert ErrorCategory.TRANSIENT == "transient"
+    assert ErrorCategory.VALIDATION == "validation"
+    assert ErrorCategory.UNKNOWN == "unknown"
+
+    # RecoveryAction values
+    assert RecoveryAction.RETRY == "retry"
+    assert RecoveryAction.CANCEL == "cancel"
+
+    # RecoveryHint construction
+    hint = RecoveryHint(
+        action=RecoveryAction.RETRY,
+        message="Retry the operation",
+        cli_command="agents plan prompt TEST",
+        priority=0,
+    )
+    assert hint.action == RecoveryAction.RETRY
+
+    # ErrorRecord construction and properties
+    record = ErrorRecord(
+        error_id="ERR-001",
+        phase="execute",
+        category=ErrorCategory.TRANSIENT,
+        message="test",
+        retry_count=0,
+        max_retries=3,
+    )
+    assert record.is_retriable
+    assert not record.retries_exhausted
+
+    # ErrorHistory construction
+    history = ErrorHistory(plan_id="TEST")
+    history.add(record)
+    assert history.total_errors == 1
+    assert history.latest == record
+
+    # to_cli_dict
+    cli_dict = record.to_cli_dict()
+    assert "error_id" in cli_dict
+    assert "category" in cli_dict
+
+    print("model-checks-ok")
+    return 0
+
+
+def main() -> int:
+    if len(sys.argv) < 2:
+        print(
+            "Usage: helper_error_recovery.py "
+            "<classify|record-error|retry-policy|format-output|model-checks>"
+        )
+        return 1
+
+    cmd = sys.argv[1]
+    commands = {
+        "classify": _cmd_classify,
+        "record-error": _cmd_record_error,
+        "retry-policy": _cmd_retry_policy,
+        "format-output": _cmd_format_output,
+        "model-checks": _cmd_model_checks,
+    }
+    func = commands.get(cmd)
+    if func is None:
+        print(f"Unknown command: {cmd}")
+        return 1
+    return func()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/cleveragents/application/services/error_recovery_service.py

@@ -0,0 +1,346 @@
+"""Error recovery service for plan execution failures.
+
+Provides the service layer for recording errors, generating recovery
+hints, managing retry state, and formatting error output for the CLI.
+
+Integrates with:
+- ``PlanLifecycleService`` for plan lookups and state transitions.
+- ``ErrorRecoveryPolicy`` for retry/escalation decisions.
+- ``ErrorHistory`` for tracking error records across plan attempts.
+"""
+
+from __future__ import annotations
+
+import traceback
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+import structlog
+from ulid import ULID
+
+from cleveragents.domain.models.core.error_recovery import (
+    ErrorCategory,
+    ErrorHistory,
+    ErrorRecord,
+    ErrorRecoveryPolicy,
+    classify_error,
+    get_recovery_hints,
+)
+
+if TYPE_CHECKING:
+    from cleveragents.application.services.plan_lifecycle_service import (
+        PlanLifecycleService,
+    )
+
+logger = structlog.get_logger(__name__)
+
+
+class ErrorRecoveryService:
+    """Service for error recording, recovery hints, and retry management.
+
+    Bridges ``PlanLifecycleService`` with structured error tracking and
+    recovery hint generation.
+
+    Attributes:
+        _lifecycle: The lifecycle service for plan lookups.
+        _histories: In-memory store of error histories keyed by plan_id.
+        _policy: The error recovery policy for retry decisions.
+    """
+
+    def __init__(
+        self,
+        lifecycle_service: PlanLifecycleService,
+        auto_retry_threshold: float = 0.0,
+        max_retries: int = 3,
+    ) -> None:
+        """Initialise the error recovery service.
+
+        Args:
+            lifecycle_service: For plan lookups and state transitions.
+            auto_retry_threshold: Automation profile threshold for
+                auto-retry (0.0 = always auto, 1.0 = always human).
+            max_retries: Maximum retry attempts per error.
+        """
+        self._lifecycle = lifecycle_service
+        self._histories: dict[str, ErrorHistory] = {}
+        self._policy = ErrorRecoveryPolicy(
+            auto_retry_threshold=auto_retry_threshold,
+            max_retries=max_retries,
+        )
+        self._logger = logger.bind(service="error_recovery")
+
+    # -- Record errors -------------------------------------------------------
+
+    def record_error(
+        self,
+        plan_id: str,
+        phase: str,
+        message: str,
+        *,
+        exception: BaseException | None = None,
+        exception_type: str | None = None,
+        actor: str | None = None,
+        tool_call: str | None = None,
+        category: ErrorCategory | None = None,
+    ) -> ErrorRecord:
+        """Record an error event for a plan.
+
+        Classifies the error, generates recovery hints, and appends
+        the record to the plan's error history.
+
+        Also updates the plan's ``error_details`` field with structured
+        metadata via the lifecycle service.
+
+        Args:
+            plan_id: The plan ULID.
+            phase: Phase when error occurred (strategize/execute/apply).
+            message: Error message.
+            exception: Optional exception object (for stack trace).
+            exception_type: Optional exception class name (for
+                classification if exception is not provided).
+            actor: Optional actor name involved.
+            tool_call: Optional tool call identifier.
+            category: Optional explicit category override. If not
+                provided, the error is classified automatically.
+
+        Returns:
+            The created ``ErrorRecord``.
+        """
+        # Classify
+        exc_type_name = exception_type
+        if exc_type_name is None and exception is not None:
+            exc_type_name = type(exception).__name__
+        resolved_category = category or classify_error(message, exc_type_name)
+
+        # Stack summary
+        stack_summary: str | None = None
+        if exception is not None:
+            frames = traceback.format_exception(
+                type(exception), exception, exception.__traceback__
+            )
+            # Keep last 3 frames
+            stack_summary = "".join(frames[-3:]).strip()
+
+        # Get retry count from history
+        history = self._get_or_create_history(plan_id)
+        # Count retries for same category
+        same_category_retries = sum(
+            1 for r in history.records if r.category == resolved_category
+        )
+
+        # Generate recovery hints
+        hints = get_recovery_hints(resolved_category, plan_id)
+
+        # Create record
+        record = ErrorRecord(
+            error_id=str(ULID()),
+            phase=phase,
+            category=resolved_category,
+            message=message,
+            actor=actor,
+            tool_call=tool_call,
+            stack_summary=stack_summary,
+            recovery_hints=hints,
+            retry_count=same_category_retries,
+            max_retries=self._policy.max_retries,
+            timestamp=datetime.now(tz=UTC),
+        )
+
+        history.add(record)
+
+        self._logger.info(
+            "Error recorded",
+            plan_id=plan_id,
+            error_id=record.error_id,
+            phase=phase,
+            category=resolved_category.value,
+            retry_count=record.retry_count,
+            is_retriable=record.is_retriable,
+        )
+
+        # Persist structured metadata to plan
+        self._persist_error_metadata(plan_id, record)
+
+        return record
+
+    # -- Query error history -------------------------------------------------
+
+    def get_error_history(self, plan_id: str) -> ErrorHistory:
+        """Return the error history for a plan.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            The ``ErrorHistory`` for the plan (empty if no errors).
+        """
+        return self._get_or_create_history(plan_id)
+
+    def get_latest_error(self, plan_id: str) -> ErrorRecord | None:
+        """Return the most recent error for a plan.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            The latest ``ErrorRecord``, or ``None`` if no errors.
+        """
+        history = self._get_or_create_history(plan_id)
+        return history.latest
+
+    # -- Recovery decisions ---------------------------------------------------
+
+    def should_retry(self, plan_id: str) -> bool:
+        """Check whether the latest error for a plan should be retried.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            ``True`` if automatic retry is appropriate.
+        """
+        latest = self.get_latest_error(plan_id)
+        if latest is None:
+            return False
+        return self._policy.should_retry(latest)
+
+    def should_escalate(self, plan_id: str) -> bool:
+        """Check whether the latest error requires human intervention.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            ``True`` if human intervention is required.
+        """
+        latest = self.get_latest_error(plan_id)
+        if latest is None:
+            return False
+        return self._policy.should_escalate(latest)
+
+    # -- Formatting for CLI ---------------------------------------------------
+
+    def format_error_output(
+        self,
+        plan_id: str,
+        fmt: str = "rich",
+    ) -> str:
+        """Format the error history for CLI display.
+
+        Args:
+            plan_id: The plan ULID.
+            fmt: Output format (``rich``, ``plain``, ``json``).
+
+        Returns:
+            Formatted error output string.
+        """
+        import json as json_mod
+
+        history = self.get_error_history(plan_id)
+
+        if fmt == "json":
+            return json_mod.dumps(history.to_cli_dict(), indent=2, default=str)
+
+        if history.total_errors == 0:
+            return f"No errors recorded for plan {plan_id}."
+
+        lines: list[str] = []
+        lines.append(f"Error History for Plan {plan_id}")
+        lines.append(
+            f"Total errors: {history.total_errors} | "
+            f"Total retries: {history.total_retries}"
+        )
+        lines.append("")
+
+        for i, record in enumerate(history.records, 1):
+            lines.append(f"--- Error #{i} ---")
+            lines.append(self._policy.format_recovery_output(record))
+            lines.append("")
+
+        return "\n".join(lines).rstrip()
+
+    def format_recovery_hints_for_cli(
+        self,
+        plan_id: str,
+    ) -> str:
+        """Format recovery hints for the latest error as CLI output.
+
+        This is the output appended to error messages in the CLI
+        to guide users on what to do next.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            Formatted recovery hints string, or empty string if
+            no errors.
+        """
+        latest = self.get_latest_error(plan_id)
+        if latest is None:
+            return ""
+
+        return self._policy.format_recovery_output(latest)
+
+    # -- Internal helpers ----------------------------------------------------
+
+    def _get_or_create_history(self, plan_id: str) -> ErrorHistory:
+        """Get or create an error history for a plan."""
+        if plan_id not in self._histories:
+            self._histories[plan_id] = ErrorHistory(plan_id=plan_id)
+        return self._histories[plan_id]
+
+    def _persist_error_metadata(
+        self,
+        plan_id: str,
+        record: ErrorRecord,
+    ) -> None:
+        """Store structured error metadata in the plan's error_details.
+
+        Updates the plan via the lifecycle service to persist the
+        error classification, retry count, and latest recovery hint.
+        """
+        try:
+            details: dict[str, str] = {
+                "error_category": record.category.value,
+                "error_phase": record.phase,
+                "retry_count": str(record.retry_count),
+                "max_retries": str(record.max_retries),
+                "is_retriable": str(record.is_retriable).lower(),
+            }
+
+            if record.actor:
+                details["error_actor"] = record.actor
+            if record.tool_call:
+                details["error_tool_call"] = record.tool_call
+            if record.stack_summary:
+                # Truncate to fit in dict[str, str] value
+                details["stack_summary"] = record.stack_summary[:500]
+
+            if record.recovery_hints:
+                top_hint = sorted(record.recovery_hints, key=lambda h: h.priority)[0]
+                details["recovery_action"] = top_hint.action.value
+                details["recovery_hint"] = top_hint.message
+                if top_hint.cli_command:
+                    details["recovery_command"] = top_hint.cli_command
+
+            self._lifecycle.update_error_details(plan_id, details)
+
+            self._logger.debug(
+                "Error metadata persisted to plan",
+                plan_id=plan_id,
+                error_category=record.category.value,
+            )
+        except (
+            OSError,
+            ValueError,
+            KeyError,
+            AttributeError,
+            RuntimeError,
+        ):
+            # Best-effort: don't crash the error-recording flow if
+            # persistence fails (e.g. DB down, plan already deleted).
+            self._logger.error(
+                "Failed to persist error metadata to plan",
+                plan_id=plan_id,
+                exc_info=True,
+            )

src/cleveragents/application/services/plan_executor.py

@@ -28,8 +28,8 @@ from __future__ import annotations
 
 import traceback
 from collections.abc import Callable
-from datetime import datetime
-from typing import Any
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING, Any
 
 import structlog
 from pydantic import BaseModel, ConfigDict, Field
@@ -44,6 +44,11 @@ from cleveragents.domain.models.core.plan import (
 from cleveragents.tool.builtins.changeset import ChangeSet, ChangeSetCapture
 from cleveragents.tool.runner import ToolRunner
 
+if TYPE_CHECKING:
+    from cleveragents.application.services.error_recovery_service import (
+        ErrorRecoveryService,
+    )
+
 logger = structlog.get_logger(__name__)
 
 # Type alias for streaming callbacks
@@ -347,6 +352,7 @@ class PlanExecutor:
         lifecycle_service: Any,
         tool_runner: ToolRunner | None = None,
         sandbox_root: str | None = None,
+        error_recovery_service: ErrorRecoveryService | None = None,
     ) -> None:
         """Initialize the plan executor.
 
@@ -354,6 +360,8 @@ class PlanExecutor:
             lifecycle_service: The ``PlanLifecycleService`` instance.
             tool_runner: Optional ``ToolRunner`` for execute phase.
             sandbox_root: Optional sandbox root directory.
+            error_recovery_service: Optional error recovery service for
+                recording errors and managing retry logic.
         """
         if lifecycle_service is None:
             raise ValidationError("lifecycle_service must not be None")
@@ -361,6 +369,7 @@ class PlanExecutor:
         self._lifecycle = lifecycle_service
         self._tool_runner = tool_runner
         self._sandbox_root = sandbox_root
+        self._error_recovery = error_recovery_service
         self._strategize_actor = StrategizeStubActor()
         self._execute_actor = ExecuteStubActor()
         self._logger = logger.bind(service="plan_executor")
@@ -413,7 +422,7 @@ class PlanExecutor:
             # Persist decision tree into plan metadata
             plan = self._lifecycle.get_plan(plan_id)
             plan.decision_root_id = result.decision_root_id
-            plan.timestamps.updated_at = datetime.now()
+            plan.timestamps.updated_at = datetime.now(tz=UTC)
 
             # Store decisions and invariant records in error_details as
             # structured metadata (Plan model uses error_details for
@@ -438,15 +447,25 @@ class PlanExecutor:
             return result
 
         except Exception as exc:
-            # Capture error details
+            # Record via error recovery service if available
+            if self._error_recovery is not None:
+                self._error_recovery.record_error(
+                    plan_id=plan_id,
+                    phase="strategize",
+                    message=str(exc),
+                    exception=exc,
+                    exception_type=type(exc).__name__,
+                )
+
+            # Capture error details via public API
             error_msg = f"{type(exc).__name__}: {exc}"
-            error_details = {
-                "exception_type": type(exc).__name__,
-                "traceback": traceback.format_exc(),
-            }
-            plan = self._lifecycle.get_plan(plan_id)
-            plan.error_details = error_details
-            self._lifecycle._commit_plan(plan)
+            self._lifecycle.update_error_details(
+                plan_id,
+                {
+                    "exception_type": type(exc).__name__,
+                    "traceback": traceback.format_exc(),
+                },
+            )
             self._lifecycle.fail_strategize(plan_id, error_msg)
             raise
 
@@ -511,47 +530,86 @@ class PlanExecutor:
         # Start execute
         self._lifecycle.start_execute(plan_id)
 
-        try:
-            # Run the execute stub actor
-            result = self._execute_actor.execute(
-                plan_id=plan_id,
-                decisions=decisions,
-                tool_runner=self._tool_runner,
-                sandbox_root=self._sandbox_root,
-                stream_callback=stream_callback,
-            )
+        # Determine max attempts: 1 (no recovery) or policy max_retries + 1.
+        max_attempts = (
+            self._error_recovery._policy.max_retries + 1
+            if self._error_recovery is not None
+            else 1
+        )
+        last_exc: Exception | None = None
 
-            # Persist execution metadata into plan
-            plan = self._lifecycle.get_plan(plan_id)
-            plan.changeset_id = result.changeset_id
-            plan.sandbox_refs = result.sandbox_refs
-            plan.error_details = {
-                "tool_calls_count": str(result.tool_calls_count),
-                "sandbox_refs_count": str(len(result.sandbox_refs)),
-            }
-            plan.timestamps.updated_at = datetime.now()
-            self._lifecycle._commit_plan(plan)
+        for attempt in range(max_attempts):
+            try:
+                # Run the execute stub actor
+                result = self._execute_actor.execute(
+                    plan_id=plan_id,
+                    decisions=decisions,
+                    tool_runner=self._tool_runner,
+                    sandbox_root=self._sandbox_root,
+                    stream_callback=stream_callback,
+                )
 
-            # Complete execute
-            self._lifecycle.complete_execute(plan_id)
+                # Persist execution metadata into plan
+                self._lifecycle.update_error_details(
+                    plan_id,
+                    {
+                        "tool_calls_count": str(result.tool_calls_count),
+                        "sandbox_refs_count": str(len(result.sandbox_refs)),
+                    },
+                )
+                plan = self._lifecycle.get_plan(plan_id)
+                plan.changeset_id = result.changeset_id
+                plan.sandbox_refs = result.sandbox_refs
+                plan.timestamps.updated_at = datetime.now(tz=UTC)
+                self._lifecycle._commit_plan(plan)
 
-            self._logger.info(
-                "Execute completed via executor",
-                plan_id=plan_id,
-                changeset_id=result.changeset_id,
-                tool_calls=result.tool_calls_count,
-            )
+                # Complete execute
+                self._lifecycle.complete_execute(plan_id)
 
-            return result
+                self._logger.info(
+                    "Execute completed via executor",
+                    plan_id=plan_id,
+                    changeset_id=result.changeset_id,
+                    tool_calls=result.tool_calls_count,
+                )
 
-        except Exception as exc:
-            error_msg = f"{type(exc).__name__}: {exc}"
-            error_details = {
-                "exception_type": type(exc).__name__,
+                return result
+
+            except Exception as exc:
+                last_exc = exc
+
+                # Record error via recovery service if available
+                if self._error_recovery is not None:
+                    self._error_recovery.record_error(
+                        plan_id=plan_id,
+                        phase="execute",
+                        message=str(exc),
+                        exception=exc,
+                        exception_type=type(exc).__name__,
+                    )
+                    # Check if retry is available
+                    if self._error_recovery.should_retry(plan_id):
+                        self._logger.info(
+                            "Retrying execute phase after recoverable error",
+                            plan_id=plan_id,
+                            attempt=attempt + 1,
+                            max_attempts=max_attempts,
+                            error=str(exc),
+                        )
+                        continue
+
+                # No recovery service or no more retries — fail
+                break
+
+        # All attempts exhausted — persist error and fail
+        assert last_exc is not None
+        error_msg = f"{type(last_exc).__name__}: {last_exc}"
+        self._lifecycle.update_error_details(
+            plan_id,
+            {
+                "exception_type": type(last_exc).__name__,
                 "traceback": traceback.format_exc(),
-            }
-            plan = self._lifecycle.get_plan(plan_id)
-            plan.error_details = error_details
-            self._lifecycle._commit_plan(plan)
-            self._lifecycle.fail_execute(plan_id, error_msg)
-            raise
+            },
+        )
+        self._lifecycle.fail_execute(plan_id, error_msg)
+        raise last_exc

src/cleveragents/application/services/plan_lifecycle_service.py

@@ -217,6 +217,29 @@ class PlanLifecycleService:
             with self.unit_of_work.transaction() as ctx:
                 self._persist_plan_update(plan, ctx)
 
+    def update_error_details(
+        self,
+        plan_id: str,
+        error_details: dict[str, str],
+    ) -> None:
+        """Merge *error_details* into the plan and persist.
+
+        This is the public API for external services (e.g.
+        ``ErrorRecoveryService``) to attach structured error metadata
+        to a plan without reaching into private persistence methods.
+
+        Args:
+            plan_id: The plan to update.
+            error_details: Key/value pairs to merge into the plan's
+                ``error_details`` dict.
+        """
+        plan = self.get_plan(plan_id)
+        existing = dict(plan.error_details) if plan.error_details else {}
+        existing.update(error_details)
+        plan.error_details = existing
+        plan.timestamps.updated_at = datetime.now()
+        self._commit_plan(plan)
+
     def _generate_ulid(self) -> str:
         """Generate a new ULID string."""
         return str(ULID())

src/cleveragents/cli/commands/plan.py

@@ -15,6 +15,7 @@ plan lifecycle.
 | ``agents plan cancel``        | Cancel a non-terminal plan              |
 | ``agents plan diff``          | Show ChangeSet as unified diff          |
 | ``agents plan artifacts``     | Show ChangeSet ID, sandbox refs, summary|
+| ``agents plan errors``        | Show error decisions + recovery hints   |
 
 ## Deprecated Legacy Commands
 
@@ -1596,6 +1597,129 @@ def plan_status(
         raise typer.Abort() from e
 
 
+@app.command("errors")
+def plan_errors(
+    plan_id: Annotated[
+        str,
+        typer.Argument(help="Plan ID to show errors for"),
+    ],
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help=_FORMAT_HELP,
+        ),
+    ] = "rich",
+) -> None:
+    """Show error decisions with recovery hints and retry history.
+
+    Displays structured error metadata from the plan, including error
+    category, phase, retry count, and recovery suggestions with CLI
+    commands.
+    """
+    from cleveragents.domain.models.core.error_recovery import (
+        ErrorCategory,
+        get_recovery_hints,
+    )
+
+    try:
+        service = _get_lifecycle_service()
+        plan = service.get_plan(plan_id)
+
+        details = plan.error_details or {}
+        has_error_recovery = "error_category" in details
+
+        # Build recovery hints from stored category
+        hints: list[dict[str, Any]] = []
+        if has_error_recovery:
+            with suppress(ValueError):
+                cat = ErrorCategory(details["error_category"])
+                for h in get_recovery_hints(cat, plan_id):
+                    entry: dict[str, Any] = {
+                        "action": h.action.value,
+                        "message": h.message,
+                    }
+                    if h.cli_command:
+                        entry["cli_command"] = h.cli_command
+                    hints.append(entry)
+
+        if fmt != OutputFormat.RICH.value:
+            data: dict[str, Any] = {
+                "plan_id": plan_id,
+                "phase": plan.phase.value,
+                "state": plan.state.value if plan.state else "unknown",
+            }
+            if plan.error_message:
+                data["error_message"] = plan.error_message
+            if has_error_recovery:
+                data["error_category"] = details.get("error_category")
+                data["error_phase"] = details.get("error_phase")
+                data["retry_count"] = details.get("retry_count")
+                data["max_retries"] = details.get("max_retries")
+                data["is_retriable"] = details.get("is_retriable")
+            if details.get("error_actor"):
+                data["error_actor"] = details["error_actor"]
+            if details.get("error_tool_call"):
+                data["error_tool_call"] = details["error_tool_call"]
+            if details.get("stack_summary"):
+                data["stack_summary"] = details["stack_summary"]
+            if hints:
+                data["recovery_hints"] = hints
+            console.print(format_output(data, fmt))
+            return
+
+        # Rich output
+        header = (
+            f"[bold]Plan:[/bold] {plan_id}\n"
+            f"[bold]Phase:[/bold] {plan.phase.value}\n"
+            f"[bold]State:[/bold] "
+            f"{plan.state.value if plan.state else 'unknown'}"
+        )
+
+        if plan.error_message:
+            header += f"\n[bold red]Error:[/bold red] {plan.error_message}"
+
+        if not has_error_recovery and not plan.error_message:
+            header += "\n\n[green]No error recovery records for this plan.[/green]"
+            console.print(Panel(header, title="Plan Errors", expand=False))
+            return
+
+        if has_error_recovery:
+            header += (
+                f"\n\n[bold]Error Category:[/bold] "
+                f"{details.get('error_category', 'unknown')}\n"
+                f"[bold]Error Phase:[/bold] "
+                f"{details.get('error_phase', 'unknown')}\n"
+                f"[bold]Retry Count:[/bold] "
+                f"{details.get('retry_count', '0')}"
+                f"/{details.get('max_retries', '3')}\n"
+                f"[bold]Retriable:[/bold] "
+                f"{details.get('is_retriable', 'false')}"
+            )
+            if details.get("error_actor"):
+                header += f"\n[bold]Actor:[/bold] {details['error_actor']}"
+            if details.get("error_tool_call"):
+                header += f"\n[bold]Tool Call:[/bold] {details['error_tool_call']}"
+
+        if hints:
+            header += "\n\n[bold]Recovery Suggestions:[/bold]"
+            for hint in hints:
+                header += f"\n  → {hint['message']}"
+                if hint.get("cli_command"):
+                    header += f"\n    [dim]$ {hint['cli_command']}[/dim]"
+
+        if details.get("stack_summary"):
+            summary = details["stack_summary"][:500]
+            header += f"\n\n[bold]Stack Summary:[/bold]\n[dim]{summary}[/dim]"
+
+        console.print(Panel(header, title="Plan Errors", expand=False))
+
+    except CleverAgentsError as e:
+        console.print(f"[red]Error:[/red] {e.message}")
+        raise typer.Abort() from e
+
+
 @app.command("lifecycle-list")
 def lifecycle_list_plans(
     regex: Annotated[

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

@@ -38,6 +38,18 @@ from cleveragents.domain.models.core.context_policy import (
     ProjectContextPolicy,
 )
 from cleveragents.domain.models.core.debug_attempt import DebugAttempt
+
+# Error recovery models
+from cleveragents.domain.models.core.error_recovery import (
+    ErrorCategory,
+    ErrorHistory,
+    ErrorRecord,
+    ErrorRecoveryPolicy,
+    RecoveryAction,
+    RecoveryHint,
+    classify_error,
+    get_recovery_hints,
+)
 from cleveragents.domain.models.core.org import (
     CloudBillingFields,
     CreditsTransaction,
@@ -170,6 +182,10 @@ __all__ = [
     "CreditsTransaction",
     "CreditsTransactionType",
     "DebugAttempt",
+    "ErrorCategory",
+    "ErrorHistory",
+    "ErrorRecord",
+    "ErrorRecoveryPolicy",
     "InMemoryChangeSetStore",
     "InMemoryInvocationTracker",
     "InvariantSource",
@@ -203,6 +219,8 @@ __all__ = [
     "ProjectLink",
     "ProjectSettings",
     "ProjectStats",
+    "RecoveryAction",
+    "RecoveryHint",
     "ResolvedToolEntry",
     "Resource",
     "ResourceAccessMode",
@@ -242,7 +260,9 @@ __all__ = [
     "Validation",
     "ValidationMode",
     "can_transition",
+    "classify_error",
     "get_builtin_profile",
+    "get_recovery_hints",
     "normalize_change_path",
     "parse_namespaced_name",
 ]

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

@@ -0,0 +1,640 @@
+"""Error recovery domain models.
+
+Provides structured error classification, recovery hints, and retry
+policy logic for plan execution failures.
+
+Core types:
+
+- **ErrorCategory**: Classifies errors as transient, validation,
+  configuration, etc.
+- **RecoveryAction**: Suggested recovery actions (retry, revert, abort).
+- **RecoveryHint**: A structured suggestion for recovering from an error,
+  including a human-readable message and a CLI command to execute.
+- **ErrorRecord**: A single recorded error with metadata (phase, category,
+  message, stack summary, recovery hints, timestamp).
+- **ErrorHistory**: An ordered collection of ErrorRecords for a plan.
+- **ErrorRecoveryPolicy**: Determines whether retry is appropriate given
+  the error category and automation profile thresholds.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+# ---------------------------------------------------------------------------
+# Error classification
+# ---------------------------------------------------------------------------
+
+
+class ErrorCategory(StrEnum):
+    """Classification of plan execution errors.
+
+    Used to determine whether an error is retriable, what recovery
+    hints to suggest, and how to display the error in CLI output.
+    """
+
+    TRANSIENT = "transient"
+    """Temporary failure (network, rate-limit, timeout)."""
+
+    VALIDATION = "validation"
+    """Validation failure during Execute or Apply phase."""
+
+    CONFIGURATION = "configuration"
+    """Bad config, missing required fields, invalid arguments."""
+
+    RESOURCE = "resource"
+    """Missing or conflicting resource (file, sandbox, etc.)."""
+
+    MERGE_CONFLICT = "merge_conflict"
+    """Sandbox merge failed with conflicts."""
+
+    AUTHENTICATION = "authentication"
+    """Auth/authz failure accessing provider or resource."""
+
+    PROVIDER = "provider"
+    """LLM provider error (model unavailable, token limit, etc.)."""
+
+    INTERNAL = "internal"
+    """Unexpected internal error (bug, assertion, etc.)."""
+
+    UNKNOWN = "unknown"
+    """Error could not be classified."""
+
+
+class RecoveryAction(StrEnum):
+    """Suggested recovery action for a plan error.
+
+    Each action maps to a concrete CLI command or user workflow.
+    """
+
+    RETRY = "retry"
+    """Retry the failed phase (transient / provider errors)."""
+
+    REVERT_STRATEGIZE = "revert_strategize"
+    """Revert to Strategize phase and re-plan."""
+
+    CORRECT = "correct"
+    """Provide corrections and re-run validation."""
+
+    PROMPT = "prompt"
+    """Resume with additional user guidance."""
+
+    CANCEL = "cancel"
+    """Cancel the plan (non-recoverable error)."""
+
+    MANUAL = "manual"
+    """Requires manual investigation (internal errors)."""
+
+
+# ---------------------------------------------------------------------------
+# Recovery hint
+# ---------------------------------------------------------------------------
+
+
+class RecoveryHint(BaseModel):
+    """A structured suggestion for recovering from an error.
+
+    Each hint includes a human-readable message explaining what to do
+    and an optional CLI command that can be copy-pasted.
+    """
+
+    action: RecoveryAction = Field(..., description="The suggested recovery action")
+    message: str = Field(..., description="Human-readable recovery suggestion")
+    cli_command: str | None = Field(
+        default=None,
+        description="CLI command to execute for recovery (copy-pasteable)",
+    )
+    priority: int = Field(
+        default=0,
+        ge=0,
+        description="Priority of this hint (0 = highest priority)",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Error record
+# ---------------------------------------------------------------------------
+
+
+class ErrorRecord(BaseModel):
+    """A single recorded error event for a plan.
+
+    Captures structured metadata about what went wrong, in which phase,
+    what actor/tool was involved, and what recovery options are available.
+    """
+
+    error_id: str = Field(..., description="Unique identifier for this error event")
+    phase: str = Field(
+        ..., description="Plan phase when error occurred (strategize/execute/apply)"
+    )
+    category: ErrorCategory = Field(..., description="Error classification")
+    message: str = Field(..., description="Error message")
+    actor: str | None = Field(
+        default=None, description="Actor name involved in the error"
+    )
+    tool_call: str | None = Field(
+        default=None, description="Tool call that triggered the error"
+    )
+    stack_summary: str | None = Field(
+        default=None,
+        description="Abbreviated stack trace (last 3 frames)",
+    )
+    recovery_hints: list[RecoveryHint] = Field(
+        default_factory=list,
+        description="Ordered recovery suggestions",
+    )
+    retry_count: int = Field(
+        default=0,
+        ge=0,
+        description="Number of retry attempts already made for this error",
+    )
+    max_retries: int = Field(
+        default=3,
+        ge=0,
+        description="Maximum retry attempts allowed",
+    )
+    timestamp: datetime = Field(
+        default_factory=lambda: datetime.now(tz=UTC),
+        description="When the error was recorded (UTC)",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+    @property
+    def is_retriable(self) -> bool:
+        """Whether this error can be retried.
+
+        An error is retriable if:
+        - Its category is in the retriable set
+        - The retry count has not exceeded max_retries
+        """
+        retriable_categories = {
+            ErrorCategory.TRANSIENT,
+            ErrorCategory.VALIDATION,
+            ErrorCategory.MERGE_CONFLICT,
+            ErrorCategory.PROVIDER,
+        }
+        return (
+            self.category in retriable_categories
+            and self.retry_count < self.max_retries
+        )
+
+    @property
+    def retries_exhausted(self) -> bool:
+        """Whether all retry attempts have been used."""
+        return self.retry_count >= self.max_retries
+
+    def to_cli_dict(self) -> dict[str, Any]:
+        """Convert to a dictionary suitable for CLI display."""
+        result: dict[str, Any] = {
+            "error_id": self.error_id,
+            "phase": self.phase,
+            "category": self.category.value,
+            "message": self.message,
+            "retry_count": self.retry_count,
+            "max_retries": self.max_retries,
+            "is_retriable": self.is_retriable,
+            "timestamp": self.timestamp.isoformat(),
+        }
+        if self.actor:
+            result["actor"] = self.actor
+        if self.tool_call:
+            result["tool_call"] = self.tool_call
+        if self.stack_summary:
+            result["stack_summary"] = self.stack_summary
+        if self.recovery_hints:
+            result["recovery_hints"] = [
+                {
+                    "action": h.action.value,
+                    "message": h.message,
+                    "cli_command": h.cli_command,
+                }
+                for h in sorted(self.recovery_hints, key=lambda h: h.priority)
+            ]
+        return result
+
+
+# ---------------------------------------------------------------------------
+# Error history
+# ---------------------------------------------------------------------------
+
+
+class ErrorHistory(BaseModel):
+    """Ordered collection of error records for a plan.
+
+    Provides convenience methods for querying error state and
+    formatting for CLI output.
+    """
+
+    plan_id: str = Field(..., description="Plan ULID")
+    records: list[ErrorRecord] = Field(
+        default_factory=list,
+        description="Error records in chronological order",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+    def add(self, record: ErrorRecord) -> None:
+        """Append an error record."""
+        self.records.append(record)
+
+    @property
+    def latest(self) -> ErrorRecord | None:
+        """Return the most recent error, or None."""
+        return self.records[-1] if self.records else None
+
+    @property
+    def total_errors(self) -> int:
+        """Total number of recorded errors."""
+        return len(self.records)
+
+    @property
+    def total_retries(self) -> int:
+        """Sum of all retry counts across records."""
+        return sum(r.retry_count for r in self.records)
+
+    @property
+    def has_retriable(self) -> bool:
+        """Whether any error in the history is still retriable."""
+        return any(r.is_retriable for r in self.records)
+
+    def errors_by_category(self) -> dict[str, int]:
+        """Count errors by category."""
+        counts: dict[str, int] = {}
+        for record in self.records:
+            key = record.category.value
+            counts[key] = counts.get(key, 0) + 1
+        return counts
+
+    def to_cli_dict(self) -> dict[str, Any]:
+        """Convert to a dictionary suitable for CLI display."""
+        return {
+            "plan_id": self.plan_id,
+            "total_errors": self.total_errors,
+            "total_retries": self.total_retries,
+            "has_retriable": self.has_retriable,
+            "errors_by_category": self.errors_by_category(),
+            "records": [r.to_cli_dict() for r in self.records],
+        }
+
+
+# ---------------------------------------------------------------------------
+# Recovery hint generation
+# ---------------------------------------------------------------------------
+
+# Maps error categories to default recovery hints
+_DEFAULT_HINTS: dict[ErrorCategory, list[RecoveryHint]] = {
+    ErrorCategory.TRANSIENT: [
+        RecoveryHint(
+            action=RecoveryAction.RETRY,
+            message="Retry the operation — the failure was transient.",
+            cli_command="agents plan prompt {plan_id}",
+            priority=0,
+        ),
+    ],
+    ErrorCategory.VALIDATION: [
+        RecoveryHint(
+            action=RecoveryAction.CORRECT,
+            message="Fix validation issues and re-run.",
+            cli_command="agents plan correct {plan_id}",
+            priority=0,
+        ),
+        RecoveryHint(
+            action=RecoveryAction.PROMPT,
+            message="Provide additional guidance to help resolve the issue.",
+            cli_command="agents plan prompt {plan_id}",
+            priority=1,
+        ),
+    ],
+    ErrorCategory.CONFIGURATION: [
+        RecoveryHint(
+            action=RecoveryAction.CANCEL,
+            message="Fix the configuration and re-create the plan.",
+            priority=0,
+        ),
+    ],
+    ErrorCategory.RESOURCE: [
+        RecoveryHint(
+            action=RecoveryAction.REVERT_STRATEGIZE,
+            message=("Revert to Strategize to re-plan with updated resources."),
+            cli_command=("agents plan revert {plan_id} --to-phase strategize"),
+            priority=0,
+        ),
+    ],
+    ErrorCategory.MERGE_CONFLICT: [
+        RecoveryHint(
+            action=RecoveryAction.REVERT_STRATEGIZE,
+            message="Resolve merge conflicts and re-run from Strategize.",
+            cli_command=("agents plan revert {plan_id} --to-phase strategize"),
+            priority=0,
+        ),
+        RecoveryHint(
+            action=RecoveryAction.RETRY,
+            message="Retry apply after resolving conflicts manually.",
+            cli_command="agents plan prompt {plan_id}",
+            priority=1,
+        ),
+    ],
+    ErrorCategory.AUTHENTICATION: [
+        RecoveryHint(
+            action=RecoveryAction.CANCEL,
+            message=(
+                "Check authentication credentials and retry. "
+                "This error is not auto-retriable."
+            ),
+            priority=0,
+        ),
+    ],
+    ErrorCategory.PROVIDER: [
+        RecoveryHint(
+            action=RecoveryAction.RETRY,
+            message="Retry — the LLM provider may be temporarily unavailable.",
+            cli_command="agents plan prompt {plan_id}",
+            priority=0,
+        ),
+    ],
+    ErrorCategory.INTERNAL: [
+        RecoveryHint(
+            action=RecoveryAction.MANUAL,
+            message=(
+                "An internal error occurred. Check logs and report "
+                "the issue if it persists."
+            ),
+            priority=0,
+        ),
+        RecoveryHint(
+            action=RecoveryAction.CANCEL,
+            message="Cancel the plan and start fresh if the error persists.",
+            cli_command="agents plan cancel {plan_id}",
+            priority=1,
+        ),
+    ],
+    ErrorCategory.UNKNOWN: [
+        RecoveryHint(
+            action=RecoveryAction.MANUAL,
+            message="Unknown error — investigate logs for details.",
+            priority=0,
+        ),
+    ],
+}
+
+
+def get_recovery_hints(
+    category: ErrorCategory,
+    plan_id: str,
+) -> list[RecoveryHint]:
+    """Return recovery hints for an error category.
+
+    Substitutes ``{plan_id}`` placeholders in CLI commands with
+    the actual plan ID.
+
+    Args:
+        category: The error classification.
+        plan_id: The plan ULID to substitute into CLI commands.
+
+    Returns:
+        List of ``RecoveryHint`` objects ordered by priority.
+    """
+    templates = _DEFAULT_HINTS.get(category, _DEFAULT_HINTS[ErrorCategory.UNKNOWN])
+    hints: list[RecoveryHint] = []
+    for tmpl in templates:
+        cli_cmd = (
+            tmpl.cli_command.replace("{plan_id}", plan_id) if tmpl.cli_command else None
+        )
+        hints.append(
+            RecoveryHint(
+                action=tmpl.action,
+                message=tmpl.message,
+                cli_command=cli_cmd,
+                priority=tmpl.priority,
+            )
+        )
+    return hints
+
+
+# ---------------------------------------------------------------------------
+# Error classification helper
+# ---------------------------------------------------------------------------
+
+# Compiled regex patterns for classifying error messages.
+# Ordered list of (category, pattern) — checked in order, first match wins.
+# More specific categories (authentication, merge_conflict) must appear
+# before broader ones (validation, resource) to avoid false matches.
+_CATEGORY_PATTERNS: list[tuple[ErrorCategory, re.Pattern[str]]] = [
+    (
+        ErrorCategory.AUTHENTICATION,
+        re.compile(
+            r"authentication|unauthorized|forbidden|permission denied",
+        ),
+    ),
+    (
+        ErrorCategory.MERGE_CONFLICT,
+        re.compile(r"merge conflict|merge failed"),
+    ),
+    (
+        ErrorCategory.TRANSIENT,
+        re.compile(
+            r"timeout|timed out|temporary|connection reset"
+            r"|connection refused|rate limit|throttl",
+        ),
+    ),
+    (
+        ErrorCategory.PROVIDER,
+        re.compile(
+            r"model not available|token limit|context length|provider error",
+        ),
+    ),
+    (
+        ErrorCategory.CONFIGURATION,
+        re.compile(
+            r"configuration|config|missing setting|not configured",
+        ),
+    ),
+    (
+        ErrorCategory.VALIDATION,
+        re.compile(
+            r"validation|invalid|constraint|does not match"
+            r"|required field|schema",
+        ),
+    ),
+    (
+        ErrorCategory.RESOURCE,
+        re.compile(r"not found|missing resource|no such file"),
+    ),
+    (
+        ErrorCategory.INTERNAL,
+        re.compile(r"internal error|assertion|unexpected|bug"),
+    ),
+]
+
+# Exception-type → category mapping (checked before keyword patterns).
+_EXCEPTION_TYPE_MAP: dict[str, ErrorCategory] = {
+    "ratelimiterror": ErrorCategory.TRANSIENT,
+    "networkerror": ErrorCategory.TRANSIENT,
+    "timeouterror": ErrorCategory.TRANSIENT,
+    "validationerror": ErrorCategory.VALIDATION,
+    "configurationerror": ErrorCategory.CONFIGURATION,
+    "missingconfigurationerror": ErrorCategory.CONFIGURATION,
+    "resourcenotfounderror": ErrorCategory.RESOURCE,
+    "resourceconflicterror": ErrorCategory.RESOURCE,
+    "authenticationerror": ErrorCategory.AUTHENTICATION,
+    "authorizationerror": ErrorCategory.AUTHENTICATION,
+    "providererror": ErrorCategory.PROVIDER,
+    "modelnotavailableerror": ErrorCategory.PROVIDER,
+    "tokenlimitexceedederror": ErrorCategory.PROVIDER,
+}
+
+
+def classify_error(
+    message: str,
+    exception_type: str | None = None,
+) -> ErrorCategory:
+    """Classify an error message into an ErrorCategory.
+
+    Uses a compiled regex per category for keyword matching on the
+    message text. If ``exception_type`` is provided, it takes
+    precedence.
+
+    Args:
+        message: The error message to classify.
+        exception_type: Optional exception class name.
+
+    Returns:
+        The best-matching ``ErrorCategory``.
+    """
+    lower_exc = (exception_type or "").lower()
+
+    # Check exception type first (more specific)
+    if lower_exc in _EXCEPTION_TYPE_MAP:
+        return _EXCEPTION_TYPE_MAP[lower_exc]
+
+    # Single compiled-regex search per category (first match wins)
+    lower_msg = message.lower()
+    for category, pattern in _CATEGORY_PATTERNS:
+        if pattern.search(lower_msg):
+            return category
+
+    return ErrorCategory.UNKNOWN
+
+
+# ---------------------------------------------------------------------------
+# Error recovery policy
+# ---------------------------------------------------------------------------
+
+
+class ErrorRecoveryPolicy:
+    """Determines whether retry is appropriate for a given error.
+
+    Uses the automation profile thresholds to decide between automatic
+    retry (threshold 0.0) and human intervention (threshold 1.0).
+
+    Attributes:
+        auto_retry_threshold: Threshold from automation profile
+            (0.0 = always auto-retry, 1.0 = always ask human).
+        max_retries: Maximum retry attempts.
+    """
+
+    def __init__(
+        self,
+        auto_retry_threshold: float = 0.0,
+        max_retries: int = 3,
+    ) -> None:
+        self.auto_retry_threshold = auto_retry_threshold
+        self.max_retries = max_retries
+
+    def should_retry(self, record: ErrorRecord) -> bool:
+        """Determine whether the error should be retried.
+
+        Returns ``True`` if:
+        - The error is in a retriable category
+        - Retries have not been exhausted
+        - The automation profile allows auto-retry (threshold < 1.0)
+
+        Args:
+            record: The error record to evaluate.
+
+        Returns:
+            ``True`` if retry should proceed automatically.
+        """
+        if not record.is_retriable:
+            return False
+        if record.retries_exhausted:
+            return False
+        # Threshold < 1.0 means auto-retry is enabled
+        return self.auto_retry_threshold < 1.0
+
+    def should_escalate(self, record: ErrorRecord) -> bool:
+        """Determine whether the error requires human intervention.
+
+        Returns ``True`` if:
+        - The error is not retriable, OR
+        - Retries are exhausted, OR
+        - The automation profile requires human decision
+
+        Args:
+            record: The error record to evaluate.
+
+        Returns:
+            ``True`` if the error should be escalated to a human.
+        """
+        if not record.is_retriable:
+            return True
+        if record.retries_exhausted:
+            return True
+        return self.auto_retry_threshold >= 1.0
+
+    def format_recovery_output(
+        self,
+        record: ErrorRecord,
+    ) -> str:
+        """Format a human-readable recovery output for CLI display.
+
+        Includes error details, recovery hints with CLI commands,
+        and retry status.
+
+        Args:
+            record: The error record to format.
+
+        Returns:
+            Formatted string for CLI output.
+        """
+        lines: list[str] = []
+        lines.append(f"Error [{record.category.value}]: {record.message}")
+        lines.append(f"  Phase: {record.phase}")
+        if record.actor:
+            lines.append(f"  Actor: {record.actor}")
+        if record.tool_call:
+            lines.append(f"  Tool:  {record.tool_call}")
+        lines.append(f"  Retry: {record.retry_count}/{record.max_retries}")
+
+        if record.recovery_hints:
+            lines.append("  Recovery suggestions:")
+            for hint in sorted(record.recovery_hints, key=lambda h: h.priority):
+                lines.append(f"    → {hint.message}")
+                if hint.cli_command:
+                    lines.append(f"      $ {hint.cli_command}")
+
+        if record.retries_exhausted:
+            lines.append("  ⚠ All retry attempts exhausted.")
+        elif self.should_retry(record):
+            lines.append("  ✓ Automatic retry available.")
+        else:
+            lines.append("  ⚠ Human intervention required.")
+
+        return "\n".join(lines)