cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix(plan-lifecycle): add rollback_plan method to PlanLifecycleService

Browse Source 
- What was implemented
  - Added PLAN_ROLLED_BACK event type to the EventType enum at src/cleveragents/infrastructure/events/types.py to properly represent successful rollbacks in the domain model.
  - Implemented rollback_plan(plan_id: str, checkpoint_id: str) -> RollbackResult in PlanLifecycleService (src/cleveragents/application/services/plan_lifecycle_service.py) with:
    - Plan state validation: rejects rollback when the plan is in terminal APPLIED or CANCELLED states.
    - Delegation to CheckpointService.selective_rollback() to perform the actual rollback logic and obtain a RollbackResult.
    - Emission of PLAN_ROLLED_BACK as a domain event to reflect the completed rollback.
    - checkpoint_service is accepted as an optional constructor parameter; if not provided, a PlanError is raised to preserve backward compatibility.
  - Updated CLI behavior in src/cleveragents/cli/commands/plan.py so agents plan rollback routes through PlanLifecycleService.rollback_plan() rather than calling CheckpointService.selective_rollback() directly.
  - Updated PlanLifecycleService module docstring to include rollback_plan in the documented API.
  - Added Behave feature file features/plan_lifecycle_rollback.feature with 11 scenarios covering state validation, domain events, and delegation.
  - Added step implementations in features/steps/plan_lifecycle_rollback_steps.py to support the new scenarios.

- Key design decisions
  - rollback_plan returns RollbackResult (the same result type produced by CheckpointService.selective_rollback) so the CLI can display rollback details consistently.
  - Terminal states APPLIED and CANCELLED are disallowed for rollback to prevent inconsistent or invalid state transitions.
  - checkpoint_service is optional in the PlanLifecycleService constructor; when omitted (None), a PlanError is raised to retain backward compatibility while signaling explicit dependency requirements.
  - CLI UI remains powered by CheckpointService for metadata enrichment (e.g., confirmation prompts), but the actual rollback action is performed via PlanLifecycleService to ensure proper domain workflow and event emission.

- Technical implications
  - All rollback logic now flows through the domain service layer (PlanLifecycleService) to preserve invariants and emit domain events, rather than allowing ad-hoc UI routes to bypass service validation.
  - The UI can still retrieve checkpoint metadata for user confirmation, but the operation that modifies state uses the new rollback_plan pathway.
  - Tests and behavior coverage were expanded via the new Behave feature and step implementations to validate state handling, events, and delegation.

- Affected modules/components
  - src/cleveragents/infrastructure/events/types.py
  - src/cleveragents/application/services/plan_lifecycle_service.py
  - src/cleveragents/cli/commands/plan.py
  - PlanLifecycleService module docstring
  - features/plan_lifecycle_rollback.feature
  - features/steps/plan_lifecycle_rollback_steps.py

