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

feat(plan): implement real revert-mode re-execution from decision point #1153

Merged
brent.edwards merged 2 commits from feat/844-revert-mode-re-execution into master 2026-03-28 00:57:51 +00:00
Conversation 1 Commits 2 Files Changed 7
+1035 -9
7 changed files with 1035 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+9
View File
@@ -77,6 +77,15 @@
`final_validation_results` fields on the result model, DI container
registration per ADR-003, and structured logging via `structlog`.
(#583)
- Implemented real revert-mode re-execution from decision point.
`CorrectionService.execute_revert` now performs checkpoint restoration via
`CheckpointService`, extracts `actor_state_ref` from the target decision's
context snapshot for reasoning rollback, generates a `user_intervention`
decision ID for guidance injection, and signals phase transition to
Strategize. Added `checkpoint_restored`, `actor_state_ref`,
`user_intervention_decision_id`, and `phase_transition_target` fields to
`CorrectionResult`. Includes 16 Behave BDD scenarios and 7 Robot Framework
integration tests. (#844)
- Added LSP resource types: `executable`, `lsp-server`, `lsp-workspace`,
`lsp-document` with parent/child hierarchy, auto-discovery rules, and
handler references. Registered in bootstrap, with YAML configurations,
features/revert_re_execution.feature
+140
View File
@@ -0,0 +1,140 @@
Feature: Revert-mode re-execution from decision point
As a plan correction user
I want revert mode to perform real checkpoint restoration, actor state
recovery, and re-execution signalling from the targeted decision point
So that the plan can resume Strategize with the corrected guidance
# ── Checkpoint restoration ─────────────────────────────────────
Scenario: Revert with checkpoint service restores sandbox checkpoint
Given revexec a correction service with a mock checkpoint service
And revexec a checkpoint exists for plan "P1" decision "D1" with ref "abc123"
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction
Then revexec the result checkpoint_restored should be true
And revexec the result status should be "applied"
Scenario: Revert without checkpoint service skips checkpoint restoration
Given revexec a correction service without checkpoint service
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction
Then revexec the result checkpoint_restored should be false
And revexec the result status should be "applied"
Scenario: Revert with no matching checkpoint skips restoration
Given revexec a correction service with a mock checkpoint service
And revexec no checkpoint exists for plan "P1" decision "D1"
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction
Then revexec the result checkpoint_restored should be false
# ── Actor state restoration ────────────────────────────────────
Scenario: Revert extracts actor_state_ref from target decision
Given revexec a correction service without checkpoint service
And revexec a decision "D1" with actor_state_ref "lg-checkpoint-42"
And revexec a correction request targeting decision "D1" in plan "P1" for revert with decisions
When revexec I execute the revert correction with decisions
Then revexec the result actor_state_ref should be "lg-checkpoint-42"
Scenario: Revert without decisions mapping returns empty actor_state_ref
Given revexec a correction service without checkpoint service
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction
Then revexec the result actor_state_ref should be ""
# ── Phase transition ───────────────────────────────────────────
Scenario: Revert signals phase transition to strategize
Given revexec a correction service without checkpoint service
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction
Then revexec the result phase_transition_target should be "strategize"
# ── User intervention decision ─────────────────────────────────
Scenario: Revert creates a user_intervention decision ID
Given revexec a correction service without checkpoint service
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction
Then revexec the result user_intervention_decision_id should not be empty
# ── Guidance injection ─────────────────────────────────────────
Scenario: Revert with guidance propagates to attempt details
Given revexec a correction service without checkpoint service
And revexec a correction request targeting decision "D1" in plan "P1" for revert with guidance "Prioritize payments"
When revexec I execute the revert correction
Then revexec the result status should be "applied"
And revexec the attempt details should contain phase_transition_target
# ── Full re-execution flow ─────────────────────────────────────
Scenario: Full revert flow with checkpoint and actor state
Given revexec a correction service with a mock checkpoint service
And revexec a checkpoint exists for plan "P1" decision "D1" with ref "commit-sha-1"
And revexec a decision "D1" with actor_state_ref "lg-state-99"
And revexec a decision tree where "D1" has children "D2,D3"
And revexec a correction request targeting decision "D1" in plan "P1" for revert with decisions
When revexec I execute the revert correction with decisions and tree
Then revexec the result checkpoint_restored should be true
And revexec the result actor_state_ref should be "lg-state-99"
And revexec the result phase_transition_target should be "strategize"
And revexec the result user_intervention_decision_id should not be empty
And revexec the reverted decisions should contain "D1"
And revexec the reverted decisions should contain "D2"
And revexec the reverted decisions should contain "D3"
# ── Subtree selectivity ────────────────────────────────────────
Scenario: Revert only affects downstream decisions from target
Given revexec a correction service without checkpoint service
And revexec a two-level tree "ROOT" children "D1,D2" then "D1" children "D3"
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction with tree
Then revexec the reverted decisions should contain "D1"
And revexec the reverted decisions should contain "D3"
And revexec the reverted decisions should not contain "ROOT"
And revexec the reverted decisions should not contain "D2"
# ── Dispatch integration ───────────────────────────────────────
Scenario: Execute correction dispatch passes decisions to revert
Given revexec a correction service without checkpoint service
And revexec a decision "D1" with actor_state_ref "dispatch-state-ref"
And revexec a correction request targeting decision "D1" in plan "P1" for revert with decisions
When revexec I dispatch execute_correction with decisions
Then revexec the result actor_state_ref should be "dispatch-state-ref"
And revexec the result phase_transition_target should be "strategize"
# ── Error paths ─────────────────────────────────────────────────
Scenario: Revert handles checkpoint list failure gracefully
Given revexec a correction service with a failing checkpoint list
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction
Then revexec the result checkpoint_restored should be false
And revexec the result status should be "applied"
Scenario: Revert handles checkpoint rollback failure gracefully
Given revexec a correction service with a failing checkpoint rollback
And revexec a correction request targeting decision "D1" in plan "P1" for revert
When revexec I execute the revert correction
Then revexec the result checkpoint_restored should be false
And revexec the result status should be "applied"
Scenario: Revert with decision not in mapping returns empty actor state
Given revexec a correction service without checkpoint service
And revexec a decision "OTHER" with actor_state_ref "lg-state-miss"
And revexec a correction request targeting decision "D1" in plan "P1" for revert with decisions
When revexec I execute the revert correction with decisions
Then revexec the result actor_state_ref should be ""
# ── CorrectionResult model fields ──────────────────────────────
Scenario: CorrectionResult default revert fields
When revexec I create a CorrectionResult with only correction_id and status
Then revexec the result checkpoint_restored should be false
And revexec the result actor_state_ref should be ""
And revexec the result user_intervention_decision_id should be ""
And revexec the result phase_transition_target should be ""
features/steps/revert_re_execution_steps.py
+384
View File
@@ -0,0 +1,384 @@
"""Step definitions for revert-mode re-execution from decision point.
Covers checkpoint restoration, actor state recovery, phase transition
signalling, user intervention decision creation, and subtree selectivity.
"""
from __future__ import annotations
import re
from typing import Any
from unittest.mock import MagicMock
from behave import given, then, when # type: ignore[import-untyped]
from behave.runner import Context # type: ignore[import-untyped]
from cleveragents.application.services.checkpoint_service import CheckpointService
from cleveragents.application.services.correction_service import CorrectionService
from cleveragents.domain.models.core.checkpoint import Checkpoint, CheckpointMetadata
from cleveragents.domain.models.core.correction import (
CorrectionMode,
CorrectionResult,
CorrectionStatus,
)
from cleveragents.domain.models.core.decision import (
ContextSnapshot,
Decision,
DecisionType,
)
from cleveragents.domain.models.core.plan import ULID_PATTERN
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_ULID_RE = re.compile(ULID_PATTERN)
def _ensure_ulid(value: str) -> str:
"""Return ``value`` when already ULID-like, otherwise generate one."""
if _ULID_RE.match(value):
return value
from ulid import ULID
return str(ULID())
def _make_decision(
decision_id: str,
plan_id: str,
actor_state_ref: str = "",
) -> Decision:
"""Create a minimal Decision for testing actor state extraction."""
from ulid import ULID
return Decision(
decision_id=str(ULID()),
plan_id=str(ULID()),
sequence_number=0,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Test question?",
chosen_option="Option A",
context_snapshot=ContextSnapshot(actor_state_ref=actor_state_ref),
)
def _make_checkpoint(
plan_id: str,
decision_id: str,
sandbox_ref: str,
) -> Checkpoint:
"""Create a Checkpoint aligned to a decision."""
from ulid import ULID
return Checkpoint(
checkpoint_id=str(ULID()),
plan_id=_ensure_ulid(plan_id),
sandbox_ref=sandbox_ref,
decision_id=decision_id,
checkpoint_type="pre_write",
metadata=CheckpointMetadata(reason="decision-aligned"),
)
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("revexec a correction service with a mock checkpoint service")
def step_given_correction_service_with_checkpoint(context: Context) -> None:
mock_cp_svc = MagicMock(spec=CheckpointService)
mock_cp_svc.list_checkpoints.return_value = []
mock_cp_svc.rollback_to_checkpoint.return_value = MagicMock(
restored_files_count=3,
changed_paths=["a.py", "b.py", "c.py"],
from_checkpoint_id="cp1",
)
context.revexec_checkpoint_service = mock_cp_svc
context.revexec_service = CorrectionService(
checkpoint_service=mock_cp_svc,
)
context.revexec_decisions = {}
context.revexec_tree = {}
@given("revexec a correction service without checkpoint service")
def step_given_correction_service_no_checkpoint(context: Context) -> None:
context.revexec_service = CorrectionService()
context.revexec_decisions = {}
context.revexec_tree = {}
@given("revexec a correction service with a failing checkpoint list")
def step_given_correction_service_failing_list(context: Context) -> None:
mock_cp_svc = MagicMock(spec=CheckpointService)
mock_cp_svc.list_checkpoints.side_effect = RuntimeError("list failed")
context.revexec_checkpoint_service = mock_cp_svc
context.revexec_service = CorrectionService(
checkpoint_service=mock_cp_svc,
)
context.revexec_decisions = {}
context.revexec_tree = {}
@given("revexec a correction service with a failing checkpoint rollback")
def step_given_correction_service_failing_rollback(context: Context) -> None:
mock_cp_svc = MagicMock(spec=CheckpointService)
cp = _make_checkpoint("P1", "D1", "ref-fail")
mock_cp_svc.list_checkpoints.return_value = [cp]
mock_cp_svc.rollback_to_checkpoint.side_effect = RuntimeError("rollback failed")
context.revexec_checkpoint_service = mock_cp_svc
context.revexec_service = CorrectionService(
checkpoint_service=mock_cp_svc,
)
context.revexec_decisions = {}
context.revexec_tree = {}
@given(
'revexec a checkpoint exists for plan "{plan_id}" decision "{decision_id}" with ref "{ref}"'
)
def step_given_checkpoint_exists(
context: Context, plan_id: str, decision_id: str, ref: str
) -> None:
cp = _make_checkpoint(plan_id, decision_id, ref)
context.revexec_checkpoint_service.list_checkpoints.return_value = [cp]
@given('revexec no checkpoint exists for plan "{plan_id}" decision "{decision_id}"')
def step_given_no_checkpoint(context: Context, plan_id: str, decision_id: str) -> None:
context.revexec_checkpoint_service.list_checkpoints.return_value = []
@given('revexec a decision "{decision_id}" with actor_state_ref "{ref}"')
def step_given_decision_with_actor_state(
context: Context, decision_id: str, ref: str
) -> None:
decision = _make_decision(decision_id, "P1", actor_state_ref=ref)
context.revexec_decisions[decision_id] = decision
@given(
'revexec a correction request targeting decision "{decision_id}" in plan "{plan_id}" for revert'
)
def step_given_correction_request(
context: Context, decision_id: str, plan_id: str
) -> None:
svc: CorrectionService = context.revexec_service
context.revexec_request = svc.request_correction(
plan_id,
decision_id,
CorrectionMode.REVERT,
)
@given(
'revexec a correction request targeting decision "{decision_id}" in plan "{plan_id}" for revert with decisions'
)
def step_given_correction_request_with_decisions(
context: Context, decision_id: str, plan_id: str
) -> None:
svc: CorrectionService = context.revexec_service
context.revexec_request = svc.request_correction(
plan_id,
decision_id,
CorrectionMode.REVERT,
)
@given(
'revexec a correction request targeting decision "{decision_id}" in plan "{plan_id}" for revert with guidance "{guidance}"'
)
def step_given_correction_request_guidance(
context: Context, decision_id: str, plan_id: str, guidance: str
) -> None:
svc: CorrectionService = context.revexec_service
context.revexec_request = svc.request_correction(
plan_id,
decision_id,
CorrectionMode.REVERT,
guidance=guidance,
)
@given('revexec a decision tree where "{parent}" has children "{children}"')
def step_given_decision_tree_simple(
context: Context, parent: str, children: str
) -> None:
child_list = [c.strip() for c in children.split(",")]
context.revexec_tree[parent] = child_list
@given(
'revexec a two-level tree "{parent}" children "{children}" then "{parent2}" children "{children2}"'
)
def step_given_decision_tree_two_level(
context: Context, parent: str, children: str, parent2: str, children2: str
) -> None:
context.revexec_tree[parent] = [c.strip() for c in children.split(",")]
context.revexec_tree[parent2] = [c.strip() for c in children2.split(",")]
# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------
@when("revexec I execute the revert correction")
def step_when_execute_revert(context: Context) -> None:
svc: CorrectionService = context.revexec_service
tree: dict[str, list[str]] = getattr(context, "revexec_tree", {})
context.revexec_result = svc.execute_revert(
context.revexec_request.correction_id,
decision_tree=tree if tree else None,
)
@when("revexec I execute the revert correction with decisions")
def step_when_execute_revert_with_decisions(context: Context) -> None:
svc: CorrectionService = context.revexec_service
context.revexec_result = svc.execute_revert(
context.revexec_request.correction_id,
decisions=context.revexec_decisions,
)
@when("revexec I execute the revert correction with decisions and tree")
def step_when_execute_revert_full(context: Context) -> None:
svc: CorrectionService = context.revexec_service
context.revexec_result = svc.execute_revert(
context.revexec_request.correction_id,
decision_tree=context.revexec_tree,
decisions=context.revexec_decisions,
)
@when("revexec I execute the revert correction with tree")
def step_when_execute_revert_with_tree(context: Context) -> None:
svc: CorrectionService = context.revexec_service
context.revexec_result = svc.execute_revert(
context.revexec_request.correction_id,
decision_tree=context.revexec_tree,
)
@when("revexec I dispatch execute_correction with decisions")
def step_when_dispatch_with_decisions(context: Context) -> None:
svc: CorrectionService = context.revexec_service
context.revexec_result = svc.execute_correction(
context.revexec_request.correction_id,
decisions=context.revexec_decisions,
)
@when("revexec I create a CorrectionResult with only correction_id and status")
def step_when_create_default_result(context: Context) -> None:
context.revexec_result = CorrectionResult(
correction_id="test-corr-1",
status=CorrectionStatus.APPLIED,
)
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("revexec the result checkpoint_restored should be {expected}")
def step_then_checkpoint_restored(context: Context, expected: str) -> None:
result: CorrectionResult = context.revexec_result
expected_bool = expected.lower() == "true"
assert result.checkpoint_restored == expected_bool, (
f"Expected checkpoint_restored={expected_bool}, "
f"got {result.checkpoint_restored}"
)
@then('revexec the result status should be "{expected}"')
def step_then_result_status(context: Context, expected: str) -> None:
result: CorrectionResult = context.revexec_result
assert str(result.status) == expected, (
f"Expected status={expected}, got {result.status}"
)
@then('revexec the result actor_state_ref should be "{expected}"')
def step_then_actor_state_ref(context: Context, expected: str) -> None:
result: CorrectionResult = context.revexec_result
assert result.actor_state_ref == expected, (
f"Expected actor_state_ref={expected!r}, got {result.actor_state_ref!r}"
)
@then('revexec the result actor_state_ref should be ""')
def step_then_actor_state_ref_empty(context: Context) -> None:
result: CorrectionResult = context.revexec_result
assert result.actor_state_ref == "", (
f"Expected actor_state_ref to be empty, got {result.actor_state_ref!r}"
)
@then('revexec the result phase_transition_target should be "{expected}"')
def step_then_phase_transition(context: Context, expected: str) -> None:
result: CorrectionResult = context.revexec_result
assert result.phase_transition_target == expected, (
f"Expected phase_transition_target={expected!r}, "
f"got {result.phase_transition_target!r}"
)
@then('revexec the result phase_transition_target should be ""')
def step_then_phase_transition_empty(context: Context) -> None:
result: CorrectionResult = context.revexec_result
assert result.phase_transition_target == "", (
"Expected phase_transition_target to be empty, "
f"got {result.phase_transition_target!r}"
)
@then("revexec the result user_intervention_decision_id should not be empty")
def step_then_intervention_not_empty(context: Context) -> None:
result: CorrectionResult = context.revexec_result
assert result.user_intervention_decision_id, (
"Expected non-empty user_intervention_decision_id"
)
@then('revexec the result user_intervention_decision_id should be ""')
def step_then_intervention_empty(context: Context) -> None:
result: CorrectionResult = context.revexec_result
assert result.user_intervention_decision_id == "", (
f"Expected empty user_intervention_decision_id, "
f"got {result.user_intervention_decision_id!r}"
)
@then('revexec the reverted decisions should contain "{decision_id}"')
def step_then_reverted_contains(context: Context, decision_id: str) -> None:
result: CorrectionResult = context.revexec_result
assert decision_id in result.reverted_decisions, (
f"Expected {decision_id} in reverted_decisions, got {result.reverted_decisions}"
)
@then('revexec the reverted decisions should not contain "{decision_id}"')
def step_then_reverted_not_contains(context: Context, decision_id: str) -> None:
result: CorrectionResult = context.revexec_result
assert decision_id not in result.reverted_decisions, (
f"Expected {decision_id} NOT in reverted_decisions, "
f"got {result.reverted_decisions}"
)
@then("revexec the attempt details should contain phase_transition_target")
def step_then_attempt_has_phase(context: Context) -> None:
svc: CorrectionService = context.revexec_service
attempts = svc.list_attempts(context.revexec_request.correction_id)
assert len(attempts) > 0, "Expected at least one attempt"
details: dict[str, Any] = attempts[0].details
assert "phase_transition_target" in details, (
f"Expected phase_transition_target in attempt details, "
f"got keys: {list(details.keys())}"
)
robot/helper_revert_re_execution.py
+212
View File
@@ -0,0 +1,212 @@
"""Helper script for revert re-execution Robot Framework integration tests.
Exercises the full revert-mode re-execution pipeline: checkpoint restoration,
actor state recovery, phase transition signalling, user intervention decision
creation, and selective subtree recomputation.
"""
from __future__ import annotations
import sys
from unittest.mock import MagicMock
from cleveragents.application.services.checkpoint_service import CheckpointService
from cleveragents.application.services.correction_service import CorrectionService
from cleveragents.domain.models.core.checkpoint import Checkpoint, CheckpointMetadata
from cleveragents.domain.models.core.correction import CorrectionMode
from cleveragents.domain.models.core.decision import (
ContextSnapshot,
Decision,
DecisionType,
)
# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------
_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"
_DECISION_D1 = "D1"
def _make_decision(
decision_id: str,
actor_state_ref: str = "",
) -> Decision:
"""Create a minimal Decision for testing."""
from ulid import ULID
return Decision(
decision_id=str(ULID()),
plan_id=str(ULID()),
sequence_number=0,
decision_type=DecisionType.STRATEGY_CHOICE,
question="Test question?",
chosen_option="Option A",
context_snapshot=ContextSnapshot(actor_state_ref=actor_state_ref),
)
def _mock_checkpoint_service(
checkpoints: list[Checkpoint] | None = None,
) -> CheckpointService:
"""Create a mock CheckpointService."""
mock = MagicMock(spec=CheckpointService)
mock.list_checkpoints.return_value = checkpoints or []
mock.rollback_to_checkpoint.return_value = MagicMock(
restored_files_count=3,
changed_paths=["a.py", "b.py", "c.py"],
from_checkpoint_id="cp1",
)
return mock
def _make_checkpoint(
plan_id: str,
decision_id: str,
sandbox_ref: str,
) -> Checkpoint:
"""Create a Checkpoint aligned to a decision."""
from ulid import ULID
return Checkpoint(
checkpoint_id=str(ULID()),
plan_id=plan_id,
sandbox_ref=sandbox_ref,
decision_id=decision_id,
checkpoint_type="pre_write",
metadata=CheckpointMetadata(reason="decision-aligned"),
)
# ---------------------------------------------------------------------------
# Command handlers
# ---------------------------------------------------------------------------
def _revert_with_checkpoint() -> None:
cp = _make_checkpoint(_PLAN_ID, _DECISION_D1, "abc123")
mock_cp_svc = _mock_checkpoint_service([cp])
svc = CorrectionService(checkpoint_service=mock_cp_svc)
req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
result = svc.execute_revert(req.correction_id)
assert result.checkpoint_restored is True, (
f"Expected checkpoint_restored=True, got {result.checkpoint_restored}"
)
assert result.status == "applied"
mock_cp_svc.rollback_to_checkpoint.assert_called_once()
print("revert-with-checkpoint-ok")
def _revert_no_checkpoint() -> None:
svc = CorrectionService()
req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
result = svc.execute_revert(req.correction_id)
assert result.checkpoint_restored is False
assert result.status == "applied"
print("revert-no-checkpoint-ok")
def _revert_actor_state() -> None:
decision = _make_decision(_DECISION_D1, actor_state_ref="lg-state-42")
decisions = {_DECISION_D1: decision}
svc = CorrectionService()
req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
result = svc.execute_revert(req.correction_id, decisions=decisions)
assert result.actor_state_ref == "lg-state-42", (
f"Expected actor_state_ref='lg-state-42', got {result.actor_state_ref!r}"
)
print("revert-actor-state-ok")
def _revert_phase_transition() -> None:
svc = CorrectionService()
req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
result = svc.execute_revert(req.correction_id)
assert result.phase_transition_target == "strategize", (
f"Expected phase_transition_target='strategize', "
f"got {result.phase_transition_target!r}"
)
print("revert-phase-transition-ok")
def _revert_intervention_id() -> None:
svc = CorrectionService()
req = svc.request_correction(_PLAN_ID, _DECISION_D1, CorrectionMode.REVERT)
result = svc.execute_revert(req.correction_id)
assert result.user_intervention_decision_id, (
"Expected non-empty user_intervention_decision_id"
)
print("revert-intervention-id-ok")
def _full_revert_flow() -> None:
cp = _make_checkpoint(_PLAN_ID, _DECISION_D1, "commit-sha-1")
mock_cp_svc = _mock_checkpoint_service([cp])
decision = _make_decision(_DECISION_D1, actor_state_ref="lg-state-99")
decisions = {_DECISION_D1: decision}
tree = {_DECISION_D1: ["D2", "D3"]}
svc = CorrectionService(checkpoint_service=mock_cp_svc)
req = svc.request_correction(
_PLAN_ID,
_DECISION_D1,
CorrectionMode.REVERT,
guidance="Prioritize payments",
)
result = svc.execute_revert(
req.correction_id,
decision_tree=tree,
decisions=decisions,
)
assert result.status == "applied"
assert result.checkpoint_restored is True
assert result.actor_state_ref == "lg-state-99"
assert result.phase_transition_target == "strategize"
assert result.user_intervention_decision_id
assert "D1" in result.reverted_decisions
assert "D2" in result.reverted_decisions
assert "D3" in result.reverted_decisions
print("full-revert-flow-ok")
def _revert_selective_subtree() -> None:
tree = {"ROOT": ["D1", "D2"], "D1": ["D3"]}
svc = CorrectionService()
req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
result = svc.execute_revert(req.correction_id, decision_tree=tree)
assert "D1" in result.reverted_decisions
assert "D3" in result.reverted_decisions
assert "ROOT" not in result.reverted_decisions
assert "D2" not in result.reverted_decisions
print("revert-selective-subtree-ok")
# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------
COMMANDS: dict[str, object] = {
"revert-with-checkpoint": _revert_with_checkpoint,
"revert-no-checkpoint": _revert_no_checkpoint,
"revert-actor-state": _revert_actor_state,
"revert-phase-transition": _revert_phase_transition,
"revert-intervention-id": _revert_intervention_id,
"full-revert-flow": _full_revert_flow,
"revert-selective-subtree": _revert_selective_subtree,
}
def main() -> None:
"""Dispatch command from sys.argv."""
if len(sys.argv) < 2:
raise SystemExit("Expected command argument")
command = sys.argv[1]
if command not in COMMANDS:
raise SystemExit(f"Unknown command: {command}")
func = COMMANDS[command]
if callable(func):
func()
if __name__ == "__main__":
main()
robot/revert_re_execution.robot
+60
View File
@@ -0,0 +1,60 @@
*** Settings ***
Documentation Integration tests for revert-mode re-execution from decision point.
... Validates checkpoint restoration, actor state recovery, phase
... transition signalling, and selective subtree recomputation.
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} robot/helper_revert_re_execution.py
*** Test Cases ***
Revert With Checkpoint Restores Sandbox
[Documentation] Verify revert delegates to CheckpointService when available
[Tags] phase2 correction revert re-execution
${result}= Run Process ${PYTHON} ${HELPER} revert-with-checkpoint cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} revert-with-checkpoint-ok
Revert Without Checkpoint Skips Restoration
[Documentation] Verify revert works without checkpoint service
[Tags] phase2 correction revert re-execution
${result}= Run Process ${PYTHON} ${HELPER} revert-no-checkpoint cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} revert-no-checkpoint-ok
Revert Extracts Actor State Ref
[Documentation] Verify actor state reference is extracted from decision
[Tags] phase2 correction revert re-execution
${result}= Run Process ${PYTHON} ${HELPER} revert-actor-state cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} revert-actor-state-ok
Revert Signals Phase Transition
[Documentation] Verify revert signals Strategize re-entry
[Tags] phase2 correction revert re-execution
${result}= Run Process ${PYTHON} ${HELPER} revert-phase-transition cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} revert-phase-transition-ok
Revert Creates User Intervention Decision
[Documentation] Verify user_intervention decision ID is generated
[Tags] phase2 correction revert re-execution
${result}= Run Process ${PYTHON} ${HELPER} revert-intervention-id cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} revert-intervention-id-ok
Full Revert Re Execution Flow
[Documentation] End-to-end revert with checkpoint + actor state + phase transition
[Tags] phase2 correction revert re-execution e2e
${result}= Run Process ${PYTHON} ${HELPER} full-revert-flow cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} full-revert-flow-ok
Revert Selective Subtree Only
[Documentation] Verify only downstream decisions are affected
[Tags] phase2 correction revert re-execution subtree
${result}= Run Process ${PYTHON} ${HELPER} revert-selective-subtree cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} revert-selective-subtree-ok
src/cleveragents/application/services/correction_service.py
+188 -8
View File
@@ -3,8 +3,21 @@
Orchestrates impact analysis using BFS subtree traversal over both the
structural tree (parent-child) and the influence DAG
(``decision_dependencies`` edges), dry-run reporting, revert execution
(decision invalidation + artifact archival), and append execution
(child plan spawning).
(checkpoint restoration + actor state recovery + re-execution signalling),
and append execution (child plan spawning).
The revert flow implements the full re-execution pipeline specified in
§ Correction Flow (Revert Mode):
1. **Resource rollback** — delegates to ``CheckpointService`` when a
decision-aligned checkpoint exists for the target decision.
2. **Reasoning rollback** — extracts ``actor_state_ref`` from the target
decision's ``ContextSnapshot`` and includes it in the result for
downstream LangGraph checkpoint restoration.
3. **Guidance injection** — creates a ``user_intervention`` decision ID
so callers can record the user's correction guidance in the tree.
4. **Phase transition** — signals that the plan should re-enter the
Strategize phase from the corrected decision point.
"""
from __future__ import annotations
@@ -31,6 +44,7 @@ from cleveragents.infrastructure.events.models import DomainEvent
from cleveragents.infrastructure.events.types import EventType
if TYPE_CHECKING:
from cleveragents.domain.models.core.decision import ContextSnapshot, Decision
from cleveragents.infrastructure.events.protocol import EventBus
logger = structlog.get_logger(__name__)
@@ -402,20 +416,45 @@ class CorrectionService:
correction_id: str,
decision_tree: dict[str, list[str]] | None = None,
influence_edges: dict[str, list[str]] | None = None,
decisions: dict[str, Decision] | None = None,
) -> CorrectionResult:
"""Execute a revert correction.
"""Execute a revert correction with full re-execution pipeline.
Marks every decision in the affected subtree (structural tree
**and** influence DAG) as reverted and records all archived
artifacts.
Implements the specification's Correction Flow (Revert Mode):
1. **Resource rollback**: When a ``CheckpointService`` is
available and the target decision has a decision-aligned
checkpoint, delegates to ``rollback_to_checkpoint`` to
execute a real ``git reset --hard`` in the sandbox.
2. **Reasoning rollback**: Extracts the ``actor_state_ref``
from the target decision's ``context_snapshot`` and returns
it in the result so downstream consumers can restore the
LangGraph actor's reasoning state.
3. **Guidance injection**: Generates a ``user_intervention``
decision ID for callers to record the user's correction
guidance in the decision tree.
4. **Phase transition**: Signals that the plan should re-enter
the Strategize phase from the corrected decision point by
setting ``phase_transition_target`` to ``"strategize"``.
Args:
correction_id: Correction request ID.
decision_tree: Optional structural tree adjacency list.
influence_edges: Optional influence DAG adjacency list.
decisions: Optional mapping of decision_id → ``Decision``
objects. When provided, the target decision's
``context_snapshot.actor_state_ref`` is extracted for
reasoning rollback, and its ``decision_id`` is used to
look up the checkpoint for resource rollback.
Returns:
``CorrectionResult`` with reverted decisions.
``CorrectionResult`` with reverted decisions plus
re-execution metadata (checkpoint_restored,
actor_state_ref, user_intervention_decision_id,
phase_transition_target).
Raises:
ResourceNotFoundError: If correction does not exist.
@@ -446,14 +485,43 @@ class CorrectionService:
)
request.status = CorrectionStatus.EXECUTING
# --- Resource rollback (spec § Mid-Execute Correction) ---
checkpoint_restored = self._try_checkpoint_restoration(
request.plan_id,
request.target_decision_id,
)
# --- Reasoning rollback (spec § actor_state_ref) ---
actor_state_ref = self._extract_actor_state_ref(
request.target_decision_id,
decisions,
)
# --- Guidance injection (spec § user_intervention) ---
user_intervention_id = str(ULID())
# --- Phase transition signal ---
# Revert always re-enters Strategize from the decision point.
phase_target = "strategize"
result = CorrectionResult(
correction_id=correction_id,
status=CorrectionStatus.APPLIED,
reverted_decisions=impact.affected_decisions,
archived_artifacts=impact.artifacts_to_archive,
checkpoint_restored=checkpoint_restored,
actor_state_ref=actor_state_ref,
user_intervention_decision_id=user_intervention_id,
phase_transition_target=phase_target,
)
request.status = CorrectionStatus.APPLIED
attempt.success = True
attempt.details = {
"checkpoint_restored": checkpoint_restored,
"actor_state_ref": actor_state_ref,
"user_intervention_decision_id": user_intervention_id,
"phase_transition_target": phase_target,
}
except Exception as exc:
logger.error(
"correction.revert_failed",
@@ -476,6 +544,9 @@ class CorrectionService:
"correction.revert_executed",
correction_id=correction_id,
status=result.status,
checkpoint_restored=result.checkpoint_restored,
actor_state_ref=result.actor_state_ref,
phase_transition_target=result.phase_transition_target,
)
self._emit_correction_applied(
correction_id, request, result, attempt_id=attempt.attempt_id
@@ -574,6 +645,7 @@ class CorrectionService:
correction_id: str,
decision_tree: dict[str, list[str]] | None = None,
influence_edges: dict[str, list[str]] | None = None,
decisions: dict[str, Decision] | None = None,
) -> CorrectionResult:
"""Execute a correction, dispatching to revert or append.
@@ -583,6 +655,8 @@ class CorrectionService:
(used for revert).
influence_edges: Optional influence DAG adjacency list
(used for revert).
decisions: Optional mapping of decision_id → ``Decision``
objects (used for revert re-execution).
Returns:
``CorrectionResult`` from the chosen strategy.
@@ -590,7 +664,9 @@ class CorrectionService:
request = self._get_request_or_raise(correction_id)
if request.mode == CorrectionMode.REVERT:
return self.execute_revert(correction_id, decision_tree, influence_edges)
return self.execute_revert(
correction_id, decision_tree, influence_edges, decisions
)
return self.execute_append(correction_id)
# ------------------------------------------------------------------
@@ -645,6 +721,110 @@ class CorrectionService:
# Internal helpers
# ------------------------------------------------------------------
# ------------------------------------------------------------------
# Revert re-execution helpers
# ------------------------------------------------------------------
def _try_checkpoint_restoration(
self,
plan_id: str,
target_decision_id: str,
) -> bool:
"""Attempt checkpoint restoration for the target decision.
Queries the ``CheckpointService`` for checkpoints belonging to
the plan and aligned to the target decision. If a matching
checkpoint is found, delegates to
``rollback_to_checkpoint`` for real ``git reset --hard``.
Args:
plan_id: Plan owning the decision tree.
target_decision_id: Decision whose checkpoint to restore.
Returns:
``True`` if a checkpoint was successfully restored,
``False`` if no checkpoint service or no matching
checkpoint was available.
"""
if self._checkpoint_service is None:
logger.info(
"correction.checkpoint_skip",
reason="no_checkpoint_service",
plan_id=plan_id,
target_decision_id=target_decision_id,
)
return False
try:
checkpoints = self._checkpoint_service.list_checkpoints(plan_id)
except Exception:
logger.warning(
"correction.checkpoint_list_failed",
plan_id=plan_id,
exc_info=True,
)
return False
# Find the checkpoint aligned to the target decision.
matching = [cp for cp in checkpoints if cp.decision_id == target_decision_id]
if not matching:
logger.info(
"correction.checkpoint_not_found",
plan_id=plan_id,
target_decision_id=target_decision_id,
)
return False
# Use the most recent matching checkpoint.
checkpoint = matching[-1]
try:
self._checkpoint_service.rollback_to_checkpoint(
plan_id, checkpoint.checkpoint_id
)
logger.info(
"correction.checkpoint_restored",
plan_id=plan_id,
checkpoint_id=checkpoint.checkpoint_id,
target_decision_id=target_decision_id,
)
return True
except Exception:
logger.warning(
"correction.checkpoint_rollback_failed",
plan_id=plan_id,
checkpoint_id=checkpoint.checkpoint_id,
exc_info=True,
)
return False
@staticmethod
def _extract_actor_state_ref(
target_decision_id: str,
decisions: dict[str, Decision] | None,
) -> str:
"""Extract the actor state reference from the target decision.
Looks up the target decision in the provided mapping and
returns its ``context_snapshot.actor_state_ref``. When the
mapping is not provided or the decision is not found, returns
an empty string.
Args:
target_decision_id: Decision to extract state from.
decisions: Optional mapping of decision_id → ``Decision``.
Returns:
The ``actor_state_ref`` string, or ``""`` if unavailable.
"""
if decisions is None:
return ""
decision = decisions.get(target_decision_id)
if decision is None:
return ""
snapshot: ContextSnapshot = decision.context_snapshot
return snapshot.actor_state_ref
def _get_request_or_raise(self, correction_id: str) -> CorrectionRequest:
"""Look up a correction or raise ``ResourceNotFoundError``."""
request = self._corrections.get(correction_id)
src/cleveragents/domain/models/core/correction.py
+42 -1
View File
@@ -177,7 +177,14 @@ class CorrectionImpact(BaseModel):
class CorrectionResult(BaseModel):
"""Outcome of executing a correction."""
"""Outcome of executing a correction.
For revert corrections, the result includes fields for checkpoint
restoration, actor state recovery, phase transition signalling,
and user-intervention decision creation — enabling full re-execution
from the decision point per the specification (§ Correction Flow,
Revert Mode).
"""
model_config = ConfigDict(frozen=True)
@@ -214,6 +221,40 @@ class CorrectionResult(BaseModel):
description="Timestamp when execution finished.",
)
# --- Revert re-execution fields (spec § Correction Flow) ---
checkpoint_restored: bool = Field(
default=False,
description=(
"Whether a sandbox checkpoint was restored during revert. "
"True when the CheckpointService successfully rolled back "
"resources to the target decision's checkpoint."
),
)
actor_state_ref: str = Field(
default="",
description=(
"LangGraph actor checkpoint reference restored from the "
"target decision's context_snapshot.actor_state_ref. "
"Downstream consumers use this to restore the actor's "
"reasoning state for re-execution."
),
)
user_intervention_decision_id: str = Field(
default="",
description=(
"ULID of the user_intervention decision created to inject "
"the correction guidance into the decision tree."
),
)
phase_transition_target: str = Field(
default="",
description=(
"Target plan phase after revert (e.g. 'strategize'). "
"Empty when no phase transition is needed."
),
)
class CorrectionAttempt(BaseModel):
"""Record of a single correction execution attempt.