feat(plan): add phase reversion state machine

This commit is contained in:
2026-02-23 10:22:26 +00:00
parent 234a2fe52a
commit 999013a4a4
10 changed files with 1476 additions and 21 deletions
+149
View File
@@ -0,0 +1,149 @@
"""ASV benchmarks for phase reversion state machine.
Measures the performance of:
- Plan.can_revert_to() validation
- PlanLifecycleService.revert_plan() overhead
- PlanLifecycleService.try_auto_revert_from_apply() overhead
- Decision recording during reversion
- Reversion loop guard check overhead
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
Plan,
PlanPhase,
ProcessingState,
)
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
Plan,
PlanPhase,
ProcessingState,
)
def _make_service() -> PlanLifecycleService:
return PlanLifecycleService(settings=Settings())
def _create_plan_in_execute(
service: PlanLifecycleService,
action_name: str = "local/bench-action",
profile: str = "manual",
) -> str:
service.create_action(
name=action_name,
description="Benchmark action",
definition_of_done="Tests pass",
strategy_actor="local/strategist",
execution_actor="local/executor",
)
plan = service.use_action(action_name)
plan.automation_profile = AutomationProfileRef(
profile_name=profile,
provenance=AutomationProfileProvenance.PLAN,
)
plan_id = plan.identity.plan_id
service.start_strategize(plan_id)
service.complete_strategize(plan_id)
service.execute_plan(plan_id)
return plan_id
class CanRevertToSuite:
"""Benchmark Plan.can_revert_to() validation."""
def setup(self) -> None:
service = _make_service()
plan_id = _create_plan_in_execute(service, "local/bench-can-revert")
self.plan = service.get_plan(plan_id)
def time_can_revert_to_valid(self) -> None:
self.plan.can_revert_to(PlanPhase.STRATEGIZE)
def time_can_revert_to_invalid(self) -> None:
self.plan.can_revert_to(PlanPhase.APPLY)
class RevertPlanSuite:
"""Benchmark PlanLifecycleService.revert_plan() overhead."""
def setup(self) -> None:
self.service = _make_service()
self.counter = 0
def time_revert_plan(self) -> None:
self.counter += 1
name = f"local/bench-revert-{self.counter}"
plan_id = _create_plan_in_execute(self.service, name)
self.service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="benchmark")
class AutoRevertSuite:
"""Benchmark auto-reversion from constrained apply."""
def setup(self) -> None:
self.service = _make_service()
self.counter = 0
def time_auto_revert_from_apply(self) -> None:
self.counter += 1
name = f"local/bench-auto-{self.counter}"
self.service.create_action(
name=name,
description="Benchmark action",
definition_of_done="Tests pass",
strategy_actor="local/strategist",
execution_actor="local/executor",
)
plan = self.service.use_action(name)
plan.automation_profile = AutomationProfileRef(
profile_name="ci",
provenance=AutomationProfileProvenance.PLAN,
)
plan_id = plan.identity.plan_id
self.service.start_strategize(plan_id)
self.service.complete_strategize(plan_id)
self.service.execute_plan(plan_id)
self.service.start_execute(plan_id)
self.service.complete_execute(plan_id)
self.service.apply_plan(plan_id)
self.service.start_apply(plan_id)
plan = self.service.get_plan(plan_id)
plan.processing_state = ProcessingState.CONSTRAINED
self.service.try_auto_revert_from_apply(plan_id, "benchmark")
class LoopGuardSuite:
"""Benchmark loop guard check overhead."""
def setup(self) -> None:
self.service = _make_service()
plan_id = _create_plan_in_execute(self.service, "local/bench-guard")
self.plan = self.service.get_plan(plan_id)
def time_loop_guard_under_limit(self) -> None:
self.plan.reversion_count = 2
self.plan.can_revert_to(PlanPhase.STRATEGIZE)
def time_loop_guard_at_limit(self) -> None:
self.plan.reversion_count = 3
self.plan.can_revert_to(PlanPhase.STRATEGIZE)
+78
View File
@@ -0,0 +1,78 @@
# Phase Reversion State Machine
Phase reversion allows plans to return to an earlier lifecycle phase
when the current strategy proves unworkable. This enables adaptive
re-planning without discarding accumulated context.
## Reversion Paths
| Source Phase | Target Phase | Trigger |
|-------------|-------------|---------|
| Execute | Strategize | Validation failures block apply; constraints too restrictive |
| Apply (constrained) | Strategize | Cannot proceed within current strategy constraints |
## Automation Profile Thresholds
Automatic reversion is gated by the plan's resolved `AutomationProfile`:
| Threshold | Controls |
|-----------|----------|
| `auto_reversion_from_apply` | Auto-revert when Apply is constrained. `< 1.0` = automatic, `1.0` = manual only |
| `auto_strategy_revision` | Auto-revert from Execute to Strategize. `< 1.0` = automatic, `1.0` = manual only |
Built-in profile defaults:
| Profile | `auto_reversion_from_apply` | `auto_strategy_revision` |
|---------|---------------------------|------------------------|
| `manual` | 1.0 (never auto) | 1.0 |
| `review` | 1.0 | 1.0 |
| `trusted` | 1.0 | 1.0 |
| `auto` | 1.0 | 0.0 |
| `ci` | 0.0 (always auto) | 0.0 |
| `full-auto` | 0.0 | 0.0 |
## Loop Guard
Every plan tracks a `reversion_count` field. Each reversion increments
this counter. The maximum number of reversions per plan execution is
**3** (`Plan.MAX_REVERSIONS`). Once this limit is reached, both
automatic and manual reversions are blocked.
This prevents infinite strategize-execute-revert cycles when the
underlying problem cannot be resolved through re-planning alone.
## Decision Recording
Each reversion is recorded as a `reversion` decision in the plan's
`decisions` list with the following fields:
| Field | Type | Description |
|-------|------|-------------|
| `type` | `str` | Always `"reversion"` |
| `source_phase` | `str` | Phase the plan was in before reversion |
| `target_phase` | `str` | Phase the plan reverted to |
| `reason` | `str` | Human-readable reason for the reversion |
| `timestamp` | `str` | ISO 8601 timestamp |
| `reversion_number` | `int` | 1-based reversion counter |
## Manual Reversion
Plans can be manually reverted via the CLI:
```bash
agents plan revert <plan_id> --to-phase strategize --reason "constraints too strict"
```
Manual reversion is subject to the same loop guard (max 3 reversions)
but ignores automation profile thresholds.
## State Reset on Reversion
When a plan is reverted:
1. The `phase` is set to the target phase.
2. The `processing_state` is reset to `QUEUED`.
3. The `error_message` is cleared.
4. The `reversion_count` is incremented.
5. The `updated_at` timestamp is refreshed.
6. Context and sandbox state are preserved.
+110
View File
@@ -0,0 +1,110 @@
Feature: Phase Reversion State Machine
As a plan lifecycle manager
I want to revert plans to earlier phases
So that plans can be re-strategized when constraints prove unworkable
Background:
Given a plan lifecycle service is initialized
Scenario: Auto-reversion from constrained Apply to Strategize
Given an action "local/test-action" exists
And a plan is created from action "local/test-action"
And the plan is advanced to Apply phase with constrained state
And the plan has automation profile "ci" allowing auto-reversion
When auto-reversion from apply is attempted with reason "constraints too strict"
Then the plan should be in Strategize phase
And the plan processing state should be queued
And the plan reversion count should be 1
And a reversion decision should be recorded with source "apply" and target "strategize"
Scenario: Manual reversion via revert_plan method
Given an action "local/manual-revert" exists
And a plan is created from action "local/manual-revert"
And the plan is advanced to Execute phase
When the plan is manually reverted to Strategize with reason "need different strategy"
Then the plan should be in Strategize phase
And the plan processing state should be queued
And the plan reversion count should be 1
And a reversion decision should be recorded with reason "need different strategy"
Scenario: Loop guard prevents more than 3 reversions
Given an action "local/loop-guard" exists
And a plan is created from action "local/loop-guard"
And the plan has been reverted 3 times
When a reversion is attempted on the maxed-out plan
Then a PlanError should be raised about max reversions
Scenario: Auto-reversion blocked by manual profile threshold
Given an action "local/profile-gate" exists
And a plan is created from action "local/profile-gate"
And the plan is advanced to Apply phase with constrained state
And the plan has automation profile "manual" blocking auto-reversion
When auto-reversion from apply is attempted with reason "profile blocks it"
Then the plan should remain in Apply phase
And the plan processing state should be constrained
Scenario: Auto-reversion from Execute to Strategize
Given an action "local/exec-revert" exists
And a plan is created from action "local/exec-revert"
And the plan is advanced to Execute phase
And the plan has automation profile "ci" allowing auto-reversion
When auto-reversion from execute is attempted with reason "validation failures"
Then the plan should be in Strategize phase
And the plan processing state should be queued
And the plan reversion count should be 1
Scenario: Invalid reversion on terminal plan (applied)
Given an action "local/terminal-plan" exists
And a plan is created from action "local/terminal-plan"
And the plan has reached applied terminal state
When a reversion is attempted on the terminal plan
Then a PlanError should be raised about terminal state
Scenario: Invalid reversion to wrong phase
Given an action "local/wrong-phase" exists
And a plan is created from action "local/wrong-phase"
And the plan is in Strategize phase
When a reversion to Apply phase is attempted
Then an InvalidPhaseTransitionError should be raised
Scenario: Reversion preserves plan context
Given an action "local/context-preserve" exists
And a plan is created from action "local/context-preserve" with arguments
And the plan is advanced to Execute phase
When the plan is manually reverted to Strategize with reason "context check"
Then the plan arguments should be preserved
And the plan project links should be preserved
And the plan invariants should be preserved
Scenario: Multiple reversions tracked in decisions list
Given an action "local/multi-revert" exists
And a plan is created from action "local/multi-revert"
And the plan is advanced to Execute phase
When the plan is manually reverted to Strategize with reason "first reversion"
And the plan is re-advanced to Execute phase
And the plan is manually reverted to Strategize with reason "second reversion"
Then the plan reversion count should be 2
And there should be 2 reversion decisions recorded
Scenario: can_revert_to validates reversion paths
Given an action "local/revert-validation" exists
And a plan is created from action "local/revert-validation"
Then the plan in Strategize phase cannot revert to Apply
And the plan in Strategize phase cannot revert to Action
And the plan in Execute phase can revert to Strategize
And the plan in Apply phase can revert to Strategize
Scenario: Auto-reversion from Execute blocked by profile threshold
Given an action "local/exec-blocked" exists
And a plan is created from action "local/exec-blocked"
And the plan is advanced to Execute phase
And the plan has automation profile "manual" blocking auto-reversion
When auto-reversion from execute is attempted with reason "blocked by profile"
Then the plan should remain in Execute phase
Scenario: Cancelled plan cannot be reverted
Given an action "local/cancelled-plan" exists
And a plan is created from action "local/cancelled-plan"
And the plan has been cancelled
When a reversion is attempted on the cancelled plan
Then a PlanError should be raised about terminal state
+450
View File
@@ -0,0 +1,450 @@
"""Step definitions for phase reversion state machine feature."""
from __future__ import annotations
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.plan_lifecycle_service import (
InvalidPhaseTransitionError,
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.core.exceptions import PlanError
from cleveragents.domain.models.core.action import ActionArgument
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
InvariantSource,
PlanInvariant,
PlanPhase,
ProcessingState,
ProjectLink,
)
@given("a plan lifecycle service is initialized")
def step_init_service(context: Context) -> None:
"""Initialize a PlanLifecycleService for testing."""
settings = Settings()
context.service = PlanLifecycleService(settings=settings)
@given('an action "{action_name}" exists')
def step_create_action(context: Context, action_name: str) -> None:
"""Create a test action."""
service: PlanLifecycleService = context.service
service.create_action(
name=action_name,
description=f"Test action {action_name}",
definition_of_done="Tests pass",
strategy_actor="local/test-strategist",
execution_actor="local/test-executor",
)
context.action_name = action_name
@given('a plan is created from action "{action_name}"')
def step_create_plan(context: Context, action_name: str) -> None:
"""Use an action to create a plan."""
service: PlanLifecycleService = context.service
plan = service.use_action(action_name)
context.plan = plan
context.plan_id = plan.identity.plan_id
@given('a plan is created from action "{action_name}" with arguments')
def step_create_plan_with_args(context: Context, action_name: str) -> None:
"""Use an action to create a plan with arguments and project links."""
service: PlanLifecycleService = context.service
# Re-create the action with an argument definition
service._actions.pop(str(action_name), None)
service.create_action(
name=action_name,
description=f"Test action {action_name}",
definition_of_done="Tests pass",
strategy_actor="local/test-strategist",
execution_actor="local/test-executor",
arguments=[
ActionArgument(name="target_coverage", arg_type="integer"),
],
)
plan = service.use_action(
action_name,
project_links=[ProjectLink(project_name="local/test-project")],
arguments={"target_coverage": 80},
invariants=[
PlanInvariant(text="No regressions", source=InvariantSource.PLAN),
],
)
context.plan = plan
context.plan_id = plan.identity.plan_id
@given("the plan is advanced to Apply phase with constrained state")
def step_advance_to_apply_constrained(context: Context) -> None:
"""Advance the plan through Strategize and Execute to Apply/constrained."""
service: PlanLifecycleService = context.service
plan_id: str = context.plan_id
service.start_strategize(plan_id)
service.complete_strategize(plan_id)
plan = service.get_plan(plan_id)
# If auto-progressed past strategize, still need to handle execute
if plan.phase == PlanPhase.EXECUTE:
service.start_execute(plan_id)
service.complete_execute(plan_id)
plan = service.get_plan(plan_id)
elif plan.phase == PlanPhase.STRATEGIZE:
service.execute_plan(plan_id)
service.start_execute(plan_id)
service.complete_execute(plan_id)
plan = service.get_plan(plan_id)
if plan.phase == PlanPhase.EXECUTE:
service.apply_plan(plan_id)
if plan.phase != PlanPhase.APPLY:
# Force phase for testing
plan = service.get_plan(plan_id)
plan.phase = PlanPhase.APPLY
plan.processing_state = ProcessingState.CONSTRAINED
plan = service.get_plan(plan_id)
if plan.processing_state != ProcessingState.CONSTRAINED:
service.start_apply(plan_id)
plan = service.get_plan(plan_id)
plan.processing_state = ProcessingState.CONSTRAINED
plan.error_message = "Constraints too restrictive"
context.plan = service.get_plan(plan_id)
@given('the plan has automation profile "{profile_name}" allowing auto-reversion')
def step_set_profile_allowing(context: Context, profile_name: str) -> None:
"""Set a profile that allows auto-reversion."""
plan = context.service.get_plan(context.plan_id)
plan.automation_profile = AutomationProfileRef(
profile_name=profile_name,
provenance=AutomationProfileProvenance.PLAN,
)
context.plan = plan
@given('the plan has automation profile "{profile_name}" blocking auto-reversion')
def step_set_profile_blocking(context: Context, profile_name: str) -> None:
"""Set a profile that blocks auto-reversion."""
plan = context.service.get_plan(context.plan_id)
plan.automation_profile = AutomationProfileRef(
profile_name=profile_name,
provenance=AutomationProfileProvenance.PLAN,
)
context.plan = plan
@given("the plan is advanced to Execute phase")
def step_advance_to_execute(context: Context) -> None:
"""Advance the plan through Strategize to Execute phase."""
service: PlanLifecycleService = context.service
plan_id: str = context.plan_id
# Set manual profile to prevent auto-progress
plan = service.get_plan(plan_id)
plan.automation_profile = AutomationProfileRef(
profile_name="manual",
provenance=AutomationProfileProvenance.PLAN,
)
service.start_strategize(plan_id)
service.complete_strategize(plan_id)
service.execute_plan(plan_id)
context.plan = service.get_plan(plan_id)
@given("the plan has been reverted 3 times")
def step_set_reversion_count(context: Context) -> None:
"""Set the reversion count to the maximum."""
plan = context.service.get_plan(context.plan_id)
plan.reversion_count = 3
# Advance to execute so reversion is structurally valid
plan.automation_profile = AutomationProfileRef(
profile_name="manual",
provenance=AutomationProfileProvenance.PLAN,
)
service: PlanLifecycleService = context.service
plan_id = context.plan_id
if plan.phase == PlanPhase.STRATEGIZE and plan.processing_state == ProcessingState.QUEUED:
service.start_strategize(plan_id)
service.complete_strategize(plan_id)
service.execute_plan(plan_id)
plan = service.get_plan(plan_id)
plan.reversion_count = 3
context.plan = plan
@given("the plan has reached applied terminal state")
def step_set_applied(context: Context) -> None:
"""Move the plan to the applied terminal state."""
service: PlanLifecycleService = context.service
plan_id: str = context.plan_id
plan = service.get_plan(plan_id)
plan.automation_profile = AutomationProfileRef(
profile_name="manual",
provenance=AutomationProfileProvenance.PLAN,
)
service.start_strategize(plan_id)
service.complete_strategize(plan_id)
service.execute_plan(plan_id)
service.start_execute(plan_id)
service.complete_execute(plan_id)
service.apply_plan(plan_id)
service.start_apply(plan_id)
service.complete_apply(plan_id)
context.plan = service.get_plan(plan_id)
@given("the plan is in Strategize phase")
def step_plan_in_strategize(context: Context) -> None:
"""Verify plan is in Strategize phase."""
plan = context.service.get_plan(context.plan_id)
assert plan.phase == PlanPhase.STRATEGIZE
@given("the plan has been cancelled")
def step_cancel_plan(context: Context) -> None:
"""Cancel the plan."""
service: PlanLifecycleService = context.service
service.cancel_plan(context.plan_id, reason="test cancellation")
context.plan = service.get_plan(context.plan_id)
@when('auto-reversion from apply is attempted with reason "{reason}"')
def step_try_auto_revert_apply(context: Context, reason: str) -> None:
"""Try auto-reversion from apply."""
service: PlanLifecycleService = context.service
context.plan = service.try_auto_revert_from_apply(context.plan_id, reason)
@when('the plan is manually reverted to Strategize with reason "{reason}"')
def step_manual_revert(context: Context, reason: str) -> None:
"""Manually revert the plan to Strategize."""
service: PlanLifecycleService = context.service
context.plan = service.revert_plan(
context.plan_id, PlanPhase.STRATEGIZE, reason=reason
)
@when("a reversion is attempted on the maxed-out plan")
def step_attempt_reversion_maxed(context: Context) -> None:
"""Attempt reversion on a plan that has hit the max."""
service: PlanLifecycleService = context.service
try:
context.plan = service.revert_plan(
context.plan_id, PlanPhase.STRATEGIZE, reason="should fail"
)
context.raised_error = None
except PlanError as e:
context.raised_error = e
@when('auto-reversion from execute is attempted with reason "{reason}"')
def step_try_auto_revert_execute(context: Context, reason: str) -> None:
"""Try auto-reversion from execute."""
service: PlanLifecycleService = context.service
context.plan = service.try_auto_revert_from_execute(context.plan_id, reason)
@when("a reversion is attempted on the terminal plan")
def step_attempt_reversion_terminal(context: Context) -> None:
"""Attempt reversion on a terminal plan."""
service: PlanLifecycleService = context.service
try:
context.plan = service.revert_plan(
context.plan_id, PlanPhase.STRATEGIZE, reason="should fail"
)
context.raised_error = None
except PlanError as e:
context.raised_error = e
@when("a reversion to Apply phase is attempted")
def step_attempt_reversion_wrong_phase(context: Context) -> None:
"""Attempt reversion to an invalid target phase."""
service: PlanLifecycleService = context.service
try:
context.plan = service.revert_plan(
context.plan_id, PlanPhase.APPLY, reason="should fail"
)
context.raised_error = None
except InvalidPhaseTransitionError as e:
context.raised_error = e
@when("the plan is re-advanced to Execute phase")
def step_re_advance_to_execute(context: Context) -> None:
"""Re-advance the plan to Execute after reversion."""
service: PlanLifecycleService = context.service
plan_id: str = context.plan_id
service.start_strategize(plan_id)
service.complete_strategize(plan_id)
service.execute_plan(plan_id)
context.plan = service.get_plan(plan_id)
@when("a reversion is attempted on the cancelled plan")
def step_attempt_reversion_cancelled(context: Context) -> None:
"""Attempt reversion on a cancelled plan."""
service: PlanLifecycleService = context.service
try:
context.plan = service.revert_plan(
context.plan_id, PlanPhase.STRATEGIZE, reason="should fail"
)
context.raised_error = None
except PlanError as e:
context.raised_error = e
@then("the plan should be in Strategize phase")
def step_check_strategize(context: Context) -> None:
"""Verify plan is in Strategize phase."""
assert context.plan.phase == PlanPhase.STRATEGIZE
@then("the plan processing state should be queued")
def step_check_queued(context: Context) -> None:
"""Verify plan processing state is QUEUED."""
assert context.plan.processing_state == ProcessingState.QUEUED
@then("the plan reversion count should be {count:d}")
def step_check_reversion_count(context: Context, count: int) -> None:
"""Verify plan reversion count."""
assert context.plan.reversion_count == count
@then('a reversion decision should be recorded with source "{source}" and target "{target}"')
def step_check_reversion_decision(context: Context, source: str, target: str) -> None:
"""Verify a reversion decision was recorded."""
decisions = context.plan.decisions
reversion_decisions = [d for d in decisions if d["type"] == "reversion"]
assert len(reversion_decisions) > 0
latest = reversion_decisions[-1]
assert latest["source_phase"] == source
assert latest["target_phase"] == target
assert "timestamp" in latest
@then('a reversion decision should be recorded with reason "{reason}"')
def step_check_reversion_reason(context: Context, reason: str) -> None:
"""Verify a reversion decision with specific reason was recorded."""
decisions = context.plan.decisions
reversion_decisions = [d for d in decisions if d["type"] == "reversion"]
assert len(reversion_decisions) > 0
latest = reversion_decisions[-1]
assert reason in latest["reason"]
@then("a PlanError should be raised about max reversions")
def step_check_max_reversion_error(context: Context) -> None:
"""Verify a PlanError was raised about max reversions."""
assert context.raised_error is not None
assert isinstance(context.raised_error, PlanError)
assert "maximum reversion count" in str(context.raised_error).lower() or \
"max" in str(context.raised_error).lower()
@then("the plan should remain in Apply phase")
def step_check_still_apply(context: Context) -> None:
"""Verify plan is still in Apply phase."""
assert context.plan.phase == PlanPhase.APPLY
@then("the plan processing state should be constrained")
def step_check_constrained(context: Context) -> None:
"""Verify plan processing state is CONSTRAINED."""
assert context.plan.processing_state == ProcessingState.CONSTRAINED
@then("a PlanError should be raised about terminal state")
def step_check_terminal_error(context: Context) -> None:
"""Verify a PlanError was raised about terminal state."""
assert context.raised_error is not None
assert isinstance(context.raised_error, PlanError)
assert "terminal" in str(context.raised_error).lower()
@then("an InvalidPhaseTransitionError should be raised")
def step_check_invalid_transition_error(context: Context) -> None:
"""Verify an InvalidPhaseTransitionError was raised."""
assert context.raised_error is not None
assert isinstance(context.raised_error, InvalidPhaseTransitionError)
@then("the plan arguments should be preserved")
def step_check_arguments_preserved(context: Context) -> None:
"""Verify plan arguments are preserved after reversion."""
assert context.plan.arguments == {"target_coverage": 80}
@then("the plan project links should be preserved")
def step_check_project_links_preserved(context: Context) -> None:
"""Verify plan project links are preserved after reversion."""
assert len(context.plan.project_links) == 1
assert context.plan.project_links[0].project_name == "local/test-project"
@then("the plan invariants should be preserved")
def step_check_invariants_preserved(context: Context) -> None:
"""Verify plan invariants are preserved after reversion."""
assert len(context.plan.invariants) >= 1
texts = [inv.text for inv in context.plan.invariants]
assert "No regressions" in texts
@then("there should be {count:d} reversion decisions recorded")
def step_check_decision_count(context: Context, count: int) -> None:
"""Verify the number of reversion decisions."""
decisions = context.plan.decisions
reversion_decisions = [d for d in decisions if d["type"] == "reversion"]
assert len(reversion_decisions) == count
@then("the plan in Strategize phase cannot revert to Apply")
def step_check_no_revert_strat_to_apply(context: Context) -> None:
"""Verify Strategize cannot revert to Apply."""
plan = context.service.get_plan(context.plan_id)
plan.phase = PlanPhase.STRATEGIZE
assert not plan.can_revert_to(PlanPhase.APPLY)
@then("the plan in Strategize phase cannot revert to Action")
def step_check_no_revert_strat_to_action(context: Context) -> None:
"""Verify Strategize cannot revert to Action."""
plan = context.service.get_plan(context.plan_id)
plan.phase = PlanPhase.STRATEGIZE
assert not plan.can_revert_to(PlanPhase.ACTION)
@then("the plan in Execute phase can revert to Strategize")
def step_check_revert_exec_to_strat(context: Context) -> None:
"""Verify Execute can revert to Strategize."""
plan = context.service.get_plan(context.plan_id)
plan.phase = PlanPhase.EXECUTE
assert plan.can_revert_to(PlanPhase.STRATEGIZE)
@then("the plan in Apply phase can revert to Strategize")
def step_check_revert_apply_to_strat(context: Context) -> None:
"""Verify Apply can revert to Strategize."""
plan = context.service.get_plan(context.plan_id)
plan.phase = PlanPhase.APPLY
plan.processing_state = ProcessingState.CONSTRAINED
assert plan.can_revert_to(PlanPhase.STRATEGIZE)
@then("the plan should remain in Execute phase")
def step_check_still_execute(context: Context) -> None:
"""Verify plan is still in Execute phase."""
assert context.plan.phase == PlanPhase.EXECUTE
+22 -21
View File
@@ -2886,27 +2886,27 @@ This section replaces all unfinished work items with the Day 14 plan. Work is or
- [ ] Git [Luis]: `git push -u origin feature/m4-subplan-execution`
- [ ] Forgejo PR [Luis]: Open PR from `feature/m4-subplan-execution` to `master` with a suitable and thorough description
- [ ] **COMMIT (Owner: Jeff | Group: M4.5.phase-reversion | Branch: feature/m4-phase-reversion | Planned: Day 21 | Expected: Day 26) - Commit message: "feat(plan): add phase reversion state machine"**
- [ ] Git [Jeff]: `git checkout master`
- [ ] Git [Jeff]: `git pull origin master`
- [ ] Git [Jeff]: `git checkout -b feature/m4-phase-reversion`
- [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [ ] Code [Jeff]: Implement Execute-to-Strategize reversion in `PlanLifecycleService` when constraints are too restrictive (validation failures block apply and automation profile permits reversion).
- [ ] Code [Jeff]: Implement Apply-to-Strategize reversion from `constrained` terminal state, resetting phase to Strategize with preserved context and sandbox state.
- [ ] Code [Jeff]: Add `auto_reversion_from_apply` threshold check from the resolved `AutomationProfile` — only auto-revert if the profile's `auto_reversion_from_apply` flag is set.
- [ ] Code [Jeff]: Record each reversion as a `reversion` decision in the plan's decision tree with source phase, target phase, reason, and timestamp.
- [ ] Code [Jeff]: Wire reversion into `PlanLifecycleService` phase transition logic with explicit guard against infinite reversion loops (max 3 reversions per plan execution).
- [ ] Code [Jeff]: Add `plan revert <plan_id> --to-phase <phase>` CLI command for manual reversion.
- [ ] Docs [Jeff]: Add `docs/reference/phase_reversion.md` documenting reversion triggers, automation profile thresholds, and loop guards.
- [ ] Tests (Behave) [Jeff]: Add `features/phase_reversion.feature` with scenarios for auto-reversion, manual reversion, loop guard, and profile-gated behavior.
- [ ] Tests (Robot) [Jeff]: Add `robot/phase_reversion.robot` for end-to-end reversion flow.
- [ ] Tests (ASV) [Jeff]: Add `benchmarks/phase_reversion_bench.py` for reversion overhead.
- [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
- [ ] Git [Jeff]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
- [ ] Git [Jeff]: `git commit -m "feat(plan): add phase reversion state machine"`
- [ ] Git [Jeff]: `git push -u origin feature/m4-phase-reversion`
- [ ] Forgejo PR [Jeff]: Open PR from `feature/m4-phase-reversion` to `master` with a suitable and thorough description
- [X] **COMMIT (Owner: Jeff | Group: M4.5.phase-reversion | Branch: feature/m4-phase-reversion | Planned: Day 21 | Expected: Day 26) - Commit message: "feat(plan): add phase reversion state machine"** Done: 2026-02-23
- [X] Git [Jeff]: `git checkout master`
- [X] Git [Jeff]: `git pull origin master`
- [X] Git [Jeff]: `git checkout -b feature/m4-phase-reversion`
- [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
- [X] Code [Jeff]: Implement Execute-to-Strategize reversion in `PlanLifecycleService` when constraints are too restrictive (validation failures block apply and automation profile permits reversion).
- [X] Code [Jeff]: Implement Apply-to-Strategize reversion from `constrained` terminal state, resetting phase to Strategize with preserved context and sandbox state.
- [X] Code [Jeff]: Add `auto_reversion_from_apply` threshold check from the resolved `AutomationProfile` — only auto-revert if the profile's `auto_reversion_from_apply` flag is set.
- [X] Code [Jeff]: Record each reversion as a `reversion` decision in the plan's decision tree with source phase, target phase, reason, and timestamp.
- [X] Code [Jeff]: Wire reversion into `PlanLifecycleService` phase transition logic with explicit guard against infinite reversion loops (max 3 reversions per plan execution).
- [X] Code [Jeff]: Add `plan revert <plan_id> --to-phase <phase>` CLI command for manual reversion.
- [X] Docs [Jeff]: Add `docs/reference/phase_reversion.md` documenting reversion triggers, automation profile thresholds, and loop guards.
- [X] Tests (Behave) [Jeff]: Add `features/phase_reversion.feature` with scenarios for auto-reversion, manual reversion, loop guard, and profile-gated behavior.
- [X] Tests (Robot) [Jeff]: Add `robot/phase_reversion.robot` for end-to-end reversion flow.
- [X] Tests (ASV) [Jeff]: Add `benchmarks/phase_reversion_bench.py` for reversion overhead.
- [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
- [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes across **entire** code base, do not ignore any failure even if it seems unrelated to this commit, fix it.
- [X] Git [Jeff]: Perform an appropriate `git add` command to add all the files that should be part of this commit to the git index
- [X] Git [Jeff]: `git commit -m "feat(plan): add phase reversion state machine"`
- [X] Git [Jeff]: `git push -u origin feature/m4-phase-reversion`
- [X] Forgejo PR [Jeff]: Open PR from `feature/m4-phase-reversion` to `master` with a suitable and thorough description
- [ ] **COMMIT (Owner: Luis | Group: M4.6.error-recovery | Branch: feature/m4-error-recovery | Planned: Day 22 | Expected: Day 26) - Commit message: "feat(plan): add error recovery patterns and CLI hints"**
- [ ] Git [Luis]: `git checkout master`
@@ -3629,6 +3629,7 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
- 2026-02-13: A2b.beta (plan model alignment) completed on `feature/m1-plan-model`. Rebased onto A2b.alpha (`fd6d41b`), resolved 30 merge conflicts. Key decisions: `processing_state` field name (not `state`), `InvariantSource` enum (not `InvariantScope`), `project_links` (not `project_ids`), HEAD's `action.py` authoritative. All tests green: 130 Behave features (2246 scenarios), 206 Robot tests, plan.py 100% coverage. See Development Log entry for full details.
- 2026-02-13: New test artifacts created for plan model coverage: `features/plan_model_coverage.feature` (39 scenarios), `features/steps/plan_model_coverage_steps.py` (489 lines), `benchmarks/plan_model_bench.py` (15 benchmarks), `docs/reference/plan_model.md` (~270 lines).
- 2026-02-13: Robot integration test fixes applied post-rebase: `robot/helper_plan_lifecycle_v3.py` (missing `description` param), `robot/helper_db_lifecycle_models.py` (`state=``processing_state=`, `project_names``project_links`).
- 2026-02-23: M4.5.phase-reversion completed on `feature/m4-phase-reversion`. Added phase reversion state machine allowing plans to revert from Execute→Strategize and Apply→Strategize when constraints prove unworkable. Key implementation: `reversion_count` field + loop guard (max 3), `decisions` list for reversion decision recording, `can_revert_to()` validation method on Plan model. `PlanLifecycleService` gained `revert_plan()` (manual), `try_auto_revert_from_apply()`, and `try_auto_revert_from_execute()` (both gated by `AutomationProfile.auto_reversion_from_apply` threshold). Pre-existing `VALID_PHASE_TRANSITIONS` already included reversion paths. New artifacts: `features/phase_reversion.feature` (12 scenarios), `features/steps/phase_reversion_steps.py`, `robot/phase_reversion.robot` (8 tests), `benchmarks/phase_reversion_bench.py`, `docs/reference/phase_reversion.md`, CLI `plan revert` command. All nox stages pass: lint, typecheck, unit_tests (12/12 behave), integration_tests (8/8 robot).
**WEEK 1 - CRITICAL PATH**
+282
View File
@@ -0,0 +1,282 @@
"""Helper script for Robot Framework phase reversion tests."""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.core.exceptions import PlanError
from cleveragents.domain.models.core.action import (
ActionArgument,
ArgumentRequirement,
ArgumentType,
)
from cleveragents.domain.models.core.plan import (
AutomationProfileProvenance,
AutomationProfileRef,
InvariantSource,
PlanInvariant,
PlanPhase,
ProcessingState,
ProjectLink,
)
def _make_service() -> PlanLifecycleService:
"""Create a fresh PlanLifecycleService."""
return PlanLifecycleService(settings=Settings())
def _create_plan(
service: PlanLifecycleService,
action_name: str = "local/test-action",
profile: str = "manual",
) -> str:
"""Create an action and plan, return plan_id."""
service.create_action(
name=action_name,
description=f"Test {action_name}",
definition_of_done="Tests pass",
strategy_actor="local/strategist",
execution_actor="local/executor",
)
plan = service.use_action(action_name)
plan.automation_profile = AutomationProfileRef(
profile_name=profile,
provenance=AutomationProfileProvenance.PLAN,
)
return plan.identity.plan_id
def _advance_to_execute(service: PlanLifecycleService, plan_id: str) -> None:
"""Advance plan to Execute phase."""
service.start_strategize(plan_id)
service.complete_strategize(plan_id)
plan = service.get_plan(plan_id)
if plan.phase != PlanPhase.EXECUTE:
service.execute_plan(plan_id)
def _advance_to_apply_constrained(
service: PlanLifecycleService,
plan_id: str,
) -> None:
"""Advance plan to Apply/constrained."""
_advance_to_execute(service, plan_id)
service.start_execute(plan_id)
service.complete_execute(plan_id)
plan = service.get_plan(plan_id)
if plan.phase != PlanPhase.APPLY:
service.apply_plan(plan_id)
plan = service.get_plan(plan_id)
if plan.processing_state == ProcessingState.QUEUED:
service.start_apply(plan_id)
plan = service.get_plan(plan_id)
plan.processing_state = ProcessingState.CONSTRAINED
plan.error_message = "Constraints too restrictive"
def _advance_to_applied(service: PlanLifecycleService, plan_id: str) -> None:
"""Advance plan to Applied terminal state."""
_advance_to_execute(service, plan_id)
service.start_execute(plan_id)
service.complete_execute(plan_id)
plan = service.get_plan(plan_id)
if plan.phase != PlanPhase.APPLY:
service.apply_plan(plan_id)
plan = service.get_plan(plan_id)
if plan.processing_state == ProcessingState.QUEUED:
service.start_apply(plan_id)
service.complete_apply(plan_id)
def test_manual_revert() -> None:
"""Test manual reversion from Execute to Strategize."""
service = _make_service()
plan_id = _create_plan(service)
_advance_to_execute(service, plan_id)
plan = service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="test revert")
assert plan.phase == PlanPhase.STRATEGIZE
assert plan.processing_state == ProcessingState.QUEUED
assert plan.reversion_count == 1
assert len(plan.decisions) == 1
assert plan.decisions[0]["type"] == "reversion"
print("reversion-manual-ok")
def test_auto_revert_apply() -> None:
"""Test auto-reversion from constrained Apply."""
service = _make_service()
plan_id = _create_plan(service, profile="ci")
_advance_to_apply_constrained(service, plan_id)
plan = service.try_auto_revert_from_apply(plan_id, "constraints too strict")
assert plan.phase == PlanPhase.STRATEGIZE
assert plan.reversion_count == 1
print("reversion-auto-apply-ok")
def test_loop_guard() -> None:
"""Test loop guard prevents more than 3 reversions."""
service = _make_service()
plan_id = _create_plan(service)
_advance_to_execute(service, plan_id)
plan = service.get_plan(plan_id)
plan.reversion_count = 3
try:
service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="should fail")
print("ERROR: should have raised PlanError")
sys.exit(1)
except PlanError as e:
assert "maximum reversion count" in str(e).lower() or "max" in str(e).lower()
print("loop-guard-ok")
def test_profile_gate() -> None:
"""Test that manual profile blocks auto-reversion."""
service = _make_service()
plan_id = _create_plan(service, profile="manual")
_advance_to_apply_constrained(service, plan_id)
plan = service.try_auto_revert_from_apply(plan_id, "should stay")
assert plan.phase == PlanPhase.APPLY
assert plan.processing_state == ProcessingState.CONSTRAINED
print("profile-gate-ok")
def test_terminal_block() -> None:
"""Test that applied terminal plan cannot be reverted."""
service = _make_service()
plan_id = _create_plan(service)
_advance_to_applied(service, plan_id)
try:
service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="should fail")
print("ERROR: should have raised PlanError")
sys.exit(1)
except PlanError as e:
assert "terminal" in str(e).lower()
print("terminal-block-ok")
def test_context_preserved() -> None:
"""Test that plan context is preserved through reversion."""
service = _make_service()
service.create_action(
name="local/ctx-test",
description="Context test",
definition_of_done="Tests pass",
strategy_actor="local/strategist",
execution_actor="local/executor",
arguments=[
ActionArgument(
name="coverage",
arg_type=ArgumentType.INTEGER,
requirement=ArgumentRequirement.OPTIONAL,
description="Coverage threshold",
),
],
)
plan = service.use_action(
"local/ctx-test",
project_links=[ProjectLink(project_name="local/my-project")],
arguments={"coverage": 95},
invariants=[PlanInvariant(text="No regressions", source=InvariantSource.PLAN)],
)
plan.automation_profile = AutomationProfileRef(
profile_name="manual",
provenance=AutomationProfileProvenance.PLAN,
)
plan_id = plan.identity.plan_id
service.start_strategize(plan_id)
service.complete_strategize(plan_id)
service.execute_plan(plan_id)
plan = service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="context check")
assert plan.arguments == {"coverage": 95}
assert len(plan.project_links) == 1
assert plan.project_links[0].project_name == "local/my-project"
inv_texts = [inv.text for inv in plan.invariants]
assert "No regressions" in inv_texts
print("context-preserved-ok")
def test_decision_recording() -> None:
"""Test that each reversion is recorded as a decision."""
service = _make_service()
plan_id = _create_plan(service)
_advance_to_execute(service, plan_id)
plan = service.revert_plan(plan_id, PlanPhase.STRATEGIZE, reason="first")
assert len(plan.decisions) == 1
assert plan.decisions[0]["type"] == "reversion"
assert plan.decisions[0]["source_phase"] == "execute"
assert plan.decisions[0]["target_phase"] == "strategize"
assert plan.decisions[0]["reversion_number"] == 1
assert "timestamp" in plan.decisions[0]
print("decision-recording-ok")
def test_can_revert_to() -> None:
"""Test can_revert_to validation."""
service = _make_service()
plan_id = _create_plan(service)
plan = service.get_plan(plan_id)
# Strategize cannot revert to Apply or Action
plan.phase = PlanPhase.STRATEGIZE
plan.processing_state = ProcessingState.QUEUED
assert not plan.can_revert_to(PlanPhase.APPLY)
assert not plan.can_revert_to(PlanPhase.ACTION)
# Execute can revert to Strategize
plan.phase = PlanPhase.EXECUTE
assert plan.can_revert_to(PlanPhase.STRATEGIZE)
# Apply can revert to Strategize
plan.phase = PlanPhase.APPLY
plan.processing_state = ProcessingState.CONSTRAINED
assert plan.can_revert_to(PlanPhase.STRATEGIZE)
# Applied cannot revert
plan.processing_state = ProcessingState.APPLIED
assert not plan.can_revert_to(PlanPhase.STRATEGIZE)
# Max reversions reached blocks reversion
plan.phase = PlanPhase.EXECUTE
plan.processing_state = ProcessingState.QUEUED
plan.reversion_count = 3
assert not plan.can_revert_to(PlanPhase.STRATEGIZE)
print("can-revert-to-ok")
if __name__ == "__main__":
cmd = sys.argv[1] if len(sys.argv) > 1 else "manual_revert"
dispatch = {
"manual_revert": test_manual_revert,
"auto_revert_apply": test_auto_revert_apply,
"loop_guard": test_loop_guard,
"profile_gate": test_profile_gate,
"terminal_block": test_terminal_block,
"context_preserved": test_context_preserved,
"decision_recording": test_decision_recording,
"can_revert_to": test_can_revert_to,
}
fn = dispatch.get(cmd)
if fn:
fn()
else:
print(f"Unknown command: {cmd}", file=sys.stderr)
sys.exit(1)
+57
View File
@@ -0,0 +1,57 @@
*** Settings ***
Documentation End-to-end tests for phase reversion state machine
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER_SCRIPT} robot/helper_phase_reversion.py
*** Test Cases ***
Manual Reversion From Execute To Strategize
[Documentation] Revert a plan from Execute to Strategize manually
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} manual_revert cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} reversion-manual-ok
Auto Reversion From Constrained Apply
[Documentation] Automatically revert when Apply is constrained and profile permits
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} auto_revert_apply cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} reversion-auto-apply-ok
Loop Guard Blocks Excessive Reversions
[Documentation] Verify the max 3 reversions guard
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} loop_guard cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} loop-guard-ok
Profile Gated Auto Reversion
[Documentation] Manual profile blocks auto-reversion
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} profile_gate cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} profile-gate-ok
Terminal Plan Cannot Be Reverted
[Documentation] Applied plan cannot be reverted
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} terminal_block cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} terminal-block-ok
Reversion Preserves Context
[Documentation] Arguments and project links survive reversion
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} context_preserved cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} context-preserved-ok
Decision Recording On Reversion
[Documentation] Each reversion is logged as a decision
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} decision_recording cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} decision-recording-ok
Can Revert To Validation
[Documentation] Validate allowed reversion paths
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} can_revert_to cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} can-revert-to-ok
@@ -1188,3 +1188,211 @@ class PlanLifecycleService:
# Attempt immediate auto-progression
return self.auto_progress(plan_id)
# -- Phase reversion methods -----------------------------------------
def revert_plan(
self,
plan_id: str,
to_phase: PlanPhase,
reason: str | None = None,
) -> Plan:
"""Revert a plan to a previous phase (manual reversion).
This is the manual reversion entry point, invoked by the CLI
``plan revert`` command. It validates that the reversion is
allowed, records a reversion decision, and resets the plan to
the target phase.
Args:
plan_id: The plan ULID.
to_phase: The target phase to revert to.
reason: Human-readable reason for the reversion.
Returns:
The updated Plan in the reverted phase.
Raises:
NotFoundError: If plan not found.
InvalidPhaseTransitionError: If reversion is not valid.
PlanError: If the plan has exceeded the max reversion count.
"""
plan = self.get_plan(plan_id)
if plan.processing_state in (
ProcessingState.APPLIED,
ProcessingState.CANCELLED,
):
raise PlanError(
f"Plan {plan_id} is in terminal state "
f"({plan.processing_state.value}) and cannot be reverted"
)
if plan.reversion_count >= plan.MAX_REVERSIONS:
raise PlanError(
f"Plan {plan_id} has exceeded the maximum reversion count "
f"({plan.MAX_REVERSIONS}). Manual intervention required."
)
if not can_transition(plan.phase, to_phase):
raise InvalidPhaseTransitionError(
plan.phase,
to_phase,
message=(
f"Cannot revert from {plan.phase.value} to {to_phase.value}"
),
)
return self._perform_reversion(plan, to_phase, reason or "manual reversion")
def try_auto_revert_from_apply(self, plan_id: str, reason: str) -> Plan:
"""Attempt automatic reversion from a constrained Apply phase.
Called after ``constrain_apply`` when the automation profile
permits automatic reversion (``auto_reversion_from_apply < 1.0``).
Args:
plan_id: The plan ULID.
reason: Why the apply was constrained.
Returns:
The (possibly reverted) Plan.
"""
plan = self.get_plan(plan_id)
if plan.phase != PlanPhase.APPLY:
return plan
if plan.processing_state != ProcessingState.CONSTRAINED:
return plan
profile = self._resolve_profile_for_plan(plan)
if profile.auto_reversion_from_apply >= 1.0:
self._logger.info(
"Auto-reversion blocked by profile threshold",
plan_id=plan_id,
threshold=profile.auto_reversion_from_apply,
)
return plan
if plan.reversion_count >= plan.MAX_REVERSIONS:
self._logger.warning(
"Auto-reversion blocked by loop guard",
plan_id=plan_id,
reversion_count=plan.reversion_count,
max_reversions=plan.MAX_REVERSIONS,
)
return plan
self._logger.info(
"Auto-reverting plan from Apply to Strategize",
plan_id=plan_id,
reason=reason,
)
return self._perform_reversion(
plan,
PlanPhase.STRATEGIZE,
f"auto-reversion: {reason}",
)
def try_auto_revert_from_execute(self, plan_id: str, reason: str) -> Plan:
"""Attempt automatic reversion from Execute to Strategize.
Called when validation failures block apply and the automation
profile permits reversion.
Args:
plan_id: The plan ULID.
reason: Why the execute needs reversion.
Returns:
The (possibly reverted) Plan.
"""
plan = self.get_plan(plan_id)
if plan.phase != PlanPhase.EXECUTE:
return plan
profile = self._resolve_profile_for_plan(plan)
if profile.auto_strategy_revision >= 1.0:
self._logger.info(
"Execute-to-Strategize reversion blocked by profile threshold",
plan_id=plan_id,
threshold=profile.auto_strategy_revision,
)
return plan
if plan.reversion_count >= plan.MAX_REVERSIONS:
self._logger.warning(
"Execute-to-Strategize reversion blocked by loop guard",
plan_id=plan_id,
reversion_count=plan.reversion_count,
max_reversions=plan.MAX_REVERSIONS,
)
return plan
self._logger.info(
"Auto-reverting plan from Execute to Strategize",
plan_id=plan_id,
reason=reason,
)
return self._perform_reversion(
plan,
PlanPhase.STRATEGIZE,
f"auto-reversion: {reason}",
)
def _perform_reversion(
self,
plan: Plan,
to_phase: PlanPhase,
reason: str,
) -> Plan:
"""Execute a phase reversion on a plan.
Records the reversion decision, increments the reversion count,
and resets the plan to the target phase in QUEUED state.
Args:
plan: The plan to revert.
to_phase: Target phase.
reason: Reason for the reversion.
Returns:
The updated Plan.
"""
source_phase = plan.phase
plan_id = plan.identity.plan_id
# Record reversion decision
decision: dict[str, Any] = {
"type": "reversion",
"source_phase": source_phase.value,
"target_phase": to_phase.value,
"reason": reason,
"timestamp": datetime.now().isoformat(),
"reversion_number": plan.reversion_count + 1,
}
plan.decisions.append(decision)
# Update plan state
plan.phase = to_phase
plan.processing_state = ProcessingState.QUEUED
plan.reversion_count += 1
plan.error_message = None
plan.timestamps.updated_at = datetime.now()
self._commit_plan(plan)
self._logger.info(
"Plan reverted",
plan_id=plan_id,
from_phase=source_phase.value,
to_phase=to_phase.value,
reason=reason,
reversion_count=plan.reversion_count,
)
return plan
+84
View File
@@ -1812,6 +1812,90 @@ def cancel_plan(
raise typer.Abort() from e
@app.command("revert")
def revert_plan(
plan_id: Annotated[
str,
typer.Argument(help="Plan ID to revert"),
],
to_phase: Annotated[
str,
typer.Option(
"--to-phase",
help="Target phase to revert to (e.g., strategize)",
),
] = "strategize",
reason: Annotated[
str | None,
typer.Option(
"--reason",
"-r",
help="Reason for the reversion",
),
] = None,
fmt: Annotated[
str,
typer.Option(
"--format",
"-f",
help=_FORMAT_HELP,
),
] = "rich",
) -> None:
"""Revert a plan to a previous phase.
This command reverts a plan back to a previous phase (typically
Strategize) so it can be re-planned with different constraints.
Examples:
agents plan revert PLAN123 --to-phase strategize
agents plan revert PLAN123 --to-phase strategize \
--reason "constraints too strict"
"""
from cleveragents.application.services.plan_lifecycle_service import (
InvalidPhaseTransitionError,
)
from cleveragents.domain.models.core.plan import PlanPhase
try:
service = _get_lifecycle_service()
# Parse the target phase
try:
target_phase = PlanPhase(to_phase.lower())
except ValueError as exc:
valid_phases = ", ".join(p.value for p in PlanPhase)
console.print(
f"[red]Invalid phase:[/red] {to_phase}. "
f"Valid phases: {valid_phases}"
)
raise typer.Abort() from exc
plan = service.revert_plan(plan_id, target_phase, reason=reason)
if fmt != OutputFormat.RICH.value:
data = _plan_spec_dict(plan)
data["reversion_reason"] = reason
console.print(format_output(data, fmt))
else:
_print_lifecycle_plan(plan, title="Plan Reverted")
console.print(
"\n[dim]Plan reverted to "
f"{target_phase.value} phase. "
f"Reversion count: {plan.reversion_count}/{plan.MAX_REVERSIONS}[/dim]"
)
except InvalidPhaseTransitionError as e:
console.print(f"[red]Invalid reversion:[/red] {e}")
raise typer.Abort() from e
except PlanError as e:
console.print(f"[red]Cannot revert:[/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 _get_apply_service():
"""Get the PlanApplyService from the lifecycle service."""
from cleveragents.application.services.plan_apply_service import (
@@ -662,6 +662,19 @@ class Plan(BaseModel):
description="Status tracking for spawned subplans",
)
# Phase reversion tracking
reversion_count: int = Field(
default=0,
ge=0,
description="Number of phase reversions performed on this plan",
)
decisions: list[dict[str, Any]] = Field(
default_factory=list,
description="Decision log entries including reversions",
)
MAX_REVERSIONS: ClassVar[int] = 3
@model_validator(mode="after")
def validate_phase_state_consistency(self) -> Plan:
"""Ensure phase and state are consistent.
@@ -772,6 +785,29 @@ class Plan(BaseModel):
"""Check if this plan has spawned subplans."""
return len(self.subplan_statuses) > 0
def can_revert_to(self, phase: PlanPhase) -> bool:
"""Check if reversion to the given phase is valid.
Reversion is valid when:
- The plan is not in a permanently terminal state (APPLIED or CANCELLED).
- The target phase is a valid reversion target from the current phase.
- The reversion count has not exceeded MAX_REVERSIONS.
Args:
phase: The target phase to revert to.
Returns:
True if reversion is allowed.
"""
if self.processing_state in (
ProcessingState.APPLIED,
ProcessingState.CANCELLED,
):
return False
if self.reversion_count >= self.MAX_REVERSIONS:
return False
return can_transition(self.phase, phase)
def as_cli_dict(self) -> dict[str, Any]:
"""Return a stable dictionary representation for CLI output.