cleveragents-core/robot/helper_influence_dag_traversal.py

"""Helper script for influence DAG traversal Robot Framework tests.

Exercises CorrectionService with both structural tree and influence
DAG edges, plus DecisionService influence dependency recording.
"""

from __future__ import annotations

import sys

from cleveragents.application.services.correction_service import CorrectionService
from cleveragents.application.services.decision_service import DecisionService
from cleveragents.domain.models.core.correction import CorrectionMode

# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------
_PLAN_ID = "01ARZ3NDEKTSV4RRFFQ69G5FAV"


def _structural_only() -> None:
    """Revert with only structural children."""
    svc = CorrectionService()
    req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
    tree = {"D1": ["D2", "D3"], "D2": ["D4"]}
    result = svc.execute_revert(req.correction_id, tree)
    assert result.status == "applied"
    assert set(result.reverted_decisions) == {"D1", "D2", "D3", "D4"}
    print("structural-only-ok")


def _influence_only() -> None:
    """Revert with only influence DAG edges."""
    svc = CorrectionService()
    req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
    tree: dict[str, list[str]] = {}
    influence = {"D1": ["D5", "D6"], "D5": ["D7"]}
    result = svc.execute_revert(req.correction_id, tree, influence)
    assert result.status == "applied"
    assert set(result.reverted_decisions) == {"D1", "D5", "D6", "D7"}
    print("influence-only-ok")


def _combined_revert() -> None:
    """Revert traverses union of structural + influence edges."""
    svc = CorrectionService()
    req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
    tree = {"D1": ["D2"]}
    influence = {"D1": ["D3"], "D2": ["D4"]}
    result = svc.execute_revert(req.correction_id, tree, influence)
    assert result.status == "applied"
    assert set(result.reverted_decisions) == {"D1", "D2", "D3", "D4"}
    print("combined-revert-ok")


def _cycle_detection() -> None:
    """Corrupt cycle in influence DAG handled gracefully."""
    svc = CorrectionService()
    req = svc.request_correction(_PLAN_ID, "D1", CorrectionMode.REVERT)
    tree: dict[str, list[str]] = {}
    influence = {"D1": ["D2"], "D2": ["D3"], "D3": ["D1"]}
    result = svc.execute_revert(req.correction_id, tree, influence)
    assert result.status == "applied"
    reverted = set(result.reverted_decisions)
    assert reverted == {"D1", "D2", "D3"}, f"Got {reverted}"
    # No duplicates
    assert len(result.reverted_decisions) == len(set(result.reverted_decisions))
    print("cycle-detection-ok")


def _record_deps() -> None:
    """record_decision with dependency_decision_ids populates DAG."""
    svc = DecisionService()
    root = svc.record_decision(
        plan_id=_PLAN_ID,
        decision_type="strategy_choice",
        question="Strategy?",
        chosen_option="Approach A",
    )
    child = svc.record_decision(
        plan_id=_PLAN_ID,
        decision_type="implementation_choice",
        question="Impl?",
        chosen_option="Option B",
        parent_decision_id=root.decision_id,
        dependency_decision_ids=[root.decision_id],
    )
    edges = svc.get_influence_edges(_PLAN_ID)
    assert root.decision_id in edges
    assert child.decision_id in edges[root.decision_id]
    print("record-deps-ok")


def _e2e_cascade() -> None:
    """Full flow: record decisions with deps, then revert cascades."""
    # 1. Record decisions with influence dependencies
    dsvc = DecisionService()
    d1 = dsvc.record_decision(
        plan_id=_PLAN_ID,
        decision_type="strategy_choice",
        question="Strategy?",
        chosen_option="Approach A",
    )
    d2 = dsvc.record_decision(
        plan_id=_PLAN_ID,
        decision_type="resource_selection",
        question="Resource?",
        chosen_option="File X",
        parent_decision_id=d1.decision_id,
    )
    d3 = dsvc.record_decision(
        plan_id=_PLAN_ID,
        decision_type="implementation_choice",
        question="Impl?",
        chosen_option="Option B",
        parent_decision_id=d2.decision_id,
        dependency_decision_ids=[d1.decision_id, d2.decision_id],
    )

    # 2. Build structural tree
    tree = {
        d1.decision_id: [d2.decision_id],
        d2.decision_id: [d3.decision_id],
    }

    # 3. Get influence edges
    influence = dsvc.get_influence_edges(_PLAN_ID)

    # 4. Revert from d1 → should cascade to d2 and d3
    csvc = CorrectionService()
    req = csvc.request_correction(_PLAN_ID, d1.decision_id, CorrectionMode.REVERT)
    result = csvc.execute_revert(req.correction_id, tree, influence)
    assert result.status == "applied"
    reverted = set(result.reverted_decisions)
    assert d1.decision_id in reverted
    assert d2.decision_id in reverted
    assert d3.decision_id in reverted
    print("e2e-cascade-ok")


# ---------------------------------------------------------------------------
# Dispatch
# ---------------------------------------------------------------------------

_COMMANDS = {
    "structural-only": _structural_only,
    "influence-only": _influence_only,
    "combined-revert": _combined_revert,
    "cycle-detection": _cycle_detection,
    "record-deps": _record_deps,
    "e2e-cascade": _e2e_cascade,
}

if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr)
        sys.exit(1)
    _COMMANDS[sys.argv[1]]()