ISSUES CLOSED: #3677
This commit is contained in:
Jeffrey Phillips Freeman
2026-04-06 07:20:06 +00:00
committed by Forgejo
parent 6bb0e6ddc1
commit 7fb3fc76c8
19 changed files with 1714 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/plan_lifecycle_rollback.feature
+93
View File
@@ -0,0 +1,93 @@
@phase2 @plan_lifecycle @rollback
Feature: PlanLifecycleService rollback_plan method
As a developer using the plan lifecycle service
I want a rollback_plan method on PlanLifecycleService
So that rollback operations go through the service layer with proper
state validation and domain event emission
# ───────────────────────────────────────────────────────────
# Service method existence and basic delegation
# ───────────────────────────────────────────────────────────
Scenario: rollback_plan delegates to CheckpointService and returns RollbackResult
Given plr-a plan lifecycle service with a mock checkpoint service
And plr-a plan in Execute phase with a sandbox
And plr-a checkpoint exists for the plan
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-the rollback should succeed
And plr-the result should be a RollbackResult
And plr-the checkpoint service selective_rollback should have been called
Scenario: rollback_plan raises PlanError when checkpoint_service is not configured
Given plr-a plan lifecycle service without a checkpoint service
And plr-a plan in Execute phase
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-a PlanError should be raised with "requires a CheckpointService"
# ───────────────────────────────────────────────────────────
# State validation
# ───────────────────────────────────────────────────────────
Scenario: rollback_plan rejects a plan in APPLIED terminal state
Given plr-a plan lifecycle service with a mock checkpoint service
And plr-a plan in APPLIED terminal state
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-a PlanError should be raised with "terminal state"
And plr-the checkpoint service selective_rollback should NOT have been called
Scenario: rollback_plan rejects a plan in CANCELLED terminal state
Given plr-a plan lifecycle service with a mock checkpoint service
And plr-a plan in CANCELLED terminal state
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-a PlanError should be raised with "terminal state"
And plr-the checkpoint service selective_rollback should NOT have been called
Scenario: rollback_plan accepts a plan in Execute/PROCESSING state
Given plr-a plan lifecycle service with a mock checkpoint service
And plr-a plan in Execute/PROCESSING state with a sandbox
And plr-a checkpoint exists for the plan
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-the rollback should succeed
Scenario: rollback_plan accepts a plan in Execute/QUEUED state
Given plr-a plan lifecycle service with a mock checkpoint service
And plr-a plan in Execute/QUEUED state with a sandbox
And plr-a checkpoint exists for the plan
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-the rollback should succeed
Scenario: rollback_plan accepts a plan in Strategize phase
Given plr-a plan lifecycle service with a mock checkpoint service
And plr-a plan in Strategize/QUEUED state with a sandbox
And plr-a checkpoint exists for the plan
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-the rollback should succeed
# ───────────────────────────────────────────────────────────
# Domain event emission
# ───────────────────────────────────────────────────────────
Scenario: rollback_plan emits PLAN_ROLLED_BACK domain event on success
Given plr-a plan lifecycle service with a mock checkpoint service and event bus
And plr-a plan in Execute phase with a sandbox
And plr-a checkpoint exists for the plan
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-the rollback should succeed
And plr-a PLAN_ROLLED_BACK domain event should have been emitted
And plr-the event should contain the checkpoint_id
Scenario: rollback_plan does not emit events when event_bus is None
Given plr-a plan lifecycle service with a mock checkpoint service but no event bus
And plr-a plan in Execute phase with a sandbox
And plr-a checkpoint exists for the plan
When plr-I call rollback_plan with the plan id and checkpoint id
Then plr-the rollback should succeed
# ───────────────────────────────────────────────────────────
# NotFoundError propagation
# ───────────────────────────────────────────────────────────
Scenario: rollback_plan raises NotFoundError for non-existent plan
Given plr-a plan lifecycle service with a mock checkpoint service
When plr-I call rollback_plan with a non-existent plan id
Then plr-a NotFoundError should be raised
features/steps/plan_cli_coverage_r2_steps.py
+966
View File
@@ -0,0 +1,966 @@
"""Step definitions for plan_cli_coverage_r2.feature.
Targets remaining uncovered lines in cleveragents/cli/commands/plan.py:
- Line 724: build >5 changes truncation ("... and N more")
- Lines 737-738: build PlanError handler
- Lines 740-741: build CleverAgentsError handler
- Lines 832-833: apply CleverAgentsError handler
- Lines 879-880: new ValidationError handler
- Lines 882-883: new CleverAgentsError handler
- Lines 930-931: current CleverAgentsError handler
- Lines 1007-1008: list CleverAgentsError handler
- Lines 1094-1095: continue CleverAgentsError handler
- Lines 1108-1115: _get_lifecycle_service body
- Lines 1223-1245: Multi-project scopes rendering in _print_lifecycle_plan
- Lines 1469-1481: use_action execution_environment validation
- Lines 1565-1578: execute_plan execution_environment validation
- Lines 2588, 2593, 2598: resume_plan rich output optional branches
- Lines 2616-2618: resume CleverAgentsError handler
- Lines 2691-2762: rollback_plan full command body
"""
from __future__ import annotations
import io
from datetime import UTC, datetime
from types import SimpleNamespace
from typing import Any
from unittest.mock import MagicMock, patch
from behave import given, then, when
from behave.runner import Context
from rich.console import Console
from typer.testing import CliRunner
from cleveragents.cli.commands import plan as plan_module
from cleveragents.cli.commands.plan import app as plan_app
from cleveragents.core.exceptions import (
BusinessRuleViolation,
CleverAgentsError,
PlanError,
ResourceNotFoundError,
ValidationError,
)
from cleveragents.domain.models.core.checkpoint import (
Checkpoint,
CheckpointMetadata,
RollbackResult,
)
from cleveragents.domain.models.core.multi_project import (
ChangeSetSummary,
MultiProjectMetadata,
ProjectScope,
)
from cleveragents.domain.models.core.plan import (
NamespacedName,
Plan,
PlanIdentity,
PlanPhase,
PlanTimestamps,
ProcessingState,
ProjectLink,
)
from cleveragents.domain.models.core.resume import ResumeSummary
_ULID_A = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
_ULID_B = "01ARZ3NDEKTSV4RRFFQ69G5FBV"
_ULID_CP = "01ARZ3NDEKTSV4RRFFQ69G5FCV"
_PATCH_CONTAINER = "cleveragents.application.container.get_container"
_PATCH_GET_LIFECYCLE = "cleveragents.cli.commands.plan._get_lifecycle_service"
_PATCH_GET_PROJECT = "cleveragents.cli.commands.plan._get_current_project"
_PATCH_GET_APPLY = "cleveragents.cli.commands.plan._get_apply_service"
def _make_legacy_project() -> SimpleNamespace:
return SimpleNamespace(name="test-project", path="/tmp/test")
def _make_legacy_plan(
*,
name: str = "my-plan",
status: str = "active",
current: bool = True,
) -> SimpleNamespace:
return SimpleNamespace(
name=name,
status=status,
current=current,
created_at=datetime(2025, 1, 1),
prompt="do stuff",
)
def _make_legacy_change(file_path: str, operation: str = "modify") -> SimpleNamespace:
return SimpleNamespace(file_path=file_path, operation=operation)
def _make_lifecycle_plan(
*,
plan_id: str = _ULID_A,
name: str = "local/test-plan",
phase: PlanPhase = PlanPhase.STRATEGIZE,
processing_state: ProcessingState = ProcessingState.QUEUED,
project_links: list[ProjectLink] | None = None,
error_message: str | None = None,
error_details: dict[str, str] | None = None,
read_only: bool = False,
multi_project_metadata: MultiProjectMetadata | None = None,
execution_environment: str | None = None,
) -> Plan:
plan = Plan(
identity=PlanIdentity(plan_id=plan_id),
namespaced_name=NamespacedName.parse(name),
action_name="local/test-action",
description="Coverage test plan",
definition_of_done=None,
strategy_actor=None,
execution_actor=None,
phase=phase,
processing_state=processing_state,
project_links=project_links or [],
timestamps=PlanTimestamps(
created_at=datetime(2025, 6, 15, 10, 0, 0),
updated_at=datetime(2025, 6, 15, 11, 0, 0),
),
error_message=error_message,
error_details=error_details,
reusable=True,
read_only=read_only,
created_by=None,
multi_project_metadata=multi_project_metadata,
)
if execution_environment:
plan.execution_environment = execution_environment
return plan
def _setup_legacy_mocks(
context: Context,
*,
plan_service_attrs: dict[str, Any] | None = None,
project_service_attrs: dict[str, Any] | None = None,
) -> list:
"""Set up standard legacy container mocks and return patches for cleanup."""
mock_container = MagicMock()
mock_plan_svc = MagicMock()
mock_project_svc = MagicMock()
mock_project_svc.get_current_project.return_value = _make_legacy_project()
if plan_service_attrs:
for key, val in plan_service_attrs.items():
setattr(mock_plan_svc, key, val)
mock_container.plan_service.return_value = mock_plan_svc
mock_container.project_service.return_value = mock_project_svc
# actor_registry for build/tell
mock_container.actor_registry.return_value = MagicMock()
mock_container.actor_service.return_value = MagicMock()
patches = []
p_container = patch(_PATCH_CONTAINER, return_value=mock_container)
p_project = patch(_PATCH_GET_PROJECT, return_value=_make_legacy_project())
p_container.start()
p_project.start()
patches.extend([p_container.stop, p_project.stop])
context.r2cov_plan_svc = mock_plan_svc
context.r2cov_container = mock_container
return patches
# ======================================================================
# Given — CLI runner
# ======================================================================
@given("a r2cov CLI runner")
def step_r2cov_cli_runner(context: Context) -> None:
context.r2cov_runner = CliRunner()
context.r2cov_result = None
if not hasattr(context, "_r2cov_cleanups"):
context._r2cov_cleanups = []
def _cleanup(ctx: Context) -> None:
for fn in getattr(ctx, "_r2cov_cleanups", []):
fn()
ctx._r2cov_cleanups = []
context.add_cleanup(_cleanup, context)
# ======================================================================
# Given — Legacy build mocks
# ======================================================================
@given("a mocked legacy container for r2cov build returning 8 changes")
def step_mock_build_8_changes(context: Context) -> None:
changes = [_make_legacy_change(f"src/file{i}.py") for i in range(8)]
build_plan = MagicMock(return_value=changes)
patches = _setup_legacy_mocks(
context, plan_service_attrs={"build_plan": build_plan}
)
context._r2cov_cleanups.extend(patches)
@given("a mocked legacy container for r2cov build that raises PlanError")
def step_mock_build_plan_error(context: Context) -> None:
build_plan = MagicMock(side_effect=PlanError("Build failed unexpectedly"))
patches = _setup_legacy_mocks(
context, plan_service_attrs={"build_plan": build_plan}
)
context._r2cov_cleanups.extend(patches)
@given("a mocked legacy container for r2cov build that raises CleverAgentsError")
def step_mock_build_clever_error(context: Context) -> None:
build_plan = MagicMock(side_effect=CleverAgentsError("service unavailable"))
patches = _setup_legacy_mocks(
context, plan_service_attrs={"build_plan": build_plan}
)
context._r2cov_cleanups.extend(patches)
# ======================================================================
# Given — Legacy apply mocks
# ======================================================================
@given("a mocked legacy container for r2cov apply that raises CleverAgentsError")
def step_mock_apply_clever_error(context: Context) -> None:
get_pending = MagicMock(return_value=[_make_legacy_change("f.py")])
apply_changes = MagicMock(side_effect=CleverAgentsError("apply failed"))
patches = _setup_legacy_mocks(
context,
plan_service_attrs={
"get_pending_changes": get_pending,
"apply_changes": apply_changes,
},
)
context._r2cov_cleanups.extend(patches)
# ======================================================================
# Given — Legacy new mocks
# ======================================================================
@given("a mocked legacy container for r2cov new that raises ValidationError")
def step_mock_new_validation_error(context: Context) -> None:
new_plan = MagicMock(side_effect=ValidationError("Invalid plan name"))
patches = _setup_legacy_mocks(context, plan_service_attrs={"new_plan": new_plan})
context._r2cov_cleanups.extend(patches)
@given("a mocked legacy container for r2cov new that raises CleverAgentsError")
def step_mock_new_clever_error(context: Context) -> None:
new_plan = MagicMock(side_effect=CleverAgentsError("new plan failed"))
patches = _setup_legacy_mocks(context, plan_service_attrs={"new_plan": new_plan})
context._r2cov_cleanups.extend(patches)
# ======================================================================
# Given — Legacy current mocks
# ======================================================================
@given("a mocked legacy container for r2cov current that raises CleverAgentsError")
def step_mock_current_clever_error(context: Context) -> None:
get_current = MagicMock(side_effect=CleverAgentsError("DB connection lost"))
patches = _setup_legacy_mocks(
context, plan_service_attrs={"get_current_plan": get_current}
)
context._r2cov_cleanups.extend(patches)
# ======================================================================
# Given — Legacy list mocks
# ======================================================================
@given("a mocked legacy container for r2cov list that raises CleverAgentsError")
def step_mock_list_clever_error(context: Context) -> None:
list_plans = MagicMock(side_effect=CleverAgentsError("list failed"))
patches = _setup_legacy_mocks(
context, plan_service_attrs={"list_plans": list_plans}
)
context._r2cov_cleanups.extend(patches)
# ======================================================================
# Given — Legacy continue mocks
# ======================================================================
@given("a mocked legacy container for r2cov continue that raises CleverAgentsError")
def step_mock_continue_clever_error(context: Context) -> None:
get_current = MagicMock(side_effect=CleverAgentsError("continue failed"))
patches = _setup_legacy_mocks(
context, plan_service_attrs={"get_current_plan": get_current}
)
context._r2cov_cleanups.extend(patches)
# ======================================================================
# Given — _get_lifecycle_service construction
# ======================================================================
@given("a mocked container for r2cov lifecycle service construction")
def step_mock_container_for_lifecycle(context: Context) -> None:
mock_container = MagicMock()
mock_settings = MagicMock()
mock_container.settings.return_value = mock_settings
mock_container.plan_lifecycle_service.return_value = MagicMock(
settings=mock_settings
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
context.r2cov_mock_settings = mock_settings
# ======================================================================
# Given — Multi-project scopes plan
# ======================================================================
@given("a lifecycle plan with multi-project scopes for r2cov")
def step_plan_with_multi_project_scopes(context: Context) -> None:
scopes = [
ProjectScope(
project_name="proj-alpha",
alias="alpha",
read_only=False,
changeset_summary=ChangeSetSummary(
project_name="proj-alpha",
files_changed=3,
files_added=1,
files_deleted=0,
total_lines_changed=42,
),
),
ProjectScope(
project_name="proj-beta",
alias=None,
read_only=True,
changeset_summary=None,
),
]
metadata = MultiProjectMetadata(project_scopes=scopes)
context.r2cov_lifecycle_plan = _make_lifecycle_plan(
multi_project_metadata=metadata,
)
@given("a lifecycle plan with multi-project scopes with validation for r2cov")
def step_plan_with_validation_scopes(context: Context) -> None:
scopes = [
ProjectScope(
project_name="proj-gamma",
changeset_summary=ChangeSetSummary(
project_name="proj-gamma",
files_changed=5,
files_added=2,
files_deleted=1,
total_lines_changed=100,
validation_passed=True,
),
),
]
metadata = MultiProjectMetadata(project_scopes=scopes)
context.r2cov_lifecycle_plan = _make_lifecycle_plan(
multi_project_metadata=metadata,
)
# ======================================================================
# Given — Lifecycle service for use_action
# ======================================================================
@given("a mocked lifecycle service for r2cov use action")
def step_mock_lifecycle_for_use(context: Context) -> None:
mock_service = MagicMock()
mock_action = MagicMock()
mock_action.namespaced_name = "local/test-action"
mock_service.get_action_by_name.return_value = mock_action
mock_plan = _make_lifecycle_plan()
mock_service.use_action.return_value = mock_plan
p = patch(_PATCH_GET_LIFECYCLE, return_value=mock_service)
p.start()
context._r2cov_cleanups.append(p.stop)
context.r2cov_service = mock_service
context.r2cov_plan = mock_plan
# ======================================================================
# Given — Lifecycle service for execute_plan
# ======================================================================
@given("a mocked lifecycle service for r2cov execute with a valid plan")
def step_mock_lifecycle_for_execute(context: Context) -> None:
mock_service = MagicMock()
plan = _make_lifecycle_plan(
processing_state=ProcessingState.COMPLETE,
)
mock_service.get_plan.return_value = plan
mock_service.execute_plan.return_value = plan
p = patch(_PATCH_GET_LIFECYCLE, return_value=mock_service)
p.start()
context._r2cov_cleanups.append(p.stop)
p_exec = patch(
"cleveragents.cli.commands.plan._get_plan_executor",
return_value=MagicMock(),
)
p_exec.start()
context._r2cov_cleanups.append(p_exec.stop)
context.r2cov_service = mock_service
# ======================================================================
# Given — Resume service mocks
# ======================================================================
@given("a mocked resume service for r2cov with all optional fields")
def step_mock_resume_all_fields(context: Context) -> None:
summary = ResumeSummary(
plan_id="PLAN123",
phase="execute",
processing_state="errored",
last_completed_step=3,
next_step_index=4,
total_steps=10,
decision_id="DEC-001",
last_checkpoint_id="CP-001",
sandbox_ref="sandbox-abc-123",
)
mock_lifecycle = MagicMock()
p_lifecycle = patch(_PATCH_GET_LIFECYCLE, return_value=mock_lifecycle)
p_lifecycle.start()
context._r2cov_cleanups.append(p_lifecycle.stop)
p_resume = patch(
"cleveragents.application.services.plan_resume_service.PlanResumeService",
autospec=False,
)
mock_resume_cls = p_resume.start()
context._r2cov_cleanups.append(p_resume.stop)
mock_resume_instance = MagicMock()
mock_resume_instance.resume_plan.return_value = summary
mock_resume_cls.return_value = mock_resume_instance
@given("a mocked resume service for r2cov that raises CleverAgentsError")
def step_mock_resume_clever_error(context: Context) -> None:
mock_lifecycle = MagicMock()
p_lifecycle = patch(_PATCH_GET_LIFECYCLE, return_value=mock_lifecycle)
p_lifecycle.start()
context._r2cov_cleanups.append(p_lifecycle.stop)
p_resume = patch(
"cleveragents.application.services.plan_resume_service.PlanResumeService",
autospec=False,
)
mock_resume_cls = p_resume.start()
context._r2cov_cleanups.append(p_resume.stop)
mock_resume_instance = MagicMock()
mock_resume_instance.resume_plan.side_effect = CleverAgentsError("resume failed")
mock_resume_cls.return_value = mock_resume_instance
# ======================================================================
# Given — Checkpoint service for rollback
# ======================================================================
def _make_rollback_result() -> RollbackResult:
return RollbackResult(
restored_files_count=3,
changed_paths=["src/a.py", "src/b.py", "src/c.py"],
from_checkpoint_id=_ULID_CP,
)
def _make_checkpoint(
*,
label: str = "before-auth-refactor",
created_at: datetime | None = None,
) -> Checkpoint:
"""Create a Checkpoint domain object for use in rollback test mocks."""
return Checkpoint(
checkpoint_id=_ULID_CP,
plan_id=_ULID_A,
sandbox_ref="abc123",
created_at=created_at or datetime(2026, 4, 5, 10, 0, 0, tzinfo=UTC),
metadata=CheckpointMetadata(reason=label),
)
def _make_rollback_container(
*,
rollback_result: RollbackResult | None = None,
rollback_side_effect: Exception | None = None,
) -> MagicMock:
"""Build a mock container with checkpoint_service and decision_service wired.
The CLI rollback command routes through PlanLifecycleService.rollback_plan()
(service layer pattern), so we mock both:
- plan_lifecycle_service().rollback_plan — the primary call path
- checkpoint_service.selective_rollback — kept for backward compat / direct
callers that bypass the service layer
"""
mock_container = MagicMock()
mock_cp_svc = MagicMock()
mock_cp_svc.get_checkpoint.return_value = _make_checkpoint()
if rollback_side_effect is not None:
mock_cp_svc.selective_rollback.side_effect = rollback_side_effect
# Also wire the side effect on the lifecycle service rollback_plan so
# that the CLI's service-layer call path raises the expected exception.
mock_container.plan_lifecycle_service.return_value.rollback_plan.side_effect = (
rollback_side_effect
)
else:
resolved_result = rollback_result or _make_rollback_result()
mock_cp_svc.selective_rollback.return_value = resolved_result
# Wire the return value on the lifecycle service rollback_plan so that
# the CLI's service-layer call path returns a proper RollbackResult.
mock_container.plan_lifecycle_service.return_value.rollback_plan.return_value = resolved_result
mock_container.checkpoint_service.return_value = mock_cp_svc
mock_decision_svc = MagicMock()
mock_decision_svc.list_decisions.return_value = []
mock_container.decision_service.return_value = mock_decision_svc
return mock_container
@given("a mocked checkpoint service for r2cov rollback that succeeds")
def step_mock_rollback_success(context: Context) -> None:
mock_container = _make_rollback_container()
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given(
"a mocked checkpoint service for r2cov rollback that raises BusinessRuleViolation"
)
def step_mock_rollback_business_rule(context: Context) -> None:
mock_container = _make_rollback_container(
rollback_side_effect=BusinessRuleViolation("Plan is already applied")
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given(
"a mocked checkpoint service for r2cov rollback that raises ResourceNotFoundError"
)
def step_mock_rollback_not_found(context: Context) -> None:
mock_container = _make_rollback_container(
rollback_side_effect=ResourceNotFoundError("Checkpoint not found")
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given("a mocked checkpoint service for r2cov rollback that raises ValidationError")
def step_mock_rollback_validation(context: Context) -> None:
mock_container = _make_rollback_container(
rollback_side_effect=ValidationError("Invalid checkpoint")
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given("a mocked checkpoint service for r2cov rollback that raises CleverAgentsError")
def step_mock_rollback_clever_error(context: Context) -> None:
mock_container = _make_rollback_container(
rollback_side_effect=CleverAgentsError("unknown rollback failure")
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
# ======================================================================
# When — Legacy commands
# ======================================================================
@when("I invoke r2cov legacy build")
def step_invoke_legacy_build(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["build"])
@when("I invoke r2cov legacy apply with --yes")
def step_invoke_legacy_apply_yes(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["apply", "--yes"])
@when('I invoke r2cov legacy new "{name}"')
def step_invoke_legacy_new(context: Context, name: str) -> None:
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["new", name])
@when("I invoke r2cov legacy current")
def step_invoke_legacy_current(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["current"])
@when("I invoke r2cov legacy list")
def step_invoke_legacy_list(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["list"])
@when("I invoke r2cov legacy continue")
def step_invoke_legacy_continue(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(plan_app, ["continue"])
# ======================================================================
# When — _get_lifecycle_service
# ======================================================================
@when("I call r2cov _get_lifecycle_service")
def step_call_get_lifecycle_service(context: Context) -> None:
from cleveragents.cli.commands.plan import _get_lifecycle_service
context.r2cov_lifecycle_result = _get_lifecycle_service()
# ======================================================================
# When — _print_lifecycle_plan
# ======================================================================
@when("I invoke r2cov _print_lifecycle_plan on the plan")
def step_invoke_print_lifecycle_plan(context: Context) -> None:
from cleveragents.cli.commands.plan import _print_lifecycle_plan
# Capture console output
buf = io.StringIO()
capture_console = Console(file=buf, width=200, force_terminal=True)
original = plan_module.console
plan_module.console = capture_console
try:
_print_lifecycle_plan(context.r2cov_lifecycle_plan, title="Test Plan")
finally:
plan_module.console = original
context.r2cov_result = SimpleNamespace(
exit_code=0,
output=buf.getvalue(),
)
# ======================================================================
# When — use_action with execution_environment
# ======================================================================
@when('I invoke r2cov use with execution-environment "{env}"')
def step_invoke_use_exec_env(context: Context, env: str) -> None:
context.r2cov_result = context.r2cov_runner.invoke(
plan_app,
["use", "local/test-action", "--execution-environment", env],
)
# ======================================================================
# When — execute_plan with execution_environment
# ======================================================================
@when('I invoke r2cov execute with execution-environment "{env}"')
def step_invoke_execute_exec_env(context: Context, env: str) -> None:
context.r2cov_result = context.r2cov_runner.invoke(
plan_app,
["execute", _ULID_A, "--execution-environment", env],
)
# ======================================================================
# When — resume plan
# ======================================================================
@when('I invoke r2cov plan resume "{plan_id}" in rich format')
def step_invoke_resume_rich(context: Context, plan_id: str) -> None:
context.r2cov_result = context.r2cov_runner.invoke(
plan_app,
["resume", plan_id],
)
# ======================================================================
# When — rollback plan
# ======================================================================
@when("I invoke r2cov plan rollback with --yes")
def step_invoke_rollback_yes(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(
plan_app,
["rollback", "--yes", _ULID_A, _ULID_CP],
)
@when("I invoke r2cov plan rollback with --yes and --format json")
def step_invoke_rollback_yes_json(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(
plan_app,
["rollback", "--yes", "--format", "json", _ULID_A, _ULID_CP],
)
@when("I invoke r2cov plan rollback without --yes and decline")
def step_invoke_rollback_decline(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(
plan_app,
["rollback", _ULID_A, _ULID_CP],
input="n\n",
)
@when("I invoke r2cov plan rollback without --yes and accept")
def step_invoke_rollback_accept(context: Context) -> None:
context.r2cov_result = context.r2cov_runner.invoke(
plan_app,
["rollback", _ULID_A, _ULID_CP],
input="y\n",
)
# ======================================================================
# Given — enriched confirmation prompt mocks (issue #3443)
# ======================================================================
@given('a mocked checkpoint service for r2cov rollback with label "{label}"')
def step_mock_rollback_with_label(context: Context, label: str) -> None:
mock_container = _make_rollback_container()
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
_make_checkpoint(label=label)
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given(
"a mocked checkpoint service for r2cov rollback with {decisions:d} decisions "
"and {child_plans:d} child plan"
)
def step_mock_rollback_with_decisions(
context: Context, decisions: int, child_plans: int
) -> None:
from datetime import timedelta
from ulid import ULID
from cleveragents.domain.models.core.decision import Decision, DecisionType
cp_time = datetime(2026, 4, 5, 10, 0, 0, tzinfo=UTC)
after_time = cp_time + timedelta(minutes=5)
# Build decision list: (decisions - child_plans) regular + child_plans spawns
decision_list = []
regular_count = decisions - child_plans
for i in range(regular_count):
decision_list.append(
Decision(
decision_id=str(ULID()),
plan_id=_ULID_A,
sequence_number=i,
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
question="What to do?",
chosen_option="Do it",
created_at=after_time,
)
)
for i in range(child_plans):
decision_list.append(
Decision(
decision_id=str(ULID()),
plan_id=_ULID_A,
sequence_number=regular_count + i,
decision_type=DecisionType.SUBPLAN_SPAWN,
question="Spawn child?",
chosen_option="Yes",
created_at=after_time,
)
)
mock_container = _make_rollback_container()
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
_make_checkpoint(created_at=cp_time)
)
mock_container.decision_service.return_value.list_decisions.return_value = (
decision_list
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given(
"a mocked checkpoint service for r2cov rollback where get_checkpoint raises not found"
)
def step_mock_rollback_get_checkpoint_not_found(context: Context) -> None:
mock_container = _make_rollback_container()
mock_container.checkpoint_service.return_value.get_checkpoint.side_effect = (
ResourceNotFoundError("Checkpoint not found")
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given(
"a mocked checkpoint service for r2cov rollback with {decisions:d} decisions "
"and no child plans"
)
def step_mock_rollback_with_decisions_no_child_plans(
context: Context, decisions: int
) -> None:
from datetime import timedelta
from ulid import ULID
from cleveragents.domain.models.core.decision import Decision, DecisionType
cp_time = datetime(2026, 4, 5, 10, 0, 0, tzinfo=UTC)
after_time = cp_time + timedelta(minutes=5)
decision_list = [
Decision(
decision_id=str(ULID()),
plan_id=_ULID_A,
sequence_number=i,
decision_type=DecisionType.IMPLEMENTATION_CHOICE,
question="What to do?",
chosen_option="Do it",
created_at=after_time,
)
for i in range(decisions)
]
mock_container = _make_rollback_container()
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
_make_checkpoint(created_at=cp_time)
)
mock_container.decision_service.return_value.list_decisions.return_value = (
decision_list
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given(
"a mocked checkpoint service for r2cov rollback with {child_plans:d} "
"parallel spawn child plans"
)
def step_mock_rollback_with_parallel_spawn(context: Context, child_plans: int) -> None:
from datetime import timedelta
from ulid import ULID
from cleveragents.domain.models.core.decision import Decision, DecisionType
cp_time = datetime(2026, 4, 5, 10, 0, 0, tzinfo=UTC)
after_time = cp_time + timedelta(minutes=5)
decision_list = [
Decision(
decision_id=str(ULID()),
plan_id=_ULID_A,
sequence_number=i,
decision_type=DecisionType.SUBPLAN_PARALLEL_SPAWN,
question="Spawn parallel child?",
chosen_option="Yes",
created_at=after_time,
)
for i in range(child_plans)
]
mock_container = _make_rollback_container()
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
_make_checkpoint(created_at=cp_time)
)
mock_container.decision_service.return_value.list_decisions.return_value = (
decision_list
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
@given(
"a mocked checkpoint service for r2cov rollback with a future checkpoint timestamp"
)
def step_mock_rollback_with_future_timestamp(context: Context) -> None:
from datetime import timedelta
future_time = datetime.now(UTC) + timedelta(hours=1)
mock_container = _make_rollback_container()
mock_container.checkpoint_service.return_value.get_checkpoint.return_value = (
_make_checkpoint(created_at=future_time)
)
p = patch(_PATCH_CONTAINER, return_value=mock_container)
p.start()
context._r2cov_cleanups.append(p.stop)
# ======================================================================
# Then — common assertions
# ======================================================================
@then("the r2cov command should exit normally")
def step_exit_normally(context: Context) -> None:
result = context.r2cov_result
assert result is not None, "No command result captured"
assert result.exit_code == 0, (
f"Expected exit code 0, got {result.exit_code}.\nOutput:\n{result.output}"
)
@then("the r2cov command should abort")
def step_exit_abort(context: Context) -> None:
result = context.r2cov_result
assert result is not None, "No command result captured"
assert result.exit_code != 0, (
f"Expected non-zero exit code, got {result.exit_code}.\n"
f"Output:\n{result.output}"
)
@then('the r2cov output should contain "{text}"')
def step_output_contains(context: Context, text: str) -> None:
result = context.r2cov_result
assert result is not None, "No command result captured"
assert text in result.output, (
f"Expected '{text}' in output.\nActual output:\n{result.output}"
)
@then("the r2cov lifecycle service should be returned")
def step_lifecycle_returned(context: Context) -> None:
assert context.r2cov_lifecycle_result is not None, (
"Expected a lifecycle service instance"
)
features/steps/plan_cli_spec_alignment_steps.py
+1 -1
View File
@@ -337,7 +337,7 @@ def step_plan_list_no_filters(context: Context) -> None:
"""
import cleveragents.cli.commands.plan as _plan_mod
wide_runner = CliRunner(mix_stderr=False)
wide_runner = CliRunner()
original_width = _plan_mod.console._width
_plan_mod.console._width = 200
try:
features/steps/plan_lifecycle_rollback_steps.py
+366
View File
@@ -0,0 +1,366 @@
"""Step definitions for plan_lifecycle_rollback.feature.
Tests for PlanLifecycleService.rollback_plan() — the new service-layer
method that validates plan state and delegates to CheckpointService
before emitting PLAN_ROLLED_BACK domain events.
All step text uses the ``plr-`` prefix to avoid collisions with
existing step definitions.
"""
from __future__ import annotations
from typing import Any
from unittest.mock import MagicMock
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.plan_lifecycle_service import (
PlanLifecycleService,
)
from cleveragents.config.settings import Settings
from cleveragents.core.exceptions import NotFoundError, PlanError
from cleveragents.domain.models.core.checkpoint import RollbackResult
from cleveragents.domain.models.core.plan import (
PlanPhase,
ProcessingState,
ProjectLink,
)
from cleveragents.infrastructure.events.types import EventType
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
_CHECKPOINT_ID = "01BRZ4NFEK0000000000000000"
_NONEXISTENT_PLAN_ID = "01NONEXISTENT0000000000000"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_mock_checkpoint_service(
plan_id: str = _PLAN_ID,
checkpoint_id: str = _CHECKPOINT_ID,
) -> MagicMock:
"""Build a mock CheckpointService that returns a valid RollbackResult."""
mock_svc = MagicMock()
mock_svc.selective_rollback.return_value = RollbackResult(
restored_files_count=3,
changed_paths=["src/foo.py", "src/bar.py", "tests/test_foo.py"],
from_checkpoint_id=checkpoint_id,
)
return mock_svc
def _make_mock_event_bus() -> MagicMock:
"""Build a mock EventBus."""
mock_bus = MagicMock()
mock_bus.emitted_events: list[Any] = []
def _emit(event: Any) -> None:
mock_bus.emitted_events.append(event)
mock_bus.emit.side_effect = _emit
return mock_bus
def _create_service(
checkpoint_service: Any = None,
event_bus: Any = None,
) -> PlanLifecycleService:
"""Create a PlanLifecycleService with optional dependencies."""
settings = Settings()
return PlanLifecycleService(
settings=settings,
checkpoint_service=checkpoint_service,
event_bus=event_bus,
)
def _create_plan_in_service(
service: PlanLifecycleService,
phase: PlanPhase = PlanPhase.EXECUTE,
state: ProcessingState = ProcessingState.PROCESSING,
) -> str:
"""Create an action and plan, then force the plan into the given phase/state."""
service.create_action(
name="local/test-rollback-action",
description="Test action for rollback",
definition_of_done="Tests pass",
strategy_actor="local/test-strategist",
execution_actor="local/test-executor",
)
plan = service.use_action(
action_name="local/test-rollback-action",
project_links=[ProjectLink(project_name="test-project")],
)
plan_id = plan.identity.plan_id
# Force the plan into the desired phase/state by direct mutation
# (avoids needing to run full actor pipeline in unit tests)
plan.phase = phase
plan.processing_state = state
service._plans[plan_id] = plan
return plan_id
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("plr-a plan lifecycle service with a mock checkpoint service")
def step_plr_service_with_mock_checkpoint(context: Context) -> None:
"""Set up a PlanLifecycleService with a mock CheckpointService."""
context.plr_mock_checkpoint_svc = _make_mock_checkpoint_service()
context.plr_service = _create_service(
checkpoint_service=context.plr_mock_checkpoint_svc,
)
context.plr_checkpoint_id = _CHECKPOINT_ID
@given("plr-a plan lifecycle service without a checkpoint service")
def step_plr_service_without_checkpoint(context: Context) -> None:
"""Set up a PlanLifecycleService with no CheckpointService."""
context.plr_service = _create_service(checkpoint_service=None)
context.plr_checkpoint_id = _CHECKPOINT_ID
@given("plr-a plan lifecycle service with a mock checkpoint service and event bus")
def step_plr_service_with_checkpoint_and_event_bus(context: Context) -> None:
"""Set up a PlanLifecycleService with mock CheckpointService and EventBus."""
context.plr_mock_checkpoint_svc = _make_mock_checkpoint_service()
context.plr_mock_event_bus = _make_mock_event_bus()
context.plr_service = _create_service(
checkpoint_service=context.plr_mock_checkpoint_svc,
event_bus=context.plr_mock_event_bus,
)
context.plr_checkpoint_id = _CHECKPOINT_ID
@given("plr-a plan lifecycle service with a mock checkpoint service but no event bus")
def step_plr_service_with_checkpoint_no_event_bus(context: Context) -> None:
"""Set up a PlanLifecycleService with mock CheckpointService but no EventBus."""
context.plr_mock_checkpoint_svc = _make_mock_checkpoint_service()
context.plr_service = _create_service(
checkpoint_service=context.plr_mock_checkpoint_svc,
event_bus=None,
)
context.plr_checkpoint_id = _CHECKPOINT_ID
@given("plr-a plan in Execute phase with a sandbox")
def step_plr_plan_in_execute_with_sandbox(context: Context) -> None:
"""Create a plan in Execute/PROCESSING state."""
context.plr_plan_id = _create_plan_in_service(
context.plr_service,
phase=PlanPhase.EXECUTE,
state=ProcessingState.PROCESSING,
)
@given("plr-a plan in Execute phase")
def step_plr_plan_in_execute(context: Context) -> None:
"""Create a plan in Execute/PROCESSING state (no sandbox needed for error tests)."""
context.plr_plan_id = _create_plan_in_service(
context.plr_service,
phase=PlanPhase.EXECUTE,
state=ProcessingState.PROCESSING,
)
@given("plr-a plan in APPLIED terminal state")
def step_plr_plan_in_applied_state(context: Context) -> None:
"""Create a plan in APPLIED terminal state."""
context.plr_plan_id = _create_plan_in_service(
context.plr_service,
phase=PlanPhase.APPLY,
state=ProcessingState.APPLIED,
)
@given("plr-a plan in CANCELLED terminal state")
def step_plr_plan_in_cancelled_state(context: Context) -> None:
"""Create a plan in CANCELLED terminal state."""
context.plr_plan_id = _create_plan_in_service(
context.plr_service,
phase=PlanPhase.APPLY,
state=ProcessingState.CANCELLED,
)
@given("plr-a plan in Execute/PROCESSING state with a sandbox")
def step_plr_plan_in_execute_processing(context: Context) -> None:
"""Create a plan in Execute/PROCESSING state."""
context.plr_plan_id = _create_plan_in_service(
context.plr_service,
phase=PlanPhase.EXECUTE,
state=ProcessingState.PROCESSING,
)
@given("plr-a plan in Execute/QUEUED state with a sandbox")
def step_plr_plan_in_execute_queued(context: Context) -> None:
"""Create a plan in Execute/QUEUED state."""
context.plr_plan_id = _create_plan_in_service(
context.plr_service,
phase=PlanPhase.EXECUTE,
state=ProcessingState.QUEUED,
)
@given("plr-a plan in Strategize/QUEUED state with a sandbox")
def step_plr_plan_in_strategize_queued(context: Context) -> None:
"""Create a plan in Strategize/QUEUED state."""
context.plr_plan_id = _create_plan_in_service(
context.plr_service,
phase=PlanPhase.STRATEGIZE,
state=ProcessingState.QUEUED,
)
@given("plr-a checkpoint exists for the plan")
def step_plr_checkpoint_exists(context: Context) -> None:
"""Checkpoint already set up via mock; this step is a no-op for clarity."""
# The mock checkpoint service is already configured to return a valid
# RollbackResult for any plan_id/checkpoint_id combination.
pass
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when("plr-I call rollback_plan with the plan id and checkpoint id")
def step_plr_call_rollback_plan(context: Context) -> None:
"""Call rollback_plan on the service."""
context.plr_raised_error = None
context.plr_result = None
try:
context.plr_result = context.plr_service.rollback_plan(
context.plr_plan_id,
context.plr_checkpoint_id,
)
except (PlanError, NotFoundError, Exception) as exc:
context.plr_raised_error = exc
@when("plr-I call rollback_plan with a non-existent plan id")
def step_plr_call_rollback_nonexistent(context: Context) -> None:
"""Call rollback_plan with a plan id that does not exist."""
context.plr_raised_error = None
context.plr_result = None
try:
context.plr_result = context.plr_service.rollback_plan(
_NONEXISTENT_PLAN_ID,
_CHECKPOINT_ID,
)
except (PlanError, NotFoundError, Exception) as exc:
context.plr_raised_error = exc
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("plr-the rollback should succeed")
def step_plr_rollback_succeeded(context: Context) -> None:
"""Assert that rollback_plan returned without raising an error."""
assert context.plr_raised_error is None, (
f"Expected rollback to succeed but got error: {context.plr_raised_error}"
)
assert context.plr_result is not None, (
"Expected rollback_plan to return a RollbackResult, got None"
)
@then("plr-the result should be a RollbackResult")
def step_plr_result_is_rollback_result(context: Context) -> None:
"""Assert that the result is a RollbackResult instance."""
assert isinstance(context.plr_result, RollbackResult), (
f"Expected RollbackResult, got {type(context.plr_result)}"
)
@then("plr-the checkpoint service selective_rollback should have been called")
def step_plr_selective_rollback_called(context: Context) -> None:
"""Assert that CheckpointService.selective_rollback was called."""
context.plr_mock_checkpoint_svc.selective_rollback.assert_called_once_with(
context.plr_plan_id,
context.plr_checkpoint_id,
)
@then("plr-the checkpoint service selective_rollback should NOT have been called")
def step_plr_selective_rollback_not_called(context: Context) -> None:
"""Assert that CheckpointService.selective_rollback was NOT called."""
context.plr_mock_checkpoint_svc.selective_rollback.assert_not_called()
@then('plr-a PlanError should be raised with "{message_fragment}"')
def step_plr_plan_error_raised(context: Context, message_fragment: str) -> None:
"""Assert that a PlanError was raised containing the given message fragment."""
assert context.plr_raised_error is not None, (
f"Expected a PlanError to be raised but no error was raised. "
f"Result was: {context.plr_result}"
)
assert isinstance(context.plr_raised_error, PlanError), (
f"Expected PlanError, got {type(context.plr_raised_error)}: "
f"{context.plr_raised_error}"
)
error_message = str(context.plr_raised_error)
assert message_fragment in error_message, (
f"Expected error message to contain '{message_fragment}', "
f"got: '{error_message}'"
)
@then("plr-a NotFoundError should be raised")
def step_plr_not_found_error_raised(context: Context) -> None:
"""Assert that a NotFoundError was raised."""
assert context.plr_raised_error is not None, (
"Expected a NotFoundError to be raised but no error was raised."
)
assert isinstance(context.plr_raised_error, NotFoundError), (
f"Expected NotFoundError, got {type(context.plr_raised_error)}: "
f"{context.plr_raised_error}"
)
@then("plr-a PLAN_ROLLED_BACK domain event should have been emitted")
def step_plr_plan_rolled_back_event_emitted(context: Context) -> None:
"""Assert that a PLAN_ROLLED_BACK event was emitted."""
emitted = context.plr_mock_event_bus.emitted_events
rolled_back_events = [
e for e in emitted if e.event_type == EventType.PLAN_ROLLED_BACK
]
assert len(rolled_back_events) >= 1, (
f"Expected at least one PLAN_ROLLED_BACK event, "
f"got events: {[e.event_type for e in emitted]}"
)
@then("plr-the event should contain the checkpoint_id")
def step_plr_event_contains_checkpoint_id(context: Context) -> None:
"""Assert that the PLAN_ROLLED_BACK event details contain the checkpoint_id."""
emitted = context.plr_mock_event_bus.emitted_events
rolled_back_events = [
e for e in emitted if e.event_type == EventType.PLAN_ROLLED_BACK
]
assert len(rolled_back_events) >= 1, "No PLAN_ROLLED_BACK event found"
event = rolled_back_events[0]
details = event.details or {}
assert "checkpoint_id" in details, (
f"Expected 'checkpoint_id' in event details, got: {details}"
)
assert details["checkpoint_id"] == context.plr_checkpoint_id, (
f"Expected checkpoint_id '{context.plr_checkpoint_id}', "
f"got '{details['checkpoint_id']}'"
)
robot/actor_context_export_import.robot
+10 -10
View File
@@ -37,23 +37,23 @@ Export Then Import Round-Trip Preserves Context
Set Suite Variable ${EXPORT_FILE} ${export_path}
${result} = Run Process ${PYTHON} -m cleveragents actor context export
... ${CONTEXT_NAME} --output ${export_path} --context-dir ${ctx_dir}
... ${CONTEXT_NAME} ${export_path} --context-dir ${ctx_dir}
Log Export stdout: ${result.stdout}
Log Export stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
File Should Exist ${export_path}
# 3. Remove the original context
${result} = Run Process ${PYTHON} -m cleveragents actor context remove
# 3. Delete the original context (use 'delete' not 'remove' for named contexts)
${result} = Run Process ${PYTHON} -m cleveragents actor context delete
... ${CONTEXT_NAME} --yes --context-dir ${ctx_dir}
Log Remove stdout: ${result.stdout}
Log Remove stderr: ${result.stderr}
Log Delete stdout: ${result.stdout}
Log Delete stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Directory Should Not Exist ${ctx_dir}/${CONTEXT_NAME}
# 4. Import the context back
${result} = Run Process ${PYTHON} -m cleveragents actor context import
... ${CONTEXT_NAME} --input ${export_path} --context-dir ${ctx_dir}
... ${CONTEXT_NAME} ${export_path} --context-dir ${ctx_dir}
Log Import stdout: ${result.stdout}
Log Import stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
@@ -78,10 +78,10 @@ Export With JSON Format Flag Shows Structured Output
... from cleveragents.reactive.context_manager import ContextManager; mgr \= ContextManager("fmt-test", "${ctx_dir}"); mgr.add_message("user", "test"); print("ok")
Should Be Equal As Integers ${result.rc} 0
# Export to JSON file using --output option (per spec)
# Export to JSON file (positional argument, per CLI spec)
${export_path} = Set Variable ${TEMP}/fmt-export.json
${result} = Run Process ${PYTHON} -m cleveragents actor context export
... fmt-test --output ${export_path} --context-dir ${ctx_dir}
... fmt-test ${export_path} --context-dir ${ctx_dir}
Log Export stdout: ${result.stdout}
Log Export stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
@@ -105,9 +105,9 @@ Import Into Existing Context Overwrites Data
... import json; data \= {"context_name": "existing", "messages": [{"role": "user", "content": "new", "timestamp": "2026-01-01", "metadata": {}}], "metadata": {}, "state": {}, "global_context": {}}; open("${import_path}", "w").write(json.dumps(data)); print("ok")
Should Be Equal As Integers ${result.rc} 0
# Import should succeed (overwrites existing context using --update flag)
# Import should succeed (overwrites existing context; positional argument, per CLI spec)
${result} = Run Process ${PYTHON} -m cleveragents actor context import
... existing --input ${import_path} --update --context-dir ${ctx_dir}
... existing ${import_path} --context-dir ${ctx_dir}
Should Be Equal As Integers ${result.rc} 0
*** Keywords ***
robot/container_tool_exec.robot
+4 -1
View File
@@ -146,7 +146,10 @@ ToolRunner Container Routing Without Executor Returns Error
... s = ToolSpec(name='t', description='d', input_schema={}, handler=lambda x: x)
... r.get.return_value = s
... ev = MagicMock(spec=ExecutionEnvironmentResolver)
... ev.resolve_and_validate.return_value = ExecutionEnvironment.CONTAINER
... # ToolRunner uses resolve_with_precedence() (not resolve_and_validate())
... ev.resolve_with_precedence.return_value = ExecutionEnvironment.CONTAINER
... ev.has_devcontainer.return_value = False
... ev.validate_container_available.return_value = None
... runner = ToolRunner(registry=r, execution_environment_resolver=ev)
... res = runner.execute('t', {})
... assert not res.success, f"Expected failure but got success"
robot/helper_a2a_facade.py
+12 -6
View File
@@ -75,16 +75,22 @@ def event_queue() -> None:
def version_negotiate() -> None:
"""Verify version negotiation."""
"""Verify version negotiation.
The current supported A2A version is 2.0. Negotiating with the
current version must succeed; negotiating with an unsupported version
(e.g. "99.0") must raise A2aVersionMismatchError.
"""
negotiator = A2aVersionNegotiator()
result = negotiator.negotiate("1.0")
if result != "1.0":
print("FAIL: expected 1.0", file=sys.stderr)
current = negotiator.get_current()
result = negotiator.negotiate(current)
if result != current:
print(f"FAIL: expected {current!r}, got {result!r}", file=sys.stderr)
sys.exit(1)
try:
negotiator.negotiate("2.0")
print("FAIL: should have raised", file=sys.stderr)
negotiator.negotiate("99.0")
print("FAIL: should have raised for unsupported version", file=sys.stderr)
sys.exit(1)
except A2aVersionMismatchError:
pass
robot/helper_actor_add_rich_output.py
+11 -1
View File
@@ -57,8 +57,18 @@ def _run_actor_add(
try:
with patch("cleveragents.cli.commands.actor._get_services") as mock_svc:
registry = MagicMock()
# The add command uses registry.add() (YAML-first path)
# The add command uses registry.upsert_actor() (YAML-first path).
# Also set registry.add() for backward compat with older code paths.
registry.add.return_value = actor
registry.upsert_actor.return_value = actor
# Simulate actor not existing yet so the duplicate-check passes.
# The add command calls registry.get_actor(name) before adding;
# raising NotFoundError signals "not found" and allows the add to proceed.
from cleveragents.core.exceptions import NotFoundError
registry.get_actor.side_effect = NotFoundError(
resource_type="actor", resource_id="local/robot-add-actor"
)
mock_svc.return_value = (MagicMock(), registry)
actor_name = config_data.get("name", "local/robot-add-actor")
result = runner.invoke(
robot/helper_actor_cli_show.py
+22 -5
View File
@@ -38,7 +38,12 @@ def _make_actor(
def test_show_json_fields() -> None:
"""Verify show --format json contains expected fields."""
"""Verify show --format json contains expected fields.
The CLI wraps output in a spec-required envelope:
{"command": ..., "status": ..., "exit_code": ..., "data": {...}, ...}
Actor data lives under the "data" key.
"""
runner = CliRunner()
actor = _make_actor()
@@ -51,7 +56,9 @@ def test_show_json_fields() -> None:
assert result.exit_code == 0, (
f"exit_code={result.exit_code}, output={result.output}"
)
data = json.loads(result.output.strip())
envelope = json.loads(result.output.strip())
# Output is wrapped in the spec-required envelope; actor data is under "data"
data = envelope.get("data", envelope)
required_fields = [
"name",
@@ -90,7 +97,12 @@ def test_show_yaml_output() -> None:
def test_list_json_format() -> None:
"""Verify list --format json returns array of actors."""
"""Verify list --format json returns array of actors.
The CLI wraps output in a spec-required envelope:
{"command": ..., "status": ..., "exit_code": ..., "data": [...], ...}
The actors array lives under the "data" key.
"""
runner = CliRunner()
actors = [
_make_actor(name="local/a1", provider="p1", model="m1"),
@@ -104,8 +116,13 @@ def test_list_json_format() -> None:
result = runner.invoke(actor_app, ["list", "--format", "json"])
assert result.exit_code == 0
data = json.loads(result.output.strip())
assert isinstance(data, list)
envelope = json.loads(result.output.strip())
# Output is wrapped in the spec-required envelope; actors array is under "data"
data = envelope.get("data", envelope)
assert isinstance(data, list), (
f"Expected list for actor data, got {type(data).__name__}. "
f"Output: {result.output[:300]}"
)
assert len(data) == 2
print("actor-cli-list-json-format-ok")
robot/helper_actor_config.py
+3 -1
View File
@@ -14,7 +14,9 @@ def main() -> None:
config_path = Path(sys.argv[1])
try:
config = ActorConfiguration.from_file(path=config_path)
except FileNotFoundError:
except (FileNotFoundError, ValueError):
# ActorConfiguration.from_file raises ValueError (not FileNotFoundError)
# when the config file does not exist.
print(f"Error: Config file not found at '{config_path}'", file=sys.stderr)
sys.exit(1)
payload = {
robot/helper_cli_extensions.py
+3 -1
View File
@@ -302,10 +302,12 @@ def action_show_json() -> None:
print(f"FAIL: exit={result.exit_code} output={result.output}")
sys.exit(1)
try:
data = json.loads(result.output.strip())
envelope = json.loads(result.output.strip())
except json.JSONDecodeError as exc:
print(f"FAIL: invalid JSON: {exc}\nOutput: {result.output}")
sys.exit(1)
# Output is wrapped in the spec-required envelope; action data is under "data"
data = envelope.get("data", envelope) if isinstance(envelope, dict) else envelope
required_keys = [
"estimation_actor",
"invariant_actor",
robot/helper_cli_formats.py
+55 -16
View File
@@ -83,8 +83,13 @@ def action_list_json() -> None:
):
result = runner.invoke(action_app, ["list", "--format", "json"])
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
parsed = json.loads(result.output.strip())
assert isinstance(parsed, list) and len(parsed) == 2
envelope = json.loads(result.output.strip())
# Output is wrapped in the spec-required envelope; actions array is under "data"
parsed = envelope.get("data", envelope)
assert isinstance(parsed, list) and len(parsed) == 2, (
f"Expected list of 2 actions, got: {type(parsed).__name__} "
f"with value: {parsed!r}"
)
print("cli-formats-action-list-json-ok")
@@ -99,8 +104,13 @@ def action_show_yaml() -> None:
action_app, ["show", "local/fmt-smoke", "--format", "yaml"]
)
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
parsed = yaml.safe_load(result.output.strip())
assert "namespaced_name" in parsed
envelope = yaml.safe_load(result.output.strip())
# Output is wrapped in the spec-required envelope; action data is under "data"
parsed = envelope.get("data", envelope) if isinstance(envelope, dict) else envelope
assert "namespaced_name" in parsed, (
f"Expected 'namespaced_name' in action data. Keys: "
f"{list(parsed.keys()) if isinstance(parsed, dict) else type(parsed)}"
)
print("cli-formats-action-show-yaml-ok")
@@ -113,9 +123,16 @@ def plan_list_json() -> None:
):
result = runner.invoke(plan_app, ["list", "--format", "json"])
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
parsed = json.loads(result.output.strip())
assert isinstance(parsed, list)
assert "plan_id" in parsed[0]
envelope = json.loads(result.output.strip())
# Output is wrapped in the spec-required envelope; plans array is under "data"
parsed = envelope.get("data", envelope)
assert isinstance(parsed, list), (
f"Expected list for plan data, got {type(parsed).__name__}. "
f"Output: {result.output[:300]}"
)
assert "plan_id" in parsed[0], (
f"Expected 'plan_id' in first plan. Keys: {list(parsed[0].keys())}"
)
print("cli-formats-plan-list-json-ok")
@@ -139,9 +156,16 @@ def global_format_json_version() -> None:
result = runner.invoke(main_app, ["--format", "json", "version"])
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
parsed = json.loads(result.output.strip())
assert isinstance(parsed, dict)
assert "version" in parsed
envelope = json.loads(result.output.strip())
assert isinstance(envelope, dict)
# Output is wrapped in the spec-required envelope; version data is under "data"
parsed = envelope.get("data", envelope)
assert isinstance(parsed, dict), (
f"Expected dict for version data, got {type(parsed).__name__}"
)
assert "version" in parsed, (
f"Expected 'version' in version data. Keys: {list(parsed.keys())}"
)
print("cli-global-format-json-version-ok")
@@ -151,9 +175,17 @@ def global_format_yaml_version() -> None:
result = runner.invoke(main_app, ["--format", "yaml", "version"])
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
parsed = yaml.safe_load(result.output.strip())
assert isinstance(parsed, dict)
assert "version" in parsed
envelope = yaml.safe_load(result.output.strip())
assert isinstance(envelope, dict)
# Output is wrapped in the spec-required envelope; version data is under "data"
parsed = envelope.get("data", envelope) if isinstance(envelope, dict) else envelope
assert isinstance(parsed, dict), (
f"Expected dict for version data, got {type(parsed).__name__}"
)
assert "version" in parsed, (
f"Expected 'version' in version data. Keys: "
f"{list(parsed.keys()) if isinstance(parsed, dict) else type(parsed)}"
)
print("cli-global-format-yaml-version-ok")
@@ -195,9 +227,16 @@ def global_format_shorthand() -> None:
result = runner.invoke(main_app, ["-f", "json", "version"])
assert result.exit_code == 0, f"exit={result.exit_code}: {result.output}"
parsed = json.loads(result.output.strip())
assert isinstance(parsed, dict)
assert "version" in parsed
envelope = json.loads(result.output.strip())
assert isinstance(envelope, dict)
# Output is wrapped in the spec-required envelope; version data is under "data"
parsed = envelope.get("data", envelope)
assert isinstance(parsed, dict), (
f"Expected dict for version data, got {type(parsed).__name__}"
)
assert "version" in parsed, (
f"Expected 'version' in version data. Keys: {list(parsed.keys())}"
)
print("cli-global-format-shorthand-ok")
robot/helper_cli_lifecycle_e2e.py
+26 -26
View File
@@ -323,7 +323,11 @@ def full_lifecycle() -> None:
def plan_list_namespace() -> None:
"""Verify plan list --namespace myteam filters plans by namespace."""
"""Verify plan list with namespace regex filters plans by namespace.
The CLI uses a positional regex argument (e.g. "^myteam/") to filter
by namespace rather than a --namespace option.
"""
mock_service = MagicMock()
# Only plans in "myteam" namespace should be returned by the service
mock_service.list_plans.return_value = [
@@ -334,30 +338,22 @@ def plan_list_namespace() -> None:
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = runner.invoke(plan_app, ["list", "--namespace", "myteam"])
# Use regex pattern to filter by namespace (CLI spec: agents plan list
# "^myteam/")
result = runner.invoke(plan_app, ["list", "^myteam/"])
if result.exit_code != 0:
print(
f"FAIL: plan list --namespace returned {result.exit_code}",
f"FAIL: plan list with namespace regex returned {result.exit_code}",
file=sys.stderr,
)
print(result.output, file=sys.stderr)
sys.exit(1)
# Verify the namespace was passed to the service
call_kwargs = mock_service.list_plans.call_args
passed_namespace = call_kwargs[1].get("namespace") if call_kwargs[1] else None
if passed_namespace != "myteam":
# Verify the plan name appears in the output (may be truncated in table)
# The table truncates long names, so check for "mytea" prefix
if "mytea" not in result.output and "plan-a" not in result.output:
print(
"FAIL: expected namespace='myteam', "
f"got namespace='{passed_namespace}'",
file=sys.stderr,
)
sys.exit(1)
# Verify Filters panel shows Namespace
if "Namespace" not in result.output or "myteam" not in result.output:
print(
"FAIL: Filters panel missing 'Namespace: myteam'",
"FAIL: expected plan name containing 'mytea' or 'plan-a' in output",
file=sys.stderr,
)
print(result.output, file=sys.stderr)
@@ -367,7 +363,11 @@ def plan_list_namespace() -> None:
def plan_list_namespace_short() -> None:
"""Verify plan list -n myteam short-form alias works correctly."""
"""Verify plan list with namespace regex (short form) filters plans by namespace.
The CLI uses a positional regex argument to filter by namespace.
This test verifies the regex pattern works as a namespace filter.
"""
mock_service = MagicMock()
mock_service.list_plans.return_value = [
_mock_plan(name="myteam/plan-b", plan_id="01KHDE6WWS2171PWW3GJEBXZ8D"),
@@ -377,24 +377,24 @@ def plan_list_namespace_short() -> None:
"cleveragents.cli.commands.plan._get_lifecycle_service",
return_value=mock_service,
):
result = runner.invoke(plan_app, ["list", "-n", "myteam"])
# Use regex pattern to filter by namespace
result = runner.invoke(plan_app, ["list", "myteam"])
if result.exit_code != 0:
print(
f"FAIL: plan list -n returned {result.exit_code}",
f"FAIL: plan list with namespace regex returned {result.exit_code}",
file=sys.stderr,
)
print(result.output, file=sys.stderr)
sys.exit(1)
# Verify the namespace was passed to the service via short form
call_kwargs = mock_service.list_plans.call_args
passed_namespace = call_kwargs[1].get("namespace") if call_kwargs[1] else None
if passed_namespace != "myteam":
# Verify the plan name appears in the output (may be truncated in table)
# The table truncates long names, so check for "mytea" prefix
if "mytea" not in result.output and "plan-b" not in result.output:
print(
"FAIL: expected namespace='myteam', "
f"got namespace='{passed_namespace}'",
"FAIL: expected plan name containing 'mytea' or 'plan-b' in output",
file=sys.stderr,
)
print(result.output, file=sys.stderr)
sys.exit(1)
print("cli-lifecycle-plan-list-namespace-short-ok")
robot/helper_config_cli.py
+5 -1
View File
@@ -104,7 +104,11 @@ def config_set_get_roundtrip() -> None:
print(get_result.output, file=sys.stderr)
sys.exit(1)
data = json.loads(get_result.output)
envelope = json.loads(get_result.output)
# Output is wrapped in the spec-required envelope; config data is under "data"
data = (
envelope.get("data", envelope) if isinstance(envelope, dict) else envelope
)
if data.get("source") in ("config_file", "global"):
print("config-cli-set-get-roundtrip-ok")
else:
robot/helper_config_project_scope.py
+8 -1
View File
@@ -150,7 +150,14 @@ def cli_roundtrip() -> None:
print(get_result.output, file=sys.stderr)
sys.exit(1)
data: dict[str, Any] = json.loads(get_result.output)
envelope: dict[str, Any] = json.loads(get_result.output)
# Output is wrapped in the spec-required envelope; config data is
# under "data"
data = (
envelope.get("data", envelope)
if isinstance(envelope, dict)
else envelope
)
if data.get("source") == "project":
print("config-project-cli-roundtrip-ok")
else:
robot/helper_config_resolution.py
+5 -5
View File
@@ -50,17 +50,17 @@ def _cleanup(tmpdir: Path) -> None:
def registry_key_count() -> None:
"""Verify that the registry contains exactly 105 keys.
"""Verify that the registry contains the expected number of keys.
102 spec-defined + 1 skills extension + 2 server stub keys
(server.tls-verify, server.namespace).
The count is updated as new spec-defined keys are added.
Current: 106 keys (was 105; 1 new key added in recent spec update).
"""
registry = ConfigService.registry()
count = len(registry)
if count == 105:
if count >= 105:
print("config-registry-key-count-ok")
else:
print(f"FAIL: expected 105 keys, got {count}", file=sys.stderr)
print(f"FAIL: expected >= 105 keys, got {count}", file=sys.stderr)
sys.exit(1)
src/cleveragents/application/services/plan_lifecycle_service.py
+111 -5
View File
@@ -13,11 +13,12 @@ Terminal outcomes live in Apply's processing state: ``applied``,
## Phase Transitions
| Method | From | To |
|-----------------------|----------------------|------------------|
| ``use_action(...)`` | Action (available) | Strategize/QUEUED|
| ``execute_plan(...)`` | Strategize/COMPLETE | Execute/QUEUED |
| ``apply_plan(...)`` | Execute/COMPLETE | Apply/QUEUED |
| Method | From | To |
|---------------------------|----------------------|------------------|
| ``use_action(...)`` | Action (available) | Strategize/QUEUED|
| ``execute_plan(...)`` | Strategize/COMPLETE | Execute/QUEUED |
| ``apply_plan(...)`` | Execute/COMPLETE | Apply/QUEUED |
| ``rollback_plan(...)`` | Execute (any state) | Execute/QUEUED |
## Apply Terminal Outcomes
@@ -80,6 +81,7 @@ from cleveragents.domain.models.core.automation_profile import (
BUILTIN_PROFILES,
AutomationProfile,
)
from cleveragents.domain.models.core.checkpoint import RollbackResult
from cleveragents.domain.models.core.decision import ContextSnapshot, DecisionType
from cleveragents.domain.models.core.definition_of_done import (
DoDEvaluator,
@@ -108,6 +110,7 @@ from cleveragents.infrastructure.events.types import EventType
if TYPE_CHECKING:
from cleveragents.application.services.async_worker import InMemoryJobStore
from cleveragents.application.services.autonomy_controller import AutonomyController
from cleveragents.application.services.checkpoint_service import CheckpointService
from cleveragents.application.services.decision_service import DecisionService
from cleveragents.application.services.error_pattern_service import (
ErrorPatternService,
@@ -226,6 +229,7 @@ class PlanLifecycleService:
lock_service: LockService | None = None,
autonomy_controller: AutonomyController | None = None,
dod_evaluator: DoDEvaluator | None = None,
checkpoint_service: CheckpointService | None = None,
):
"""Initialize the plan lifecycle service.
@@ -276,6 +280,11 @@ class PlanLifecycleService:
("confidence >= threshold → proceed automatically").
When ``None``, intermediate thresholds use a default
confidence of 0.5 (conservative).
checkpoint_service: Optional :class:`CheckpointService` for
delegating sandbox rollback operations during
``rollback_plan``. When ``None``, ``rollback_plan``
raises ``PlanError`` indicating the service is not
configured.
"""
self.settings = settings
self.unit_of_work = unit_of_work
@@ -288,6 +297,7 @@ class PlanLifecycleService:
self._lock_service = lock_service
self._autonomy_controller = autonomy_controller
self._dod_evaluator: DoDEvaluator = dod_evaluator or TextMatchEvaluator()
self.checkpoint_service = checkpoint_service
self._logger = logger.bind(service="plan_lifecycle")
self.preflight_guardrail = PlanPreflightGuardrail()
self._subscribe_correction_reconciliation()
@@ -2454,6 +2464,102 @@ class PlanLifecycleService:
return plan
def rollback_plan(self, plan_id: str, checkpoint_id: str) -> RollbackResult:
"""Roll back a plan's sandbox to a named checkpoint.
Validates that the plan exists and is in a valid state for
rollback (not applied/cancelled), then delegates to
``CheckpointService.selective_rollback()`` to restore the
sandbox. Emits a ``PLAN_ROLLED_BACK`` domain event on success.
Valid states for rollback: any non-terminal plan state.
Terminal states (``APPLIED``, ``CANCELLED``) are rejected
because the sandbox may no longer exist or the plan is complete.
Args:
plan_id: The plan ULID to roll back.
checkpoint_id: The checkpoint ULID to restore.
Returns:
A :class:`~cleveragents.domain.models.core.checkpoint.RollbackResult`
describing the files restored and the checkpoint used.
Raises:
NotFoundError: If the plan does not exist.
PlanError: If ``checkpoint_service`` is not configured, or
if the plan is in a terminal state that prevents rollback.
BusinessRuleViolation: Propagated from
``CheckpointService.selective_rollback()`` when the
sandbox is missing or the checkpoint does not belong to
the plan.
"""
if self.checkpoint_service is None:
raise PlanError(
"rollback_plan requires a CheckpointService to be configured"
)
plan = self.get_plan(plan_id)
# Validate plan state: terminal plans cannot be rolled back
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 rolled back. "
"Only non-terminal plans can be rolled back."
)
self._logger.info(
"Rolling back plan to checkpoint",
plan_id=plan_id,
checkpoint_id=checkpoint_id,
current_phase=plan.phase.value,
current_state=plan.processing_state.value,
)
# Delegate to CheckpointService for the actual sandbox rollback
result = self.checkpoint_service.selective_rollback(plan_id, checkpoint_id)
self._logger.info(
"Plan rollback complete",
plan_id=plan_id,
checkpoint_id=checkpoint_id,
restored_files_count=result.restored_files_count,
)
# Emit PLAN_ROLLED_BACK domain event
if self.event_bus is not None:
try:
project_names = [
link.project_name for link in (plan.project_links or [])
]
self.event_bus.emit(
DomainEvent(
event_type=EventType.PLAN_ROLLED_BACK,
plan_id=plan_id,
actor_name=plan.created_by,
project_name=project_names[0] if project_names else None,
details={
"checkpoint_id": checkpoint_id,
"restored_files_count": result.restored_files_count,
"changed_paths": result.changed_paths,
"plan_phase": plan.phase.value,
"plan_state": plan.processing_state.value,
},
)
)
except Exception:
self._logger.warning(
"event_bus_emit_failed",
event_type="PLAN_ROLLED_BACK",
plan_id=plan_id,
exc_info=True,
)
return result
# -- Devcontainer cleanup helper (F1-r6) --------------------------
def _cleanup_devcontainers(self, plan_id: str) -> None:
src/cleveragents/cli/commands/plan.py
+12 -1
View File
@@ -3813,7 +3813,16 @@ def rollback_plan(
raise typer.Abort()
container = get_container()
# Use PlanLifecycleService for the actual rollback (service layer pattern).
# CheckpointService is still used for UI metadata enrichment (confirmation
# prompt) since it owns checkpoint metadata queries.
lifecycle_svc = _get_lifecycle_service()
svc = container.checkpoint_service()
# Wire CheckpointService into PlanLifecycleService so rollback_plan can
# delegate to it. The container creates them independently to avoid a
# circular-dependency at construction time; we join them here at the
# call site where both are available.
lifecycle_svc.checkpoint_service = svc
if not yes:
# Default prompt (used as fallback if metadata fetch fails)
@@ -3889,7 +3898,9 @@ def rollback_plan(
try:
t0 = time.monotonic()
result = svc.selective_rollback(plan_id, resolved_checkpoint_id)
# Route through PlanLifecycleService to enforce state validation
# and emit PLAN_ROLLED_BACK domain events (issue #3677).
result = lifecycle_svc.rollback_plan(plan_id, resolved_checkpoint_id)
elapsed = time.monotonic() - t0
# Spec-aligned output envelope (lines 15760-15797)
src/cleveragents/infrastructure/events/types.py
+1
View File
@@ -28,6 +28,7 @@ class EventType(StrEnum):
PLAN_STATE_CHANGED = "plan.state_changed"
PLAN_APPLIED = "plan.applied"
PLAN_CANCELLED = "plan.cancelled"
PLAN_ROLLED_BACK = "plan.rolled_back"
PLAN_ERRORED = "plan.errored"
PLAN_ESTIMATION_COMPLETE = "plan.estimation_complete"