cleveragents-core/robot/helper_decision_model.py

"""Helper script for Robot Framework decision model smoke tests."""

from __future__ import annotations

import sys
from pathlib import Path

# Ensure src is importable when run from workspace root
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

from ulid import ULID

from cleveragents.domain.models.core.decision import (
    ArtifactRef,
    ContextSnapshot,
    Decision,
    DecisionType,
    ResourceRef,
)


def _test_create_root() -> None:
    """Create a root prompt_definition decision."""
    d = Decision(
        plan_id=str(ULID()),
        sequence_number=0,
        decision_type=DecisionType.PROMPT_DEFINITION,
        question="What should we build?",
        chosen_option="A REST API for user management",
    )
    assert d.is_root
    assert d.decision_type == DecisionType.PROMPT_DEFINITION
    assert d.parent_decision_id is None
    print("decision-create-root-ok")


def _test_create_child() -> None:
    """Create a strategy_choice child decision."""
    parent_id = str(ULID())
    d = Decision(
        plan_id=str(ULID()),
        parent_decision_id=parent_id,
        sequence_number=1,
        decision_type=DecisionType.STRATEGY_CHOICE,
        question="Which framework?",
        chosen_option="FastAPI",
        alternatives_considered=["Flask", "Django"],
        confidence_score=0.9,
    )
    assert not d.is_root
    assert d.parent_decision_id == parent_id
    assert d.confidence_score == 0.9
    assert len(d.alternatives_considered) == 2
    print("decision-create-child-ok")


def _test_enum_values() -> None:
    """Verify all 11 decision type enum values."""
    expected = {
        "prompt_definition",
        "invariant_enforced",
        "strategy_choice",
        "implementation_choice",
        "resource_selection",
        "subplan_spawn",
        "subplan_parallel_spawn",
        "tool_invocation",
        "error_recovery",
        "validation_response",
        "user_intervention",
    }
    actual = {dt.value for dt in DecisionType}
    assert actual == expected, f"Mismatch: {actual.symmetric_difference(expected)}"
    assert len(DecisionType) == 11
    print("decision-enums-ok")


def _test_context_snapshot() -> None:
    """Create a decision with a populated context snapshot."""
    snap = ContextSnapshot(
        hot_context_hash="sha256:abc123def",
        hot_context_ref="store://snapshots/abc123def",
        relevant_resources=[
            ResourceRef(resource_id=str(ULID()), path="src/main.py"),
            ResourceRef(resource_id=str(ULID())),
        ],
        actor_state_ref="checkpoint://actor/001",
    )
    d = Decision(
        plan_id=str(ULID()),
        sequence_number=0,
        decision_type=DecisionType.PROMPT_DEFINITION,
        question="What approach?",
        chosen_option="Microservices",
        context_snapshot=snap,
    )
    assert d.context_snapshot.hot_context_hash == "sha256:abc123def"
    assert len(d.context_snapshot.relevant_resources) == 2
    assert d.context_snapshot.actor_state_ref == "checkpoint://actor/001"
    print("decision-snapshot-ok")


def _test_correction() -> None:
    """Create a correction decision and verify metadata."""
    original_id = str(ULID())
    d = Decision(
        plan_id=str(ULID()),
        parent_decision_id=str(ULID()),
        sequence_number=3,
        decision_type=DecisionType.STRATEGY_CHOICE,
        question="Which framework?",
        chosen_option="Django instead",
        is_correction=True,
        corrects_decision_id=original_id,
        correction_reason="FastAPI lacks admin panel",
    )
    assert d.is_correction
    assert d.corrects_decision_id == original_id
    assert d.correction_reason == "FastAPI lacks admin panel"
    assert not d.is_superseded
    print("decision-correction-ok")


def _test_roundtrip() -> None:
    """Verify model_dump / model_validate round-trip."""
    d = Decision(
        plan_id=str(ULID()),
        parent_decision_id=str(ULID()),
        sequence_number=5,
        decision_type=DecisionType.IMPLEMENTATION_CHOICE,
        question="How to implement auth?",
        chosen_option="JWT tokens",
        confidence_score=0.85,
        context_snapshot=ContextSnapshot(
            hot_context_hash="sha256:roundtrip",
            relevant_resources=[ResourceRef(resource_id=str(ULID()))],
        ),
        artifacts_produced=[
            ArtifactRef(artifact_path="src/auth.py", artifact_type="file"),
        ],
    )
    data = d.model_dump()
    restored = Decision.model_validate(data)
    assert restored.decision_id == d.decision_id
    assert restored.decision_type == d.decision_type
    assert restored.confidence_score == d.confidence_score
    assert restored.context_snapshot.hot_context_hash == "sha256:roundtrip"
    assert len(restored.artifacts_produced) == 1
    print("decision-roundtrip-ok")


def _test_cli_dict() -> None:
    """Verify as_cli_dict produces expected keys."""
    d = Decision(
        plan_id=str(ULID()),
        sequence_number=0,
        decision_type=DecisionType.PROMPT_DEFINITION,
        question="Build what?",
        chosen_option="API",
    )
    cli = d.as_cli_dict()
    required_keys = {
        "decision_id",
        "plan_id",
        "type",
        "sequence",
        "question",
        "chosen",
        "confidence",
        "parent",
        "is_correction",
        "superseded",
    }
    assert required_keys.issubset(cli.keys()), (
        f"Missing keys: {required_keys - cli.keys()}"
    )
    print("decision-cli-dict-ok")


_TESTS = {
    "create_root": _test_create_root,
    "create_child": _test_create_child,
    "enum_values": _test_enum_values,
    "context_snapshot": _test_context_snapshot,
    "correction": _test_correction,
    "roundtrip": _test_roundtrip,
    "cli_dict": _test_cli_dict,
}

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]} <test_name>")
        print(f"Available tests: {', '.join(sorted(_TESTS))}")
        sys.exit(1)

    test_name = sys.argv[1]
    if test_name not in _TESTS:
        print(f"Unknown test: {test_name}")
        print(f"Available: {', '.join(sorted(_TESTS))}")
        sys.exit(1)

    _TESTS[test_name]()