From c4f71e930dcdea36b371a5e1d09ff35c872dfefe Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Wed, 25 Feb 2026 00:55:47 +0000 Subject: [PATCH] feat(concurrency): add plan resume 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 --- benchmarks/plan_resume_bench.py | 159 +++++++ docs/adr/ADR-006-plan-lifecycle.md | 73 +++ ...n_lifecycle_service_coverage_boost.feature | 8 +- features/plan_resume.feature | 152 ++++++ ..._lifecycle_service_coverage_boost_steps.py | 31 +- features/steps/plan_resume_steps.py | 446 ++++++++++++++++++ robot/helper_plan_resume.py | 338 +++++++++++++ robot/plan_resume.robot | 81 ++++ .../services/plan_lifecycle_service.py | 43 +- .../services/plan_resume_service.py | 387 +++++++++++++++ src/cleveragents/cli/commands/plan.py | 105 +++++ .../domain/models/core/__init__.py | 14 + src/cleveragents/domain/models/core/plan.py | 18 + src/cleveragents/domain/models/core/resume.py | 233 +++++++++ 14 files changed, 2034 insertions(+), 54 deletions(-) create mode 100644 benchmarks/plan_resume_bench.py create mode 100644 features/plan_resume.feature create mode 100644 features/steps/plan_resume_steps.py create mode 100644 robot/helper_plan_resume.py create mode 100644 robot/plan_resume.robot create mode 100644 src/cleveragents/application/services/plan_resume_service.py create mode 100644 src/cleveragents/domain/models/core/resume.py diff --git a/benchmarks/plan_resume_bench.py b/benchmarks/plan_resume_bench.py new file mode 100644 index 00000000..82fe9028 --- /dev/null +++ b/benchmarks/plan_resume_bench.py @@ -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 diff --git a/docs/adr/ADR-006-plan-lifecycle.md b/docs/adr/ADR-006-plan-lifecycle.md index 9066e73c..66f7a5e9 100644 --- a/docs/adr/ADR-006-plan-lifecycle.md +++ b/docs/adr/ADR-006-plan-lifecycle.md @@ -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 ``. + +### 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). diff --git a/features/plan_lifecycle_service_coverage_boost.feature b/features/plan_lifecycle_service_coverage_boost.feature index fc62f38e..fe9837f0 100644 --- a/features/plan_lifecycle_service_coverage_boost.feature +++ b/features/plan_lifecycle_service_coverage_boost.feature @@ -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 diff --git a/features/plan_resume.feature b/features/plan_resume.feature new file mode 100644 index 00000000..e4b07a67 --- /dev/null +++ b/features/plan_resume.feature @@ -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 diff --git a/features/steps/plan_lifecycle_service_coverage_boost_steps.py b/features/steps/plan_lifecycle_service_coverage_boost_steps.py index 701e06fd..0409dc11 100644 --- a/features/steps/plan_lifecycle_service_coverage_boost_steps.py +++ b/features/steps/plan_lifecycle_service_coverage_boost_steps.py @@ -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 diff --git a/features/steps/plan_resume_steps.py b/features/steps/plan_resume_steps.py new file mode 100644 index 00000000..f71accba --- /dev/null +++ b/features/steps/plan_resume_steps.py @@ -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 diff --git a/robot/helper_plan_resume.py b/robot/helper_plan_resume.py new file mode 100644 index 00000000..433af958 --- /dev/null +++ b/robot/helper_plan_resume.py @@ -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]} ", 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() diff --git a/robot/plan_resume.robot b/robot/plan_resume.robot new file mode 100644 index 00000000..b8c7290f --- /dev/null +++ b/robot/plan_resume.robot @@ -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 diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index c48d981c..ce92b74c 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -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, diff --git a/src/cleveragents/application/services/plan_resume_service.py b/src/cleveragents/application/services/plan_resume_service.py new file mode 100644 index 00000000..39498f4e --- /dev/null +++ b/src/cleveragents/application/services/plan_resume_service.py @@ -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 diff --git a/src/cleveragents/cli/commands/plan.py b/src/cleveragents/cli/commands/plan.py index 3c206041..39419b5e 100644 --- a/src/cleveragents/cli/commands/plan.py +++ b/src/cleveragents/cli/commands/plan.py @@ -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. diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index 4a2eeab5..c5157e1b 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -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", diff --git a/src/cleveragents/domain/models/core/plan.py b/src/cleveragents/domain/models/core/plan.py index 757704c1..b52efd4b 100644 --- a/src/cleveragents/domain/models/core/plan.py +++ b/src/cleveragents/domain/models/core/plan.py @@ -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( diff --git a/src/cleveragents/domain/models/core/resume.py b/src/cleveragents/domain/models/core/resume.py new file mode 100644 index 00000000..b5e066ad --- /dev/null +++ b/src/cleveragents/domain/models/core/resume.py @@ -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