cleveragents-core/robot/helper_invariant_reconciliation_autowire.py

"""Helper for invariant_reconciliation_autowire.robot.

Each function creates a PlanLifecycleService with invariant
reconciliation wired, exercises one scenario, and prints a
deterministic JSON result for the Robot test assertions.
"""

from __future__ import annotations

import json
import sys

from cleveragents.application.services.decision_service import DecisionService
from cleveragents.application.services.invariant_service import InvariantService
from cleveragents.application.services.plan_lifecycle_service import (
    PlanLifecycleService,
    ReconciliationBlockedError,
)
from cleveragents.config.settings import Settings
from cleveragents.domain.models.core.invariant import InvariantScope
from cleveragents.domain.models.core.plan import (
    PlanPhase,
    ProcessingState,
)
from cleveragents.infrastructure.events.models import DomainEvent
from cleveragents.infrastructure.events.types import EventType


class _MockEventBus:
    """Minimal event bus for integration tests."""

    def __init__(self) -> None:
        self.events: list[DomainEvent] = []
        self._subscribers: dict[EventType, list[object]] = {}

    def emit(self, event: DomainEvent) -> None:
        self.events.append(event)
        for handler in self._subscribers.get(event.event_type, []):
            handler(event)  # type: ignore[operator]

    def subscribe(self, event_type: EventType, handler: object) -> None:
        self._subscribers.setdefault(event_type, []).append(handler)


def _create_service(
    invariant_actor: str | None = "builtin/invariant-reconciliation",
) -> tuple[PlanLifecycleService, InvariantService, _MockEventBus, str]:
    """Create a wired PlanLifecycleService and return (svc, inv_svc, bus, plan_id)."""
    Settings._instance = None
    settings = Settings()
    inv_svc = InvariantService()
    dec_svc = DecisionService()
    bus = _MockEventBus()
    svc = PlanLifecycleService(
        settings=settings,
        decision_service=dec_svc,
        event_bus=bus,
        invariant_service=inv_svc,
    )

    # Create action and plan
    action = svc.create_action(
        name="test-autowire-robot",
        description="Robot integration test action",
        definition_of_done="All checks pass",
        strategy_actor="openai/gpt-4",
        execution_actor="openai/gpt-4",
        invariant_actor=invariant_actor,
    )
    plan = svc.use_action(action_name=str(action.namespaced_name))
    plan_id = plan.identity.plan_id
    return svc, inv_svc, bus, plan_id


def _has_event(bus: _MockEventBus, event_type: EventType) -> bool:
    return any(e.event_type == event_type for e in bus.events)


def test_strategize() -> None:
    """Test reconciliation at start_strategize."""
    svc, inv_svc, bus, plan_id = _create_service()
    inv_svc.add_invariant("All tests must pass", InvariantScope.GLOBAL, "system")

    svc.start_strategize(plan_id)
    plan = svc.get_plan(plan_id)

    reconciled = _has_event(bus, EventType.INVARIANT_RECONCILED)
    result = {
        "phase": plan.phase.value,
        "state": plan.state.value if plan.state else "None",
        "reconciliation_invoked": reconciled,
    }
    print(f"autowire-strategize-ok {json.dumps(result)}")


def test_execute() -> None:
    """Test reconciliation at execute_plan transition."""
    svc, inv_svc, bus, plan_id = _create_service()
    inv_svc.add_invariant("No unsafe ops", InvariantScope.GLOBAL, "system")

    # Advance to Strategize/COMPLETE
    plan = svc.get_plan(plan_id)
    plan.processing_state = ProcessingState.PROCESSING
    plan.processing_state = ProcessingState.COMPLETE
    svc.commit_plan(plan)

    svc.execute_plan(plan_id)
    plan = svc.get_plan(plan_id)

    reconciled = _has_event(bus, EventType.INVARIANT_RECONCILED)
    result = {
        "phase": plan.phase.value,
        "state": plan.state.value if plan.state else "None",
        "reconciliation_invoked": reconciled,
    }
    print(f"autowire-execute-ok {json.dumps(result)}")


def test_apply() -> None:
    """Test reconciliation at apply_plan transition."""
    svc, inv_svc, bus, plan_id = _create_service()
    inv_svc.add_invariant("Review first", InvariantScope.GLOBAL, "system")

    # Advance to Execute/COMPLETE
    plan = svc.get_plan(plan_id)
    plan.processing_state = ProcessingState.COMPLETE
    svc.commit_plan(plan)
    plan.processing_state = ProcessingState.QUEUED
    plan.phase = PlanPhase.EXECUTE
    svc.commit_plan(plan)
    plan.processing_state = ProcessingState.PROCESSING
    plan.processing_state = ProcessingState.COMPLETE
    svc.commit_plan(plan)

    svc.apply_plan(plan_id)
    plan = svc.get_plan(plan_id)

    reconciled = _has_event(bus, EventType.INVARIANT_RECONCILED)
    result = {
        "phase": plan.phase.value,
        "state": plan.state.value if plan.state else "None",
        "reconciliation_invoked": reconciled,
    }
    print(f"autowire-apply-ok {json.dumps(result)}")


def test_block() -> None:
    """Test that reconciliation failure blocks transition."""
    svc, inv_svc, _bus, plan_id = _create_service()

    # Make invariant service raise
    def _raise(*args: object, **kwargs: object) -> list[object]:
        raise RuntimeError("Simulated failure")

    inv_svc.list_invariants = _raise  # type: ignore[assignment]

    try:
        svc.start_strategize(plan_id)
        print("autowire-block-FAILED: no error raised")
    except ReconciliationBlockedError:
        plan = svc.get_plan(plan_id)
        result = {
            "error": "ReconciliationBlockedError",
            "phase": plan.phase.value,
            "state": plan.state.value if plan.state else "None",
        }
        print(f"autowire-block-ok {json.dumps(result)}")


def test_skip() -> None:
    """Test that reconciliation is skipped when actor is unset."""
    svc, inv_svc, bus, plan_id = _create_service(invariant_actor=None)
    inv_svc.add_invariant("Test invariant", InvariantScope.GLOBAL, "system")

    svc.start_strategize(plan_id)
    plan = svc.get_plan(plan_id)

    reconciled = _has_event(bus, EventType.INVARIANT_RECONCILED)
    result = {
        "phase": plan.phase.value,
        "state": plan.state.value if plan.state else "None",
        "reconciliation_skipped": not reconciled,
    }
    print(f"autowire-skip-ok {json.dumps(result)}")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "strategize"
    dispatch = {
        "strategize": test_strategize,
        "execute": test_execute,
        "apply": test_apply,
        "block": test_block,
        "skip": test_skip,
    }
    fn = dispatch.get(cmd, test_strategize)
    fn()