freemo
dd3770f930
CI / security (push) Has been cancelled
CI / benchmark-publish (push) Has been cancelled
CI / lint (push) Has been cancelled
CI / unit_tests (push) Has been cancelled
CI / build (push) Has been cancelled
CI / typecheck (push) Has been cancelled
CI / quality (push) Has been cancelled
CI / e2e_tests (push) Has been cancelled
CI / integration_tests (push) Has been cancelled
CI / helm (push) Has been cancelled
CI / coverage (push) Has been cancelled
CI / benchmark-regression (push) Has been cancelled
CI / status-check (push) Has been cancelled
CI / docker (push) Has been cancelled
Remove three top-level fields from CorrectionDryRunReport that duplicated data already present in the embedded CorrectionImpact object: - excluded_decisions → now accessed via report.impact.excluded_decisions - rollback_tier_depth → now accessed via report.impact.rollback_tier_depth - child_plans_to_rollback → now accessed via report.impact.affected_child_plans The redundant fields created a divergence risk: both models are frozen=True Pydantic models, so post-construction mutation is impossible, but nothing prevented constructing an instance where the top-level copies disagreed with the impact sub-object. Approach: Option B (nest-only) — remove the duplicated top-level fields and update all consumers to use the impact object's fields instead. Closes #1087 Co-authored-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me> Co-committed-by: Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
461 lines
16 KiB
Python
461 lines
16 KiB
Python
"""Helper script for correction subtree isolation Robot Framework tests.
|
|
|
|
Validates that CorrectionService correctly isolates correction scope:
|
|
- excluded_decisions populated after impact analysis
|
|
- rollback_tier_depth computation
|
|
- dry-run report enhancements
|
|
- subtree isolation validation
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from collections.abc import Callable
|
|
from unittest.mock import patch
|
|
|
|
from cleveragents.application.services.correction_service import CorrectionService
|
|
from cleveragents.core.exceptions import ValidationError
|
|
from cleveragents.domain.models.core.correction import CorrectionMode, CorrectionStatus
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Shared fixtures
|
|
# ---------------------------------------------------------------------------
|
|
_PLAN_ID = "01ROBOTSUBTREEPLAN0000000001"
|
|
|
|
|
|
def _sample_tree() -> dict[str, list[str]]:
|
|
return {
|
|
"root": ["A", "B"],
|
|
"A": ["C", "D"],
|
|
"B": ["E"],
|
|
}
|
|
|
|
|
|
def _deep_tree(depth: int) -> dict[str, list[str]]:
|
|
tree: dict[str, list[str]] = {}
|
|
for i in range(depth):
|
|
parent = f"depth_{i}" if i > 0 else "root"
|
|
child = f"depth_{i + 1}"
|
|
tree[parent] = [child]
|
|
return tree
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Command handlers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _leaf_isolation() -> None:
|
|
"""Target a leaf node and verify only it is affected."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
req = svc.request_correction(_PLAN_ID, "D", CorrectionMode.REVERT)
|
|
impact = svc.analyze_impact(req.correction_id, tree)
|
|
|
|
assert impact.affected_decisions == ["D"], (
|
|
f"Expected ['D'], got {impact.affected_decisions}"
|
|
)
|
|
for excluded in ["root", "A", "B", "C", "E"]:
|
|
assert excluded in impact.excluded_decisions, (
|
|
f"Expected '{excluded}' in excluded={impact.excluded_decisions}"
|
|
)
|
|
print("leaf-isolation-ok")
|
|
|
|
|
|
def _middle_node_isolation() -> None:
|
|
"""Target a middle node and verify subtree is affected."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
req = svc.request_correction(_PLAN_ID, "A", CorrectionMode.REVERT)
|
|
impact = svc.analyze_impact(req.correction_id, tree)
|
|
|
|
for affected in ["A", "C", "D"]:
|
|
assert affected in impact.affected_decisions, (
|
|
f"Expected '{affected}' in affected={impact.affected_decisions}"
|
|
)
|
|
for excluded in ["root", "B", "E"]:
|
|
assert excluded in impact.excluded_decisions, (
|
|
f"Expected '{excluded}' in excluded={impact.excluded_decisions}"
|
|
)
|
|
print("middle-node-isolation-ok")
|
|
|
|
|
|
def _root_targets_all() -> None:
|
|
"""Target root and verify all decisions are affected."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
req = svc.request_correction(_PLAN_ID, "root", CorrectionMode.REVERT)
|
|
impact = svc.analyze_impact(req.correction_id, tree)
|
|
|
|
for node in ["root", "A", "B", "C", "D", "E"]:
|
|
assert node in impact.affected_decisions, (
|
|
f"Expected '{node}' in affected={impact.affected_decisions}"
|
|
)
|
|
assert len(impact.excluded_decisions) == 0, (
|
|
f"Expected empty excluded, got {impact.excluded_decisions}"
|
|
)
|
|
assert impact.rollback_tier_depth == 0
|
|
print("root-targets-all-ok")
|
|
|
|
|
|
def _rollback_tier_computation() -> None:
|
|
"""Verify rollback tier at various depths."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
|
|
assert svc.compute_rollback_tier("root", _PLAN_ID, tree) == 0
|
|
assert svc.compute_rollback_tier("A", _PLAN_ID, tree) == 1
|
|
assert svc.compute_rollback_tier("B", _PLAN_ID, tree) == 1
|
|
assert svc.compute_rollback_tier("C", _PLAN_ID, tree) == 2
|
|
assert svc.compute_rollback_tier("D", _PLAN_ID, tree) == 2
|
|
assert svc.compute_rollback_tier("E", _PLAN_ID, tree) == 2
|
|
|
|
# Deep tree test
|
|
deep = _deep_tree(4)
|
|
assert svc.compute_rollback_tier("depth_3", _PLAN_ID, deep) == 3
|
|
print("rollback-tier-computation-ok")
|
|
|
|
|
|
def _dry_run_report_enhanced() -> None:
|
|
"""Verify dry-run report includes excluded decisions and tier depth via impact."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
req = svc.request_correction(_PLAN_ID, "A", CorrectionMode.REVERT)
|
|
report = svc.generate_dry_run_report(req.correction_id, tree)
|
|
|
|
assert report.impact.rollback_tier_depth == 1, (
|
|
f"Expected tier_depth=1, got {report.impact.rollback_tier_depth}"
|
|
)
|
|
assert len(report.impact.excluded_decisions) > 0, (
|
|
"Expected non-empty excluded_decisions"
|
|
)
|
|
for excluded in ["root", "B", "E"]:
|
|
assert excluded in report.impact.excluded_decisions, (
|
|
f"Expected '{excluded}' in report "
|
|
f"excluded={report.impact.excluded_decisions}"
|
|
)
|
|
print("dry-run-report-enhanced-ok")
|
|
|
|
|
|
def _dry_run_tier_zero_warning() -> None:
|
|
"""Verify tier-0 warning when root is targeted."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
req = svc.request_correction(_PLAN_ID, "root", CorrectionMode.REVERT)
|
|
report = svc.generate_dry_run_report(req.correction_id, tree)
|
|
|
|
all_warnings = " ".join(report.warnings)
|
|
assert "Tier 0" in all_warnings, f"Expected 'Tier 0' in warnings: {report.warnings}"
|
|
assert report.impact.rollback_tier_depth == 0
|
|
print("dry-run-tier-zero-warning-ok")
|
|
|
|
|
|
def _subtree_isolation_validation() -> None:
|
|
"""Verify subtree isolation validation passes for non-root targets."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
|
|
assert svc.validate_subtree_isolation("A", tree) is True
|
|
assert svc.validate_subtree_isolation("D", tree) is True
|
|
assert svc.validate_subtree_isolation("E", tree) is True
|
|
print("subtree-isolation-validation-ok")
|
|
|
|
|
|
def _edge_case_empty_tree() -> None:
|
|
"""Verify edge cases with empty tree."""
|
|
svc = CorrectionService()
|
|
|
|
# validate_subtree_isolation with empty tree
|
|
assert svc.validate_subtree_isolation("X", {}) is True
|
|
|
|
# _find_root with empty tree
|
|
assert CorrectionService._find_root({}) is None
|
|
|
|
# _find_parent for root node (not a child of anything)
|
|
tree = _sample_tree()
|
|
assert CorrectionService._find_parent("root", tree) is None
|
|
|
|
# compute_rollback_tier for unknown node
|
|
assert svc.compute_rollback_tier("UNKNOWN", _PLAN_ID, tree) == 0
|
|
print("edge-case-empty-tree-ok")
|
|
|
|
|
|
def _influence_dag_traversal() -> None:
|
|
"""Verify influence edges pull extra decisions into affected set."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
|
|
# D -> E via influence DAG (E is a sibling-subtree node)
|
|
req = svc.request_correction(_PLAN_ID, "D", CorrectionMode.REVERT)
|
|
influence = {"D": ["E"]}
|
|
impact = svc.analyze_impact(req.correction_id, tree, influence_edges=influence)
|
|
|
|
assert "D" in impact.affected_decisions, (
|
|
f"Expected 'D' in affected={impact.affected_decisions}"
|
|
)
|
|
assert "E" in impact.affected_decisions, (
|
|
f"Expected 'E' in affected (via influence)={impact.affected_decisions}"
|
|
)
|
|
assert "root" in impact.excluded_decisions
|
|
assert "A" in impact.excluded_decisions
|
|
|
|
# Cycle detection: C -> D -> C via influence DAG
|
|
svc2 = CorrectionService()
|
|
req2 = svc2.request_correction(_PLAN_ID, "C", CorrectionMode.REVERT)
|
|
cycle_influence = {"C": ["D"], "D": ["C"]}
|
|
impact2 = svc2.analyze_impact(
|
|
req2.correction_id, tree, influence_edges=cycle_influence
|
|
)
|
|
assert "C" in impact2.affected_decisions
|
|
assert "D" in impact2.affected_decisions
|
|
|
|
# Isolation validation passes even when influence edges reach sibling
|
|
assert svc.validate_subtree_isolation("D", tree, influence_edges=influence) is True
|
|
|
|
print("influence-dag-traversal-ok")
|
|
|
|
|
|
def _append_mode_correction() -> None:
|
|
"""Verify append mode spawns a child plan."""
|
|
svc = CorrectionService()
|
|
req = svc.request_correction(_PLAN_ID, "A", CorrectionMode.APPEND)
|
|
result = svc.execute_append(req.correction_id)
|
|
|
|
assert result.status.value == "applied", (
|
|
f"Expected status='applied', got '{result.status.value}'"
|
|
)
|
|
assert result.spawned_child_plan_id is not None, (
|
|
"Expected spawned_child_plan_id, got None"
|
|
)
|
|
assert len(result.new_decisions) > 0, (
|
|
f"Expected new_decisions, got {result.new_decisions}"
|
|
)
|
|
print("append-mode-correction-ok")
|
|
|
|
|
|
def _dry_run_enforcement() -> None:
|
|
"""Verify that executing a dry-run correction raises ValidationError."""
|
|
svc = CorrectionService()
|
|
|
|
# Revert dry-run should not be executable
|
|
req_revert = svc.request_correction(
|
|
_PLAN_ID, "A", CorrectionMode.REVERT, dry_run=True
|
|
)
|
|
try:
|
|
svc.execute_correction(req_revert.correction_id, _sample_tree())
|
|
raise AssertionError("Expected ValidationError for dry-run revert execution")
|
|
except ValidationError as exc:
|
|
assert "dry-run" in str(exc).lower(), (
|
|
f"Expected 'dry-run' in error message, got: {exc}"
|
|
)
|
|
|
|
# Append dry-run should not be executable
|
|
svc2 = CorrectionService()
|
|
req_append = svc2.request_correction(
|
|
_PLAN_ID, "A", CorrectionMode.APPEND, dry_run=True
|
|
)
|
|
try:
|
|
svc2.execute_correction(req_append.correction_id)
|
|
raise AssertionError("Expected ValidationError for dry-run append execution")
|
|
except ValidationError as exc:
|
|
assert "dry-run" in str(exc).lower(), (
|
|
f"Expected 'dry-run' in error message, got: {exc}"
|
|
)
|
|
|
|
print("dry-run-enforcement-ok")
|
|
|
|
|
|
def _execute_revert_non_dry_run() -> None:
|
|
"""Verify execute_revert works end-to-end for a non-dry-run correction."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
req = svc.request_correction(_PLAN_ID, "A", CorrectionMode.REVERT)
|
|
result = svc.execute_revert(req.correction_id, tree)
|
|
|
|
assert result.status.value == "applied", (
|
|
f"Expected status='applied', got '{result.status.value}'"
|
|
)
|
|
assert sorted(result.reverted_decisions) == sorted(["A", "C", "D"]), (
|
|
f"Expected reverted=['A','C','D'], got {result.reverted_decisions}"
|
|
)
|
|
assert len(result.archived_artifacts) > 0, (
|
|
f"Expected non-empty archived_artifacts, got {result.archived_artifacts}"
|
|
)
|
|
print("execute-revert-non-dry-run-ok")
|
|
|
|
|
|
def _status_guard_enforcement() -> None:
|
|
"""Verify status guards prevent re-execution and cancelled execution."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
|
|
# Re-execution of applied correction
|
|
req = svc.request_correction(_PLAN_ID, "A", CorrectionMode.REVERT)
|
|
svc.execute_revert(req.correction_id, tree)
|
|
try:
|
|
svc.execute_correction(req.correction_id, tree)
|
|
raise AssertionError("Expected ValidationError for re-execution")
|
|
except ValidationError:
|
|
pass
|
|
|
|
# Execution of cancelled correction
|
|
svc2 = CorrectionService()
|
|
req2 = svc2.request_correction(_PLAN_ID, "B", CorrectionMode.REVERT)
|
|
svc2.cancel_correction(req2.correction_id)
|
|
try:
|
|
svc2.execute_correction(req2.correction_id, tree)
|
|
raise AssertionError("Expected ValidationError for cancelled execution")
|
|
except ValidationError:
|
|
pass
|
|
|
|
print("status-guard-enforcement-ok")
|
|
|
|
|
|
def _mode_mismatch_enforcement() -> None:
|
|
"""Verify mode mismatch raises ValidationError."""
|
|
svc = CorrectionService()
|
|
|
|
# execute_revert on an append correction
|
|
req_a = svc.request_correction(_PLAN_ID, "A", CorrectionMode.APPEND)
|
|
try:
|
|
svc.execute_revert(req_a.correction_id, _sample_tree())
|
|
raise AssertionError("Expected ValidationError for mode mismatch")
|
|
except ValidationError as exc:
|
|
assert "REVERT" in str(exc), f"Expected 'REVERT' in error, got: {exc}"
|
|
|
|
# execute_append on a revert correction
|
|
svc2 = CorrectionService()
|
|
req_r = svc2.request_correction(_PLAN_ID, "A", CorrectionMode.REVERT)
|
|
try:
|
|
svc2.execute_append(req_r.correction_id)
|
|
raise AssertionError("Expected ValidationError for mode mismatch")
|
|
except ValidationError as exc:
|
|
assert "APPEND" in str(exc), f"Expected 'APPEND' in error, got: {exc}"
|
|
|
|
print("mode-mismatch-enforcement-ok")
|
|
|
|
|
|
def _diamond_dag_topology() -> None:
|
|
"""Verify convergent (diamond) DAG topology does not duplicate visits."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
|
|
# Both C and D (children of A) influence E via DAG — diamond pattern
|
|
req = svc.request_correction(_PLAN_ID, "A", CorrectionMode.REVERT)
|
|
influence = {"C": ["E"], "D": ["E"]}
|
|
impact = svc.analyze_impact(req.correction_id, tree, influence_edges=influence)
|
|
|
|
assert "E" in impact.affected_decisions, (
|
|
f"Expected 'E' in affected={impact.affected_decisions}"
|
|
)
|
|
count = impact.affected_decisions.count("E")
|
|
assert count == 1, (
|
|
f"Expected 'E' exactly once, found {count} times: {impact.affected_decisions}"
|
|
)
|
|
print("diamond-dag-topology-ok")
|
|
|
|
|
|
def _dry_run_exception_recovery() -> None:
|
|
"""Verify dry-run report restores status when analyze_impact fails."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
req = svc.request_correction(_PLAN_ID, "A", CorrectionMode.REVERT)
|
|
|
|
assert req.status == CorrectionStatus.PENDING, (
|
|
f"Expected PENDING before dry-run, got {req.status}"
|
|
)
|
|
|
|
try:
|
|
with patch.object(
|
|
CorrectionService,
|
|
"_compute_affected_subtree",
|
|
side_effect=RuntimeError("simulated BFS failure"),
|
|
):
|
|
svc.generate_dry_run_report(req.correction_id, tree)
|
|
except RuntimeError:
|
|
pass # Expected — exception propagates from analyze_impact
|
|
|
|
assert req.status == CorrectionStatus.PENDING, (
|
|
f"Expected PENDING after failed dry-run, got {req.status}"
|
|
)
|
|
print("dry-run-exception-recovery-ok")
|
|
|
|
|
|
def _execute_revert_with_influence() -> None:
|
|
"""Verify execute_revert works with influence edges."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
|
|
# D influences E via DAG — revert of A should include E
|
|
influence = {"D": ["E"]}
|
|
req = svc.request_correction(_PLAN_ID, "A", CorrectionMode.REVERT)
|
|
result = svc.execute_revert(req.correction_id, tree, influence_edges=influence)
|
|
|
|
assert result.status.value == "applied", (
|
|
f"Expected status='applied', got '{result.status.value}'"
|
|
)
|
|
assert "E" in result.reverted_decisions, (
|
|
f"Expected 'E' in reverted={result.reverted_decisions}"
|
|
)
|
|
print("execute-revert-with-influence-ok")
|
|
|
|
|
|
def _dag_only_nodes_excluded() -> None:
|
|
"""Verify nodes in the DAG (not in tree) appear in excluded set."""
|
|
svc = CorrectionService()
|
|
tree = _sample_tree()
|
|
|
|
# X_EXT and Y_EXT exist only in the influence DAG, not in the tree
|
|
influence = {"X_EXT": ["Y_EXT"]}
|
|
req = svc.request_correction(_PLAN_ID, "D", CorrectionMode.REVERT)
|
|
impact = svc.analyze_impact(req.correction_id, tree, influence_edges=influence)
|
|
|
|
assert "X_EXT" in impact.excluded_decisions, (
|
|
f"Expected 'X_EXT' in excluded={impact.excluded_decisions}"
|
|
)
|
|
assert "Y_EXT" in impact.excluded_decisions, (
|
|
f"Expected 'Y_EXT' in excluded={impact.excluded_decisions}"
|
|
)
|
|
print("dag-only-nodes-excluded-ok")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dispatch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
COMMANDS: dict[str, Callable[[], None]] = {
|
|
"leaf-isolation": _leaf_isolation,
|
|
"middle-node-isolation": _middle_node_isolation,
|
|
"root-targets-all": _root_targets_all,
|
|
"rollback-tier-computation": _rollback_tier_computation,
|
|
"dry-run-report-enhanced": _dry_run_report_enhanced,
|
|
"dry-run-tier-zero-warning": _dry_run_tier_zero_warning,
|
|
"subtree-isolation-validation": _subtree_isolation_validation,
|
|
"edge-case-empty-tree": _edge_case_empty_tree,
|
|
"influence-dag-traversal": _influence_dag_traversal,
|
|
"append-mode-correction": _append_mode_correction,
|
|
"dry-run-enforcement": _dry_run_enforcement,
|
|
"execute-revert-non-dry-run": _execute_revert_non_dry_run,
|
|
"status-guard-enforcement": _status_guard_enforcement,
|
|
"mode-mismatch-enforcement": _mode_mismatch_enforcement,
|
|
"diamond-dag-topology": _diamond_dag_topology,
|
|
"dry-run-exception-recovery": _dry_run_exception_recovery,
|
|
"execute-revert-with-influence": _execute_revert_with_influence,
|
|
"dag-only-nodes-excluded": _dag_only_nodes_excluded,
|
|
}
|
|
|
|
|
|
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]
|
|
func()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|