HAL9000/temp
0
0
Fork 0
forked from HAL9000/cleveragents-core
Code Pull Requests Activity

feat(concurrency): add plan resume

Browse Source 
Implement step-level progress persistence and plan resume with graceful
shutdown handling.

- Add ResumeCheckpoint, ResumeMetadata, ResumeEligibility, ResumeSummary
  domain models (resume.py)
- Add PlanResumeService with validate_eligibility(), build_resume_summary(),
  resume_plan(dry_run), record_step_checkpoint(), record_shutdown()
- Add last_completed_step and last_checkpoint_id fields to Plan model
- Add 'plan resume' CLI command with --dry-run flag
- Update plan lifecycle docs (ADR-006) with resume behavior section
- Add Behave tests (24 scenarios in plan_resume.feature)
- Add Robot Framework integration tests (10 tests)
- Add ASV benchmarks for resume overhead

Closes #328
This commit is contained in:
Jeffrey Phillips Freeman
2026-02-25 00:55:47 +00:00
parent 134dce896a
commit c4f71e930d
14 changed files with 2034 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/plan_resume_bench.py
+159
View File
@@ -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
docs/adr/ADR-006-plan-lifecycle.md
+73
View File
@@ -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).
features/plan_lifecycle_service_coverage_boost.feature
+4 -4
View File
@@ -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
features/plan_resume.feature
+152
View File
@@ -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
features/steps/plan_lifecycle_service_coverage_boost_steps.py
+10 -21
View File
@@ -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
features/steps/plan_resume_steps.py
+446
View File
@@ -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
robot/helper_plan_resume.py
+338
View File
@@ -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()
robot/plan_resume.robot
+81
View File
@@ -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
src/cleveragents/application/services/plan_lifecycle_service.py
+14 -29
View File
@@ -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,
src/cleveragents/application/services/plan_resume_service.py
+387
View File
@@ -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
src/cleveragents/cli/commands/plan.py
+105
View File
@@ -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.
src/cleveragents/domain/models/core/__init__.py
+14
View File
@@ -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",
src/cleveragents/domain/models/core/plan.py
+18
View File
@@ -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(
src/cleveragents/domain/models/core/resume.py
+233
View File
@@ -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