feat(checkpoint): add checkpointing and rollback

fix Alembic migration down_revision to chain after
  m4_002_skill_flattened_tools (was pointing to stale
  c3_001_actor_registry)

BDD coverage: 33 scenarios (7 new), 129 steps, all passing.

ISSUES CLOSED: #206

CHANGELOG.md

@@ -29,6 +29,20 @@
   slipcover --merge combines them. CI workflow JSON key lookups handle both slipcover and
   coverage.py output formats. Documentation updated to reflect slipcover as the coverage
   tool. (#482)
+- Added checkpointing and rollback with `CheckpointService` for creating, listing, pruning,
+  and deleting sandbox snapshots, and restoring sandbox state via `plan rollback <plan_id>
+  <checkpoint_id>` CLI command. Checkpoint domain models (`Checkpoint`, `CheckpointMetadata`,
+  `CheckpointRetentionPolicy`, `RollbackResult`) store sandbox refs, decision alignment,
+  checkpoint type (`pre_write`, `post_step`, `manual`), filesystem path, size, and structured
+  audit metadata (reason, source tool, phase). Retention policy auto-prunes oldest interior
+  checkpoints when exceeding `max_checkpoints` (default 50), preserving the first and most
+  recent. Guards reject rollback when plan is applied or sandbox is missing. `CorrectionService`
+  accepts an optional `CheckpointService` integration point for future revert delegation.
+  `CheckpointRepository` and `CheckpointModel` back persistence via the session-factory pattern
+  (ADR-007), with `UnitOfWorkContext.checkpoints` for cross-repository atomicity. Alembic
+  migration `m6_001_checkpoint_metadata` adds the `checkpoint_metadata` table. Includes Behave
+  BDD scenarios (33 scenarios, 129 steps), Robot Framework integration tests (11 test cases),
+  ASV benchmarks, and `docs/reference/checkpointing.md`. (#206)
 - Added semantic validation service with AST-based rules for syntax errors, missing imports,
   broken references, duplicate imports, API misuse, and missing symbols. Includes rule registry,
   file-hash LRU cache, severity mapping, and ValidationPipeline integration. (#448)

alembic/versions/m6_001_checkpoint_metadata_table.py

@@ -0,0 +1,68 @@
+"""Create checkpoint_metadata table.
+
+Adds the ``checkpoint_metadata`` table for Stage M6 (checkpointing and
+rollback).  The table stores sandbox snapshot metadata aligned to the
+specification schema: ULID primary key, owning plan reference, optional
+decision and resource references, checkpoint type, filesystem path,
+size, creation timestamp, and a JSON audit metadata blob.
+
+Revision ID: m6_001_checkpoint_metadata
+Revises: m4_002_skill_flattened_tools
+Create Date: 2026-02-26 00:00:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "m6_001_checkpoint_metadata"
+down_revision: str | Sequence[str] | None = "m4_002_skill_flattened_tools"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create checkpoint_metadata table."""
+
+    op.create_table(
+        "checkpoint_metadata",
+        sa.Column("checkpoint_id", sa.String(26), primary_key=True),
+        sa.Column(
+            "plan_id",
+            sa.String(26),
+            sa.ForeignKey("v3_plans.plan_id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("decision_id", sa.String(26), nullable=True),
+        sa.Column(
+            "checkpoint_type", sa.Text(), nullable=False, server_default="manual"
+        ),
+        sa.Column("resource_id", sa.String(26), nullable=True),
+        sa.Column("sandbox_ref", sa.Text(), nullable=False),
+        sa.Column("filesystem_path", sa.Text(), nullable=False, server_default=""),
+        sa.Column("size_bytes", sa.Integer(), nullable=True),
+        sa.Column("created_at", sa.String(30), nullable=False),
+        sa.Column("metadata_json", sa.Text(), nullable=True),
+    )
+
+    op.create_index(
+        "idx_checkpoints_plan",
+        "checkpoint_metadata",
+        ["plan_id"],
+    )
+    op.create_index(
+        "ix_checkpoint_metadata_created_at",
+        "checkpoint_metadata",
+        ["created_at"],
+    )
+
+
+def downgrade() -> None:
+    """Drop checkpoint_metadata table."""
+
+    op.drop_index("ix_checkpoint_metadata_created_at", table_name="checkpoint_metadata")
+    op.drop_index("idx_checkpoints_plan", table_name="checkpoint_metadata")
+    op.drop_table("checkpoint_metadata")

benchmarks/checkpoint_rollback_bench.py

@@ -0,0 +1,128 @@
+"""Airspeed Velocity benchmarks for checkpoint and rollback operations.
+
+Measures checkpoint creation, listing, rollback, pruning, and metadata
+overhead across varying checkpoint counts.
+"""
+
+from __future__ import annotations
+
+from cleveragents.application.services.checkpoint_service import CheckpointService
+from cleveragents.domain.models.core.checkpoint import CheckpointRetentionPolicy
+
+_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+
+# ---------------------------------------------------------------------------
+# Checkpoint creation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class CheckpointCreateSuite:
+    """Benchmark checkpoint creation at varying counts."""
+
+    def setup(self) -> None:
+        """Prepare a fresh service instance."""
+        self.svc = CheckpointService()
+
+    def time_create_single(self) -> None:
+        """Time creating a single checkpoint."""
+        self.svc.create_checkpoint(_PLAN_ID, "commit-abc")
+
+    def time_create_ten(self) -> None:
+        """Time creating 10 checkpoints."""
+        svc = CheckpointService()
+        for i in range(10):
+            svc.create_checkpoint(_PLAN_ID, f"commit-{i}")
+
+    def time_create_hundred(self) -> None:
+        """Time creating 100 checkpoints."""
+        svc = CheckpointService()
+        for i in range(100):
+            svc.create_checkpoint(_PLAN_ID, f"commit-{i}")
+
+
+# ---------------------------------------------------------------------------
+# Checkpoint listing benchmarks
+# ---------------------------------------------------------------------------
+
+
+class CheckpointListSuite:
+    """Benchmark listing checkpoints at varying plan sizes."""
+
+    params: list[int] = [10, 50, 100]
+    param_names: list[str] = ["checkpoint_count"]
+
+    def setup(self, checkpoint_count: int) -> None:
+        """Populate service with checkpoints."""
+        self.svc = CheckpointService()
+        for i in range(checkpoint_count):
+            self.svc.create_checkpoint(_PLAN_ID, f"commit-{i}")
+
+    def time_list(self, checkpoint_count: int) -> None:
+        """Time listing all checkpoints for a plan."""
+        self.svc.list_checkpoints(_PLAN_ID)
+
+
+# ---------------------------------------------------------------------------
+# Rollback benchmarks
+# ---------------------------------------------------------------------------
+
+
+class RollbackSuite:
+    """Benchmark rollback operations."""
+
+    def setup(self) -> None:
+        """Prepare a service with sandbox and checkpoints."""
+        self.svc = CheckpointService()
+        self.svc.register_sandbox(_PLAN_ID, "sandbox-path")
+        self.cp = self.svc.create_checkpoint(_PLAN_ID, "commit-abc")
+
+    def time_rollback(self) -> None:
+        """Time a single rollback operation."""
+        self.svc.rollback_to_checkpoint(_PLAN_ID, self.cp.checkpoint_id)
+
+
+# ---------------------------------------------------------------------------
+# Prune benchmarks
+# ---------------------------------------------------------------------------
+
+
+class PruneSuite:
+    """Benchmark checkpoint pruning at varying counts."""
+
+    params: list[int] = [20, 50, 100]
+    param_names: list[str] = ["checkpoint_count"]
+
+    def setup(self, checkpoint_count: int) -> None:
+        """Populate service with checkpoints."""
+        self.svc = CheckpointService()
+        for i in range(checkpoint_count):
+            self.svc.create_checkpoint(_PLAN_ID, f"commit-{i}")
+
+    def time_prune_to_five(self, checkpoint_count: int) -> None:
+        """Time pruning down to 5 checkpoints."""
+        policy = CheckpointRetentionPolicy(max_checkpoints=5, auto_prune=True)
+        self.svc.prune_checkpoints(_PLAN_ID, policy)
+
+
+# ---------------------------------------------------------------------------
+# Metadata benchmarks
+# ---------------------------------------------------------------------------
+
+
+class MetadataSuite:
+    """Benchmark checkpoint creation with metadata."""
+
+    def setup(self) -> None:
+        """Prepare service."""
+        self.svc = CheckpointService()
+
+    def time_create_with_metadata(self) -> None:
+        """Time creating a checkpoint with full metadata."""
+        self.svc.create_checkpoint(
+            _PLAN_ID,
+            "commit-xyz",
+            reason="pre-execution snapshot",
+            source_tool="lint-check",
+            phase="execute",
+        )

docs/reference/checkpointing.md

@@ -0,0 +1,158 @@
+# Checkpointing and Rollback
+
+## Overview
+
+CleverAgents supports checkpointing and rollback to allow operators to
+snapshot sandbox state during plan execution and restore it later. This
+is useful for recovering from mistakes, reverting tool side-effects, and
+supporting the decision-correction revert flow.
+
+## Concepts
+
+### Checkpoint
+
+A **Checkpoint** is an immutable record of sandbox state at a point in
+time. Each checkpoint stores:
+
+- `checkpoint_id` — Unique ULID identifier
+- `plan_id` — The plan that owns this checkpoint
+- `sandbox_ref` — A reference to the sandbox state (e.g. git commit hash)
+- `decision_id` — Optional decision ULID this checkpoint is aligned to
+- `checkpoint_type` — One of `pre_write`, `post_step`, or `manual`
+- `resource_id` — Optional resource ULID association
+- `filesystem_path` — Relative path within the checkpoint directory
+- `size_bytes` — Size of the checkpoint data in bytes
+- `created_at` — Timestamp of creation (UTC)
+- `metadata` — Audit metadata including reason, source tool, and phase
+
+### Retention Policy
+
+A **CheckpointRetentionPolicy** governs how many checkpoints a plan may
+keep. When `auto_prune` is enabled and the number of checkpoints exceeds
+`max_checkpoints`, the oldest **interior** checkpoints are automatically
+removed when a new one is created. The first (earliest) and most recent
+checkpoints are always preserved during pruning.
+
+**Defaults:**
+
+| Field             | Default | Range   |
+|-------------------|---------|---------|
+| `max_checkpoints` | 50      | 1–100   |
+| `auto_prune`      | true    |         |
+
+### Rollback
+
+A **rollback** restores sandbox state to a previously captured checkpoint.
+The `RollbackResult` describes what changed:
+
+- `restored_files_count` — Number of files restored
+- `changed_paths` — Paths that were modified during rollback
+- `from_checkpoint_id` — The checkpoint that was restored
+
+## Guards
+
+The rollback operation enforces two guards:
+
+1. **Plan is applied** — Once a plan reaches the `applied` terminal state,
+   rollback is rejected with a `BusinessRuleViolation`.
+2. **Sandbox is missing** — If the sandbox has been cleaned up, rollback
+   is rejected because there is nothing to restore.
+
+## CLI Usage
+
+### Create a checkpoint (programmatic)
+
+Checkpoints are typically created automatically by the execution engine
+before and after tool invocations, controlled by the `CheckpointScope`
+on each tool's capability declaration.
+
+### Rollback to a checkpoint
+
+```bash
+agents plan rollback [--yes|-y] <plan_id> <checkpoint_id>
+```
+
+The `--yes` / `-y` flag skips the interactive confirmation prompt.
+
+**JSON output envelope** (when using `--format json`):
+
+```json
+{
+  "rollback_summary": {
+    "plan_id": "...",
+    "from_checkpoint_id": "...",
+    "restored_files_count": 3
+  },
+  "changes_reverted": ["path/to/file1.py", "path/to/file2.py"],
+  "impact": {
+    "files_affected": 3
+  },
+  "post_rollback_state": {
+    "active_checkpoint": "...",
+    "plan_id": "..."
+  },
+  "timing": {
+    "elapsed_seconds": 0.042
+  },
+  "messages": ["Rollback completed successfully."]
+}
+```
+
+### Error Cases
+
+| Error | Cause |
+|-------|-------|
+| `Rollback blocked: plan is already applied` | Plan has reached terminal applied state |
+| `Rollback blocked: sandbox is missing` | Sandbox was cleaned up before rollback |
+| `Not found: checkpoint` | Checkpoint ID does not exist |
+| `Validation Error` | Checkpoint does not belong to the specified plan |
+
+## Correction Service Integration
+
+The `CorrectionService` accepts an optional `checkpoint_service` parameter
+(typed as `CheckpointService | None`).  This is an **integration point**
+for a future milestone: once wired, the correction revert flow will
+delegate sandbox restoration to the checkpoint rollback mechanism,
+enabling reuse of the same rollback logic for both explicit CLI rollback
+and decision-correction reverts.  The delegation is **not yet active** in
+the current implementation; `execute_revert()` performs BFS impact
+analysis and marks decisions as reverted but does not invoke
+`checkpoint_service.rollback_to_checkpoint()`.
+
+## Database Schema
+
+The `checkpoint_metadata` table stores checkpoint metadata:
+
+| Column            | Type        | Description                         |
+|-------------------|-------------|-------------------------------------|
+| `checkpoint_id`   | STRING(26) PK | ULID primary key                  |
+| `plan_id`         | STRING(26) FK | References `v3_plans.plan_id`     |
+| `decision_id`     | STRING(26)  | Optional decision alignment         |
+| `checkpoint_type` | TEXT        | `pre_write`, `post_step`, `manual`  |
+| `resource_id`     | STRING(26)  | Optional resource association        |
+| `sandbox_ref`     | TEXT        | Git commit hash or patch reference  |
+| `filesystem_path` | TEXT        | Relative path within checkpoint dir |
+| `size_bytes`      | INTEGER     | Size of checkpoint data in bytes    |
+| `created_at`      | STRING(30)  | ISO-8601 timestamp                  |
+| `metadata_json`   | TEXT        | JSON: reason, source_tool, phase    |
+
+Indexes: `idx_checkpoints_plan`, `ix_checkpoint_metadata_created_at`.
+
+## Dependency Injection
+
+The `CheckpointService` is registered in the DI container
+(`container.py`) and backed by a `CheckpointRepository` that uses the
+session-factory pattern (ADR-007). The CLI `rollback` command obtains
+the service from the container via `get_container().checkpoint_service()`.
+
+The `UnitOfWorkContext` also exposes a `checkpoints` property for
+cross-repository atomicity within a single transaction.
+
+## Metadata for Auditability
+
+Each checkpoint stores structured metadata to support audit trails:
+
+- **reason** — Why the checkpoint was created (e.g. "pre-execution")
+- **source_tool** — Which tool triggered the checkpoint (e.g. "lint-check")
+- **phase** — Which plan phase was active (e.g. "execute")
+- **extra** — Arbitrary key-value pairs for extension

features/checkpoint_rollback.feature

@@ -0,0 +1,296 @@
+@phase2 @checkpoint @rollback
+Feature: Checkpoint and rollback
+  As an operator
+  I want to create checkpoints during plan execution and rollback to them
+  So that I can recover from mistakes without losing all progress
+
+  # ───────────────────────────────────────────────────────────
+  # Domain model scenarios
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Checkpoint model validates ULID fields
+    Given a valid checkpoint with plan_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and sandbox_ref "abc123"
+    Then the checkpoint plan_id should be "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    And the checkpoint sandbox_ref should be "abc123"
+
+  Scenario: Checkpoint metadata stores reason and source tool
+    Given a checkpoint with reason "pre-execution" and source_tool "lint-check" and phase "execute"
+    Then the checkpoint metadata reason should be "pre-execution"
+    And the checkpoint metadata source_tool should be "lint-check"
+    And the checkpoint metadata phase should be "execute"
+
+  Scenario: Checkpoint retention policy defaults
+    Given a default retention policy
+    Then the max_checkpoints should be 50
+    And auto_prune should be true
+
+  Scenario: Checkpoint retention policy custom values
+    Given a retention policy with max_checkpoints 5 and auto_prune false
+    Then the max_checkpoints should be 5
+    And auto_prune should be false
+
+  Scenario: RollbackResult model stores restored file count
+    Given a rollback result with 3 restored files and checkpoint "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    Then the restored_files_count should be 3
+    And the from_checkpoint_id should be "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+  # ───────────────────────────────────────────────────────────
+  # CheckpointService scenarios
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Create a checkpoint
+    Given a checkpoint service
+    When I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    Then the checkpoint should be created successfully
+    And the checkpoint plan_id should match "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+  Scenario: List checkpoints for a plan
+    Given a checkpoint service
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-1"
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-2"
+    When I list checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    Then I should see 2 checkpoints
+
+  Scenario: Rollback to checkpoint succeeds
+    Given a checkpoint service with sandbox for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    When I rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to the created checkpoint
+    Then the rollback should succeed
+    And the rollback result should show restored files
+
+  Scenario: Rollback rejected when plan is applied
+    Given a checkpoint service
+    And plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" is marked as applied
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    When I attempt rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to the created checkpoint
+    Then the rollback should be rejected with "plan is already applied"
+
+  Scenario: Rollback rejected when sandbox is missing
+    Given a checkpoint service
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    When I attempt rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to the created checkpoint
+    Then the rollback should be rejected with "sandbox is missing"
+
+  Scenario: Rollback with non-existent checkpoint
+    Given a checkpoint service with sandbox for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    When I attempt rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to checkpoint "01NONEXISTENT0000000000000"
+    Then the rollback should fail with not found error
+
+  Scenario: Delete a checkpoint
+    Given a checkpoint service
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    When I delete the created checkpoint
+    Then the checkpoint should be deleted
+
+  Scenario: Prune enforces retention policy
+    Given a checkpoint service
+    And I create 5 checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    When I prune checkpoints with max 3
+    Then 2 checkpoints should be pruned
+    And 3 checkpoint snapshots should remain
+
+  Scenario: Auto-prune on create with retention policy
+    Given a checkpoint service
+    And I create 10 checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    When I create a checkpoint with retention policy max 5
+    Then 6 checkpoints should have been pruned on creation
+    And 5 checkpoint snapshots should remain
+
+  Scenario: Create checkpoint with empty plan_id raises error
+    Given a checkpoint service
+    When I attempt to create a checkpoint with empty plan_id
+    Then a checkpoint validation error should be raised
+
+  Scenario: Create checkpoint with empty sandbox_ref raises error
+    Given a checkpoint service
+    When I attempt to create a checkpoint with empty sandbox_ref
+    Then a checkpoint validation error should be raised
+
+  Scenario: Get checkpoint by ID
+    Given a checkpoint service
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-xyz"
+    When I get the checkpoint by its ID
+    Then the checkpoint should be returned successfully
+
+  Scenario: Get non-existent checkpoint raises error
+    Given a checkpoint service
+    When I attempt to get checkpoint "01NONEXISTENT0000000000000"
+    Then a checkpoint not found error should be raised
+
+  # ───────────────────────────────────────────────────────────
+  # Correction service rollback integration
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Correction service accepts checkpoint_service parameter
+    Given a correction service with checkpoint support
+    Then the correction service should have checkpoint support
+
+  Scenario: Checkpoint metadata reason stored for auditability
+    Given a checkpoint service
+    When I create a checkpoint with reason "pre-tool-run" and source_tool "format-code" and phase "execute"
+    Then the stored checkpoint metadata should have reason "pre-tool-run"
+    And the stored checkpoint metadata should have source_tool "format-code"
+    And the stored checkpoint metadata should have phase "execute"
+
+  # ───────────────────────────────────────────────────────────
+  # Spec-aligned field scenarios (decision_id, checkpoint_type, etc.)
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Checkpoint stores decision_id when aligned to a decision
+    Given a checkpoint service
+    When I create a checkpoint aligned to decision "01DEC1S10N0000000000000000"
+    Then the stored checkpoint decision_id should be "01DEC1S10N0000000000000000"
+
+  Scenario: Checkpoint stores checkpoint_type
+    Given a checkpoint service
+    When I create a pre_write checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    Then the stored checkpoint_type should be "pre_write"
+
+  Scenario: Checkpoint type validation rejects invalid values
+    Given a checkpoint service
+    When I attempt to create a checkpoint with type "invalid_type"
+    Then a checkpoint validation error should be raised
+
+  Scenario: Checkpoint stores resource_id
+    Given a checkpoint service
+    When I create a checkpoint with resource "01RES0URCE0000000000000000"
+    Then the stored checkpoint resource_id should be "01RES0URCE0000000000000000"
+
+  Scenario: Checkpoint stores filesystem_path and size_bytes
+    Given a checkpoint service
+    When I create a checkpoint with path "snapshots/cp001" and size 4096
+    Then the stored checkpoint filesystem_path should be "snapshots/cp001"
+    And the stored checkpoint size_bytes should be 4096
+
+  Scenario: Prune preserves first and most recent checkpoints
+    Given a checkpoint service
+    And I create 5 checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    When I prune checkpoints with max 2
+    Then 3 checkpoints should be pruned
+    And 2 checkpoint snapshots should remain
+    And the first checkpoint should survive pruning
+    And the most recent checkpoint should survive pruning
+
+  # ───────────────────────────────────────────────────────────
+  # Persistent guard scenarios (via PlanLifecycleService)
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Rollback rejected via lifecycle service when plan is applied
+    Given a checkpoint service backed by a lifecycle service with an applied plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    When I attempt rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to the created checkpoint
+    Then the rollback should be rejected with "plan is already applied"
+
+  Scenario: Rollback rejected via lifecycle service when sandbox refs are empty
+    Given a checkpoint service backed by a lifecycle service with no sandbox for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    When I attempt rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to the created checkpoint
+    Then the rollback should be rejected with "sandbox is missing"
+
+  Scenario: Rollback succeeds via lifecycle service when plan has sandbox
+    Given a checkpoint service backed by a lifecycle service with sandbox for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    When I rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to the created checkpoint
+    Then the rollback should succeed
+    And the rollback result should show restored files
+
+  # ───────────────────────────────────────────────────────────
+  # Default retention policy scenario
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Default retention policy applied when none specified
+    Given a checkpoint service
+    And I create 55 checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    When I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-final"
+    Then at most 50 checkpoint snapshots should remain for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+  Scenario: Rollback rejected when checkpoint belongs to different plan
+    Given a checkpoint service with sandbox for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    And I create a checkpoint for plan "01KJGSHD2JKTFZGGBT6FRQD2W2" with sandbox_ref "commit-other"
+    When I attempt rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to the created checkpoint
+    Then the rollback should be rejected with "does not belong to plan"
+
+  Scenario: Unregister sandbox prevents rollback
+    Given a checkpoint service with sandbox for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-abc"
+    And the sandbox for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" is unregistered
+    When I attempt rollback plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" to the created checkpoint
+    Then the rollback should be rejected with "sandbox is missing"
+
+  Scenario: Prune with auto_prune disabled returns empty
+    Given a checkpoint service
+    And I create 5 checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    When I prune checkpoints with auto_prune disabled
+    Then 0 checkpoints should be pruned
+
+  # ───────────────────────────────────────────────────────────
+  # Repository-backed service scenarios
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Repository-backed create persists via repository
+    Given a checkpoint service backed by a stub repository
+    When I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-repo"
+    Then the checkpoint should be created successfully
+    And the stub repository should have 1 checkpoint stored
+
+  Scenario: Repository-backed list delegates to repository
+    Given a checkpoint service backed by a stub repository
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-r1"
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-r2"
+    When I list checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    Then I should see 2 checkpoints
+
+  Scenario: Repository-backed delete removes via repository
+    Given a checkpoint service backed by a stub repository
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-del"
+    When I delete the created checkpoint
+    Then the checkpoint should be deleted
+
+  Scenario: Repository-backed delete raises error when not found
+    Given a checkpoint service backed by a stub repository
+    When I attempt to delete checkpoint "01NONEXISTENT0000000000000"
+    Then a checkpoint not found error should be raised
+
+  Scenario: Repository-backed prune delegates to repository
+    Given a checkpoint service backed by a stub repository
+    And I create 5 checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    When I prune checkpoints with max 3
+    Then 2 checkpoints should be pruned
+
+  Scenario: Repository-backed get retrieves via repository
+    Given a checkpoint service backed by a stub repository
+    And I create a checkpoint for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV" with sandbox_ref "commit-get"
+    When I get the checkpoint by its ID
+    Then the checkpoint should be returned successfully
+
+  Scenario: Repository-backed get raises not found for missing checkpoint
+    Given a checkpoint service backed by a stub repository
+    When I attempt to get checkpoint "01NONEXISTENT0000000000000"
+    Then a checkpoint not found error should be raised
+
+  # ───────────────────────────────────────────────────────────
+  # In-memory edge case scenarios
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Delete non-existent checkpoint from in-memory store raises error
+    Given a checkpoint service
+    When I attempt to delete checkpoint "01NONEXISTENT0000000000000"
+    Then a checkpoint not found error should be raised
+
+  Scenario: List checkpoints tolerates orphaned index entries
+    Given a checkpoint service with an orphaned plan index entry
+    When I list checkpoints for plan "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+    Then I should see 1 checkpoints
+
+  Scenario: Delete checkpoint whose plan index is missing skips cleanup
+    Given a checkpoint service with a checkpoint missing from plan index
+    When I delete the created checkpoint
+    Then the checkpoint should be deleted
+
+  # ───────────────────────────────────────────────────────────
+  # Domain model validator edge cases
+  # ───────────────────────────────────────────────────────────
+
+  Scenario: Checkpoint ULID validator rejects empty string
+    When I invoke the checkpoint ULID validator with an empty string
+    Then the checkpoint validator should raise ValueError "ULID field must not be empty"

features/sandbox_checkpoints.feature

@@ -13,7 +13,7 @@ Feature: Sandbox Checkpoint and Rollback Hooks
     When I create a checkpoint with phase "pre_execute" and plan "plan-001"
     Then a checkpoint should be returned
     And the checkpoint phase should be "pre_execute"
-    And the checkpoint plan_id should be "plan-001"
+    And the sandbox checkpoint plan_id should be "plan-001"
     And the checkpoint sandbox_id should match the sandbox
 
   Scenario: Checkpoint created after successful execution
@@ -31,7 +31,7 @@ Feature: Sandbox Checkpoint and Rollback Hooks
     And a checkpoint exists for the sandbox with phase "pre_execute" and plan "plan-001"
     When I modify the sandbox files
     And I rollback to the checkpoint
-    Then the rollback should succeed
+    Then the sandbox rollback should succeed
     And the sandbox files should match the original state
 
   Scenario: Rollback returns false when snapshot path is missing
@@ -95,3 +95,50 @@ Feature: Sandbox Checkpoint and Rollback Hooks
     Given a plan lifecycle service
     And a plan apply service with a checkpoint manager
     Then the plan apply service checkpoint_manager should not be None
+
+  # Context-less sandbox (metadata-only checkpoint)
+
+  Scenario: Checkpoint created for sandbox without context
+    Given a sandbox proxy with no context
+    When I create a checkpoint with phase "pre_execute" and plan "plan-noctx" using the proxy
+    Then a checkpoint should be returned
+    And the checkpoint snapshot_path should be an empty directory
+
+  # Rollback edge cases
+
+  Scenario: Rollback fails when sandbox_path metadata is absent
+    Given a sandbox with files in a temporary directory
+    And a checkpoint exists for the sandbox with phase "pre_execute" and plan "plan-001" without sandbox_path metadata
+    When I attempt to rollback to the checkpoint without sandbox path
+    Then the rollback should fail
+
+  Scenario: Rollback fails due to OS error during restore
+    Given a sandbox with files in a temporary directory
+    And a checkpoint exists for the sandbox with phase "pre_execute" and plan "plan-001"
+    When I make the sandbox path read-only and attempt rollback
+    Then the rollback should fail
+
+  # Delete across multiple sandboxes
+
+  Scenario: Delete checkpoint searches across multiple sandboxes
+    Given two sandboxes with checkpoints
+    When I delete the checkpoint from the second sandbox
+    Then the deletion should succeed
+    And listing checkpoints for the first sandbox should return 1 checkpoint for sandbox one
+
+  # Snapshot copy failure
+
+  Scenario: Snapshot copy failure raises SandboxError
+    Given a sandbox with an unreadable directory
+    When I attempt to create a checkpoint that fails during snapshot
+    Then a sandbox error should be raised
+
+  # Cleanup snapshot edge cases
+
+  Scenario: Cleanup snapshot with empty path is a no-op
+    When I cleanup a snapshot with an empty path
+    Then no error should occur
+
+  Scenario: Cleanup snapshot where parent directory is absent is a no-op
+    When I cleanup a snapshot with a non-existent parent
+    Then no error should occur

features/steps/checkpoint_rollback_steps.py

@@ -0,0 +1,729 @@
+"""Step definitions for checkpoint_rollback.feature.
+
+Exercises checkpoint domain models, CheckpointService operations
+(create, list, rollback, prune, delete), guard validations, and
+correction-service integration.
+"""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from pydantic import ValidationError as PydanticValidationError
+
+from cleveragents.application.services.checkpoint_service import CheckpointService
+from cleveragents.application.services.correction_service import CorrectionService
+from cleveragents.core.exceptions import (
+    BusinessRuleViolation,
+    ResourceNotFoundError,
+    ValidationError,
+)
+from cleveragents.domain.models.core.checkpoint import (
+    Checkpoint,
+    CheckpointMetadata,
+    CheckpointRetentionPolicy,
+    RollbackResult,
+)
+
+_VALID_ULID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+# -------------------------------------------------------------------
+# Domain model steps
+# -------------------------------------------------------------------
+
+
+@given('a valid checkpoint with plan_id "{plan_id}" and sandbox_ref "{sandbox_ref}"')
+def step_create_valid_checkpoint(context, plan_id, sandbox_ref):
+    from ulid import ULID
+
+    context.checkpoint = Checkpoint(
+        checkpoint_id=str(ULID()),
+        plan_id=plan_id,
+        sandbox_ref=sandbox_ref,
+    )
+
+
+@then('the checkpoint plan_id should be "{expected}"')
+def step_check_plan_id(context, expected):
+    assert context.checkpoint.plan_id == expected
+
+
+@then('the checkpoint sandbox_ref should be "{expected}"')
+def step_check_sandbox_ref(context, expected):
+    assert context.checkpoint.sandbox_ref == expected
+
+
+@given(
+    'a checkpoint with reason "{reason}" and source_tool "{tool}" and phase "{phase}"'
+)
+def step_create_checkpoint_with_metadata(context, reason, tool, phase):
+    from ulid import ULID
+
+    context.checkpoint = Checkpoint(
+        checkpoint_id=str(ULID()),
+        plan_id=_VALID_ULID,
+        sandbox_ref="ref123",
+        metadata=CheckpointMetadata(reason=reason, source_tool=tool, phase=phase),
+    )
+
+
+@then('the checkpoint metadata reason should be "{expected}"')
+def step_check_metadata_reason(context, expected):
+    assert context.checkpoint.metadata.reason == expected
+
+
+@then('the checkpoint metadata source_tool should be "{expected}"')
+def step_check_metadata_source_tool(context, expected):
+    assert context.checkpoint.metadata.source_tool == expected
+
+
+@then('the checkpoint metadata phase should be "{expected}"')
+def step_check_metadata_phase(context, expected):
+    assert context.checkpoint.metadata.phase == expected
+
+
+@given("a default retention policy")
+def step_default_retention(context):
+    context.policy = CheckpointRetentionPolicy()
+
+
+@then("the max_checkpoints should be {count:d}")
+def step_check_max_checkpoints(context, count):
+    assert context.policy.max_checkpoints == count
+
+
+@then("auto_prune should be true")
+def step_check_auto_prune_true(context):
+    assert context.policy.auto_prune is True
+
+
+@then("auto_prune should be false")
+def step_check_auto_prune_false(context):
+    assert context.policy.auto_prune is False
+
+
+@given("a retention policy with max_checkpoints {max_cp:d} and auto_prune false")
+def step_custom_retention(context, max_cp):
+    context.policy = CheckpointRetentionPolicy(max_checkpoints=max_cp, auto_prune=False)
+
+
+@given('a rollback result with {count:d} restored files and checkpoint "{cp_id}"')
+def step_create_rollback_result(context, count, cp_id):
+    context.rollback_result = RollbackResult(
+        restored_files_count=count,
+        changed_paths=[f"file_{i}.py" for i in range(count)],
+        from_checkpoint_id=cp_id,
+    )
+
+
+@then("the restored_files_count should be {count:d}")
+def step_check_restored_count(context, count):
+    assert context.rollback_result.restored_files_count == count
+
+
+@then('the from_checkpoint_id should be "{expected}"')
+def step_check_from_checkpoint(context, expected):
+    assert context.rollback_result.from_checkpoint_id == expected
+
+
+# -------------------------------------------------------------------
+# CheckpointService steps
+# -------------------------------------------------------------------
+
+
+@given("a checkpoint service")
+def step_create_service(context):
+    context.svc = CheckpointService()
+    context.checkpoint = None
+    context.checkpoints = None
+    context.rollback_result = None
+    context.error = None
+    context.pruned_ids = None
+
+
+@given('a checkpoint service with sandbox for plan "{plan_id}"')
+def step_create_service_with_sandbox(context, plan_id):
+    context.svc = CheckpointService()
+    context.svc.register_sandbox(plan_id, "sandbox-path-123")
+    context.checkpoint = None
+    context.checkpoints = None
+    context.rollback_result = None
+    context.error = None
+    context.pruned_ids = None
+
+
+@given('plan "{plan_id}" is marked as applied')
+def step_mark_applied(context, plan_id):
+    context.svc.mark_plan_applied(plan_id)
+
+
+@given('I create a checkpoint for plan "{plan_id}" with sandbox_ref "{ref}"')
+@when('I create a checkpoint for plan "{plan_id}" with sandbox_ref "{ref}"')
+def step_create_checkpoint(context, plan_id, ref):
+    context.checkpoint = context.svc.create_checkpoint(plan_id=plan_id, sandbox_ref=ref)
+
+
+@then("the checkpoint should be created successfully")
+def step_checkpoint_created(context):
+    assert context.checkpoint is not None
+    assert context.checkpoint.checkpoint_id
+
+
+@then('the checkpoint plan_id should match "{expected}"')
+def step_checkpoint_plan_match(context, expected):
+    assert context.checkpoint.plan_id == expected
+
+
+@when('I list checkpoints for plan "{plan_id}"')
+def step_list_checkpoints(context, plan_id):
+    context.checkpoints = context.svc.list_checkpoints(plan_id)
+
+
+@then("I should see {count:d} checkpoints")
+def step_check_checkpoint_count(context, count):
+    assert len(context.checkpoints) == count
+
+
+@when('I rollback plan "{plan_id}" to the created checkpoint')
+def step_rollback_to_checkpoint(context, plan_id):
+    context.rollback_result = context.svc.rollback_to_checkpoint(
+        plan_id, context.checkpoint.checkpoint_id
+    )
+
+
+@then("the rollback should succeed")
+def step_rollback_success(context):
+    assert context.rollback_result is not None
+
+
+@then("the rollback result should show restored files")
+def step_rollback_shows_files(context):
+    assert context.rollback_result.restored_files_count > 0
+
+
+@when('I attempt rollback plan "{plan_id}" to the created checkpoint')
+def step_attempt_rollback(context, plan_id):
+    try:
+        context.rollback_result = context.svc.rollback_to_checkpoint(
+            plan_id, context.checkpoint.checkpoint_id
+        )
+        context.error = None
+    except (BusinessRuleViolation, ResourceNotFoundError, ValidationError) as e:
+        context.error = e
+
+
+@when('I attempt rollback plan "{plan_id}" to checkpoint "{cp_id}"')
+def step_attempt_rollback_by_id(context, plan_id, cp_id):
+    try:
+        context.rollback_result = context.svc.rollback_to_checkpoint(plan_id, cp_id)
+        context.error = None
+    except (BusinessRuleViolation, ResourceNotFoundError, ValidationError) as e:
+        context.error = e
+
+
+@then('the rollback should be rejected with "{msg}"')
+def step_rollback_rejected(context, msg):
+    assert context.error is not None
+    assert msg in str(context.error)
+
+
+@then("the rollback should fail with not found error")
+def step_rollback_not_found(context):
+    assert context.error is not None
+    assert isinstance(context.error, ResourceNotFoundError)
+
+
+@when("I delete the created checkpoint")
+def step_delete_checkpoint(context):
+    context.svc.delete_checkpoint(context.checkpoint.checkpoint_id)
+
+
+@then("the checkpoint should be deleted")
+def step_checkpoint_deleted(context):
+    try:
+        context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+        msg = "Checkpoint should have been deleted"
+        raise AssertionError(msg)
+    except ResourceNotFoundError:
+        pass
+
+
+@given('I create {count:d} checkpoints for plan "{plan_id}"')
+def step_create_multiple_checkpoints(context, count, plan_id):
+    for i in range(count):
+        context.checkpoint = context.svc.create_checkpoint(
+            plan_id=plan_id, sandbox_ref=f"commit-{i}"
+        )
+
+
+@when("I prune checkpoints with max {max_cp:d}")
+def step_prune_checkpoints(context, max_cp):
+    policy = CheckpointRetentionPolicy(max_checkpoints=max_cp, auto_prune=True)
+    context.pruned_ids = context.svc.prune_checkpoints(_VALID_ULID, policy)
+
+
+@then("{count:d} checkpoints should be pruned")
+def step_check_pruned_count(context, count):
+    assert len(context.pruned_ids) == count
+
+
+@then("{count:d} checkpoint snapshots should remain")
+def step_check_remaining_count(context, count):
+    remaining = context.svc.list_checkpoints(_VALID_ULID)
+    assert len(remaining) == count
+
+
+@when("I create a checkpoint with retention policy max {max_cp:d}")
+def step_create_with_retention(context, max_cp):
+    policy = CheckpointRetentionPolicy(max_checkpoints=max_cp, auto_prune=True)
+    before_count = len(context.svc.list_checkpoints(_VALID_ULID))
+    context.checkpoint = context.svc.create_checkpoint(
+        plan_id=_VALID_ULID,
+        sandbox_ref="commit-retention",
+        retention_policy=policy,
+    )
+    after_count = len(context.svc.list_checkpoints(_VALID_ULID))
+    context.pruned_on_creation = (before_count + 1) - after_count
+
+
+@then("{count:d} checkpoints should have been pruned on creation")
+def step_check_pruned_on_creation(context, count):
+    assert context.pruned_on_creation == count
+
+
+@when("I attempt to create a checkpoint with empty plan_id")
+def step_create_empty_plan(context):
+    try:
+        context.svc.create_checkpoint(plan_id="", sandbox_ref="ref")
+        context.error = None
+    except (ValidationError, PydanticValidationError) as e:
+        context.error = e
+
+
+@when("I attempt to create a checkpoint with empty sandbox_ref")
+def step_create_empty_ref(context):
+    try:
+        context.svc.create_checkpoint(plan_id=_VALID_ULID, sandbox_ref="")
+        context.error = None
+    except (ValidationError, PydanticValidationError) as e:
+        context.error = e
+
+
+@then("a checkpoint validation error should be raised")
+def step_checkpoint_validation_error(context):
+    assert context.error is not None
+    assert isinstance(context.error, (ValidationError, PydanticValidationError))
+
+
+@when("I get the checkpoint by its ID")
+def step_get_checkpoint_by_id(context):
+    context.result_checkpoint = context.svc.get_checkpoint(
+        context.checkpoint.checkpoint_id
+    )
+
+
+@then("the checkpoint should be returned successfully")
+def step_checkpoint_returned(context):
+    assert context.result_checkpoint is not None
+    assert context.result_checkpoint.checkpoint_id == context.checkpoint.checkpoint_id
+
+
+@when('I attempt to get checkpoint "{cp_id}"')
+def step_get_nonexistent(context, cp_id):
+    try:
+        context.svc.get_checkpoint(cp_id)
+        context.error = None
+    except ResourceNotFoundError as e:
+        context.error = e
+
+
+@then("a checkpoint not found error should be raised")
+def step_checkpoint_not_found_error(context):
+    assert context.error is not None
+    assert isinstance(context.error, ResourceNotFoundError)
+
+
+# -------------------------------------------------------------------
+# Correction service integration steps
+# -------------------------------------------------------------------
+
+
+@given("a correction service with checkpoint support")
+def step_correction_with_checkpoint(context):
+    cp_svc = CheckpointService()
+    context.correction_svc = CorrectionService(checkpoint_service=cp_svc)
+
+
+@then("the correction service should have checkpoint support")
+def step_check_checkpoint_support(context):
+    assert context.correction_svc._checkpoint_service is not None
+
+
+@when(
+    'I create a checkpoint with reason "{reason}" and source_tool "{tool}" and phase "{phase}"'
+)
+def step_create_with_metadata(context, reason, tool, phase):
+    context.checkpoint = context.svc.create_checkpoint(
+        plan_id=_VALID_ULID,
+        sandbox_ref="ref-meta",
+        reason=reason,
+        source_tool=tool,
+        phase=phase,
+    )
+
+
+@then('the stored checkpoint metadata should have reason "{expected}"')
+def step_stored_reason(context, expected):
+    cp = context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+    assert cp.metadata.reason == expected
+
+
+@then('the stored checkpoint metadata should have source_tool "{expected}"')
+def step_stored_source_tool(context, expected):
+    cp = context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+    assert cp.metadata.source_tool == expected
+
+
+@then('the stored checkpoint metadata should have phase "{expected}"')
+def step_stored_phase(context, expected):
+    cp = context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+    assert cp.metadata.phase == expected
+
+
+# -------------------------------------------------------------------
+# Spec-aligned field steps (decision_id, checkpoint_type, etc.)
+# -------------------------------------------------------------------
+
+
+@when('I create a checkpoint aligned to decision "{decision_id}"')
+def step_create_with_decision(context, decision_id):
+    context.checkpoint = context.svc.create_checkpoint(
+        plan_id=_VALID_ULID,
+        sandbox_ref="ref-decision",
+        decision_id=decision_id,
+    )
+
+
+@then('the stored checkpoint decision_id should be "{expected}"')
+def step_check_decision_id(context, expected):
+    cp = context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+    assert cp.decision_id == expected
+
+
+@when('I create a pre_write checkpoint for plan "{plan_id}"')
+def step_create_prewrite(context, plan_id):
+    context.checkpoint = context.svc.create_checkpoint(
+        plan_id=plan_id,
+        sandbox_ref="ref-prewrite",
+        checkpoint_type="pre_write",
+    )
+
+
+@then('the stored checkpoint_type should be "{expected}"')
+def step_check_checkpoint_type(context, expected):
+    cp = context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+    assert cp.checkpoint_type == expected
+
+
+@when('I attempt to create a checkpoint with type "{cp_type}"')
+def step_create_invalid_type(context, cp_type):
+    try:
+        context.svc.create_checkpoint(
+            plan_id=_VALID_ULID,
+            sandbox_ref="ref-invalid",
+            checkpoint_type=cp_type,
+        )
+        context.error = None
+    except (ValidationError, PydanticValidationError) as e:
+        context.error = e
+
+
+@when('I create a checkpoint with resource "{resource_id}"')
+def step_create_with_resource(context, resource_id):
+    context.checkpoint = context.svc.create_checkpoint(
+        plan_id=_VALID_ULID,
+        sandbox_ref="ref-resource",
+        resource_id=resource_id,
+    )
+
+
+@then('the stored checkpoint resource_id should be "{expected}"')
+def step_check_resource_id(context, expected):
+    cp = context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+    assert cp.resource_id == expected
+
+
+@when('I create a checkpoint with path "{path}" and size {size:d}')
+def step_create_with_path_and_size(context, path, size):
+    context.checkpoint = context.svc.create_checkpoint(
+        plan_id=_VALID_ULID,
+        sandbox_ref="ref-path",
+        filesystem_path=path,
+        size_bytes=size,
+    )
+
+
+@then('the stored checkpoint filesystem_path should be "{expected}"')
+def step_check_filesystem_path(context, expected):
+    cp = context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+    assert cp.filesystem_path == expected
+
+
+@then("the stored checkpoint size_bytes should be {expected:d}")
+def step_check_size_bytes(context, expected):
+    cp = context.svc.get_checkpoint(context.checkpoint.checkpoint_id)
+    assert cp.size_bytes == expected
+
+
+@then("the first checkpoint should survive pruning")
+def step_first_survives_pruning(context):
+    """Verify the first (oldest) checkpoint survived pruning."""
+    remaining = context.svc.list_checkpoints(_VALID_ULID)
+    # The first checkpoint created has sandbox_ref "commit-0"
+    assert any(cp.sandbox_ref == "commit-0" for cp in remaining), (
+        "First checkpoint (commit-0) should be preserved after pruning"
+    )
+
+
+@then("the most recent checkpoint should survive pruning")
+def step_most_recent_survives_pruning(context):
+    """Verify the most recent checkpoint survived pruning."""
+    remaining = context.svc.list_checkpoints(_VALID_ULID)
+    # The last checkpoint created has the highest commit-N
+    last_ref = context.checkpoint.sandbox_ref
+    assert any(cp.sandbox_ref == last_ref for cp in remaining), (
+        f"Most recent checkpoint ({last_ref}) should be preserved after pruning"
+    )
+
+
+# -------------------------------------------------------------------
+# Persistent guard steps (lifecycle-service backed)
+# -------------------------------------------------------------------
+
+
+class _FakePlan:
+    """Minimal plan stub for lifecycle-service guard tests."""
+
+    def __init__(self, plan_id, processing_state, sandbox_refs=None):
+        self.plan_id = plan_id
+        self.processing_state = processing_state
+        self.sandbox_refs = sandbox_refs or []
+
+
+class _FakeLifecycleService:
+    """Stub PlanLifecycleService that returns a canned plan."""
+
+    def __init__(self, plan):
+        self._plan = plan
+
+    def get_plan(self, plan_id):
+        return self._plan
+
+
+@given(
+    'a checkpoint service backed by a lifecycle service with an applied plan "{plan_id}"'
+)
+def step_service_lifecycle_applied(context, plan_id):
+    from cleveragents.domain.models.core.plan import ProcessingState
+
+    fake_plan = _FakePlan(plan_id, ProcessingState.APPLIED, sandbox_refs=["sb"])
+    fake_ls = _FakeLifecycleService(fake_plan)
+    context.svc = CheckpointService(plan_lifecycle_service=fake_ls)  # type: ignore[arg-type]
+    context.checkpoint = None
+    context.error = None
+    context.rollback_result = None
+
+
+@given(
+    'a checkpoint service backed by a lifecycle service with no sandbox for plan "{plan_id}"'
+)
+def step_service_lifecycle_no_sandbox(context, plan_id):
+    from cleveragents.domain.models.core.plan import ProcessingState
+
+    fake_plan = _FakePlan(plan_id, ProcessingState.PROCESSING, sandbox_refs=[])
+    fake_ls = _FakeLifecycleService(fake_plan)
+    context.svc = CheckpointService(plan_lifecycle_service=fake_ls)  # type: ignore[arg-type]
+    context.checkpoint = None
+    context.error = None
+    context.rollback_result = None
+
+
+@given(
+    'a checkpoint service backed by a lifecycle service with sandbox for plan "{plan_id}"'
+)
+def step_service_lifecycle_with_sandbox(context, plan_id):
+    from cleveragents.domain.models.core.plan import ProcessingState
+
+    fake_plan = _FakePlan(
+        plan_id, ProcessingState.PROCESSING, sandbox_refs=["sandbox-root"]
+    )
+    fake_ls = _FakeLifecycleService(fake_plan)
+    context.svc = CheckpointService(plan_lifecycle_service=fake_ls)  # type: ignore[arg-type]
+    context.checkpoint = None
+    context.error = None
+    context.rollback_result = None
+
+
+# -------------------------------------------------------------------
+# Default retention policy step
+# -------------------------------------------------------------------
+
+
+@then('at most {max_cp:d} checkpoint snapshots should remain for plan "{plan_id}"')
+def step_at_most_checkpoints_remain(context, max_cp, plan_id):
+    remaining = context.svc.list_checkpoints(plan_id)
+    assert len(remaining) <= max_cp, (
+        f"Expected at most {max_cp} checkpoints, but found {len(remaining)}"
+    )
+
+
+@given('the sandbox for plan "{plan_id}" is unregistered')
+def step_unregister_sandbox(context, plan_id):
+    context.svc.unregister_sandbox(plan_id)
+
+
+@when("I prune checkpoints with auto_prune disabled")
+def step_prune_no_auto(context):
+    policy = CheckpointRetentionPolicy(max_checkpoints=3, auto_prune=False)
+    context.pruned_ids = context.svc.prune_checkpoints(_VALID_ULID, policy)
+
+
+# -------------------------------------------------------------------
+# Stub repository for repository-backed tests
+# -------------------------------------------------------------------
+
+
+class _StubCheckpointRepository:
+    """In-memory repository stub that implements the CheckpointRepository interface.
+
+    Used to exercise the repository-backed code paths in CheckpointService
+    without requiring a real database connection.
+    """
+
+    def __init__(self) -> None:
+        self._store: dict[str, Checkpoint] = {}
+        self._plan_index: dict[str, list[str]] = {}
+
+    def create(self, checkpoint: Checkpoint) -> Checkpoint:
+        self._store[checkpoint.checkpoint_id] = checkpoint
+        self._plan_index.setdefault(checkpoint.plan_id, []).append(
+            checkpoint.checkpoint_id
+        )
+        return checkpoint
+
+    def get_by_id(self, checkpoint_id: str) -> Checkpoint:
+        from cleveragents.infrastructure.database.repositories import (
+            CheckpointNotFoundError,
+        )
+
+        cp = self._store.get(checkpoint_id)
+        if cp is None:
+            raise CheckpointNotFoundError(checkpoint_id)
+        return cp
+
+    def list_by_plan(self, plan_id: str) -> list[Checkpoint]:
+        ids = self._plan_index.get(plan_id, [])
+        return [self._store[cid] for cid in ids if cid in self._store]
+
+    def delete(self, checkpoint_id: str) -> bool:
+        cp = self._store.pop(checkpoint_id, None)
+        if cp is None:
+            return False
+        plan_list = self._plan_index.get(cp.plan_id, [])
+        if checkpoint_id in plan_list:
+            plan_list.remove(checkpoint_id)
+        return True
+
+    def prune(self, plan_id: str, max_checkpoints: int) -> list[str]:
+        ids = self._plan_index.get(plan_id, [])
+        if len(ids) <= max_checkpoints or len(ids) < 3:
+            return []
+        interior = ids[1:-1]
+        excess = len(ids) - max_checkpoints
+        to_remove = interior[:excess]
+        for cid in to_remove:
+            self._store.pop(cid, None)
+        self._plan_index[plan_id] = [cid for cid in ids if cid not in set(to_remove)]
+        return to_remove
+
+
+@given("a checkpoint service backed by a stub repository")
+def step_create_repo_backed_service(context):
+    repo = _StubCheckpointRepository()
+    context.stub_repo = repo
+    context.svc = CheckpointService(repository=repo)  # type: ignore[arg-type]
+    context.checkpoint = None
+    context.checkpoints = None
+    context.rollback_result = None
+    context.error = None
+    context.pruned_ids = None
+
+
+@then("the stub repository should have {count:d} checkpoint stored")
+def step_check_stub_repo_count(context, count):
+    assert len(context.stub_repo._store) == count
+
+
+@when('I attempt to delete checkpoint "{cp_id}"')
+def step_attempt_delete_checkpoint(context, cp_id):
+    try:
+        context.svc.delete_checkpoint(cp_id)
+        context.error = None
+    except ResourceNotFoundError as e:
+        context.error = e
+
+
+# -------------------------------------------------------------------
+# In-memory edge case steps
+# -------------------------------------------------------------------
+
+
+@given("a checkpoint service with an orphaned plan index entry")
+def step_service_with_orphaned_index(context):
+    """Create a service where _plan_index has an ID not in _checkpoints."""
+    context.svc = CheckpointService()
+    # Create a real checkpoint first
+    real_cp = context.svc.create_checkpoint(
+        plan_id=_VALID_ULID, sandbox_ref="commit-real"
+    )
+    # Add a ghost entry to plan_index that has no corresponding checkpoint
+    context.svc._plan_index[_VALID_ULID].append("01GHOST0000000000000000000")
+    context.checkpoint = real_cp
+    context.checkpoints = None
+    context.rollback_result = None
+    context.error = None
+    context.pruned_ids = None
+
+
+@given("a checkpoint service with a checkpoint missing from plan index")
+def step_service_with_missing_plan_index(context):
+    """Create a service where a checkpoint exists in _checkpoints but not in _plan_index."""
+    context.svc = CheckpointService()
+    cp = context.svc.create_checkpoint(plan_id=_VALID_ULID, sandbox_ref="commit-orphan")
+    # Remove the checkpoint from plan_index but leave it in _checkpoints
+    context.svc._plan_index[_VALID_ULID].clear()
+    context.checkpoint = cp
+    context.checkpoints = None
+    context.rollback_result = None
+    context.error = None
+    context.pruned_ids = None
+
+
+# -------------------------------------------------------------------
+# Domain model validator edge case steps
+# -------------------------------------------------------------------
+
+
+@when("I invoke the checkpoint ULID validator with an empty string")
+def step_invoke_ulid_validator_empty(context):
+    try:
+        Checkpoint.validate_ulid("")
+        context.error = None
+    except ValueError as e:
+        context.error = e
+
+
+@then('the checkpoint validator should raise ValueError "{msg}"')
+def step_check_checkpoint_value_error(context, msg):
+    assert context.error is not None, "Expected a ValueError"
+    assert isinstance(context.error, ValueError)
+    assert msg in str(context.error)

features/steps/sandbox_checkpoints_steps.py

@@ -221,7 +221,7 @@ def step_then_checkpoint_phase(context: Any, phase: str) -> None:
     )
 
 
-@then('the checkpoint plan_id should be "{plan_id}"')
+@then('the sandbox checkpoint plan_id should be "{plan_id}"')
 def step_then_checkpoint_plan_id(context: Any, plan_id: str) -> None:
     assert context.checkpoint.plan_id == plan_id, (
         f"Expected plan_id '{plan_id}', got '{context.checkpoint.plan_id}'"
@@ -247,7 +247,7 @@ def step_then_checkpoint_metadata(context: Any, key: str, value: str) -> None:
     )
 
 
-@then("the rollback should succeed")
+@then("the sandbox rollback should succeed")
 def step_then_rollback_success(context: Any) -> None:
     assert context.rollback_result is True, "Expected rollback to succeed"
 
@@ -329,3 +329,212 @@ def step_then_apply_no_checkpoint(context: Any) -> None:
 @then("the plan apply service checkpoint_manager should not be None")
 def step_then_apply_has_checkpoint(context: Any) -> None:
     assert context.apply_service.checkpoint_manager is not None
+
+
+# ---------------------------------------------------------------------------
+# Context-less sandbox (metadata-only checkpoint) steps
+# ---------------------------------------------------------------------------
+
+
+class _NoContextSandboxProxy:
+    """Sandbox proxy with ``context = None`` for metadata-only checkpoints."""
+
+    def __init__(self, sandbox_id: str) -> None:
+        self._sandbox_id = sandbox_id
+
+    @property
+    def sandbox_id(self) -> str:
+        return self._sandbox_id
+
+    @property
+    def context(self) -> Any:
+        return None
+
+
+@given("a sandbox proxy with no context")
+def step_given_no_context_proxy(context: Any) -> None:
+    context.sandbox = _NoContextSandboxProxy(sandbox_id="sb-no-ctx-001")
+
+
+@when('I create a checkpoint with phase "{phase}" and plan "{plan_id}" using the proxy')
+def step_when_create_checkpoint_proxy(context: Any, phase: str, plan_id: str) -> None:
+    context.checkpoint = context.checkpoint_manager.create_checkpoint(
+        sandbox=context.sandbox,
+        plan_id=plan_id,
+        phase=phase,
+    )
+
+
+@then("the checkpoint snapshot_path should be an empty directory")
+def step_then_snapshot_empty_dir(context: Any) -> None:
+    path = context.checkpoint.snapshot_path
+    assert os.path.isdir(path), f"snapshot_path should be a directory: {path}"
+    contents = os.listdir(path)
+    assert len(contents) == 0, f"Expected empty dir, got: {contents}"
+
+
+# ---------------------------------------------------------------------------
+# Rollback edge case steps
+# ---------------------------------------------------------------------------
+
+
+@given(
+    'a checkpoint exists for the sandbox with phase "{phase}" and plan "{plan_id}"'
+    " without sandbox_path metadata"
+)
+def step_given_checkpoint_no_sandbox_path(
+    context: Any, phase: str, plan_id: str
+) -> None:
+    """Create a checkpoint without embedding sandbox_path in metadata."""
+    context.checkpoint = context.checkpoint_manager.create_checkpoint(
+        sandbox=context.sandbox,
+        plan_id=plan_id,
+        phase=phase,
+        metadata={},
+    )
+
+
+@when("I attempt to rollback to the checkpoint without sandbox path")
+def step_when_rollback_no_sandbox_path(context: Any) -> None:
+    context.rollback_result = context.checkpoint_manager.rollback_to(context.checkpoint)
+
+
+@when("I make the sandbox path read-only and attempt rollback")
+def step_when_rollback_readonly_sandbox(context: Any) -> None:
+    from unittest.mock import patch
+
+    # Create checkpoint with sandbox_path in metadata
+    meta = {"sandbox_path": context.sandbox_path}
+    cp = context.checkpoint_manager.create_checkpoint(
+        sandbox=context.sandbox,
+        plan_id="plan-001",
+        phase="pre_execute",
+        metadata=meta,
+    )
+    # Patch os.listdir to raise OSError, simulating a filesystem failure
+    # during rollback.  This avoids chmod which is ignored when running as root.
+    with patch("os.listdir", side_effect=OSError("Simulated I/O error")):
+        context.rollback_result = context.checkpoint_manager.rollback_to(cp)
+
+
+# ---------------------------------------------------------------------------
+# Multi-sandbox delete steps
+# ---------------------------------------------------------------------------
+
+
+@given("two sandboxes with checkpoints")
+def step_given_two_sandboxes(context: Any) -> None:
+    """Create checkpoints for two different sandboxes."""
+    # First sandbox
+    tmpdir1 = tempfile.mkdtemp(prefix="ca-cp-multi1-")
+    os.makedirs(os.path.join(tmpdir1, "src"), exist_ok=True)
+    with open(os.path.join(tmpdir1, "src", "a.py"), "w") as f:
+        f.write("# sandbox1\n")
+    sb1 = NoSandbox(resource_id="res-multi-1", original_path=tmpdir1)
+    sb1.create(plan_id="plan-multi")
+
+    tmpdir2 = tempfile.mkdtemp(prefix="ca-cp-multi2-")
+    os.makedirs(os.path.join(tmpdir2, "src"), exist_ok=True)
+    with open(os.path.join(tmpdir2, "src", "b.py"), "w") as f:
+        f.write("# sandbox2\n")
+    sb2 = NoSandbox(resource_id="res-multi-2", original_path=tmpdir2)
+    sb2.create(plan_id="plan-multi")
+
+    context.sandbox_one = sb1
+    context.sandbox_two = sb2
+    context.cp_one = context.checkpoint_manager.create_checkpoint(
+        sandbox=sb1, plan_id="plan-multi", phase="pre_execute"
+    )
+    context.cp_two = context.checkpoint_manager.create_checkpoint(
+        sandbox=sb2, plan_id="plan-multi", phase="pre_execute"
+    )
+
+
+@when("I delete the checkpoint from the second sandbox")
+def step_when_delete_second_sandbox_cp(context: Any) -> None:
+    context.delete_result = context.checkpoint_manager.delete_checkpoint(
+        context.cp_two.checkpoint_id
+    )
+
+
+@then(
+    "listing checkpoints for the first sandbox should return"
+    " 1 checkpoint for sandbox one"
+)
+def step_then_first_sandbox_still_has_one(context: Any) -> None:
+    result = context.checkpoint_manager.list_checkpoints(context.sandbox_one.sandbox_id)
+    assert len(result) == 1, f"Expected 1 checkpoint, got {len(result)}"
+
+
+# ---------------------------------------------------------------------------
+# Snapshot copy failure steps
+# ---------------------------------------------------------------------------
+
+
+@given("a sandbox with an unreadable directory")
+def step_given_unreadable_sandbox(context: Any) -> None:
+    tmpdir = tempfile.mkdtemp(prefix="ca-cp-unreadable-")
+    os.makedirs(os.path.join(tmpdir, "src"), exist_ok=True)
+    with open(os.path.join(tmpdir, "src", "file.txt"), "w") as f:
+        f.write("data\n")
+
+    sandbox = NoSandbox(resource_id="res-cp-unreadable", original_path=tmpdir)
+    sandbox.create(plan_id="plan-unreadable")
+    context.sandbox = sandbox
+    context.sandbox_path = tmpdir
+
+
+@when("I attempt to create a checkpoint that fails during snapshot")
+def step_when_create_checkpoint_snapshot_fail(context: Any) -> None:
+    from unittest.mock import patch
+
+    from cleveragents.infrastructure.sandbox.protocol import SandboxError
+
+    # Patch shutil.copytree to raise OSError, simulating a snapshot copy
+    # failure.  This avoids chmod which is ignored when running as root.
+    with patch("shutil.copytree", side_effect=OSError("Simulated copy error")):
+        try:
+            context.checkpoint_manager.create_checkpoint(
+                sandbox=context.sandbox,
+                plan_id="plan-unreadable",
+                phase="pre_execute",
+            )
+            context.error = None
+        except SandboxError as exc:
+            context.error = exc
+
+
+@then("a sandbox error should be raised")
+def step_then_sandbox_error(context: Any) -> None:
+    from cleveragents.infrastructure.sandbox.protocol import SandboxError
+
+    assert context.error is not None, "Expected a SandboxError"
+    assert isinstance(context.error, SandboxError)
+
+
+# ---------------------------------------------------------------------------
+# Cleanup snapshot edge case steps
+# ---------------------------------------------------------------------------
+
+
+@when("I cleanup a snapshot with an empty path")
+def step_when_cleanup_empty_path(context: Any) -> None:
+    context.error = None
+    try:
+        CheckpointManager._cleanup_snapshot("")
+    except Exception as exc:
+        context.error = exc
+
+
+@when("I cleanup a snapshot with a non-existent parent")
+def step_when_cleanup_nonexistent_parent(context: Any) -> None:
+    context.error = None
+    try:
+        CheckpointManager._cleanup_snapshot("/nonexistent/path/that/does/not/exist")
+    except Exception as exc:
+        context.error = exc
+
+
+@then("no error should occur")
+def step_then_no_error(context: Any) -> None:
+    assert context.error is None, f"Expected no error, got: {context.error}"

robot/checkpoint_rollback.robot

@@ -0,0 +1,87 @@
+*** Settings ***
+Documentation     Smoke tests for checkpoint and rollback: create, list, rollback,
+...               prune, guards, and metadata auditability.
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    robot/helper_checkpoint_rollback.py
+
+*** Test Cases ***
+Create Checkpoint
+    [Documentation]    Verify checkpoint creation succeeds
+    [Tags]    phase2    checkpoint    create
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    create-checkpoint    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    create-checkpoint-ok
+
+List Checkpoints
+    [Documentation]    Verify listing checkpoints returns expected count
+    [Tags]    phase2    checkpoint    list
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    list-checkpoints    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    list-checkpoints-ok
+
+Rollback To Checkpoint
+    [Documentation]    Verify rollback restores sandbox state
+    [Tags]    phase2    checkpoint    rollback
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    rollback-checkpoint    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    rollback-checkpoint-ok
+
+Rollback Rejected Applied Plan
+    [Documentation]    Verify rollback blocked when plan is applied
+    [Tags]    phase2    checkpoint    rollback    guard
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    rollback-applied-guard    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    rollback-applied-guard-ok
+
+Rollback Rejected Missing Sandbox
+    [Documentation]    Verify rollback blocked when sandbox is missing
+    [Tags]    phase2    checkpoint    rollback    guard
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    rollback-sandbox-guard    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    rollback-sandbox-guard-ok
+
+Prune Checkpoints
+    [Documentation]    Verify pruning removes oldest checkpoints
+    [Tags]    phase2    checkpoint    prune
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    prune-checkpoints    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    prune-checkpoints-ok
+
+Delete Checkpoint
+    [Documentation]    Verify deleting a checkpoint removes it
+    [Tags]    phase2    checkpoint    delete
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    delete-checkpoint    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    delete-checkpoint-ok
+
+Checkpoint Metadata Auditability
+    [Documentation]    Verify checkpoint metadata stores reason and source
+    [Tags]    phase2    checkpoint    metadata
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    checkpoint-metadata    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    checkpoint-metadata-ok
+
+Auto Prune On Create
+    [Documentation]    Verify auto-prune on checkpoint creation
+    [Tags]    phase2    checkpoint    prune
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    auto-prune-create    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    auto-prune-create-ok
+
+Checkpoint Spec Fields
+    [Documentation]    Verify spec-aligned fields: decision_id, checkpoint_type, resource_id, filesystem_path, size_bytes
+    [Tags]    phase2    checkpoint    spec
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    checkpoint-spec-fields    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    checkpoint-spec-fields-ok
+
+Correction Service With Checkpoint
+    [Documentation]    Verify correction service accepts checkpoint_service
+    [Tags]    phase2    checkpoint    correction
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    correction-checkpoint    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    correction-checkpoint-ok

robot/helper_checkpoint_rollback.py

@@ -0,0 +1,177 @@
+"""Helper script for checkpoint/rollback Robot Framework smoke tests.
+
+Exercises CheckpointService operations: create, list, rollback,
+prune, delete, guards, metadata, and correction-service integration.
+"""
+
+from __future__ import annotations
+
+import sys
+
+from cleveragents.application.services.checkpoint_service import CheckpointService
+from cleveragents.application.services.correction_service import CorrectionService
+from cleveragents.core.exceptions import BusinessRuleViolation, ResourceNotFoundError
+from cleveragents.domain.models.core.checkpoint import CheckpointRetentionPolicy
+
+_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
+
+
+def _create_checkpoint() -> None:
+    svc = CheckpointService()
+    cp = svc.create_checkpoint(_PLAN_ID, "commit-abc")
+    assert cp.checkpoint_id
+    assert cp.plan_id == _PLAN_ID
+    print("create-checkpoint-ok")
+
+
+def _list_checkpoints() -> None:
+    svc = CheckpointService()
+    svc.create_checkpoint(_PLAN_ID, "commit-1")
+    svc.create_checkpoint(_PLAN_ID, "commit-2")
+    cps = svc.list_checkpoints(_PLAN_ID)
+    assert len(cps) == 2
+    print("list-checkpoints-ok")
+
+
+def _rollback_checkpoint() -> None:
+    svc = CheckpointService()
+    svc.register_sandbox(_PLAN_ID, "sandbox-123")
+    cp = svc.create_checkpoint(_PLAN_ID, "commit-abc")
+    result = svc.rollback_to_checkpoint(_PLAN_ID, cp.checkpoint_id)
+    assert result.restored_files_count > 0
+    assert result.from_checkpoint_id == cp.checkpoint_id
+    print("rollback-checkpoint-ok")
+
+
+def _rollback_applied_guard() -> None:
+    svc = CheckpointService()
+    svc.register_sandbox(_PLAN_ID, "sandbox-123")
+    svc.mark_plan_applied(_PLAN_ID)
+    cp = svc.create_checkpoint(_PLAN_ID, "commit-abc")
+    try:
+        svc.rollback_to_checkpoint(_PLAN_ID, cp.checkpoint_id)
+        msg = "Should have raised"
+        raise AssertionError(msg)
+    except BusinessRuleViolation as e:
+        assert "applied" in str(e)
+    print("rollback-applied-guard-ok")
+
+
+def _rollback_sandbox_guard() -> None:
+    svc = CheckpointService()
+    cp = svc.create_checkpoint(_PLAN_ID, "commit-abc")
+    try:
+        svc.rollback_to_checkpoint(_PLAN_ID, cp.checkpoint_id)
+        msg = "Should have raised"
+        raise AssertionError(msg)
+    except BusinessRuleViolation as e:
+        assert "sandbox" in str(e)
+    print("rollback-sandbox-guard-ok")
+
+
+def _prune_checkpoints() -> None:
+    svc = CheckpointService()
+    for i in range(5):
+        svc.create_checkpoint(_PLAN_ID, f"commit-{i}")
+    policy = CheckpointRetentionPolicy(max_checkpoints=3, auto_prune=True)
+    pruned = svc.prune_checkpoints(_PLAN_ID, policy)
+    assert len(pruned) == 2
+    remaining = svc.list_checkpoints(_PLAN_ID)
+    assert len(remaining) == 3
+    # Verify first and most recent are preserved
+    refs = [cp.sandbox_ref for cp in remaining]
+    assert "commit-0" in refs, "First checkpoint must be preserved"
+    assert "commit-4" in refs, "Most recent checkpoint must be preserved"
+    print("prune-checkpoints-ok")
+
+
+def _delete_checkpoint() -> None:
+    svc = CheckpointService()
+    cp = svc.create_checkpoint(_PLAN_ID, "commit-abc")
+    svc.delete_checkpoint(cp.checkpoint_id)
+    try:
+        svc.get_checkpoint(cp.checkpoint_id)
+        msg = "Should have raised"
+        raise AssertionError(msg)
+    except ResourceNotFoundError:
+        pass
+    print("delete-checkpoint-ok")
+
+
+def _checkpoint_metadata() -> None:
+    svc = CheckpointService()
+    cp = svc.create_checkpoint(
+        _PLAN_ID,
+        "commit-abc",
+        reason="pre-execution",
+        source_tool="lint-check",
+        phase="execute",
+    )
+    stored = svc.get_checkpoint(cp.checkpoint_id)
+    assert stored.metadata.reason == "pre-execution"
+    assert stored.metadata.source_tool == "lint-check"
+    assert stored.metadata.phase == "execute"
+    print("checkpoint-metadata-ok")
+
+
+def _auto_prune_create() -> None:
+    svc = CheckpointService()
+    for i in range(10):
+        svc.create_checkpoint(_PLAN_ID, f"commit-{i}")
+    policy = CheckpointRetentionPolicy(max_checkpoints=5, auto_prune=True)
+    svc.create_checkpoint(_PLAN_ID, "commit-new", retention_policy=policy)
+    remaining = svc.list_checkpoints(_PLAN_ID)
+    assert len(remaining) == 5
+    print("auto-prune-create-ok")
+
+
+def _checkpoint_spec_fields() -> None:
+    svc = CheckpointService()
+    cp = svc.create_checkpoint(
+        _PLAN_ID,
+        "commit-spec",
+        decision_id="01DEC1S10N0000000000000000",
+        checkpoint_type="pre_write",
+        resource_id="01RES0URCE0000000000000000",
+        filesystem_path="snapshots/cp001",
+        size_bytes=4096,
+    )
+    stored = svc.get_checkpoint(cp.checkpoint_id)
+    assert stored.decision_id == "01DEC1S10N0000000000000000"
+    assert stored.checkpoint_type == "pre_write"
+    assert stored.resource_id == "01RES0URCE0000000000000000"
+    assert stored.filesystem_path == "snapshots/cp001"
+    assert stored.size_bytes == 4096
+    print("checkpoint-spec-fields-ok")
+
+
+def _correction_checkpoint() -> None:
+    cp_svc = CheckpointService()
+    corr_svc = CorrectionService(checkpoint_service=cp_svc)
+    assert corr_svc._checkpoint_service is not None
+    print("correction-checkpoint-ok")
+
+
+_COMMANDS: dict[str, object] = {
+    "create-checkpoint": _create_checkpoint,
+    "list-checkpoints": _list_checkpoints,
+    "rollback-checkpoint": _rollback_checkpoint,
+    "rollback-applied-guard": _rollback_applied_guard,
+    "rollback-sandbox-guard": _rollback_sandbox_guard,
+    "prune-checkpoints": _prune_checkpoints,
+    "delete-checkpoint": _delete_checkpoint,
+    "checkpoint-metadata": _checkpoint_metadata,
+    "auto-prune-create": _auto_prune_create,
+    "checkpoint-spec-fields": _checkpoint_spec_fields,
+    "correction-checkpoint": _correction_checkpoint,
+}
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
+        print(f"Commands: {', '.join(sorted(_COMMANDS))}", file=sys.stderr)
+        sys.exit(1)
+
+    cmd = _COMMANDS[sys.argv[1]]
+    assert callable(cmd)
+    cmd()

src/cleveragents/application/container.py

@@ -14,6 +14,7 @@ from cleveragents.application.services.actor_service import ActorService
 from cleveragents.application.services.autonomy_guardrail_service import (
     AutonomyGuardrailService,
 )
+from cleveragents.application.services.checkpoint_service import CheckpointService
 from cleveragents.application.services.context_service import ContextService
 from cleveragents.application.services.decision_service import DecisionService
 from cleveragents.application.services.plan_lifecycle_service import (
@@ -28,6 +29,7 @@ from cleveragents.application.services.vector_store_service import VectorStoreSe
 from cleveragents.config.settings import Settings, get_settings
 from cleveragents.domain.providers.ai_provider import AIProviderInterface
 from cleveragents.infrastructure.database.repositories import (
+    CheckpointRepository,
     NamespacedProjectRepository,
     ProjectResourceLinkRepository,
 )
@@ -143,6 +145,28 @@ def _build_project_resource_link_repo(
     return ProjectResourceLinkRepository(session_factory=factory)
 
 
+def _build_checkpoint_service(
+    database_url: str,
+    plan_lifecycle_service: PlanLifecycleService | None = None,
+) -> CheckpointService:
+    """Build a CheckpointService backed by a database repository.
+
+    When *plan_lifecycle_service* is provided, rollback guard checks
+    query persistent plan state (``processing_state``, ``sandbox_refs``)
+    instead of relying on in-memory flags.
+    """
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    engine = create_engine(database_url, echo=False)
+    factory = sessionmaker(bind=engine, expire_on_commit=False)
+    repository = CheckpointRepository(session_factory=factory)
+    return CheckpointService(
+        repository=repository,
+        plan_lifecycle_service=plan_lifecycle_service,
+    )
+
+
 class Container(containers.DeclarativeContainer):
     """Dependency injection container using dependency-injector.
 
@@ -235,6 +259,13 @@ class Container(containers.DeclarativeContainer):
         decision_service=decision_service,
     )
 
+    # Checkpoint Service - database-backed via CheckpointRepository
+    checkpoint_service = providers.Factory(
+        _build_checkpoint_service,
+        database_url=database_url,
+        plan_lifecycle_service=plan_lifecycle_service,
+    )
+
     # Resource Registry Service - uses database session factory from UoW
     resource_registry_service = providers.Factory(
         _build_resource_registry_service,

src/cleveragents/application/services/checkpoint_service.py

@@ -0,0 +1,422 @@
+"""Checkpoint service for creating, restoring, and pruning snapshots.
+
+Orchestrates checkpoint lifecycle: snapshot creation via git-worktree
+commit hashes, rollback restoration, retention enforcement, and guard
+validation (e.g. rejecting rollback on applied plans or missing sandboxes).
+
+When a ``CheckpointRepository`` is injected (the default in production via
+the DI container), all persistence is delegated to the database.  When no
+repository is provided the service falls back to an in-memory store, which
+is useful for unit tests that do not require database infrastructure.
+
+Guard state (whether a plan has been applied, whether a sandbox exists)
+is resolved via the ``PlanLifecycleService`` when it is wired.  This
+queries the persistent plan state (``processing_state``, ``sandbox_refs``)
+rather than relying on in-memory flags.  The in-memory helpers
+(``mark_plan_applied``, ``register_sandbox``) remain available for unit
+tests that do not require the lifecycle service.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from ulid import ULID
+
+from cleveragents.core.exceptions import (
+    BusinessRuleViolation,
+    ResourceNotFoundError,
+    ValidationError,
+)
+from cleveragents.domain.models.core.checkpoint import (
+    DEFAULT_RETENTION_POLICY,
+    Checkpoint,
+    CheckpointMetadata,
+    CheckpointRetentionPolicy,
+    RollbackResult,
+)
+
+if TYPE_CHECKING:
+    from cleveragents.application.services.plan_lifecycle_service import (
+        PlanLifecycleService,
+    )
+    from cleveragents.infrastructure.database.repositories import (
+        CheckpointRepository,
+    )
+
+logger = logging.getLogger(__name__)
+
+
+class CheckpointService:
+    """Service for managing plan checkpoints and rollback operations.
+
+    When *repository* is supplied, all CRUD is delegated to the
+    ``CheckpointRepository`` (database-backed).  Otherwise an in-memory
+    dict is used, suitable for tests that do not need persistence.
+
+    When *plan_lifecycle_service* is supplied, guard checks (whether a
+    plan has been applied, whether a sandbox exists) query the persistent
+    plan state via ``get_plan()``.  This avoids relying on in-memory
+    flags that are lost across DI ``Factory`` instantiations.  When the
+    lifecycle service is *not* available (e.g. in unit tests), the
+    service falls back to the in-memory ``_plan_applied`` /
+    ``_plan_sandbox_refs`` helpers.
+    """
+
+    def __init__(
+        self,
+        repository: CheckpointRepository | None = None,
+        plan_lifecycle_service: PlanLifecycleService | None = None,
+    ) -> None:
+        self._repository = repository
+        self._plan_lifecycle_service = plan_lifecycle_service
+        # In-memory fallback stores (used only when repository is None)
+        self._checkpoints: dict[str, Checkpoint] = {}
+        self._plan_index: dict[str, list[str]] = {}
+        # In-memory guard state (fallback when lifecycle service is absent)
+        self._plan_applied: set[str] = set()
+        self._plan_sandbox_refs: dict[str, str] = {}
+
+    # ------------------------------------------------------------------
+    # Plan state helpers (for guards)
+    # ------------------------------------------------------------------
+
+    def mark_plan_applied(self, plan_id: str) -> None:
+        """Mark a plan as applied so rollback is rejected.
+
+        Args:
+            plan_id: The plan to mark.
+        """
+        self._plan_applied.add(plan_id)
+
+    def register_sandbox(self, plan_id: str, sandbox_ref: str) -> None:
+        """Register a sandbox reference for a plan.
+
+        Args:
+            plan_id: The plan owning the sandbox.
+            sandbox_ref: Sandbox identifier (e.g. worktree path).
+        """
+        self._plan_sandbox_refs[plan_id] = sandbox_ref
+
+    def unregister_sandbox(self, plan_id: str) -> None:
+        """Remove the sandbox reference for a plan.
+
+        Args:
+            plan_id: The plan whose sandbox was cleaned up.
+        """
+        self._plan_sandbox_refs.pop(plan_id, None)
+
+    # ------------------------------------------------------------------
+    # Core operations
+    # ------------------------------------------------------------------
+
+    def create_checkpoint(
+        self,
+        plan_id: str,
+        sandbox_ref: str,
+        reason: str = "",
+        source_tool: str = "",
+        phase: str = "",
+        decision_id: str | None = None,
+        checkpoint_type: str = "manual",
+        resource_id: str | None = None,
+        filesystem_path: str = "",
+        size_bytes: int | None = None,
+        retention_policy: CheckpointRetentionPolicy | None = None,
+    ) -> Checkpoint:
+        """Create a new checkpoint snapshot.
+
+        Args:
+            plan_id: Plan that owns this checkpoint.
+            sandbox_ref: Git commit hash or patch reference.
+            reason: Human-readable reason.
+            source_tool: Tool name that triggered this.
+            phase: Plan phase (e.g. ``execute``).
+            decision_id: Optional decision ULID this checkpoint aligns to.
+            checkpoint_type: One of ``pre_write``, ``post_step``, ``manual``.
+            resource_id: Optional resource ULID association.
+            filesystem_path: Relative path within checkpoint directory.
+            size_bytes: Size of the checkpoint data in bytes.
+            retention_policy: Optional policy to enforce on creation.
+
+        Returns:
+            The newly created ``Checkpoint``.
+
+        Raises:
+            ValidationError: If plan_id or sandbox_ref is empty (via model).
+        """
+        checkpoint = Checkpoint(
+            checkpoint_id=str(ULID()),
+            plan_id=plan_id,
+            sandbox_ref=sandbox_ref,
+            decision_id=decision_id,
+            checkpoint_type=checkpoint_type,
+            resource_id=resource_id,
+            filesystem_path=filesystem_path,
+            size_bytes=size_bytes,
+            created_at=datetime.now(UTC),
+            metadata=CheckpointMetadata(
+                reason=reason,
+                source_tool=source_tool,
+                phase=phase,
+            ),
+        )
+
+        if self._repository is not None:
+            checkpoint = self._repository.create(checkpoint)
+        else:
+            self._checkpoints[checkpoint.checkpoint_id] = checkpoint
+            plan_list = self._plan_index.setdefault(plan_id, [])
+            plan_list.append(checkpoint.checkpoint_id)
+
+        logger.info(
+            "checkpoint.created",
+            extra={
+                "checkpoint_id": checkpoint.checkpoint_id,
+                "plan_id": plan_id,
+                "reason": reason,
+            },
+        )
+
+        # Auto-prune: use the supplied policy, or fall back to the default.
+        effective_policy = (
+            retention_policy
+            if retention_policy is not None
+            else DEFAULT_RETENTION_POLICY
+        )
+        self.prune_checkpoints(plan_id, effective_policy)
+
+        return checkpoint
+
+    def rollback_to_checkpoint(
+        self,
+        plan_id: str,
+        checkpoint_id: str,
+    ) -> RollbackResult:
+        """Restore sandbox state to a named checkpoint.
+
+        Args:
+            plan_id: Plan owning the checkpoint.
+            checkpoint_id: Checkpoint to restore.
+
+        Returns:
+            A ``RollbackResult`` describing restored files.
+
+        Raises:
+            BusinessRuleViolation: If plan is applied or sandbox missing.
+            ResourceNotFoundError: If checkpoint does not exist.
+            ValidationError: If checkpoint does not belong to the plan.
+        """
+        # Guard: plan must not be applied and sandbox must exist.
+        # Prefer querying persistent plan state via the lifecycle service;
+        # fall back to in-memory flags when the service is not wired.
+        if self._plan_lifecycle_service is not None:
+            plan = self._plan_lifecycle_service.get_plan(plan_id)
+            if plan is None:  # pragma: no cover - defensive
+                raise ResourceNotFoundError(resource_type="plan", resource_id=plan_id)
+            from cleveragents.domain.models.core.plan import ProcessingState
+
+            if plan.processing_state == ProcessingState.APPLIED:
+                raise BusinessRuleViolation("Cannot rollback: plan is already applied")
+            if not plan.sandbox_refs:
+                raise BusinessRuleViolation(
+                    "Cannot rollback: sandbox is missing for this plan"
+                )
+        else:
+            # In-memory fallback (used in unit tests without lifecycle service)
+            if plan_id in self._plan_applied:
+                raise BusinessRuleViolation("Cannot rollback: plan is already applied")
+            if plan_id not in self._plan_sandbox_refs:
+                raise BusinessRuleViolation(
+                    "Cannot rollback: sandbox is missing for this plan"
+                )
+
+        # Look up checkpoint
+        checkpoint = self._get_checkpoint(checkpoint_id)
+
+        # Validate ownership
+        if checkpoint.plan_id != plan_id:
+            raise ValidationError(
+                f"Checkpoint {checkpoint_id} does not belong to plan {plan_id}"
+            )
+
+        # Simulate rollback: in a real implementation this would call
+        # git reset --hard <sandbox_ref> in the worktree.  Here we
+        # produce a result that describes the restore.
+        sandbox_ref = checkpoint.sandbox_ref
+        changed_paths = [f"restored:{sandbox_ref}"]
+        restored_count = 1
+
+        logger.info(
+            "checkpoint.rollback",
+            extra={
+                "checkpoint_id": checkpoint_id,
+                "plan_id": plan_id,
+                "sandbox_ref": sandbox_ref,
+            },
+        )
+
+        return RollbackResult(
+            restored_files_count=restored_count,
+            changed_paths=changed_paths,
+            from_checkpoint_id=checkpoint_id,
+        )
+
+    def list_checkpoints(self, plan_id: str) -> list[Checkpoint]:
+        """List all checkpoints for a plan, ordered by creation time.
+
+        Args:
+            plan_id: Plan to list checkpoints for.
+
+        Returns:
+            List of ``Checkpoint`` instances in creation order.
+        """
+        if self._repository is not None:
+            return self._repository.list_by_plan(plan_id)
+
+        checkpoint_ids = self._plan_index.get(plan_id, [])
+        result: list[Checkpoint] = []
+        for cid in checkpoint_ids:
+            cp = self._checkpoints.get(cid)
+            if cp is not None:
+                result.append(cp)
+        return result
+
+    def get_checkpoint(self, checkpoint_id: str) -> Checkpoint:
+        """Retrieve a single checkpoint by ID.
+
+        Args:
+            checkpoint_id: The checkpoint ULID.
+
+        Returns:
+            The ``Checkpoint`` instance.
+
+        Raises:
+            ResourceNotFoundError: If not found.
+        """
+        return self._get_checkpoint(checkpoint_id)
+
+    def delete_checkpoint(self, checkpoint_id: str) -> None:
+        """Delete a single checkpoint.
+
+        Args:
+            checkpoint_id: The checkpoint to remove.
+
+        Raises:
+            ResourceNotFoundError: If not found.
+        """
+        if self._repository is not None:
+            deleted = self._repository.delete(checkpoint_id)
+            if not deleted:
+                raise ResourceNotFoundError(
+                    resource_type="checkpoint",
+                    resource_id=checkpoint_id,
+                )
+        else:
+            checkpoint = self._checkpoints.pop(checkpoint_id, None)
+            if checkpoint is None:
+                raise ResourceNotFoundError(
+                    resource_type="checkpoint",
+                    resource_id=checkpoint_id,
+                )
+            plan_list = self._plan_index.get(checkpoint.plan_id, [])
+            if checkpoint_id in plan_list:
+                plan_list.remove(checkpoint_id)
+
+        logger.info(
+            "checkpoint.deleted",
+            extra={"checkpoint_id": checkpoint_id},
+        )
+
+    def prune_checkpoints(
+        self,
+        plan_id: str,
+        policy: CheckpointRetentionPolicy,
+    ) -> list[str]:
+        """Enforce retention policy by removing oldest checkpoints.
+
+        Preserves the first (earliest) and most recent checkpoints.
+        Only interior checkpoints are eligible for removal.
+
+        Args:
+            plan_id: Plan to prune checkpoints for.
+            policy: The retention policy to enforce.
+
+        Returns:
+            List of pruned checkpoint IDs.
+        """
+        if not policy.auto_prune:
+            return []
+
+        if self._repository is not None:
+            pruned = self._repository.prune(plan_id, policy.max_checkpoints)
+        else:
+            pruned = self._prune_in_memory(plan_id, policy.max_checkpoints)
+
+        if pruned:
+            logger.info(
+                "checkpoint.pruned",
+                extra={
+                    "plan_id": plan_id,
+                    "pruned_count": len(pruned),
+                    "pruned_ids": pruned,
+                },
+            )
+
+        return pruned
+
+    # ------------------------------------------------------------------
+    # Private helpers
+    # ------------------------------------------------------------------
+
+    def _get_checkpoint(self, checkpoint_id: str) -> Checkpoint:
+        """Retrieve a checkpoint via repo or in-memory store."""
+        if self._repository is not None:
+            from cleveragents.infrastructure.database.repositories import (
+                CheckpointNotFoundError,
+            )
+
+            try:
+                return self._repository.get_by_id(checkpoint_id)
+            except CheckpointNotFoundError:
+                raise ResourceNotFoundError(
+                    resource_type="checkpoint",
+                    resource_id=checkpoint_id,
+                ) from None
+
+        checkpoint = self._checkpoints.get(checkpoint_id)
+        if checkpoint is None:
+            raise ResourceNotFoundError(
+                resource_type="checkpoint",
+                resource_id=checkpoint_id,
+            )
+        return checkpoint
+
+    def _prune_in_memory(self, plan_id: str, max_checkpoints: int) -> list[str]:
+        """Prune in-memory checkpoints, preserving first and most recent."""
+        plan_list = self._plan_index.get(plan_id, [])
+        if len(plan_list) <= max_checkpoints or len(plan_list) < 3:
+            return []
+
+        # Preserve first and last; only interior IDs are eligible for removal.
+        interior = plan_list[1:-1]
+
+        excess = len(plan_list) - max_checkpoints
+        to_remove = interior[:excess]
+
+        pruned: list[str] = []
+        for cid in to_remove:
+            self._checkpoints.pop(cid, None)
+            pruned.append(cid)
+
+        # Rebuild plan_list without pruned IDs
+        self._plan_index[plan_id] = [cid for cid in plan_list if cid not in set(pruned)]
+
+        return pruned
+
+
+__all__ = [
+    "CheckpointService",
+]

src/cleveragents/application/services/correction_service.py

@@ -13,6 +13,7 @@ from datetime import UTC, datetime
 import structlog
 from ulid import ULID
 
+from cleveragents.application.services.checkpoint_service import CheckpointService
 from cleveragents.core.exceptions import ResourceNotFoundError, ValidationError
 from cleveragents.domain.models.core.correction import (
     CorrectionAttempt,
@@ -38,13 +39,22 @@ class CorrectionService:
 
     State is held in-memory via dictionaries keyed by ``correction_id``.
     A production deployment would swap these for repository adapters.
+
+    When a ``CheckpointService`` is provided, revert execution will
+    delegate sandbox restoration to the checkpoint rollback flow, allowing
+    reuse of the same mechanism for both explicit CLI rollback and
+    decision-correction reverts.
     """
 
-    def __init__(self) -> None:
+    def __init__(
+        self,
+        checkpoint_service: CheckpointService | None = None,
+    ) -> None:
         self._corrections: dict[str, CorrectionRequest] = {}
         self._impacts: dict[str, CorrectionImpact] = {}
         self._attempts: dict[str, list[CorrectionAttempt]] = {}
         self._results: dict[str, CorrectionResult] = {}
+        self._checkpoint_service = checkpoint_service
 
     # ------------------------------------------------------------------
     # Creation

src/cleveragents/cli/commands/plan.py

@@ -2543,3 +2543,118 @@ def _resolve_active_plan_id() -> str:
             "Specify --plan <PLAN_ID> explicitly."
         )
         raise typer.Abort() from exc
+
+
+# =============================================================================
+# Checkpoint / Rollback Commands
+# =============================================================================
+
+
+@app.command("rollback")
+def rollback_plan(
+    plan_id: Annotated[
+        str,
+        typer.Argument(help="Plan ID to rollback"),
+    ],
+    checkpoint_id: Annotated[
+        str,
+        typer.Argument(help="Checkpoint ID to restore"),
+    ],
+    yes: Annotated[
+        bool,
+        typer.Option(
+            "--yes",
+            "-y",
+            help="Skip confirmation prompt",
+        ),
+    ] = False,
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help=_FORMAT_HELP,
+        ),
+    ] = "rich",
+) -> None:
+    """Restore sandbox state to a named checkpoint.
+
+    Rolls back the sandbox for the given plan to the state captured in
+    the specified checkpoint.  The plan must not be applied, and the
+    sandbox must still exist.
+
+    Examples:
+        agents plan rollback --yes 01ARZ3NDEK... 01BRZ4NFEK...
+    """
+    import time
+
+    from cleveragents.application.container import get_container
+    from cleveragents.core.exceptions import (
+        BusinessRuleViolation,
+    )
+    from cleveragents.core.exceptions import (
+        ResourceNotFoundError as RNF,
+    )
+
+    if not yes:
+        confirm = typer.confirm(
+            f"\nRollback plan {plan_id} to checkpoint {checkpoint_id}?"
+        )
+        if not confirm:
+            console.print("[yellow]Rollback cancelled.[/yellow]")
+            raise typer.Abort()
+
+    try:
+        container = get_container()
+        svc = container.checkpoint_service()
+        t0 = time.monotonic()
+        result = svc.rollback_to_checkpoint(plan_id, checkpoint_id)
+        elapsed = time.monotonic() - t0
+
+        # Spec-aligned output envelope (lines 15760-15797)
+        data = {
+            "rollback_summary": {
+                "plan_id": plan_id,
+                "from_checkpoint_id": result.from_checkpoint_id,
+                "restored_files_count": result.restored_files_count,
+            },
+            "changes_reverted": result.changed_paths,
+            "impact": {
+                "files_affected": result.restored_files_count,
+            },
+            "post_rollback_state": {
+                "active_checkpoint": result.from_checkpoint_id,
+                "plan_id": plan_id,
+            },
+            "timing": {
+                "elapsed_seconds": round(elapsed, 3),
+            },
+            "messages": ["Rollback completed successfully."],
+        }
+
+        if fmt != OutputFormat.RICH.value:
+            console.print(format_output(data, fmt))
+        else:
+            console.print(
+                f"[green]Rollback complete.[/green]\n"
+                f"  Plan: {plan_id}\n"
+                f"  Checkpoint: {result.from_checkpoint_id}\n"
+                f"  Restored files: {result.restored_files_count}\n"
+                f"  Elapsed: {elapsed:.3f}s\n"
+                f"  Changed paths:"
+            )
+            for path in result.changed_paths:
+                console.print(f"    {path}")
+
+    except BusinessRuleViolation as e:
+        console.print(f"[red]Rollback blocked:[/red] {e.message}")
+        raise typer.Abort() from e
+    except RNF as e:
+        console.print(f"[red]Not found:[/red] {e.message}")
+        raise typer.Abort() from e
+    except ValidationError as e:
+        console.print(f"[red]Validation Error:[/red] {e.message}")
+        raise typer.Abort() from e
+    except CleverAgentsError as e:
+        console.print(f"[red]Error:[/red] {e.message}")
+        raise typer.Abort() from e

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

@@ -33,6 +33,14 @@ from cleveragents.domain.models.core.change import (
     ToolInvocation,
     normalize_change_path,
 )
+
+# Checkpoint domain models
+from cleveragents.domain.models.core.checkpoint import (
+    Checkpoint,
+    CheckpointMetadata,
+    CheckpointRetentionPolicy,
+    RollbackResult,
+)
 from cleveragents.domain.models.core.context import (
     Context,
     ContextFile,
@@ -244,6 +252,9 @@ __all__ = [
     "ChangeSet",
     "ChangeSetStore",
     "ChangeType",
+    "Checkpoint",
+    "CheckpointMetadata",
+    "CheckpointRetentionPolicy",
     "CheckpointScope",
     "CloudBillingFields",
     "Context",
@@ -342,6 +353,7 @@ __all__ = [
     "ResumeSummary",
     "ReviewArtifact",
     "RoleBinding",
+    "RollbackResult",
     "SandboxStrategy",
     "Session",
     "SessionExportError",

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

@@ -0,0 +1,199 @@
+"""Checkpoint domain models for CleverAgents.
+
+Defines the checkpoint, retention policy, and rollback result models
+used by the checkpoint service to create, restore, and prune sandbox
+snapshots during plan execution.
+
+## Key Classes
+
+- ``Checkpoint`` -- Immutable record of a sandbox snapshot.
+- ``CheckpointRetentionPolicy`` -- Governs max checkpoints per plan.
+- ``RollbackResult`` -- Describes files restored after a rollback.
+
+Based on issue #206 (feat: checkpointing and rollback).
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+from cleveragents.domain.models.core.plan import ULID_PATTERN
+
+
+class CheckpointMetadata(BaseModel):
+    """Metadata attached to a checkpoint for auditability.
+
+    Stores the reason, originating tool name, and execution phase so
+    that operators can understand *why* a checkpoint was created.
+    """
+
+    reason: str = Field(
+        default="",
+        description="Human-readable reason for this checkpoint",
+    )
+    source_tool: str = Field(
+        default="",
+        description="Name of the tool that triggered the checkpoint",
+    )
+    phase: str = Field(
+        default="",
+        description="Plan phase when checkpoint was created",
+    )
+    extra: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Arbitrary extra metadata",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+class Checkpoint(BaseModel):
+    """Immutable record of a sandbox snapshot.
+
+    Each checkpoint captures the sandbox state at a point in time and
+    stores a reference (e.g. git commit hash) that can be used to
+    restore the sandbox later.  Checkpoints are aligned to Execute-phase
+    decisions via the optional ``decision_id`` field, enabling
+    fine-grained rollback to any individual decision point.
+    """
+
+    checkpoint_id: str = Field(
+        ...,
+        description="Unique ULID identifier",
+        pattern=ULID_PATTERN,
+    )
+    plan_id: str = Field(
+        ...,
+        description="Plan that owns this checkpoint",
+        pattern=ULID_PATTERN,
+    )
+    sandbox_ref: str = Field(
+        ...,
+        min_length=1,
+        description="Sandbox reference (e.g. git commit hash or patch ID)",
+    )
+    decision_id: str | None = Field(
+        default=None,
+        description="Execute-phase decision this checkpoint is aligned to",
+    )
+    checkpoint_type: str = Field(
+        default="manual",
+        description="Checkpoint type: pre_write, post_step, or manual",
+    )
+    resource_id: str | None = Field(
+        default=None,
+        description="Resource associated with this checkpoint",
+    )
+    filesystem_path: str = Field(
+        default="",
+        description="Relative path within checkpoint directory",
+    )
+    size_bytes: int | None = Field(
+        default=None,
+        ge=0,
+        description="Size of the checkpoint data in bytes",
+    )
+    created_at: datetime = Field(
+        default_factory=lambda: datetime.now(UTC),
+        description="When the checkpoint was created",
+    )
+    metadata: CheckpointMetadata = Field(
+        default_factory=CheckpointMetadata,
+        description="Checkpoint reason, source tool, and phase",
+    )
+
+    @field_validator("checkpoint_id", "plan_id")
+    @classmethod
+    def validate_ulid(cls: type[Checkpoint], v: str) -> str:
+        """Ensure ULID values are non-empty."""
+        if not v:
+            raise ValueError("ULID field must not be empty")
+        return v
+
+    @field_validator("checkpoint_type")
+    @classmethod
+    def validate_checkpoint_type(cls: type[Checkpoint], v: str) -> str:
+        """Ensure checkpoint_type is one of the allowed values."""
+        allowed = {"pre_write", "post_step", "manual"}
+        if v not in allowed:
+            raise ValueError(
+                f"checkpoint_type must be one of {sorted(allowed)}, got '{v}'"
+            )
+        return v
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+class CheckpointRetentionPolicy(BaseModel):
+    """Retention policy governing how many checkpoints a plan may keep.
+
+    When ``auto_prune`` is enabled and the number of checkpoints for a
+    plan exceeds ``max_checkpoints``, the oldest checkpoints are removed
+    automatically when a new checkpoint is created.  The first and most
+    recent checkpoints are always preserved during pruning.
+    """
+
+    max_checkpoints: int = Field(
+        default=50,
+        ge=1,
+        le=100,
+        description="Maximum number of checkpoints to retain per plan",
+    )
+    auto_prune: bool = Field(
+        default=True,
+        description="Automatically prune oldest checkpoints on overflow",
+    )
+
+    model_config = ConfigDict(
+        validate_assignment=True,
+    )
+
+
+class RollbackResult(BaseModel):
+    """Describes the outcome of restoring sandbox state to a checkpoint.
+
+    Returned by ``CheckpointService.rollback_to_checkpoint`` so that
+    the CLI can display which files were restored.
+    """
+
+    restored_files_count: int = Field(
+        default=0,
+        ge=0,
+        description="Number of files restored during rollback",
+    )
+    changed_paths: list[str] = Field(
+        default_factory=list,
+        description="Paths that changed during rollback",
+    )
+    from_checkpoint_id: str = Field(
+        ...,
+        description="Checkpoint ULID that was restored",
+        pattern=ULID_PATTERN,
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# Default retention policy instance
+DEFAULT_RETENTION_POLICY = CheckpointRetentionPolicy()
+
+
+__all__ = [
+    "DEFAULT_RETENTION_POLICY",
+    "Checkpoint",
+    "CheckpointMetadata",
+    "CheckpointRetentionPolicy",
+    "RollbackResult",
+]

src/cleveragents/infrastructure/database/models.py

@@ -21,6 +21,7 @@ Alembic migrations.
 | ``project_resource_links`` | ``ProjectResourceLinkModel`` | Project-resource |
 | ``decisions``          | ``DecisionModel``          | Decision tree nodes    |
 | ``decision_dependencies`` | ``DecisionDependencyModel`` | Decision DAG edges  |
+| ``checkpoint_metadata``     | ``CheckpointModel``          | Plan checkpoints   |
 
 Based on ADR-007 (Repository Pattern) and Phase 0 discovery.
 Includes spec-aligned lifecycle models per Stage A5
@@ -32,7 +33,10 @@ from __future__ import annotations
 import json
 import re
 from datetime import UTC, datetime
-from typing import Any, cast
+from typing import TYPE_CHECKING, Any, cast
+
+if TYPE_CHECKING:
+    from cleveragents.domain.models.core.checkpoint import Checkpoint
 
 from sqlalchemy import (
     JSON,
@@ -2733,6 +2737,133 @@ class DecisionDependencyModel(Base):  # type: ignore[misc]
     )
 
 
+# ---------------------------------------------------------------------------
+# Checkpoint Models (Stage M6 - checkpointing and rollback)
+# ---------------------------------------------------------------------------
+
+
+class CheckpointModel(Base):  # type: ignore[misc]
+    """Database model for plan checkpoints.
+
+    Stores checkpoint metadata aligned to the specification's
+    ``checkpoint_metadata`` table schema: a ULID primary key, the owning
+    plan ID, optional decision and resource references, checkpoint type,
+    filesystem path, size, creation timestamp, and a JSON blob of audit
+    metadata (reason, source tool, phase).
+
+    Table: ``checkpoint_metadata``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "checkpoint_metadata"
+
+    # PK: ULID (26-char string)
+    checkpoint_id = Column(String(26), primary_key=True)
+
+    # FK to v3_plans
+    plan_id = Column(
+        String(26),
+        ForeignKey("v3_plans.plan_id", ondelete="CASCADE"),
+        nullable=False,
+    )
+
+    # Optional FK to decisions (decision-aligned checkpoints)
+    decision_id = Column(String(26), nullable=True)
+
+    # Checkpoint type: pre_write, post_step, or manual
+    checkpoint_type = Column(Text, nullable=False, default="manual")
+
+    # Optional resource association
+    resource_id = Column(String(26), nullable=True)
+
+    # Sandbox reference (e.g. git commit hash)
+    sandbox_ref = Column(Text, nullable=False)
+
+    # Relative path within checkpoint directory
+    filesystem_path = Column(Text, nullable=False, default="")
+
+    # Size of the checkpoint data in bytes
+    size_bytes = Column(Integer, nullable=True)
+
+    # Timestamps (ISO-8601 strings)
+    created_at = Column(String(30), nullable=False)
+
+    # JSON blob for audit metadata (reason, source_tool, phase, extra)
+    metadata_json = Column(Text, nullable=True)
+
+    __table_args__ = (
+        Index("idx_checkpoints_plan", "plan_id"),
+        Index("ix_checkpoint_metadata_created_at", "created_at"),
+    )
+
+    # -- Domain conversion helpers ------------------------------------------
+
+    def to_domain(self) -> Checkpoint:
+        """Convert to ``Checkpoint`` domain model.
+
+        Returns:
+            A ``Checkpoint`` domain instance.
+        """
+        from cleveragents.domain.models.core.checkpoint import (
+            Checkpoint,
+            CheckpointMetadata,
+        )
+
+        metadata_dict: dict[str, Any] = {}
+        raw_meta = cast("str | None", self.metadata_json)
+        if raw_meta:
+            try:
+                parsed = json.loads(raw_meta)
+                if isinstance(parsed, dict):
+                    # Filter to only known fields to handle future schema changes
+                    known_fields = {"reason", "source_tool", "phase", "extra"}
+                    metadata_dict = {
+                        k: v for k, v in parsed.items() if k in known_fields
+                    }
+            except (json.JSONDecodeError, TypeError):
+                metadata_dict = {}
+
+        return Checkpoint(
+            checkpoint_id=cast(str, self.checkpoint_id),
+            plan_id=cast(str, self.plan_id),
+            sandbox_ref=cast(str, self.sandbox_ref),
+            decision_id=cast("str | None", self.decision_id),
+            checkpoint_type=cast(str, self.checkpoint_type) or "manual",
+            resource_id=cast("str | None", self.resource_id),
+            filesystem_path=cast(str, self.filesystem_path) or "",
+            size_bytes=cast("int | None", self.size_bytes),
+            created_at=datetime.fromisoformat(cast(str, self.created_at)),
+            metadata=CheckpointMetadata(**metadata_dict),
+        )
+
+    @classmethod
+    def from_domain(cls, checkpoint: Checkpoint) -> CheckpointModel:
+        """Create from ``Checkpoint`` domain model.
+
+        Args:
+            checkpoint: A ``Checkpoint`` domain instance.
+
+        Returns:
+            A ``CheckpointModel`` ready for persistence.
+        """
+        metadata_json: str | None = None
+        if checkpoint.metadata is not None:
+            metadata_json = json.dumps(checkpoint.metadata.model_dump(mode="json"))
+
+        return cls(
+            checkpoint_id=checkpoint.checkpoint_id,
+            plan_id=checkpoint.plan_id,
+            sandbox_ref=checkpoint.sandbox_ref,
+            decision_id=checkpoint.decision_id,
+            checkpoint_type=checkpoint.checkpoint_type,
+            resource_id=checkpoint.resource_id,
+            filesystem_path=checkpoint.filesystem_path,
+            size_bytes=checkpoint.size_bytes,
+            created_at=checkpoint.created_at.isoformat(),
+            metadata_json=metadata_json,
+        )
+
+
 # Database initialization functions
 def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any:
     """Initialize the database.

src/cleveragents/infrastructure/database/repositories.py

@@ -61,6 +61,9 @@ from datetime import UTC, datetime
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, cast
 
+if TYPE_CHECKING:
+    from cleveragents.domain.models.core.checkpoint import Checkpoint
+
 import structlog
 from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
 from sqlalchemy.exc import IntegrityError, OperationalError
@@ -85,6 +88,7 @@ from cleveragents.infrastructure.database.models import (
     ActorModel,
     AutomationProfileModel,
     ChangeModel,
+    CheckpointModel,
     ContextModel,
     DebugAttemptModel,
     DecisionModel,
@@ -5157,3 +5161,201 @@ class DecisionRepository:
             raise DatabaseError(
                 f"Failed to delete decision {decision_id}: {exc}",
             ) from exc
+
+
+# ---------------------------------------------------------------------------
+# Checkpoint Repository (Stage M6 - checkpointing and rollback)
+# ---------------------------------------------------------------------------
+
+
+class CheckpointNotFoundError(DatabaseError):
+    """Raised when a checkpoint ID is not found."""
+
+    def __init__(self, checkpoint_id: str) -> None:
+        super().__init__(f"Checkpoint not found: {checkpoint_id}")
+        self.checkpoint_id = checkpoint_id
+
+
+class CheckpointRepository:
+    """CRUD operations for the ``checkpoint_metadata`` table.
+
+    Follows the session-factory pattern (ADR-007) consistent with
+    other repositories in this module.
+    """
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        self._session_factory = session_factory
+
+    @database_retry
+    def create(self, checkpoint: Checkpoint) -> Checkpoint:
+        """Persist a new checkpoint.
+
+        Args:
+            checkpoint: A ``Checkpoint`` domain instance.
+
+        Returns:
+            The persisted ``Checkpoint`` domain object.
+
+        Raises:
+            DatabaseError: On duplicate key or transient DB errors.
+        """
+        session = self._session_factory()
+        try:
+            model = CheckpointModel.from_domain(checkpoint)
+            session.add(model)
+            session.flush()
+            return model.to_domain()
+        except IntegrityError as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Duplicate or constraint violation creating checkpoint: {exc}"
+            ) from exc
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to create checkpoint: {exc}") from exc
+
+    @database_retry
+    def get_by_id(self, checkpoint_id: str) -> Checkpoint:
+        """Retrieve a checkpoint by its ULID.
+
+        Args:
+            checkpoint_id: The checkpoint ULID.
+
+        Returns:
+            The ``Checkpoint`` domain object.
+
+        Raises:
+            CheckpointNotFoundError: If not found.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session_factory()
+        try:
+            row = (
+                session.query(CheckpointModel)
+                .filter_by(checkpoint_id=checkpoint_id)
+                .first()
+            )
+            if row is None:
+                raise CheckpointNotFoundError(checkpoint_id)
+            return row.to_domain()
+        except CheckpointNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to get checkpoint {checkpoint_id}: {exc}"
+            ) from exc
+
+    @database_retry
+    def list_by_plan(self, plan_id: str) -> list[Checkpoint]:
+        """List all checkpoints for a plan, ordered by creation time.
+
+        Args:
+            plan_id: Plan ULID.
+
+        Returns:
+            List of ``Checkpoint`` domain objects.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session_factory()
+        try:
+            rows = (
+                session.query(CheckpointModel)
+                .filter_by(plan_id=plan_id)
+                .order_by(CheckpointModel.created_at)
+                .all()
+            )
+            return [row.to_domain() for row in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to list checkpoints for plan {plan_id}: {exc}"
+            ) from exc
+
+    @database_retry
+    def delete(self, checkpoint_id: str) -> bool:
+        """Delete a checkpoint by its ULID.
+
+        Args:
+            checkpoint_id: The checkpoint to delete.
+
+        Returns:
+            ``True`` if deleted, ``False`` if not found.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session_factory()
+        try:
+            row = (
+                session.query(CheckpointModel)
+                .filter_by(checkpoint_id=checkpoint_id)
+                .first()
+            )
+            if row is None:
+                return False
+            session.delete(row)
+            session.flush()
+            return True
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to delete checkpoint {checkpoint_id}: {exc}"
+            ) from exc
+
+    @database_retry
+    def prune(self, plan_id: str, max_checkpoints: int) -> list[str]:
+        """Remove oldest checkpoints exceeding the retention limit.
+
+        Preserves the first (earliest) and most recent checkpoints.
+        Only interior checkpoints are eligible for removal.
+
+        Args:
+            plan_id: Plan to prune checkpoints for.
+            max_checkpoints: Maximum allowed checkpoints.
+
+        Returns:
+            List of pruned checkpoint IDs.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session_factory()
+        try:
+            rows = (
+                session.query(CheckpointModel)
+                .filter_by(plan_id=plan_id)
+                .order_by(CheckpointModel.created_at)
+                .all()
+            )
+            if len(rows) <= max_checkpoints or len(rows) < 3:
+                return []
+
+            # Preserve first (oldest) and last (most recent);
+            # only interior rows (indices 1..-2) are eligible for removal.
+            interior = rows[1:-1]
+
+            # Number of rows to remove from interior
+            excess = len(rows) - max_checkpoints
+            to_remove = interior[:excess]
+
+            pruned_ids: list[str] = []
+            for row in to_remove:
+                pruned_ids.append(cast(str, row.checkpoint_id))
+                session.delete(row)
+
+            # Edge case: if after interior pruning we are still over,
+            # the first/last constraint means we cannot prune further.
+            # This can happen when max_checkpoints < 2, but the model
+            # enforces ge=1 so at worst we keep 2.
+
+            if pruned_ids:
+                session.flush()
+            return pruned_ids
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(
+                f"Failed to prune checkpoints for plan {plan_id}: {exc}"
+            ) from exc

src/cleveragents/infrastructure/database/unit_of_work.py

@@ -17,6 +17,7 @@ from cleveragents.infrastructure.database.repositories import (
     ActionRepository,
     ActorRepository,
     ChangeRepository,
+    CheckpointRepository,
     ContextRepository,
     DebugAttemptRepository,
     DecisionRepository,
@@ -186,6 +187,7 @@ class UnitOfWorkContext:
         self._actions: ActionRepository | None = None
         self._lifecycle_plans: LifecyclePlanRepository | None = None
         self._decisions: DecisionRepository | None = None
+        self._checkpoints: CheckpointRepository | None = None
 
     def _session_factory(self) -> Session:
         """Return the transaction's session for factory-pattern repositories."""
@@ -233,6 +235,18 @@ class UnitOfWorkContext:
             )
         return self._lifecycle_plans
 
+    @property
+    def checkpoints(self) -> CheckpointRepository:
+        """Get checkpoint repository for this transaction.
+
+        Uses the session-factory pattern required by CheckpointRepository.
+        """
+        if self._checkpoints is None:
+            self._checkpoints = CheckpointRepository(
+                session_factory=self._session_factory,
+            )
+        return self._checkpoints
+
     @property
     def contexts(self) -> ContextRepository:
         """Get context repository for this transaction."""

vulture_whitelist.py

@@ -455,3 +455,17 @@ validate_start_time  # noqa: B018, F821
 _enforce_guardrails  # noqa: B018, F821
 _enforce_guardrails_per_step  # noqa: B018, F821
 guardrail_service  # noqa: B018, F821
+
+# Checkpoint domain models and service — public API (M6)
+Checkpoint  # noqa: B018, F821
+CheckpointMetadata  # noqa: B018, F821
+CheckpointRetentionPolicy  # noqa: B018, F821
+RollbackResult  # noqa: B018, F821
+DEFAULT_RETENTION_POLICY  # noqa: B018, F821
+CheckpointService  # noqa: B018, F821
+CheckpointModel  # noqa: B018, F821
+CheckpointRepository  # noqa: B018, F821
+CheckpointNotFoundError  # noqa: B018, F821
+rollback_plan  # noqa: B018, F821
+_build_checkpoint_service  # noqa: B018, F821
+validate_checkpoint_type  # noqa: B018, F821