Merge branch 'master' into develop-brent-4

2026-02-25 20:05:18 +00:00
parent 583ff60bf3 c4f71e930d
commit d055b59d7c
14 changed files with 2034 additions and 54 deletions
@@ -0,0 +1,159 @@
+"""ASV benchmarks for plan resume overhead baseline.
+
+Measures the cost of:
+- Resume eligibility validation
+- Step checkpoint recording
+- Resume summary building (dry-run)
+- Live resume state transitions
+"""
+
+from __future__ import annotations
+
+from ulid import ULID
+
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.application.services.plan_resume_service import (
+    PlanResumeService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.core.plan import (
+    ProjectLink,
+)
+from cleveragents.domain.models.core.resume import ResumeMetadata
+
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_bench_counter = 9000
+
+
+def _bench_ulid() -> str:
+    global _bench_counter
+    _bench_counter += 1
+    n = _bench_counter
+    suffix = ""
+    for _ in range(8):
+        suffix = _CB32[n % 32] + suffix
+        n //= 32
+    return f"01HGZ6FE0AQDYTR4BX{suffix}"
+
+
+def _make_services() -> tuple[PlanLifecycleService, PlanResumeService]:
+    settings = Settings()
+    lifecycle = PlanLifecycleService(settings=settings)
+    resume = PlanResumeService(lifecycle_service=lifecycle)
+    return lifecycle, resume
+
+
+def _create_action(lifecycle: PlanLifecycleService, suffix: str) -> str:
+    name = f"local/bench-resume-{suffix}"
+    lifecycle.create_action(
+        name=name,
+        description="Benchmark resume action",
+        definition_of_done="Step 1\nStep 2\nStep 3",
+        strategy_actor="local/stub-strategy",
+        execution_actor="local/stub-execute",
+    )
+    return name
+
+
+def _make_errored_plan(lifecycle: PlanLifecycleService, action_name: str) -> str:
+    plan = lifecycle.use_action(
+        action_name=action_name,
+        project_links=[ProjectLink(project_name="local/bench-proj")],
+    )
+    pid = plan.identity.plan_id
+    lifecycle.start_strategize(pid)
+    p = lifecycle.get_plan(pid)
+    p.decision_root_id = str(ULID())
+    lifecycle._commit_plan(p)
+    lifecycle.complete_strategize(pid)
+    lifecycle.execute_plan(pid)
+    lifecycle.start_execute(pid)
+    lifecycle.fail_execute(pid, "bench error")
+    return pid
+
+
+class TimeResumeEligibility:
+    """Benchmark resume eligibility validation."""
+
+    timeout = 30.0
+
+    def setup(self) -> None:
+        self.lifecycle, self.resume = _make_services()
+        action = _create_action(self.lifecycle, _bench_ulid())
+        self.plan_id = _make_errored_plan(self.lifecycle, action)
+
+    def time_validate_eligibility(self) -> None:
+        self.resume.validate_eligibility(self.plan_id)
+
+
+class TimeCheckpointRecording:
+    """Benchmark step checkpoint recording."""
+
+    timeout = 30.0
+
+    def setup(self) -> None:
+        self.lifecycle, self.resume = _make_services()
+        action = _create_action(self.lifecycle, _bench_ulid())
+        plan = self.lifecycle.use_action(
+            action_name=action,
+            project_links=[ProjectLink(project_name="local/bench-proj")],
+        )
+        pid = plan.identity.plan_id
+        self.lifecycle.start_strategize(pid)
+        p = self.lifecycle.get_plan(pid)
+        p.decision_root_id = str(ULID())
+        self.lifecycle._commit_plan(p)
+        self.lifecycle.complete_strategize(pid)
+        self.lifecycle.execute_plan(pid)
+        self.lifecycle.start_execute(pid)
+        self.plan_id = pid
+        self.resume.set_total_steps(pid, 100)
+        self.step_counter = 0
+
+    def time_record_checkpoint(self) -> None:
+        idx = self.step_counter
+        self.step_counter += 1
+        if idx >= 100:
+            return
+        self.resume.record_step_checkpoint(
+            self.plan_id, idx, f"DEC-{idx}", f"Step {idx}"
+        )
+
+
+class TimeDryRunResume:
+    """Benchmark dry-run resume summary building."""
+
+    timeout = 30.0
+
+    def setup(self) -> None:
+        self.lifecycle, self.resume = _make_services()
+        action = _create_action(self.lifecycle, _bench_ulid())
+        self.plan_id = _make_errored_plan(self.lifecycle, action)
+        self.resume.set_total_steps(self.plan_id, 10)
+        self.resume.record_step_checkpoint(self.plan_id, 5, "DEC-5", "Step 5")
+
+    def time_dry_run_resume(self) -> None:
+        self.resume.resume_plan(self.plan_id, dry_run=True)
+
+
+class TimeResumeMetadataProperties:
+    """Benchmark ResumeMetadata property access."""
+
+    timeout = 30.0
+
+    def setup(self) -> None:
+        self.metadata = ResumeMetadata(
+            last_completed_step=42,
+            total_steps=100,
+        )
+
+    def time_has_progress(self) -> None:
+        _ = self.metadata.has_progress
+
+    def time_next_step_index(self) -> None:
+        _ = self.metadata.next_step_index
+
+    def time_is_complete(self) -> None:
+        _ = self.metadata.is_complete
@@ -103,6 +103,79 @@ Actions carry: `name` (namespaced), `short_description`, `long_description`, `de

 None — specification-driven requirement. The four-phase lifecycle with reversion, hierarchy, and automation-controlled transitions is the core orchestration model prescribed by the specification.

+## Plan Resume Behavior
+
+Plans that are interrupted (e.g., process shutdown) or fail mid-execution can
+be resumed from their last checkpoint using ``agents plan resume <plan_id>``.
+
+### Resume Eligibility
+
+Only non-terminal plans can be resumed. Terminal states that block resume:
+
+| State | Reason |
+|-------|--------|
+| ``applied`` | Plan has already been committed successfully |
+| ``cancelled`` | Plan was explicitly cancelled |
+| ``constrained`` | Plan cannot proceed within constraints |
+
+Plans in ``errored``, ``processing``, ``queued``, or ``complete`` state within
+the Strategize, Execute, or Apply phases are eligible for resume.
+
+### Resume Metadata
+
+Each plan tracks two resume-related fields:
+
+- ``last_completed_step`` -- zero-based index of the last step that completed successfully (-1 if none).
+- ``last_checkpoint_id`` -- ULID of the most recent ``ResumeCheckpoint``.
+
+Checkpoints are recorded at each completed execution step and are tied to:
+
+- A decision ID (from the strategy decision tree)
+- A sandbox reference (for filesystem state)
+
+### Resume Flow
+
+1. **Validate** -- ``PlanResumeService.validate_eligibility()`` checks that the plan is not in a terminal state.
+2. **Summarize** -- ``build_resume_summary()`` returns the phase, next step index, decision ID, and sandbox reference.
+3. **Dry-run** -- ``plan resume --dry-run`` shows the summary without changing state.
+4. **Live resume** -- Resets ``processing_state`` to ``PROCESSING`` (clearing any error), then the caller re-drives execution from step ``last_completed_step + 1``.
+
+### Graceful Shutdown
+
+When the system detects a shutdown signal, ``record_shutdown()`` persists the
+current step index and marks the resume metadata as ``interrupted``. This
+ensures that the next ``plan resume`` picks up exactly where execution stopped.
+
+### Example
+
+```
+$ agents plan resume 01HGZ6FE0AQDYTR4BXEXAMPLE --dry-run
+Resume Summary
+  Plan ID: 01HGZ6FE0AQDYTR4BXEXAMPLE
+  Phase: execute
+  State: errored
+  Next Step: 3 / 5
+  Decision ID: DEC-003
+  Checkpoint: 01HGZ6FE0AQDYTR4BXCHKPT03
+  No state changes made (dry-run mode).
+
+$ agents plan resume 01HGZ6FE0AQDYTR4BXEXAMPLE
+Resume Summary
+  Plan ID: 01HGZ6FE0AQDYTR4BXEXAMPLE
+  Phase: execute
+  State: processing
+  Next Step: 3 / 5
+  Plan resumed. Execution will continue from step 3.
+```
+
+### Error Cases
+
+| Scenario | Error |
+|----------|-------|
+| Resume applied plan | ``PlanError: Plan ... is in terminal state 'applied' and cannot be resumed.`` |
+| Resume cancelled plan | ``PlanError: Plan ... is in terminal state 'cancelled' and cannot be resumed.`` |
+| Empty plan ID | ``ValidationError: plan_id must not be empty`` |
+
 ## Compliance

 - **Lifecycle state machine tests**: Unit tests verify all valid phase transitions and reject invalid ones (e.g., Action → Execute without passing through Strategize).
@@ -12,10 +12,10 @@ Feature: Plan Lifecycle Service coverage boost
  # namespaced_name string representation does.
  # ---------------------------------------------------------------

-  Scenario: get_action_by_name falls back to linear scan when dict key does not match directly
-    Given an action stored under a mismatched dict key
-    When I look up the action by its namespaced name via get_action_by_name
-    Then the action should be found via linear scan fallback
+  Scenario: get_action_by_name normalises case so mixed-case input finds lowercase-stored action
+    Given an action stored under its normalised lowercase key
+    When I look up the action with mixed-case input via get_action_by_name
+    Then the action should be found via case-insensitive lookup

  # ---------------------------------------------------------------
  # Line 420: use_action appends PlanInvariant from action.invariants
@@ -0,0 +1,152 @@
+Feature: Plan Resume
+  As a developer
+  I want to resume interrupted or errored plan executions
+  So that I don't lose progress when something goes wrong
+
+  Background:
+    Given I have a plan lifecycle service for resume tests
+    And I have a plan resume service
+
+  # Resume eligibility validation
+
+  Scenario: Eligible plan in errored state can be resumed
+    Given I have a resume-test plan in execute phase with errored state
+    When I validate resume eligibility
+    Then the plan should be eligible for resume
+
+  Scenario: Eligible plan in processing state can be resumed
+    Given I have a resume-test plan in execute phase with processing state
+    When I validate resume eligibility
+    Then the plan should be eligible for resume
+
+  Scenario: Eligible plan in queued state can be resumed
+    Given I have a resume-test plan in strategize phase with queued state
+    When I validate resume eligibility
+    Then the plan should be eligible for resume
+
+  Scenario: Terminal applied plan cannot be resumed
+    Given I have a resume-test plan in applied terminal state
+    When I validate resume eligibility
+    Then the plan should not be eligible for resume
+    And the ineligible reason should be "terminal_applied"
+
+  Scenario: Terminal cancelled plan cannot be resumed
+    Given I have a resume-test plan in cancelled terminal state
+    When I validate resume eligibility
+    Then the plan should not be eligible for resume
+    And the ineligible reason should be "terminal_cancelled"
+
+  Scenario: Terminal constrained plan cannot be resumed
+    Given I have a resume-test plan in constrained terminal state
+    When I validate resume eligibility
+    Then the plan should not be eligible for resume
+    And the ineligible reason should be "terminal_constrained"
+
+  Scenario: Action phase plan cannot be resumed
+    Given I have a resume-test plan in action phase
+    When I validate resume eligibility
+    Then the plan should not be eligible for resume
+    And the ineligible reason should be "action_phase"
+
+  # Resume metadata and checkpoints
+
+  Scenario: Record step checkpoint during execution
+    Given I have a resume-test plan in execute phase with processing state
+    When I record a resume step checkpoint at index 0 with decision "DEC001"
+    Then the resume-test plan should have last completed step 0
+    And the resume-test plan should have a checkpoint ID set
+
+  Scenario: Record multiple checkpoints
+    Given I have a resume-test plan in execute phase with processing state
+    When I record a resume step checkpoint at index 0 with decision "DEC001"
+    And I record a resume step checkpoint at index 1 with decision "DEC002"
+    And I record a resume step checkpoint at index 2 with decision "DEC003"
+    Then the resume-test plan should have last completed step 2
+    And the resume metadata should have 3 checkpoints
+
+  Scenario: Resume metadata tracks next step correctly
+    Given I have a resume-test plan in execute phase with processing state
+    When I record a resume step checkpoint at index 2 with decision "DEC003"
+    Then the resume metadata next step should be 3
+
+  # Resume summary (dry-run)
+
+  Scenario: Dry-run shows resume point without changing state
+    Given I have a resume-test plan in execute phase with errored state
+    And the resume-test plan has checkpoint at step 2 with decision "DEC003"
+    When I resume the plan with dry-run
+    Then I should get a resume summary
+    And the resume summary next step should be 3
+    And the resume-test plan processing state should still be errored
+
+  Scenario: Resume summary includes decision ID from checkpoint
+    Given I have a resume-test plan in execute phase with errored state
+    And the resume-test plan has checkpoint at step 1 with decision "DEC002"
+    When I resume the plan with dry-run
+    Then the resume summary decision ID should be "DEC002"
+
+  # Live resume
+
+  Scenario: Resume errored plan resets to processing
+    Given I have a resume-test plan in execute phase with errored state
+    And the resume-test plan has checkpoint at step 1 with decision "DEC002"
+    When I resume the plan without dry-run
+    Then the resume-test plan processing state should be processing
+    And the resume-test plan error message should be cleared
+
+  Scenario: Resume queued plan transitions to processing
+    Given I have a resume-test plan in strategize phase with queued state
+    When I resume the plan without dry-run
+    Then the resume-test plan processing state should be processing
+
+  # Graceful shutdown
+
+  Scenario: Record graceful shutdown marks interrupted
+    Given I have a resume-test plan in execute phase with processing state
+    When I record a resume step checkpoint at index 1 with decision "DEC002"
+    And I record a graceful shutdown for resume
+    Then the resume metadata should be marked as interrupted
+
+  # Error handling
+
+  Scenario: Resume with empty plan ID raises validation error
+    When I try to resume a plan with empty ID
+    Then a resume validation error should be raised
+
+  Scenario: Validate eligibility with empty plan ID raises error
+    When I try to validate eligibility with empty plan ID
+    Then a resume validation error should be raised
+
+  Scenario: Resume terminal plan raises plan error
+    Given I have a resume-test plan in applied terminal state
+    When I try to resume the terminal plan for resume test
+    Then a plan error should be raised for resume
+
+  # Step count tracking
+
+  Scenario: Set total steps for a plan
+    Given I have a resume-test plan in execute phase with processing state
+    When I set the resume total steps to 5
+    Then the resume metadata total steps should be 5
+
+  Scenario: Set total steps with invalid values raises error
+    When I try to set resume total steps with empty plan ID
+    Then a resume validation error should be raised
+
+  # ResumeMetadata model properties
+
+  Scenario: Resume metadata has_progress when steps completed
+    Given I have resume metadata with last completed step 2
+    Then the resume metadata has_progress should be true
+
+  Scenario: Resume metadata has no progress initially
+    Given I have fresh resume metadata
+    Then the resume metadata has_progress should be false
+
+  Scenario: Resume metadata is_complete when all steps done
+    Given I have resume metadata with 3 total steps and last step 2
+    Then the resume metadata is_complete should be true
+
+  Scenario: Resume metadata is not complete with remaining steps
+    Given I have resume metadata with 5 total steps and last step 2
+    Then the resume metadata is_complete should be false
@@ -1,7 +1,7 @@
 """Step definitions for plan_lifecycle_service_coverage_boost.feature.

 Targets uncovered lines and branches in PlanLifecycleService:
- Line 284:  get_action_by_name linear scan fallback
+- Line 284:  get_action / get_action_by_name case-insensitive lookup
 - Line 420:  use_action PlanInvariant append from action.invariants
 - Line 643:  execute_plan InvalidPhaseTransitionError
 - Lines 829-846: constrain_apply method (happy + wrong-phase)
@@ -83,36 +83,25 @@ def _create_plan_in_phase(context: Context, target_phase: PlanPhase):


 # =================================================================
-# Scenario: get_action_by_name linear scan fallback (line 284)
+# Scenario: get_action_by_name case-insensitive lookup (line 284)
 # =================================================================


-@given("an action stored under a mismatched dict key")
-def step_store_action_under_wrong_key(context: Context) -> None:
-    """Create an action normally, then re-key it so the dict lookup misses.
-
-    The action's ``namespaced_name`` stays correct, but the dict key
-    in ``_actions`` is changed to something different.  This forces
-    ``get_action_by_name`` to fall through to the linear scan.
-    """
+@given("an action stored under its normalised lowercase key")
+def step_store_action_under_normalised_key(context: Context) -> None:
+    """Create an action normally — it is stored under its lowercase key."""
    action = _create_action(context, "local/scan-target")
-    correct_key = str(action.namespaced_name)  # "local/scan-target"
-
-    # Remove from its correct key and store under a bogus key
-    del context.service._actions[correct_key]
-    context.service._actions["__mismatched_key__"] = action
-
    context.expected_action = action


-@when("I look up the action by its namespaced name via get_action_by_name")
+@when("I look up the action with mixed-case input via get_action_by_name")
 def step_lookup_via_get_action_by_name(context: Context) -> None:
-    """Call get_action_by_name with the real namespaced name."""
-    context.found_action = context.service.get_action_by_name("local/scan-target")
+    """Call get_action_by_name with mixed-case input."""
+    context.found_action = context.service.get_action_by_name("Local/Scan-Target")


-@then("the action should be found via linear scan fallback")
-def step_verify_linear_scan_found(context: Context) -> None:
+@then("the action should be found via case-insensitive lookup")
+def step_verify_case_insensitive_found(context: Context) -> None:
    """Verify the action returned is the one we stored."""
    assert context.found_action is not None
    assert context.found_action is context.expected_action
@@ -0,0 +1,446 @@
+"""Step definitions for Plan Resume feature tests."""
+
+from behave import given, then, when
+from behave.runner import Context
+from ulid import ULID
+
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.application.services.plan_resume_service import (
+    PlanResumeService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.core.exceptions import PlanError, ValidationError
+from cleveragents.domain.models.core.plan import (
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.domain.models.core.resume import ResumeMetadata
+
+# --- Background ---
+
+
+@given("I have a plan lifecycle service for resume tests")
+def step_create_resume_lifecycle_service(context: Context) -> None:
+    """Create lifecycle service for resume tests."""
+    settings = Settings()
+    context.lifecycle_service = PlanLifecycleService(settings=settings)
+    context.error = None
+    context.plan = None
+    context.plan_id = None
+    context.eligibility = None
+    context.summary = None
+
+
+@given("I have a plan resume service")
+def step_create_resume_service(context: Context) -> None:
+    """Create plan resume service."""
+    context.resume_service = PlanResumeService(
+        lifecycle_service=context.lifecycle_service,
+    )
+
+
+# --- Helper to create plans in specific states ---
+
+
+def _create_action(context: Context) -> str:
+    """Create and return an action name for testing."""
+    action_name = f"local/resume-test-{ULID()!s}"[:30].lower()
+    context.lifecycle_service.create_action(
+        name=action_name,
+        description="Resume test action",
+        definition_of_done="Step 1\nStep 2\nStep 3",
+        strategy_actor="local/stub-strategy",
+        execution_actor="local/stub-execute",
+    )
+    return action_name
+
+
+def _create_plan_in_state(
+    context: Context,
+    phase: PlanPhase,
+    state: ProcessingState,
+) -> str:
+    """Create a plan and move it to the given phase/state."""
+    action_name = _create_action(context)
+    plan = context.lifecycle_service.use_action(
+        action_name=action_name,
+        project_links=[ProjectLink(project_name="local/test-proj")],
+    )
+    plan_id = plan.identity.plan_id
+
+    if phase == PlanPhase.STRATEGIZE and state == ProcessingState.QUEUED:
+        pass
+    elif phase == PlanPhase.STRATEGIZE and state == ProcessingState.PROCESSING:
+        context.lifecycle_service.start_strategize(plan_id)
+    elif phase == PlanPhase.EXECUTE and state == ProcessingState.PROCESSING:
+        context.lifecycle_service.start_strategize(plan_id)
+        p = context.lifecycle_service.get_plan(plan_id)
+        p.decision_root_id = str(ULID())
+        context.lifecycle_service._commit_plan(p)
+        context.lifecycle_service.complete_strategize(plan_id)
+        context.lifecycle_service.execute_plan(plan_id)
+        context.lifecycle_service.start_execute(plan_id)
+    elif phase == PlanPhase.EXECUTE and state == ProcessingState.ERRORED:
+        context.lifecycle_service.start_strategize(plan_id)
+        p = context.lifecycle_service.get_plan(plan_id)
+        p.decision_root_id = str(ULID())
+        context.lifecycle_service._commit_plan(p)
+        context.lifecycle_service.complete_strategize(plan_id)
+        context.lifecycle_service.execute_plan(plan_id)
+        context.lifecycle_service.start_execute(plan_id)
+        context.lifecycle_service.fail_execute(plan_id, "Test error")
+    elif phase == PlanPhase.APPLY and state == ProcessingState.APPLIED:
+        context.lifecycle_service.start_strategize(plan_id)
+        p = context.lifecycle_service.get_plan(plan_id)
+        p.decision_root_id = str(ULID())
+        context.lifecycle_service._commit_plan(p)
+        context.lifecycle_service.complete_strategize(plan_id)
+        context.lifecycle_service.execute_plan(plan_id)
+        context.lifecycle_service.start_execute(plan_id)
+        context.lifecycle_service.complete_execute(plan_id)
+        context.lifecycle_service.apply_plan(plan_id)
+        context.lifecycle_service.start_apply(plan_id)
+        context.lifecycle_service.complete_apply(plan_id)
+    elif phase == PlanPhase.APPLY and state == ProcessingState.CONSTRAINED:
+        context.lifecycle_service.start_strategize(plan_id)
+        p = context.lifecycle_service.get_plan(plan_id)
+        p.decision_root_id = str(ULID())
+        context.lifecycle_service._commit_plan(p)
+        context.lifecycle_service.complete_strategize(plan_id)
+        context.lifecycle_service.execute_plan(plan_id)
+        context.lifecycle_service.start_execute(plan_id)
+        context.lifecycle_service.complete_execute(plan_id)
+        context.lifecycle_service.apply_plan(plan_id)
+        context.lifecycle_service.start_apply(plan_id)
+        p2 = context.lifecycle_service.get_plan(plan_id)
+        p2.processing_state = ProcessingState.PROCESSING
+        context.lifecycle_service._commit_plan(p2)
+        context.lifecycle_service.constrain_apply(plan_id, "constraints violated")
+
+    context.plan_id = plan_id
+    context.plan = context.lifecycle_service.get_plan(plan_id)
+    return plan_id
+
+
+# --- Given steps ---
+
+
+@given("I have a resume-test plan in execute phase with errored state")
+def step_resume_plan_execute_errored(context: Context) -> None:
+    """Create a plan in execute/errored state for resume tests."""
+    _create_plan_in_state(context, PlanPhase.EXECUTE, ProcessingState.ERRORED)
+
+
+@given("I have a resume-test plan in execute phase with processing state")
+def step_resume_plan_execute_processing(context: Context) -> None:
+    """Create a plan in execute/processing state for resume tests."""
+    _create_plan_in_state(context, PlanPhase.EXECUTE, ProcessingState.PROCESSING)
+
+
+@given("I have a resume-test plan in strategize phase with queued state")
+def step_resume_plan_strategize_queued(context: Context) -> None:
+    """Create a plan in strategize/queued state for resume tests."""
+    _create_plan_in_state(context, PlanPhase.STRATEGIZE, ProcessingState.QUEUED)
+
+
+@given("I have a resume-test plan in applied terminal state")
+def step_resume_plan_applied(context: Context) -> None:
+    """Create a plan in applied terminal state for resume tests."""
+    _create_plan_in_state(context, PlanPhase.APPLY, ProcessingState.APPLIED)
+
+
+@given("I have a resume-test plan in cancelled terminal state")
+def step_resume_plan_cancelled(context: Context) -> None:
+    """Create a plan and cancel it for resume tests."""
+    _create_plan_in_state(context, PlanPhase.EXECUTE, ProcessingState.PROCESSING)
+    context.lifecycle_service.cancel_plan(context.plan_id, "test cancel")
+    context.plan = context.lifecycle_service.get_plan(context.plan_id)
+
+
+@given("I have a resume-test plan in constrained terminal state")
+def step_resume_plan_constrained(context: Context) -> None:
+    """Create a plan in constrained terminal state for resume tests."""
+    _create_plan_in_state(context, PlanPhase.APPLY, ProcessingState.CONSTRAINED)
+
+
+@given("I have a resume-test plan in action phase")
+def step_resume_plan_action_phase(context: Context) -> None:
+    """Create a plan and set it to action phase for resume tests."""
+    _create_plan_in_state(context, PlanPhase.STRATEGIZE, ProcessingState.QUEUED)
+    plan = context.lifecycle_service.get_plan(context.plan_id)
+    plan.phase = PlanPhase.ACTION
+    context.lifecycle_service._commit_plan(plan)
+    context.plan = plan
+
+
+@given('the resume-test plan has checkpoint at step {index:d} with decision "{dec_id}"')
+def step_resume_plan_has_checkpoint(context: Context, index: int, dec_id: str) -> None:
+    """Add a checkpoint to the plan at given step."""
+    context.resume_service.set_total_steps(context.plan_id, 5)
+    context.resume_service.record_step_checkpoint(
+        plan_id=context.plan_id,
+        step_index=index,
+        decision_id=dec_id,
+        step_text=f"Step {index}",
+    )
+
+
+@given("I have resume metadata with last completed step {step:d}")
+def step_resume_metadata_with_step(context: Context, step: int) -> None:
+    """Create resume metadata with given step."""
+    context.resume_metadata = ResumeMetadata(
+        last_completed_step=step,
+        total_steps=5,
+    )
+
+
+@given("I have fresh resume metadata")
+def step_fresh_resume_metadata(context: Context) -> None:
+    """Create fresh resume metadata with no progress."""
+    context.resume_metadata = ResumeMetadata()
+
+
+@given("I have resume metadata with {total:d} total steps and last step {last:d}")
+def step_resume_metadata_with_totals(context: Context, total: int, last: int) -> None:
+    """Create resume metadata with total and last step."""
+    context.resume_metadata = ResumeMetadata(
+        last_completed_step=last,
+        total_steps=total,
+    )
+
+
+# --- When steps ---
+
+
+@when("I validate resume eligibility")
+def step_validate_resume_eligibility(context: Context) -> None:
+    """Validate resume eligibility for the current plan."""
+    context.eligibility = context.resume_service.validate_eligibility(
+        context.plan_id,
+    )
+
+
+@when('I record a resume step checkpoint at index {index:d} with decision "{dec_id}"')
+def step_record_resume_checkpoint(context: Context, index: int, dec_id: str) -> None:
+    """Record a step checkpoint."""
+    context.resume_service.set_total_steps(context.plan_id, 5)
+    context.resume_service.record_step_checkpoint(
+        plan_id=context.plan_id,
+        step_index=index,
+        decision_id=dec_id,
+        step_text=f"Step {index}",
+    )
+    context.plan = context.lifecycle_service.get_plan(context.plan_id)
+
+
+@when("I resume the plan with dry-run")
+def step_resume_dry_run(context: Context) -> None:
+    """Resume the plan in dry-run mode."""
+    context.summary = context.resume_service.resume_plan(context.plan_id, dry_run=True)
+
+
+@when("I resume the plan without dry-run")
+def step_resume_live(context: Context) -> None:
+    """Resume the plan in live mode."""
+    context.summary = context.resume_service.resume_plan(context.plan_id, dry_run=False)
+    context.plan = context.lifecycle_service.get_plan(context.plan_id)
+
+
+@when("I record a graceful shutdown for resume")
+def step_record_resume_shutdown(context: Context) -> None:
+    """Record a graceful shutdown."""
+    context.resume_service.record_shutdown(context.plan_id)
+
+
+@when("I try to resume a plan with empty ID")
+def step_resume_empty_id(context: Context) -> None:
+    """Try to resume a plan with empty ID."""
+    try:
+        context.resume_service.resume_plan("")
+    except ValidationError as e:
+        context.error = e
+
+
+@when("I try to validate eligibility with empty plan ID")
+def step_validate_eligibility_empty_id(context: Context) -> None:
+    """Try to validate eligibility with empty plan ID."""
+    try:
+        context.resume_service.validate_eligibility("")
+    except ValidationError as e:
+        context.error = e
+
+
+@when("I try to resume the terminal plan for resume test")
+def step_resume_terminal_plan(context: Context) -> None:
+    """Try to resume a terminal plan."""
+    try:
+        context.resume_service.resume_plan(context.plan_id)
+    except PlanError as e:
+        context.error = e
+
+
+@when("I set the resume total steps to {total:d}")
+def step_set_resume_total_steps(context: Context, total: int) -> None:
+    """Set total steps for the plan."""
+    context.resume_service.set_total_steps(context.plan_id, total)
+
+
+@when("I try to set resume total steps with empty plan ID")
+def step_set_resume_total_steps_empty(context: Context) -> None:
+    """Try to set total steps with empty plan ID."""
+    try:
+        context.resume_service.set_total_steps("", 5)
+    except ValidationError as e:
+        context.error = e
+
+
+# --- Then steps ---
+
+
+@then("the plan should be eligible for resume")
+def step_assert_eligible(context: Context) -> None:
+    """Assert the plan is eligible for resume."""
+    assert context.eligibility is not None
+    assert context.eligibility.eligible is True, (
+        f"Expected eligible=True, got: {context.eligibility.reason}"
+    )
+
+
+@then("the plan should not be eligible for resume")
+def step_assert_not_eligible(context: Context) -> None:
+    """Assert the plan is not eligible for resume."""
+    assert context.eligibility is not None
+    assert context.eligibility.eligible is False
+
+
+@then('the ineligible reason should be "{reason}"')
+def step_assert_ineligible_reason(context: Context, reason: str) -> None:
+    """Assert the ineligible reason code."""
+    assert context.eligibility is not None
+    assert context.eligibility.ineligible_reason is not None
+    assert context.eligibility.ineligible_reason.value == reason
+
+
+@then("the resume-test plan should have last completed step {step:d}")
+def step_assert_resume_last_step(context: Context, step: int) -> None:
+    """Assert the plan's last completed step."""
+    plan = context.lifecycle_service.get_plan(context.plan_id)
+    assert plan.last_completed_step == step
+
+
+@then("the resume-test plan should have a checkpoint ID set")
+def step_assert_resume_checkpoint_id(context: Context) -> None:
+    """Assert the plan has a checkpoint ID."""
+    plan = context.lifecycle_service.get_plan(context.plan_id)
+    assert plan.last_checkpoint_id is not None
+
+
+@then("the resume metadata should have {count:d} checkpoints")
+def step_assert_resume_checkpoint_count(context: Context, count: int) -> None:
+    """Assert the resume metadata checkpoint count."""
+    metadata = context.resume_service.get_resume_metadata(context.plan_id)
+    assert len(metadata.checkpoints) == count
+
+
+@then("the resume metadata next step should be {step:d}")
+def step_assert_resume_next_step(context: Context, step: int) -> None:
+    """Assert the resume metadata next step index."""
+    metadata = context.resume_service.get_resume_metadata(context.plan_id)
+    assert metadata.next_step_index == step
+
+
+@then("I should get a resume summary")
+def step_assert_resume_summary_exists(context: Context) -> None:
+    """Assert a resume summary was returned."""
+    assert context.summary is not None
+
+
+@then("the resume summary next step should be {step:d}")
+def step_assert_resume_summary_next_step(context: Context, step: int) -> None:
+    """Assert the resume summary next step."""
+    assert context.summary is not None
+    assert context.summary.next_step_index == step
+
+
+@then("the resume-test plan processing state should still be errored")
+def step_assert_resume_still_errored(context: Context) -> None:
+    """Assert the plan is still in errored state."""
+    plan = context.lifecycle_service.get_plan(context.plan_id)
+    assert plan.processing_state == ProcessingState.ERRORED
+
+
+@then('the resume summary decision ID should be "{dec_id}"')
+def step_assert_resume_summary_decision(context: Context, dec_id: str) -> None:
+    """Assert the resume summary decision ID."""
+    assert context.summary is not None
+    assert context.summary.decision_id == dec_id
+
+
+@then("the resume-test plan processing state should be processing")
+def step_assert_resume_processing(context: Context) -> None:
+    """Assert the plan is in processing state."""
+    plan = context.lifecycle_service.get_plan(context.plan_id)
+    assert plan.processing_state == ProcessingState.PROCESSING
+
+
+@then("the resume-test plan error message should be cleared")
+def step_assert_resume_error_cleared(context: Context) -> None:
+    """Assert the plan error message is None."""
+    plan = context.lifecycle_service.get_plan(context.plan_id)
+    assert plan.error_message is None
+
+
+@then("the resume metadata should be marked as interrupted")
+def step_assert_resume_interrupted(context: Context) -> None:
+    """Assert the resume metadata is marked interrupted."""
+    metadata = context.resume_service.get_resume_metadata(context.plan_id)
+    assert metadata.interrupted is True
+    assert metadata.interrupted_at is not None
+
+
+@then("a resume validation error should be raised")
+def step_assert_resume_validation_error(context: Context) -> None:
+    """Assert a ValidationError was raised."""
+    assert context.error is not None
+    assert isinstance(context.error, ValidationError)
+
+
+@then("a plan error should be raised for resume")
+def step_assert_resume_plan_error(context: Context) -> None:
+    """Assert a PlanError was raised."""
+    assert context.error is not None
+    assert isinstance(context.error, PlanError)
+
+
+@then("the resume metadata total steps should be {total:d}")
+def step_assert_resume_total_steps(context: Context, total: int) -> None:
+    """Assert the resume metadata total steps."""
+    metadata = context.resume_service.get_resume_metadata(context.plan_id)
+    assert metadata.total_steps == total
+
+
+@then("the resume metadata has_progress should be true")
+def step_assert_resume_has_progress_true(context: Context) -> None:
+    """Assert has_progress is True."""
+    assert context.resume_metadata.has_progress is True
+
+
+@then("the resume metadata has_progress should be false")
+def step_assert_resume_has_progress_false(context: Context) -> None:
+    """Assert has_progress is False."""
+    assert context.resume_metadata.has_progress is False
+
+
+@then("the resume metadata is_complete should be true")
+def step_assert_resume_is_complete_true(context: Context) -> None:
+    """Assert is_complete is True."""
+    assert context.resume_metadata.is_complete is True
+
+
+@then("the resume metadata is_complete should be false")
+def step_assert_resume_is_complete_false(context: Context) -> None:
+    """Assert is_complete is False."""
+    assert context.resume_metadata.is_complete is False
@@ -0,0 +1,338 @@
+"""Helper script for plan_resume.robot integration tests.
+
+Each sub-command exercises a specific aspect of the plan resume
+functionality and prints a success token on completion.
+"""
+
+from __future__ import annotations
+
+import sys
+
+from ulid import ULID
+
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.application.services.plan_resume_service import (
+    PlanResumeService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.core.exceptions import PlanError, ValidationError
+from cleveragents.domain.models.core.plan import (
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.domain.models.core.resume import (
+    ResumeMetadata,
+    ResumeSummary,
+)
+
+
+def _make_services() -> tuple[PlanLifecycleService, PlanResumeService]:
+    settings = Settings()
+    lifecycle = PlanLifecycleService(settings=settings)
+    resume = PlanResumeService(lifecycle_service=lifecycle)
+    return lifecycle, resume
+
+
+def _create_action(lifecycle: PlanLifecycleService) -> str:
+    name = f"local/resume-robot-{ULID()!s}"[:30].lower()
+    lifecycle.create_action(
+        name=name,
+        description="Robot resume test",
+        definition_of_done="Step A\nStep B\nStep C",
+        strategy_actor="local/stub-strategy",
+        execution_actor="local/stub-execute",
+    )
+    return name
+
+
+def _make_errored_plan(
+    lifecycle: PlanLifecycleService,
+) -> str:
+    action = _create_action(lifecycle)
+    plan = lifecycle.use_action(
+        action_name=action,
+        project_links=[ProjectLink(project_name="local/robot-proj")],
+    )
+    pid = plan.identity.plan_id
+    lifecycle.start_strategize(pid)
+    p = lifecycle.get_plan(pid)
+    p.decision_root_id = str(ULID())
+    lifecycle._commit_plan(p)
+    lifecycle.complete_strategize(pid)
+    lifecycle.execute_plan(pid)
+    lifecycle.start_execute(pid)
+    lifecycle.fail_execute(pid, "robot test error")
+    return pid
+
+
+def _make_processing_plan(
+    lifecycle: PlanLifecycleService,
+) -> str:
+    action = _create_action(lifecycle)
+    plan = lifecycle.use_action(
+        action_name=action,
+        project_links=[ProjectLink(project_name="local/robot-proj")],
+    )
+    pid = plan.identity.plan_id
+    lifecycle.start_strategize(pid)
+    p = lifecycle.get_plan(pid)
+    p.decision_root_id = str(ULID())
+    lifecycle._commit_plan(p)
+    lifecycle.complete_strategize(pid)
+    lifecycle.execute_plan(pid)
+    lifecycle.start_execute(pid)
+    return pid
+
+
+def _make_applied_plan(
+    lifecycle: PlanLifecycleService,
+) -> str:
+    action = _create_action(lifecycle)
+    plan = lifecycle.use_action(
+        action_name=action,
+        project_links=[ProjectLink(project_name="local/robot-proj")],
+    )
+    pid = plan.identity.plan_id
+    lifecycle.start_strategize(pid)
+    p = lifecycle.get_plan(pid)
+    p.decision_root_id = str(ULID())
+    lifecycle._commit_plan(p)
+    lifecycle.complete_strategize(pid)
+    lifecycle.execute_plan(pid)
+    lifecycle.start_execute(pid)
+    lifecycle.complete_execute(pid)
+    lifecycle.apply_plan(pid)
+    lifecycle.start_apply(pid)
+    lifecycle.complete_apply(pid)
+    return pid
+
+
+def test_eligibility_errored() -> None:
+    lifecycle, resume = _make_services()
+    pid = _make_errored_plan(lifecycle)
+    result = resume.validate_eligibility(pid)
+    assert result.eligible is True
+    print("eligibility-errored-ok")
+
+
+def test_eligibility_terminal() -> None:
+    lifecycle, resume = _make_services()
+
+    # Applied
+    pid_applied = _make_applied_plan(lifecycle)
+    r1 = resume.validate_eligibility(pid_applied)
+    assert r1.eligible is False
+    assert r1.ineligible_reason is not None
+    assert r1.ineligible_reason.value == "terminal_applied"
+
+    # Cancelled
+    pid_cancel = _make_processing_plan(lifecycle)
+    lifecycle.cancel_plan(pid_cancel, "test")
+    r2 = resume.validate_eligibility(pid_cancel)
+    assert r2.eligible is False
+    assert r2.ineligible_reason is not None
+    assert r2.ineligible_reason.value == "terminal_cancelled"
+
+    print("eligibility-terminal-ok")
+
+
+def test_record_checkpoints() -> None:
+    lifecycle, resume = _make_services()
+    pid = _make_processing_plan(lifecycle)
+
+    resume.set_total_steps(pid, 3)
+    resume.record_step_checkpoint(pid, 0, "DEC-A", "Step A")
+    cp1 = resume.record_step_checkpoint(pid, 1, "DEC-B", "Step B")
+
+    plan = lifecycle.get_plan(pid)
+    assert plan.last_completed_step == 1
+    assert plan.last_checkpoint_id == cp1.checkpoint_id
+
+    meta = resume.get_resume_metadata(pid)
+    assert len(meta.checkpoints) == 2
+    assert meta.next_step_index == 2
+
+    print("record-checkpoints-ok")
+
+
+def test_dry_run_resume() -> None:
+    lifecycle, resume = _make_services()
+    pid = _make_errored_plan(lifecycle)
+
+    resume.set_total_steps(pid, 5)
+    resume.record_step_checkpoint(pid, 2, "DEC-C", "Step C")
+
+    summary = resume.resume_plan(pid, dry_run=True)
+    assert summary.next_step_index == 3
+    assert summary.total_steps == 5
+    assert summary.decision_id == "DEC-C"
+
+    # State should NOT change
+    plan = lifecycle.get_plan(pid)
+    assert plan.processing_state == ProcessingState.ERRORED
+
+    print("dry-run-resume-ok")
+
+
+def test_live_resume() -> None:
+    lifecycle, resume = _make_services()
+    pid = _make_errored_plan(lifecycle)
+
+    resume.set_total_steps(pid, 5)
+    resume.record_step_checkpoint(pid, 1, "DEC-B", "Step B")
+
+    summary = resume.resume_plan(pid, dry_run=False)
+    assert summary.next_step_index == 2
+
+    plan = lifecycle.get_plan(pid)
+    assert plan.processing_state == ProcessingState.PROCESSING
+    assert plan.error_message is None
+
+    print("live-resume-ok")
+
+
+def test_graceful_shutdown() -> None:
+    lifecycle, resume = _make_services()
+    pid = _make_processing_plan(lifecycle)
+
+    resume.set_total_steps(pid, 3)
+    resume.record_step_checkpoint(pid, 0, "DEC-A", "Step A")
+    resume.record_shutdown(pid)
+
+    meta = resume.get_resume_metadata(pid)
+    assert meta.interrupted is True
+    assert meta.interrupted_at is not None
+
+    print("graceful-shutdown-ok")
+
+
+def test_metadata_properties() -> None:
+    # Fresh metadata
+    m1 = ResumeMetadata()
+    assert m1.has_progress is False
+    assert m1.is_complete is False
+    assert m1.next_step_index == 0
+
+    # With progress
+    m2 = ResumeMetadata(last_completed_step=2, total_steps=5)
+    assert m2.has_progress is True
+    assert m2.is_complete is False
+    assert m2.next_step_index == 3
+
+    # Complete
+    m3 = ResumeMetadata(last_completed_step=4, total_steps=5)
+    assert m3.has_progress is True
+    assert m3.is_complete is True
+    assert m3.next_step_index == 5
+
+    print("metadata-properties-ok")
+
+
+def test_summary_cli_dict() -> None:
+    summary = ResumeSummary(
+        plan_id="01HGZ6FE0AQDYTR4BXEXAMPLE1",
+        phase="execute",
+        processing_state="errored",
+        last_completed_step=2,
+        next_step_index=3,
+        total_steps=5,
+        decision_id="DEC-001",
+        last_checkpoint_id="01HGZ6FE0AQDYTR4BXCHECKPT1",
+        sandbox_ref="/tmp/sandbox-test",
+    )
+    d = summary.as_cli_dict()
+    assert d["plan_id"] == "01HGZ6FE0AQDYTR4BXEXAMPLE1"
+    assert d["next_step_index"] == 3
+    assert d["decision_id"] == "DEC-001"
+    assert d["sandbox_ref"] == "/tmp/sandbox-test"
+
+    print("summary-cli-dict-ok")
+
+
+def test_validation_errors() -> None:
+    lifecycle, resume = _make_services()
+
+    # Empty plan_id
+    try:
+        resume.resume_plan("")
+        raise AssertionError("Should have raised")
+    except ValidationError:
+        pass
+
+    try:
+        resume.validate_eligibility("")
+        raise AssertionError("Should have raised")
+    except ValidationError:
+        pass
+
+    try:
+        resume.set_total_steps("", 5)
+        raise AssertionError("Should have raised")
+    except ValidationError:
+        pass
+
+    try:
+        resume.record_step_checkpoint("", 0, "D", "text")
+        raise AssertionError("Should have raised")
+    except ValidationError:
+        pass
+
+    # Terminal plan
+    pid = _make_applied_plan(lifecycle)
+    try:
+        resume.resume_plan(pid)
+        raise AssertionError("Should have raised")
+    except PlanError:
+        pass
+
+    print("validation-errors-ok")
+
+
+def test_checkpoint_sandbox() -> None:
+    lifecycle, resume = _make_services()
+    pid = _make_processing_plan(lifecycle)
+
+    resume.set_total_steps(pid, 3)
+    cp = resume.record_step_checkpoint(
+        pid, 0, "DEC-A", "Step A", sandbox_ref="/tmp/sandbox-123"
+    )
+    assert cp.sandbox_ref == "/tmp/sandbox-123"
+
+    summary = resume.resume_plan(pid, dry_run=True)
+    assert summary.sandbox_ref == "/tmp/sandbox-123"
+
+    print("checkpoint-sandbox-ok")
+
+
+TESTS = {
+    "eligibility-errored": test_eligibility_errored,
+    "eligibility-terminal": test_eligibility_terminal,
+    "record-checkpoints": test_record_checkpoints,
+    "dry-run-resume": test_dry_run_resume,
+    "live-resume": test_live_resume,
+    "graceful-shutdown": test_graceful_shutdown,
+    "metadata-properties": test_metadata_properties,
+    "summary-cli-dict": test_summary_cli_dict,
+    "validation-errors": test_validation_errors,
+    "checkpoint-sandbox": test_checkpoint_sandbox,
+}
+
+
+def main() -> None:
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <test-name>", file=sys.stderr)
+        sys.exit(1)
+
+    test_name = sys.argv[1]
+    if test_name not in TESTS:
+        print(f"Unknown test: {test_name}", file=sys.stderr)
+        print(f"Available: {', '.join(sorted(TESTS))}", file=sys.stderr)
+        sys.exit(1)
+
+    TESTS[test_name]()
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,81 @@
+*** Settings ***
+Documentation     Integration tests for plan resume functionality.
+...               Covers resume eligibility validation, checkpoint recording,
+...               dry-run preview, live resume, and graceful shutdown.
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_plan_resume.py
+
+*** Test Cases ***
+Resume Eligibility For Errored Plan
+    [Documentation]    Verify errored plans are eligible for resume
+    [Tags]    resume    eligibility    critical
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    eligibility-errored    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    eligibility-errored-ok
+
+Resume Eligibility Rejects Terminal Plans
+    [Documentation]    Verify applied/cancelled/constrained plans cannot be resumed
+    [Tags]    resume    eligibility    terminal
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    eligibility-terminal    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    eligibility-terminal-ok
+
+Record Step Checkpoints
+    [Documentation]    Verify step-level checkpoints are recorded correctly
+    [Tags]    resume    checkpoint
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    record-checkpoints    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    record-checkpoints-ok
+
+Dry Run Resume Shows Resume Point
+    [Documentation]    Verify dry-run shows resume summary without changing state
+    [Tags]    resume    dry-run
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    dry-run-resume    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    dry-run-resume-ok
+
+Live Resume Resets Errored Plan
+    [Documentation]    Verify live resume transitions errored plan to processing
+    [Tags]    resume    live    critical
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    live-resume    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    live-resume-ok
+
+Graceful Shutdown Recording
+    [Documentation]    Verify graceful shutdown marks metadata as interrupted
+    [Tags]    resume    shutdown
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    graceful-shutdown    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    graceful-shutdown-ok
+
+Resume Metadata Properties
+    [Documentation]    Verify ResumeMetadata model computed properties
+    [Tags]    resume    model
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    metadata-properties    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metadata-properties-ok
+
+Resume Summary CLI Dictionary
+    [Documentation]    Verify ResumeSummary.as_cli_dict() returns correct keys
+    [Tags]    resume    cli
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    summary-cli-dict    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    summary-cli-dict-ok
+
+Resume Validation Errors
+    [Documentation]    Verify proper error handling for invalid inputs
+    [Tags]    resume    errors
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    validation-errors    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    validation-errors-ok
+
+Resume Checkpoint With Sandbox Ref
+    [Documentation]    Verify checkpoint records sandbox reference correctly
+    [Tags]    resume    checkpoint    sandbox
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    checkpoint-sandbox    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    checkpoint-sandbox-ok
@@ -341,7 +341,8 @@ class PlanLifecycleService:
        """Get an action by namespaced name.

        Checks in-memory cache first, then falls back to the persistence
-        layer if a UnitOfWork is available.
+        layer if a UnitOfWork is available.  The name is normalised through
+        ``NamespacedName.parse`` so callers do not need to pre-lowercase.

        Args:
            action_name: The namespaced name (e.g., 'local/code-coverage')
@@ -352,15 +353,19 @@ class PlanLifecycleService:
        Raises:
            NotFoundError: If action not found
        """
-        action = self._actions.get(action_name)
+        # Normalise so lookups are case-insensitive, consistent with
+        # create_action which stores keys via NamespacedName.parse().
+        normalised = str(NamespacedName.parse(action_name))
+
+        action = self._actions.get(normalised)
        if action is not None:
            return action
        # Try persistence layer
        if self._persisted and self.unit_of_work is not None:
            with self.unit_of_work.transaction() as ctx:
-                persisted = ctx.actions.get_by_name(action_name)
+                persisted = ctx.actions.get_by_name(normalised)
            if persisted is not None:
-                self._actions[action_name] = persisted
+                self._actions[normalised] = persisted
                return persisted
        raise NotFoundError(
            resource_type="action",
@@ -370,9 +375,10 @@ class PlanLifecycleService:
    def get_action_by_name(self, name: str) -> Action:
        """Get an action by namespaced name string.

-        This is an alias for ``get_action`` -- provided for backwards
-        compatibility with CLI callers that used the old ID-based lookup
-        alongside a separate name-based lookup.
+        Delegates to ``get_action`` which normalises the name through
+        ``NamespacedName.parse``.  Retained for backwards compatibility
+        with CLI callers that used the old ID-based lookup alongside a
+        separate name-based lookup.

        Args:
            name: The namespaced name (e.g., 'local/code-coverage')
@@ -383,28 +389,7 @@ class PlanLifecycleService:
        Raises:
            NotFoundError: If action not found
        """
-        # Normalise through NamespacedName.parse so that partial names
-        # (e.g. "code-coverage" without namespace) still match.
-        parsed = NamespacedName.parse(name)
-        parsed_str = str(parsed)
-        action = self._actions.get(parsed_str)
-        if action is not None:
-            return action
-        # Fall back to linear scan in case of key mismatch
-        for act in self._actions.values():
-            if str(act.namespaced_name) == parsed_str:
-                return act
-        # Try persistence layer
-        if self._persisted and self.unit_of_work is not None:
-            with self.unit_of_work.transaction() as ctx:
-                persisted = ctx.actions.get_by_name(parsed_str)
-            if persisted is not None:
-                self._actions[parsed_str] = persisted
-                return persisted
-        raise NotFoundError(
-            resource_type="action",
-            resource_id=name,
-        )
+        return self.get_action(name)

    def list_actions(
        self,
@@ -0,0 +1,387 @@
+"""Plan Resume Service for CleverAgents v3.
+
+Implements step-level progress persistence and plan resume with
+graceful shutdown handling.
+
+## Resume Flow
+
+1. Validate resume eligibility (non-terminal plans only).
+2. Build a ``ResumeSummary`` (phase, step, decision_id).
+3. In dry-run mode, return the summary without changing state.
+4. In live mode, reset the plan to ``PROCESSING`` and restart
+   execution from ``last_completed_step + 1``.
+
+## Checkpoints
+
+Checkpoints are recorded at each completed step, tied to a
+decision ID and sandbox reference. On resume, execution picks up
+after the last checkpoint.
+
+## Graceful Shutdown
+
+``record_shutdown`` persists the current step index before the
+process exits so that ``plan resume`` can pick up later.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import TYPE_CHECKING
+
+import structlog
+from ulid import ULID
+
+from cleveragents.core.exceptions import PlanError, ValidationError
+from cleveragents.domain.models.core.plan import (
+    PlanPhase,
+    ProcessingState,
+)
+from cleveragents.domain.models.core.resume import (
+    ResumeCheckpoint,
+    ResumeEligibility,
+    ResumeIneligibleReason,
+    ResumeMetadata,
+    ResumeSummary,
+)
+
+if TYPE_CHECKING:
+    from cleveragents.application.services.plan_lifecycle_service import (
+        PlanLifecycleService,
+    )
+
+logger = structlog.get_logger(__name__)
+
+
+class PlanResumeService:
+    """Service for resuming interrupted or errored plan executions.
+
+    Works with ``PlanLifecycleService`` to validate state and
+    ``PlanExecutor`` to re-drive execution from the resume point.
+    """
+
+    def __init__(
+        self,
+        lifecycle_service: PlanLifecycleService,
+    ) -> None:
+        """Initialize the plan resume service.
+
+        Args:
+            lifecycle_service: The plan lifecycle service for plan access.
+
+        Raises:
+            ValidationError: If lifecycle_service is None.
+        """
+        if lifecycle_service is None:
+            raise ValidationError("lifecycle_service must not be None")
+        self._lifecycle = lifecycle_service
+        self._logger = logger.bind(service="plan_resume")
+        self._resume_metadata: dict[str, ResumeMetadata] = {}
+
+    def get_resume_metadata(self, plan_id: str) -> ResumeMetadata:
+        """Get or create resume metadata for a plan.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            The ResumeMetadata for the plan.
+
+        Raises:
+            ValidationError: If plan_id is empty.
+        """
+        if not plan_id:
+            raise ValidationError("plan_id must not be empty")
+
+        plan = self._lifecycle.get_plan(plan_id)
+
+        if plan_id not in self._resume_metadata:
+            self._resume_metadata[plan_id] = ResumeMetadata(
+                last_completed_step=plan.last_completed_step,
+                last_checkpoint_id=plan.last_checkpoint_id,
+            )
+        return self._resume_metadata[plan_id]
+
+    def validate_eligibility(self, plan_id: str) -> ResumeEligibility:
+        """Validate whether a plan can be resumed.
+
+        A plan is eligible for resume when it is NOT in a terminal state
+        (applied, cancelled, constrained) and has been started.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            A ResumeEligibility result.
+
+        Raises:
+            ValidationError: If plan_id is empty.
+        """
+        if not plan_id:
+            raise ValidationError("plan_id must not be empty")
+
+        plan = self._lifecycle.get_plan(plan_id)
+
+        # Terminal states cannot be resumed
+        if plan.processing_state == ProcessingState.APPLIED:
+            return ResumeEligibility(
+                eligible=False,
+                reason=(
+                    f"Plan {plan_id} is in terminal state 'applied' "
+                    "and cannot be resumed."
+                ),
+                ineligible_reason=ResumeIneligibleReason.TERMINAL_APPLIED,
+            )
+
+        if plan.processing_state == ProcessingState.CANCELLED:
+            return ResumeEligibility(
+                eligible=False,
+                reason=(
+                    f"Plan {plan_id} is in terminal state 'cancelled' "
+                    "and cannot be resumed."
+                ),
+                ineligible_reason=ResumeIneligibleReason.TERMINAL_CANCELLED,
+            )
+
+        if plan.processing_state == ProcessingState.CONSTRAINED:
+            return ResumeEligibility(
+                eligible=False,
+                reason=(
+                    f"Plan {plan_id} is in terminal state 'constrained' "
+                    "and cannot be resumed."
+                ),
+                ineligible_reason=ResumeIneligibleReason.TERMINAL_CONSTRAINED,
+            )
+
+        # ACTION phase cannot be resumed (it's a template phase)
+        if plan.phase == PlanPhase.ACTION:
+            return ResumeEligibility(
+                eligible=False,
+                reason=(
+                    f"Plan {plan_id} is in ACTION phase "
+                    "(template phase) and cannot be resumed."
+                ),
+                ineligible_reason=ResumeIneligibleReason.ACTION_PHASE,
+            )
+
+        # Eligible for resume
+        return ResumeEligibility(
+            eligible=True,
+            reason=f"Plan {plan_id} is eligible for resume.",
+        )
+
+    def build_resume_summary(self, plan_id: str) -> ResumeSummary:
+        """Build a summary of where execution will resume.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            A ResumeSummary with phase, step, decision_id details.
+
+        Raises:
+            ValidationError: If plan_id is empty.
+            PlanError: If plan is not eligible for resume.
+        """
+        if not plan_id:
+            raise ValidationError("plan_id must not be empty")
+
+        eligibility = self.validate_eligibility(plan_id)
+        if not eligibility.eligible:
+            raise PlanError(eligibility.reason)
+
+        plan = self._lifecycle.get_plan(plan_id)
+        metadata = self.get_resume_metadata(plan_id)
+
+        # Resolve decision_id from last checkpoint
+        decision_id: str | None = None
+        sandbox_ref: str | None = None
+        if metadata.checkpoints:
+            last_cp = metadata.checkpoints[-1]
+            decision_id = last_cp.decision_id
+            sandbox_ref = last_cp.sandbox_ref
+
+        return ResumeSummary(
+            plan_id=plan.identity.plan_id,
+            phase=plan.phase.value,
+            processing_state=plan.processing_state.value,
+            last_completed_step=metadata.last_completed_step,
+            next_step_index=metadata.next_step_index,
+            total_steps=metadata.total_steps,
+            decision_id=decision_id,
+            last_checkpoint_id=metadata.last_checkpoint_id,
+            sandbox_ref=sandbox_ref,
+        )
+
+    def resume_plan(
+        self,
+        plan_id: str,
+        dry_run: bool = False,
+    ) -> ResumeSummary:
+        """Resume a plan from its last checkpoint.
+
+        In dry-run mode, shows where execution will resume without
+        changing any state. In live mode, resets the plan to
+        PROCESSING and returns the summary for the caller to
+        re-drive execution.
+
+        Args:
+            plan_id: The plan ULID.
+            dry_run: If True, show resume point without executing.
+
+        Returns:
+            A ResumeSummary describing the resume point.
+
+        Raises:
+            ValidationError: If plan_id is empty.
+            PlanError: If plan is not eligible for resume.
+        """
+        if not plan_id:
+            raise ValidationError("plan_id must not be empty")
+
+        summary = self.build_resume_summary(plan_id)
+
+        if dry_run:
+            self._logger.info(
+                "Resume dry-run",
+                plan_id=plan_id,
+                next_step=summary.next_step_index,
+                total_steps=summary.total_steps,
+            )
+            return summary
+
+        # Live resume: reset to PROCESSING if currently ERRORED or QUEUED
+        plan = self._lifecycle.get_plan(plan_id)
+
+        if plan.processing_state == ProcessingState.ERRORED:
+            plan.processing_state = ProcessingState.PROCESSING
+            plan.error_message = None
+            plan.timestamps.updated_at = datetime.now()
+            self._lifecycle._commit_plan(plan)
+
+        elif plan.processing_state == ProcessingState.QUEUED:
+            plan.processing_state = ProcessingState.PROCESSING
+            plan.timestamps.updated_at = datetime.now()
+            self._lifecycle._commit_plan(plan)
+
+        self._logger.info(
+            "Plan resumed",
+            plan_id=plan_id,
+            phase=plan.phase.value,
+            next_step=summary.next_step_index,
+            total_steps=summary.total_steps,
+        )
+
+        return summary
+
+    def record_step_checkpoint(
+        self,
+        plan_id: str,
+        step_index: int,
+        decision_id: str,
+        step_text: str,
+        sandbox_ref: str | None = None,
+    ) -> ResumeCheckpoint:
+        """Record a checkpoint after a step completes.
+
+        Updates both the in-memory metadata and the plan's
+        persistent fields (last_completed_step, last_checkpoint_id).
+
+        Args:
+            plan_id: The plan ULID.
+            step_index: Zero-based index of the completed step.
+            decision_id: Decision node ID for this step.
+            step_text: Description of the step.
+            sandbox_ref: Optional sandbox reference.
+
+        Returns:
+            The created ResumeCheckpoint.
+
+        Raises:
+            ValidationError: If required arguments are invalid.
+        """
+        if not plan_id:
+            raise ValidationError("plan_id must not be empty")
+        if step_index < 0:
+            raise ValidationError("step_index must be >= 0")
+        if not decision_id:
+            raise ValidationError("decision_id must not be empty")
+        if not step_text:
+            raise ValidationError("step_text must not be empty")
+
+        checkpoint_id = str(ULID())
+        checkpoint = ResumeCheckpoint(
+            checkpoint_id=checkpoint_id,
+            plan_id=plan_id,
+            decision_id=decision_id,
+            step_index=step_index,
+            step_text=step_text,
+            sandbox_ref=sandbox_ref,
+        )
+
+        metadata = self.get_resume_metadata(plan_id)
+        metadata.record_checkpoint(checkpoint)
+
+        # Update plan persistent fields
+        plan = self._lifecycle.get_plan(plan_id)
+        plan.last_completed_step = step_index
+        plan.last_checkpoint_id = checkpoint_id
+        plan.timestamps.updated_at = datetime.now()
+        self._lifecycle._commit_plan(plan)
+
+        self._logger.debug(
+            "Step checkpoint recorded",
+            plan_id=plan_id,
+            step_index=step_index,
+            checkpoint_id=checkpoint_id,
+            decision_id=decision_id,
+        )
+
+        return checkpoint
+
+    def record_shutdown(self, plan_id: str) -> None:
+        """Record graceful shutdown for a plan.
+
+        Marks the plan's resume metadata as interrupted so the next
+        ``plan resume`` knows the execution was not completed cleanly.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Raises:
+            ValidationError: If plan_id is empty.
+        """
+        if not plan_id:
+            raise ValidationError("plan_id must not be empty")
+
+        metadata = self.get_resume_metadata(plan_id)
+        metadata.interrupted = True
+        metadata.interrupted_at = datetime.now()
+
+        # Persist the interrupted state on the plan
+        plan = self._lifecycle.get_plan(plan_id)
+        plan.timestamps.updated_at = datetime.now()
+        self._lifecycle._commit_plan(plan)
+
+        self._logger.info(
+            "Graceful shutdown recorded",
+            plan_id=plan_id,
+            last_completed_step=metadata.last_completed_step,
+        )
+
+    def set_total_steps(self, plan_id: str, total: int) -> None:
+        """Set the total number of steps for a plan.
+
+        Args:
+            plan_id: The plan ULID.
+            total: Total number of steps.
+
+        Raises:
+            ValidationError: If arguments are invalid.
+        """
+        if not plan_id:
+            raise ValidationError("plan_id must not be empty")
+        if total < 0:
+            raise ValidationError("total must be >= 0")
+
+        metadata = self.get_resume_metadata(plan_id)
+        metadata.total_steps = total
@@ -148,6 +148,10 @@ def _plan_spec_dict(plan: Any) -> dict[str, object]:
            }
        if plan.error_message:
            result["error_message"] = plan.error_message
+        if plan.last_completed_step >= 0:
+            result["last_completed_step"] = plan.last_completed_step
+        if plan.last_checkpoint_id:
+            result["last_checkpoint_id"] = plan.last_checkpoint_id
        return result

    # Legacy plan fallback
@@ -1205,6 +1209,12 @@ def _print_lifecycle_plan(plan: Any, title: str = "Plan") -> None:
        for inv in plan.invariants:
            details += f"  [{inv.source.value}] {inv.text}\n"

+    # Resume metadata
+    if plan.last_completed_step >= 0:
+        details += f"[bold]Last Completed Step:[/bold] {plan.last_completed_step}\n"
+    if plan.last_checkpoint_id:
+        details += f"[bold]Last Checkpoint:[/bold] {plan.last_checkpoint_id}\n"
+
    details += f"[bold]Terminal:[/bold] {'yes' if plan.is_terminal else 'no'}\n"

    # Timestamps
@@ -2395,6 +2405,101 @@ def correct_decision(
        raise typer.Abort() from e


+@app.command("resume")
+def resume_plan_cmd(
+    plan_id: Annotated[
+        str,
+        typer.Argument(help="Plan ID to resume"),
+    ],
+    dry_run: Annotated[
+        bool,
+        typer.Option(
+            "--dry-run",
+            help="Show where execution will resume without changing state",
+        ),
+    ] = False,
+    fmt: Annotated[
+        str,
+        typer.Option(
+            "--format",
+            "-f",
+            help=_FORMAT_HELP,
+        ),
+    ] = "rich",
+) -> None:
+    """Resume a plan from its last checkpoint.
+
+    Shows a resume summary (phase, step, decision_id) before executing.
+    Use ``--dry-run`` to preview the resume point without changing state.
+
+    Only non-terminal plans (not applied, cancelled, or constrained)
+    can be resumed.
+
+    Examples::
+
+        agents plan resume PLAN123
+        agents plan resume PLAN123 --dry-run
+        agents plan resume PLAN123 --format json
+    """
+    from cleveragents.application.services.plan_resume_service import (
+        PlanResumeService,
+    )
+
+    try:
+        service = _get_lifecycle_service()
+        resume_service = PlanResumeService(lifecycle_service=service)
+
+        summary = resume_service.resume_plan(plan_id, dry_run=dry_run)
+
+        if fmt != OutputFormat.RICH.value:
+            console.print(format_output(summary.as_cli_dict(), fmt))
+        else:
+            mode_label = "[dim](dry-run)[/dim] " if dry_run else ""
+            console.print(
+                Panel(
+                    f"[bold]Plan ID:[/bold] {summary.plan_id}\n"
+                    f"[bold]Phase:[/bold] {summary.phase}\n"
+                    f"[bold]State:[/bold] {summary.processing_state}\n"
+                    f"[bold]Last Completed Step:[/bold] "
+                    f"{summary.last_completed_step}\n"
+                    f"[bold]Next Step:[/bold] {summary.next_step_index}"
+                    f" / {summary.total_steps}\n"
+                    + (
+                        f"[bold]Decision ID:[/bold] {summary.decision_id}\n"
+                        if summary.decision_id
+                        else ""
+                    )
+                    + (
+                        f"[bold]Checkpoint:[/bold] {summary.last_checkpoint_id}\n"
+                        if summary.last_checkpoint_id
+                        else ""
+                    )
+                    + (
+                        f"[bold]Sandbox:[/bold] {summary.sandbox_ref}\n"
+                        if summary.sandbox_ref
+                        else ""
+                    ),
+                    title=f"{mode_label}Resume Summary",
+                    expand=False,
+                )
+            )
+            if dry_run:
+                console.print("[dim]No state changes made (dry-run mode).[/dim]")
+            else:
+                console.print(
+                    "[green]Plan resumed.[/green] "
+                    "Execution will continue from "
+                    f"step {summary.next_step_index}."
+                )
+
+    except PlanError as e:
+        console.print(f"[red]Cannot resume:[/red] {e.message}")
+        raise typer.Abort() from e
+    except CleverAgentsError as e:
+        console.print(f"[red]Error:[/red] {e.message}")
+        raise typer.Abort() from e
+
+
 def _resolve_active_plan_id() -> str:
    """Resolve the active plan ID when none is explicitly provided.

@@ -156,6 +156,15 @@ from cleveragents.domain.models.core.resource_type import (
    SandboxStrategy as ResourceTypeSandboxStrategy,
 )

+# Resume models
+from cleveragents.domain.models.core.resume import (
+    ResumeCheckpoint,
+    ResumeEligibility,
+    ResumeIneligibleReason,
+    ResumeMetadata,
+    ResumeSummary,
+)
+
 # Session domain model
 from cleveragents.domain.models.core.session import (
    MessageRole,
@@ -293,6 +302,11 @@ __all__ = [
    "ResourceTypeArgument",
    "ResourceTypeSandboxStrategy",
    "ResourceTypeSpec",
+    "ResumeCheckpoint",
+    "ResumeEligibility",
+    "ResumeIneligibleReason",
+    "ResumeMetadata",
+    "ResumeSummary",
    "ReviewArtifact",
    "SandboxStrategy",
    "Session",
@@ -670,6 +670,20 @@ class Plan(BaseModel):
        description="Status tracking for spawned subplans",
    )

+    # Resume metadata (step-level progress for plan resume)
+    last_completed_step: int = Field(
+        default=-1,
+        ge=-1,
+        description=(
+            "Index of the last successfully completed step. "
+            "-1 means no steps completed."
+        ),
+    )
+    last_checkpoint_id: str | None = Field(
+        default=None,
+        description="ULID of the most recent resume checkpoint",
+    )
+
    # Phase reversion tracking
    reversion_count: int = Field(
        default=0,
@@ -871,6 +885,10 @@ class Plan(BaseModel):
            result["subplan_count"] = len(self.subplan_statuses)
        if self.cost_metadata is not None:
            result["cost"] = self.cost_metadata.as_display_dict()
+        if self.last_completed_step >= 0:
+            result["last_completed_step"] = self.last_completed_step
+        if self.last_checkpoint_id:
+            result["last_checkpoint_id"] = self.last_checkpoint_id
        return result

    model_config = ConfigDict(
@@ -0,0 +1,233 @@
+"""Resume domain models for plan execution.
+
+Defines the models supporting plan resume functionality:
+
+- ``ResumeCheckpoint`` -- a checkpoint recorded at each completed step
+- ``ResumeMetadata`` -- resume state carried on a Plan
+- ``ResumeSummary`` -- output shown before resume execution
+- ``ResumeEligibility`` -- result of resume eligibility validation
+
+Checkpoints are tied to decision IDs and sandbox state to ensure
+deterministic resume points.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from enum import StrEnum
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cleveragents.domain.models.core.plan import ULID_PATTERN
+
+
+class ResumeIneligibleReason(StrEnum):
+    """Why a plan cannot be resumed."""
+
+    TERMINAL_APPLIED = "terminal_applied"
+    TERMINAL_CANCELLED = "terminal_cancelled"
+    TERMINAL_CONSTRAINED = "terminal_constrained"
+    NO_PROGRESS = "no_progress"
+    ACTION_PHASE = "action_phase"
+
+
+class ResumeCheckpoint(BaseModel):
+    """A checkpoint recorded when a plan step completes.
+
+    Ties a completed step to a decision ID and sandbox snapshot,
+    enabling deterministic resume from that point.
+    """
+
+    checkpoint_id: str = Field(
+        ...,
+        min_length=1,
+        description="Unique checkpoint ID (ULID)",
+        pattern=ULID_PATTERN,
+    )
+    plan_id: str = Field(
+        ...,
+        min_length=1,
+        description="Plan this checkpoint belongs to",
+        pattern=ULID_PATTERN,
+    )
+    decision_id: str = Field(
+        ...,
+        min_length=1,
+        description="Decision node ID this checkpoint is tied to",
+    )
+    step_index: int = Field(
+        ...,
+        ge=0,
+        description="Zero-based index of the completed step",
+    )
+    step_text: str = Field(
+        ...,
+        min_length=1,
+        description="Description of the completed step",
+    )
+    sandbox_ref: str | None = Field(
+        default=None,
+        description="Sandbox reference at checkpoint time",
+    )
+    created_at: datetime = Field(
+        default_factory=datetime.now,
+        description="When the checkpoint was recorded",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+class ResumeMetadata(BaseModel):
+    """Resume state carried on a Plan.
+
+    Tracks the last completed step and checkpoint to enable
+    execution resume after interruption or error.
+    """
+
+    last_completed_step: int = Field(
+        default=-1,
+        ge=-1,
+        description=(
+            "Index of the last successfully completed step. "
+            "-1 means no steps completed."
+        ),
+    )
+    last_checkpoint_id: str | None = Field(
+        default=None,
+        description="ULID of the most recent checkpoint",
+    )
+    total_steps: int = Field(
+        default=0,
+        ge=0,
+        description="Total number of steps in the plan",
+    )
+    checkpoints: list[ResumeCheckpoint] = Field(
+        default_factory=list,
+        description="All recorded checkpoints for this plan",
+    )
+    interrupted: bool = Field(
+        default=False,
+        description="Whether the plan was interrupted (e.g., shutdown)",
+    )
+    interrupted_at: datetime | None = Field(
+        default=None,
+        description="When the plan was interrupted",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+    @property
+    def has_progress(self) -> bool:
+        """Whether any steps have been completed."""
+        return self.last_completed_step >= 0
+
+    @property
+    def next_step_index(self) -> int:
+        """Index of the next step to execute."""
+        return self.last_completed_step + 1
+
+    @property
+    def is_complete(self) -> bool:
+        """Whether all steps have been completed."""
+        return self.total_steps > 0 and self.last_completed_step >= self.total_steps - 1
+
+    def record_checkpoint(self, checkpoint: ResumeCheckpoint) -> None:
+        """Record a new checkpoint.
+
+        Args:
+            checkpoint: The checkpoint to record.
+
+        Raises:
+            ValueError: If the checkpoint step_index is invalid.
+        """
+        if checkpoint.step_index < 0:
+            raise ValueError("Checkpoint step_index must be >= 0")
+        self.checkpoints.append(checkpoint)
+        self.last_completed_step = checkpoint.step_index
+        self.last_checkpoint_id = checkpoint.checkpoint_id
+
+
+class ResumeEligibility(BaseModel):
+    """Result of resume eligibility validation."""
+
+    eligible: bool = Field(
+        ...,
+        description="Whether the plan can be resumed",
+    )
+    reason: str = Field(
+        default="",
+        description="Why the plan can or cannot be resumed",
+    )
+    ineligible_reason: ResumeIneligibleReason | None = Field(
+        default=None,
+        description="Structured reason code for ineligibility",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+class ResumeSummary(BaseModel):
+    """Summary output shown before resume execution.
+
+    Displayed by ``plan resume`` and ``plan resume --dry-run``.
+    """
+
+    plan_id: str = Field(..., description="Plan ULID")
+    phase: str = Field(..., description="Current plan phase")
+    processing_state: str = Field(..., description="Current processing state")
+    last_completed_step: int = Field(
+        ...,
+        description="Last completed step index (-1 if none)",
+    )
+    next_step_index: int = Field(
+        ...,
+        description="Next step to execute",
+    )
+    total_steps: int = Field(
+        ...,
+        description="Total steps in the plan",
+    )
+    decision_id: str | None = Field(
+        default=None,
+        description="Decision ID at the resume point",
+    )
+    last_checkpoint_id: str | None = Field(
+        default=None,
+        description="Last checkpoint ID",
+    )
+    sandbox_ref: str | None = Field(
+        default=None,
+        description="Sandbox reference at resume point",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+    def as_cli_dict(self) -> dict[str, object]:
+        """Return a dictionary for CLI rendering."""
+        result: dict[str, object] = {
+            "plan_id": self.plan_id,
+            "phase": self.phase,
+            "processing_state": self.processing_state,
+            "last_completed_step": self.last_completed_step,
+            "next_step_index": self.next_step_index,
+            "total_steps": self.total_steps,
+        }
+        if self.decision_id:
+            result["decision_id"] = self.decision_id
+        if self.last_checkpoint_id:
+            result["last_checkpoint_id"] = self.last_checkpoint_id
+        if self.sandbox_ref:
+            result["sandbox_ref"] = self.sandbox_ref
+        return result