diff --git a/CHANGELOG.md b/CHANGELOG.md index 6fa75d9fb..b8d8162ae 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ### Added +- Wired Invariant Reconciliation Actor auto-invocation into + `PlanLifecycleService` phase transitions (`start_strategize`, + `execute_plan`, `apply_plan`). Reconciliation failures now block + the transition with `ReconciliationBlockedError` and emit + `INVARIANT_VIOLATED` events. Post-correction reconciliation runs + via `CORRECTION_APPLIED` event subscription (best-effort). Added + `InvariantService` Singleton provider in the DI container. - **TUI — Shell danger detection**: The TUI shell mode (`!` prefix) now detects dangerous command patterns before execution. A configurable pattern registry classifies commands by danger level (warning, critical) and surfaces a user diff --git a/features/invariant_reconciliation_autowire.feature b/features/invariant_reconciliation_autowire.feature new file mode 100644 index 000000000..20412fc2c --- /dev/null +++ b/features/invariant_reconciliation_autowire.feature @@ -0,0 +1,98 @@ +Feature: Invariant Reconciliation Actor Auto-Invocation + As a plan lifecycle manager + I want invariant reconciliation to run automatically at phase transitions + So that invariant violations are detected before they propagate + + Background: + Given a fresh PlanLifecycleService with invariant reconciliation wired + + # === Auto-invocation at Strategize start === + + @autowire @strategize + Scenario: Reconciliation runs at start_strategize when invariant_actor is set + Given a plan in Strategize/QUEUED with invariant_actor "builtin/invariant-reconciliation" + And a global invariant "All tests must pass" is registered + When I call start_strategize on the plan + Then the plan should transition to Strategize/PROCESSING + And invariant reconciliation should have been invoked + And an INVARIANT_RECONCILED event should have been emitted + + @autowire @strategize + Scenario: Reconciliation is skipped at start_strategize when invariant_actor is None + Given a plan in Strategize/QUEUED with invariant_actor unset + When I call start_strategize on the plan + Then the plan should transition to Strategize/PROCESSING + And invariant reconciliation should not have been invoked + + @autowire @strategize + Scenario: Reconciliation is skipped when invariant_actor is "__optional__" + Given a plan in Strategize/QUEUED with invariant_actor "__optional__" + When I call start_strategize on the plan + Then the plan should transition to Strategize/PROCESSING + And invariant reconciliation should not have been invoked + + # === Auto-invocation at Execute transition === + + @autowire @execute + Scenario: Reconciliation runs at execute_plan transition + Given a plan in Strategize/COMPLETE with invariant_actor "builtin/invariant-reconciliation" + And a global invariant "No unsafe operations" is registered + When I call execute_plan on the plan + Then the plan should transition to Execute/QUEUED + And invariant reconciliation should have been invoked + + # === Auto-invocation at Apply transition === + + @autowire @apply + Scenario: Reconciliation runs at apply_plan transition + Given a plan in Execute/COMPLETE with invariant_actor "builtin/invariant-reconciliation" + And a global invariant "Review before merge" is registered + When I call apply_plan on the plan + Then the plan should transition to Apply/QUEUED + And invariant reconciliation should have been invoked + + # === Transition blocking on failure === + + @autowire @blocking + Scenario: Reconciliation failure blocks start_strategize + Given a plan in Strategize/QUEUED with invariant_actor "builtin/invariant-reconciliation" + And the invariant service is configured to raise an error + When I attempt to call start_strategize on the plan + Then a ReconciliationBlockedError should be raised + And the plan should remain in Strategize/QUEUED + And an INVARIANT_VIOLATED event should have been emitted + + @autowire @blocking + Scenario: Reconciliation failure blocks execute_plan + Given a plan in Strategize/COMPLETE with invariant_actor "builtin/invariant-reconciliation" + And the invariant service is configured to raise an error + When I attempt to call execute_plan on the plan + Then a ReconciliationBlockedError should be raised + And the plan should remain in Strategize/COMPLETE + + @autowire @blocking + Scenario: Reconciliation failure blocks apply_plan + Given a plan in Execute/COMPLETE with invariant_actor "builtin/invariant-reconciliation" + And the invariant service is configured to raise an error + When I attempt to call apply_plan on the plan + Then a ReconciliationBlockedError should be raised + And the plan should remain in Execute/COMPLETE + + # === Decision recording === + + @autowire @decisions + Scenario: Reconciliation records invariant_enforced decisions in decision tree + Given a plan in Strategize/QUEUED with invariant_actor "builtin/invariant-reconciliation" + And a global invariant "Maintain backward compatibility" is registered + And a plan invariant "Use Python 3.12 only" is registered for the plan + When I call start_strategize on the plan without mocking + Then 2 invariant_enforced decisions should be recorded for the plan + + # === Post-correction reconciliation === + + @autowire @correction + Scenario: Reconciliation runs after correction is applied via event subscription + Given a plan in Strategize/PROCESSING with invariant_actor "builtin/invariant-reconciliation" + And a global invariant "All tests must pass" is registered + When a CORRECTION_APPLIED event is emitted for the plan + Then invariant reconciliation should have been invoked for the plan diff --git a/features/steps/invariant_reconciliation_autowire_steps.py b/features/steps/invariant_reconciliation_autowire_steps.py new file mode 100644 index 000000000..7185ec26b --- /dev/null +++ b/features/steps/invariant_reconciliation_autowire_steps.py @@ -0,0 +1,518 @@ +"""Step definitions for invariant_reconciliation_autowire.feature. + +Tests the auto-invocation of the Invariant Reconciliation Actor +during PlanLifecycleService phase transitions: start_strategize, +execute_plan, and apply_plan, as well as post-correction +reconciliation via event subscription. +""" + +from __future__ import annotations + +from unittest.mock import patch + +from behave import given, then, when # type: ignore[import-untyped] +from behave.runner import Context # type: ignore[import-untyped] + +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 + +# ================================================================ +# Mock Event Bus +# ================================================================ + + +class _MockEventBus: + """Minimal event bus that records emitted events and dispatches subscribers.""" + + def __init__(self) -> None: + self.events: list[DomainEvent] = [] + self._subscribers: dict[EventType, list[object]] = {} + + def emit(self, event: DomainEvent) -> None: + self.events.append(event) + handlers = self._subscribers.get(event.event_type, []) + for handler in handlers: + handler(event) # type: ignore[operator] + + def subscribe(self, event_type: EventType, handler: object) -> None: + self._subscribers.setdefault(event_type, []).append(handler) + + +# ================================================================ +# Helpers +# ================================================================ + + +def _create_service( + context: Context, + invariant_service: InvariantService | None = None, + event_bus: _MockEventBus | None = None, +) -> PlanLifecycleService: + """Create a PlanLifecycleService with invariant reconciliation wired.""" + Settings._instance = None + settings = Settings() + inv_svc = invariant_service or InvariantService() + dec_svc = DecisionService() + bus = event_bus or _MockEventBus() + service = PlanLifecycleService( + settings=settings, + decision_service=dec_svc, + event_bus=bus, + invariant_service=inv_svc, + ) + context.service = service + context.invariant_service = inv_svc + context.decision_service = dec_svc + context.event_bus = bus + context.error = None + context.reconciliation_invoked = False + return service + + +def _create_plan_at_phase( + context: Context, + phase: PlanPhase, + state: ProcessingState, + invariant_actor: str | None, +) -> str: + """Create a plan and advance it to the given phase/state.""" + svc: PlanLifecycleService = context.service + + # Create action + action = svc.create_action( + name="test-action-autowire", + description="Test action for reconciliation autowire", + definition_of_done="All tests pass", + strategy_actor="openai/gpt-4", + execution_actor="openai/gpt-4", + invariant_actor=invariant_actor, + ) + + # Use action to create plan (in Strategize/QUEUED) + action_name = str(action.namespaced_name) + plan = svc.use_action(action_name=action_name) + plan_id = plan.identity.plan_id + + # Advance plan to desired phase/state by directly setting fields + if phase == PlanPhase.STRATEGIZE and state == ProcessingState.QUEUED: + # Already there after use_action + pass + elif phase == PlanPhase.STRATEGIZE and state == ProcessingState.COMPLETE: + plan.processing_state = ProcessingState.PROCESSING + plan.processing_state = ProcessingState.COMPLETE + svc._commit_plan(plan) + elif phase == PlanPhase.STRATEGIZE and state == ProcessingState.PROCESSING: + plan.processing_state = ProcessingState.PROCESSING + svc._commit_plan(plan) + elif phase == PlanPhase.EXECUTE and state == ProcessingState.COMPLETE: + plan.processing_state = ProcessingState.COMPLETE + svc._commit_plan(plan) + # Transition to Execute + plan.processing_state = ProcessingState.QUEUED + plan.phase = PlanPhase.EXECUTE + svc._commit_plan(plan) + # Advance to complete + plan.processing_state = ProcessingState.PROCESSING + plan.processing_state = ProcessingState.COMPLETE + svc._commit_plan(plan) + + context.plan = svc.get_plan(plan_id) + context.plan_id = plan_id + return plan_id + + +# ================================================================ +# Background +# ================================================================ + + +@given("a fresh PlanLifecycleService with invariant reconciliation wired") +def step_fresh_service(context: Context) -> None: + """Create a PlanLifecycleService with invariant reconciliation configured.""" + _create_service(context) + + +# ================================================================ +# Plan creation at various phases +# ================================================================ + + +@given('a plan in Strategize/QUEUED with invariant_actor "{actor}"') +def step_plan_strategize_queued_with_actor(context: Context, actor: str) -> None: + """Create a plan in Strategize/QUEUED with the given invariant_actor.""" + _create_plan_at_phase(context, PlanPhase.STRATEGIZE, ProcessingState.QUEUED, actor) + + +@given("a plan in Strategize/QUEUED with invariant_actor unset") +def step_plan_strategize_queued_no_actor(context: Context) -> None: + """Create a plan in Strategize/QUEUED with no invariant_actor.""" + _create_plan_at_phase(context, PlanPhase.STRATEGIZE, ProcessingState.QUEUED, None) + + +@given('a plan in Strategize/COMPLETE with invariant_actor "{actor}"') +def step_plan_strategize_complete_with_actor(context: Context, actor: str) -> None: + """Create a plan in Strategize/COMPLETE.""" + _create_plan_at_phase( + context, PlanPhase.STRATEGIZE, ProcessingState.COMPLETE, actor + ) + + +@given('a plan in Execute/COMPLETE with invariant_actor "{actor}"') +def step_plan_execute_complete_with_actor(context: Context, actor: str) -> None: + """Create a plan in Execute/COMPLETE.""" + _create_plan_at_phase(context, PlanPhase.EXECUTE, ProcessingState.COMPLETE, actor) + + +@given('a plan in Strategize/PROCESSING with invariant_actor "{actor}"') +def step_plan_strategize_processing_with_actor(context: Context, actor: str) -> None: + """Create a plan in Strategize/PROCESSING.""" + _create_plan_at_phase( + context, PlanPhase.STRATEGIZE, ProcessingState.PROCESSING, actor + ) + + +# ================================================================ +# Invariant setup +# ================================================================ + + +@given('a global invariant "{text}" is registered') +def step_register_global_invariant(context: Context, text: str) -> None: + """Register a global invariant in the invariant service.""" + context.invariant_service.add_invariant( + text=text, + scope=InvariantScope.GLOBAL, + source_name="system", + ) + + +@given('a plan invariant "{text}" is registered for the plan') +def step_register_plan_invariant(context: Context, text: str) -> None: + """Register a plan-scoped invariant for the current plan.""" + context.invariant_service.add_invariant( + text=text, + scope=InvariantScope.PLAN, + source_name=context.plan_id, + ) + + +@given("the invariant service is configured to raise an error") +def step_invariant_service_raises(context: Context) -> None: + """Configure the invariant service to raise on list_invariants.""" + + def _raise_error(*args: object, **kwargs: object) -> list[object]: + raise RuntimeError("Simulated invariant service failure") + + context.invariant_service.list_invariants = _raise_error # type: ignore[assignment] + + +# ================================================================ +# Actions: successful calls +# ================================================================ + + +@when("I call start_strategize on the plan") +def step_call_start_strategize(context: Context) -> None: + """Call start_strategize and track reconciliation invocation.""" + svc: PlanLifecycleService = context.service + with patch( + "cleveragents.actor.reconciliation.InvariantReconciliationActor.run", + wraps=None, + ) as mock_run: + # Set up the mock to return a valid result + from cleveragents.actor.reconciliation import ReconciliationResult + from cleveragents.domain.models.core.invariant import InvariantSet + + mock_run.return_value = ReconciliationResult( + reconciled_set=InvariantSet(invariants=[]), + conflicts=[], + enforced_decision_ids=[], + ) + try: + context.plan = svc.start_strategize(context.plan_id) + except Exception as exc: + context.error = exc + context.reconciliation_invoked = mock_run.called + + +@when("I call start_strategize on the plan without mocking") +def step_call_start_strategize_real(context: Context) -> None: + """Call start_strategize with real reconciliation (no mocking).""" + svc: PlanLifecycleService = context.service + try: + context.plan = svc.start_strategize(context.plan_id) + except Exception as exc: + context.error = exc + context.reconciliation_invoked = True + + +@when("I call execute_plan on the plan") +def step_call_execute_plan(context: Context) -> None: + """Call execute_plan and track reconciliation invocation.""" + svc: PlanLifecycleService = context.service + with patch( + "cleveragents.actor.reconciliation.InvariantReconciliationActor.run", + wraps=None, + ) as mock_run: + from cleveragents.actor.reconciliation import ReconciliationResult + from cleveragents.domain.models.core.invariant import InvariantSet + + mock_run.return_value = ReconciliationResult( + reconciled_set=InvariantSet(invariants=[]), + conflicts=[], + enforced_decision_ids=[], + ) + try: + context.plan = svc.execute_plan(context.plan_id) + except Exception as exc: + context.error = exc + context.reconciliation_invoked = mock_run.called + + +@when("I call apply_plan on the plan") +def step_call_apply_plan(context: Context) -> None: + """Call apply_plan and track reconciliation invocation.""" + svc: PlanLifecycleService = context.service + with patch( + "cleveragents.actor.reconciliation.InvariantReconciliationActor.run", + wraps=None, + ) as mock_run: + from cleveragents.actor.reconciliation import ReconciliationResult + from cleveragents.domain.models.core.invariant import InvariantSet + + mock_run.return_value = ReconciliationResult( + reconciled_set=InvariantSet(invariants=[]), + conflicts=[], + enforced_decision_ids=[], + ) + try: + context.plan = svc.apply_plan(context.plan_id) + except Exception as exc: + context.error = exc + context.reconciliation_invoked = mock_run.called + + +# ================================================================ +# Actions: failure scenarios +# ================================================================ + + +@when("I attempt to call start_strategize on the plan") +def step_attempt_start_strategize(context: Context) -> None: + """Call start_strategize expecting it to fail.""" + svc: PlanLifecycleService = context.service + try: + context.plan = svc.start_strategize(context.plan_id) + except Exception as exc: + context.error = exc + + +@when("I attempt to call execute_plan on the plan") +def step_attempt_execute_plan(context: Context) -> None: + """Call execute_plan expecting it to fail.""" + svc: PlanLifecycleService = context.service + try: + context.plan = svc.execute_plan(context.plan_id) + except Exception as exc: + context.error = exc + + +@when("I attempt to call apply_plan on the plan") +def step_attempt_apply_plan(context: Context) -> None: + """Call apply_plan expecting it to fail.""" + svc: PlanLifecycleService = context.service + try: + context.plan = svc.apply_plan(context.plan_id) + except Exception as exc: + context.error = exc + + +# ================================================================ +# Post-correction event +# ================================================================ + + +@when("a CORRECTION_APPLIED event is emitted for the plan") +def step_emit_correction_applied(context: Context) -> None: + """Simulate a CORRECTION_APPLIED event to trigger reconciliation.""" + event = DomainEvent( + event_type=EventType.CORRECTION_APPLIED, + plan_id=context.plan_id, + details={ + "correction_id": "test-correction-id", + "target_decision_id": "test-decision-id", + "mode": "revert", + "guidance": "test guidance", + }, + ) + context.event_bus.emit(event) + + +# ================================================================ +# Assertions: plan state +# ================================================================ + + +@then("the plan should transition to Strategize/PROCESSING") +def step_plan_strategize_processing(context: Context) -> None: + """Verify plan is in Strategize/PROCESSING.""" + plan = context.service.get_plan(context.plan_id) + assert plan.phase == PlanPhase.STRATEGIZE, f"Expected STRATEGIZE, got {plan.phase}" + assert plan.state == ProcessingState.PROCESSING, ( + f"Expected PROCESSING, got {plan.state}" + ) + + +@then("the plan should transition to Execute/QUEUED") +def step_plan_execute_queued(context: Context) -> None: + """Verify plan is in Execute/QUEUED.""" + plan = context.service.get_plan(context.plan_id) + assert plan.phase == PlanPhase.EXECUTE, f"Expected EXECUTE, got {plan.phase}" + assert plan.state == ProcessingState.QUEUED, f"Expected QUEUED, got {plan.state}" + + +@then("the plan should transition to Apply/QUEUED") +def step_plan_apply_queued(context: Context) -> None: + """Verify plan is in Apply/QUEUED.""" + plan = context.service.get_plan(context.plan_id) + assert plan.phase == PlanPhase.APPLY, f"Expected APPLY, got {plan.phase}" + assert plan.state == ProcessingState.QUEUED, f"Expected QUEUED, got {plan.state}" + + +@then("the plan should remain in Strategize/QUEUED") +def step_plan_remain_strategize_queued(context: Context) -> None: + """Verify plan is still in Strategize/QUEUED (transition blocked).""" + plan = context.service.get_plan(context.plan_id) + assert plan.phase == PlanPhase.STRATEGIZE, f"Expected STRATEGIZE, got {plan.phase}" + assert plan.state == ProcessingState.QUEUED, f"Expected QUEUED, got {plan.state}" + + +@then("the plan should remain in Strategize/COMPLETE") +def step_plan_remain_strategize_complete(context: Context) -> None: + """Verify plan is still in Strategize/COMPLETE (transition blocked).""" + plan = context.service.get_plan(context.plan_id) + assert plan.phase == PlanPhase.STRATEGIZE, f"Expected STRATEGIZE, got {plan.phase}" + assert plan.state == ProcessingState.COMPLETE, ( + f"Expected COMPLETE, got {plan.state}" + ) + + +@then("the plan should remain in Execute/COMPLETE") +def step_plan_remain_execute_complete(context: Context) -> None: + """Verify plan is still in Execute/COMPLETE (transition blocked).""" + plan = context.service.get_plan(context.plan_id) + assert plan.phase == PlanPhase.EXECUTE, f"Expected EXECUTE, got {plan.phase}" + assert plan.state == ProcessingState.COMPLETE, ( + f"Expected COMPLETE, got {plan.state}" + ) + + +# ================================================================ +# Assertions: reconciliation invocation +# ================================================================ + + +@then("invariant reconciliation should have been invoked") +def step_reconciliation_invoked(context: Context) -> None: + """Verify that the reconciliation actor was invoked.""" + assert context.reconciliation_invoked, "Expected reconciliation to be invoked" + + +@then("invariant reconciliation should not have been invoked") +def step_reconciliation_not_invoked(context: Context) -> None: + """Verify that the reconciliation actor was NOT invoked.""" + assert not context.reconciliation_invoked, ( + "Expected reconciliation NOT to be invoked" + ) + + +@then("invariant reconciliation should have been invoked for the plan") +def step_reconciliation_invoked_for_plan(context: Context) -> None: + """Verify reconciliation ran for the plan (via event handler).""" + # Check that INVARIANT_RECONCILED or INVARIANT_VIOLATED event was emitted + # after the CORRECTION_APPLIED event + events = context.event_bus.events + reconciled_events = [ + e + for e in events + if e.event_type + in (EventType.INVARIANT_RECONCILED, EventType.INVARIANT_VIOLATED) + and e.plan_id == context.plan_id + ] + assert len(reconciled_events) > 0, ( + "Expected INVARIANT_RECONCILED or INVARIANT_VIOLATED event after correction" + ) + + +# ================================================================ +# Assertions: events +# ================================================================ + + +@then("an INVARIANT_RECONCILED event should have been emitted") +def step_invariant_reconciled_event(context: Context) -> None: + """Verify an INVARIANT_RECONCILED event was emitted.""" + events = context.event_bus.events + reconciled = [e for e in events if e.event_type == EventType.INVARIANT_RECONCILED] + assert len(reconciled) >= 1, ( + f"Expected INVARIANT_RECONCILED event, got events: " + f"{[e.event_type for e in events]}" + ) + + +@then("an INVARIANT_VIOLATED event should have been emitted") +def step_invariant_violated_event(context: Context) -> None: + """Verify an INVARIANT_VIOLATED event was emitted.""" + events = context.event_bus.events + violated = [e for e in events if e.event_type == EventType.INVARIANT_VIOLATED] + assert len(violated) >= 1, ( + f"Expected INVARIANT_VIOLATED event, got events: " + f"{[e.event_type for e in events]}" + ) + + +# ================================================================ +# Assertions: errors +# ================================================================ + + +@then("a ReconciliationBlockedError should be raised") +def step_reconciliation_blocked_error(context: Context) -> None: + """Verify that a ReconciliationBlockedError was raised.""" + assert context.error is not None, "Expected an error to be raised" + assert isinstance(context.error, ReconciliationBlockedError), ( + f"Expected ReconciliationBlockedError, got {type(context.error).__name__}: " + f"{context.error}" + ) + + +# ================================================================ +# Assertions: decisions +# ================================================================ + + +@then("{count:d} invariant_enforced decisions should be recorded for the plan") +def step_invariant_enforced_decisions_count(context: Context, count: int) -> None: + """Verify the number of invariant_enforced decisions recorded.""" + from cleveragents.domain.models.core.decision import DecisionType + + dec_svc: DecisionService = context.decision_service + decisions = dec_svc.list_decisions(plan_id=context.plan_id) + enforced = [ + d for d in decisions if d.decision_type == DecisionType.INVARIANT_ENFORCED + ] + assert len(enforced) == count, ( + f"Expected {count} invariant_enforced decisions, got {len(enforced)}" + ) diff --git a/robot/helper_invariant_reconciliation_autowire.py b/robot/helper_invariant_reconciliation_autowire.py new file mode 100644 index 000000000..3179b695b --- /dev/null +++ b/robot/helper_invariant_reconciliation_autowire.py @@ -0,0 +1,197 @@ +"""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() diff --git a/robot/invariant_reconciliation_autowire.robot b/robot/invariant_reconciliation_autowire.robot new file mode 100644 index 000000000..5174cdf86 --- /dev/null +++ b/robot/invariant_reconciliation_autowire.robot @@ -0,0 +1,56 @@ +*** Settings *** +Documentation Integration tests for Invariant Reconciliation Actor auto-invocation +... Verifies that the reconciliation actor is automatically invoked during +... plan lifecycle phase transitions and blocks transitions on failure. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_invariant_reconciliation_autowire.py + +*** Test Cases *** +Reconciliation Auto-Invocation At Strategize Start + [Documentation] Reconciliation runs automatically when start_strategize is called + ${result}= Run Process ${PYTHON} ${HELPER} strategize cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} autowire-strategize-ok + Should Contain ${result.stdout} reconciliation_invoked + +Reconciliation Auto-Invocation At Execute Transition + [Documentation] Reconciliation runs automatically at execute_plan transition + ${result}= Run Process ${PYTHON} ${HELPER} execute cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} autowire-execute-ok + Should Contain ${result.stdout} reconciliation_invoked + +Reconciliation Auto-Invocation At Apply Transition + [Documentation] Reconciliation runs automatically at apply_plan transition + ${result}= Run Process ${PYTHON} ${HELPER} apply cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} autowire-apply-ok + Should Contain ${result.stdout} reconciliation_invoked + +Reconciliation Blocks Transition On Failure + [Documentation] When reconciliation fails, the phase transition is blocked + ${result}= Run Process ${PYTHON} ${HELPER} block cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} autowire-block-ok + Should Contain ${result.stdout} ReconciliationBlockedError + +Reconciliation Skipped When Actor Unset + [Documentation] No reconciliation when invariant_actor is None + ${result}= Run Process ${PYTHON} ${HELPER} skip cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} autowire-skip-ok + Should Contain ${result.stdout} reconciliation_skipped diff --git a/src/cleveragents/application/container.py b/src/cleveragents/application/container.py index 9c99d39a6..cf7a17523 100644 --- a/src/cleveragents/application/container.py +++ b/src/cleveragents/application/container.py @@ -54,6 +54,7 @@ from cleveragents.application.services.faiss_vector_backend import ( from cleveragents.application.services.fix_then_revalidate import ( FixThenRevalidateOrchestrator, ) +from cleveragents.application.services.invariant_service import InvariantService from cleveragents.application.services.multi_project_service import ( MultiProjectService, ) @@ -619,6 +620,11 @@ class Container(containers.DeclarativeContainer): event_bus=event_bus, ) + # Invariant Service - Singleton (in-memory invariant management) + invariant_service = providers.Singleton( + InvariantService, + ) + # Plan Lifecycle Service - Factory (v3 four-phase lifecycle) plan_lifecycle_service = providers.Factory( PlanLifecycleService, @@ -626,6 +632,7 @@ class Container(containers.DeclarativeContainer): unit_of_work=unit_of_work, decision_service=decision_service, event_bus=event_bus, + invariant_service=invariant_service, ) # Checkpoint Service - database-backed via CheckpointRepository diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 111248b6b..f63a4215d 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -100,6 +100,7 @@ if TYPE_CHECKING: from cleveragents.application.services.error_pattern_service import ( ErrorPatternService, ) + from cleveragents.application.services.invariant_service import InvariantService from cleveragents.config.settings import Settings from cleveragents.infrastructure.database.unit_of_work import UnitOfWork from cleveragents.infrastructure.events.protocol import EventBus @@ -151,6 +152,23 @@ class PlanNotReadyError(BusinessRuleViolation): self.state = state +class ReconciliationBlockedError(BusinessRuleViolation): + """Raised when invariant reconciliation fails, blocking a transition. + + The Invariant Reconciliation Actor encountered an error during + execution, preventing the plan from proceeding to the next phase. + """ + + def __init__(self, plan_id: str, phase: PlanPhase, reason: str): + super().__init__( + f"Invariant reconciliation failed for plan {plan_id} " + f"during {phase.value} phase: {reason}" + ) + self.plan_id = plan_id + self.phase = phase + self.reason = reason + + class PlanLifecycleService: """Service for managing v3 plan lifecycle. @@ -171,6 +189,7 @@ class PlanLifecycleService: job_store: InMemoryJobStore | None = None, error_pattern_service: ErrorPatternService | None = None, config_service: ConfigService | None = None, + invariant_service: InvariantService | None = None, ): """Initialize the plan lifecycle service. @@ -199,6 +218,13 @@ class PlanLifecycleService: consulted before the Execute phase to inject preventive guidance. When ``None``, predictive prevention is silently skipped. + invariant_service: Optional + :class:`InvariantService` for invariant reconciliation + during phase transitions. When provided and the plan + has an ``invariant_actor`` configured, the Invariant + Reconciliation Actor is auto-invoked at each phase + transition to verify invariants hold. When ``None``, + reconciliation is silently skipped. """ self.settings = settings self.unit_of_work = unit_of_work @@ -207,8 +233,10 @@ class PlanLifecycleService: self._job_store = job_store self.error_pattern_service = error_pattern_service self._config_service = config_service + self.invariant_service = invariant_service self._logger = logger.bind(service="plan_lifecycle") self.preflight_guardrail = PlanPreflightGuardrail() + self._subscribe_correction_reconciliation() # In-memory fallback storage (used only when no UoW is provided) self._actions: dict[str, Action] = {} @@ -357,6 +385,189 @@ class PlanLifecycleService: exc_info=True, ) + def _run_invariant_reconciliation(self, plan: Plan) -> None: + """Run the Invariant Reconciliation Actor if configured. + + Invoked at phase transitions (start of Strategize, before + Execute, before Apply, and after corrections) to verify that + plan invariants hold. + + Unlike ``_run_estimation`` which is informational-only, a + reconciliation failure **blocks** the phase transition and + raises ``ReconciliationBlockedError``. This ensures invariant + violations are surfaced immediately. + + The reconciliation actor is skipped when: + - ``invariant_service`` is ``None`` + - ``decision_service`` is ``None`` + - ``plan.invariant_actor`` is ``None`` or ``"__optional__"`` + + Args: + plan: The plan to reconcile invariants for. + + Raises: + ReconciliationBlockedError: If reconciliation encounters an + error, blocking the phase transition. + """ + # Guard: skip when services are not wired or actor is disabled + if self.invariant_service is None or self.decision_service is None: + return + + actor_name = plan.invariant_actor + if not actor_name or actor_name == "__optional__": + return + + plan_id = plan.identity.plan_id + + try: + from cleveragents.actor.reconciliation import ( + InvariantReconciliationActor, + ReconciliationResult, + ) + + reconciliation_actor = InvariantReconciliationActor( + invariant_service=self.invariant_service, + decision_service=self.decision_service, + ) + + # Derive project and action names from the plan + project_name: str | None = None + if plan.project_links: + project_name = plan.project_links[0].project_name + + result: ReconciliationResult = reconciliation_actor.run( + plan_id=plan_id, + project_name=project_name, + action_name=plan.action_name, + parent_decision_id=plan.decision_root_id, + ) + + self._logger.info( + "invariant_reconciliation_complete", + plan_id=plan_id, + phase=plan.phase.value, + effective_count=len(result.reconciled_set.invariants), + conflict_count=len(result.conflicts), + decision_count=len(result.enforced_decision_ids), + ) + + # Emit INVARIANT_RECONCILED event on success + if self.event_bus is not None: + try: + self.event_bus.emit( + DomainEvent( + event_type=EventType.INVARIANT_RECONCILED, + plan_id=plan_id, + actor_name=actor_name, + project_name=project_name, + details={ + "phase": plan.phase.value, + "effective_count": len( + result.reconciled_set.invariants + ), + "conflict_count": len(result.conflicts), + "enforced_decision_ids": (result.enforced_decision_ids), + }, + ) + ) + except Exception: + self._logger.warning( + "event_bus_emit_failed", + event_type="INVARIANT_RECONCILED", + plan_id=plan_id, + exc_info=True, + ) + + except Exception as exc: + self._logger.error( + "invariant_reconciliation_failed", + plan_id=plan_id, + phase=plan.phase.value, + invariant_actor=actor_name, + exc_info=True, + ) + + # Emit INVARIANT_VIOLATED event on failure + if self.event_bus is not None: + try: + self.event_bus.emit( + DomainEvent( + event_type=EventType.INVARIANT_VIOLATED, + plan_id=plan_id, + actor_name=actor_name, + details={ + "phase": plan.phase.value, + "error": str(exc), + }, + ) + ) + except Exception: + self._logger.warning( + "event_bus_emit_failed", + event_type="INVARIANT_VIOLATED", + plan_id=plan_id, + exc_info=True, + ) + + raise ReconciliationBlockedError( + plan_id=plan_id, + phase=plan.phase, + reason=str(exc), + ) from exc + + def _subscribe_correction_reconciliation(self) -> None: + """Subscribe to CORRECTION_APPLIED events for post-correction reconciliation. + + When a correction is successfully applied, invariant + reconciliation is re-run on the affected plan to verify that + the correction did not break any invariants. + + Post-correction reconciliation is best-effort: failures are + logged but do not reverse the correction. + """ + if self.event_bus is None: + return + + try: + self.event_bus.subscribe( + EventType.CORRECTION_APPLIED, + self._handle_correction_applied, + ) + except Exception: + self._logger.warning( + "correction_reconciliation_subscription_failed", + exc_info=True, + ) + + def _handle_correction_applied(self, event: DomainEvent) -> None: + """Handle a CORRECTION_APPLIED event by re-running reconciliation. + + This is a best-effort operation — failures are logged but + never re-raised, since the correction has already been applied + and cannot be undone from this handler. + """ + plan_id = event.plan_id + if not plan_id: + return + + try: + plan = self.get_plan(plan_id) + # Run reconciliation; suppress ReconciliationBlockedError + # since we cannot undo the correction from here. + self._run_invariant_reconciliation(plan) + except ReconciliationBlockedError: + self._logger.warning( + "post_correction_reconciliation_blocked", + plan_id=plan_id, + exc_info=True, + ) + except Exception: + self._logger.warning( + "post_correction_reconciliation_failed", + plan_id=plan_id, + exc_info=True, + ) + def _persist_action_create(self, action: Action, ctx: Any) -> None: """Persist a new action via the repository. @@ -1163,6 +1374,12 @@ class PlanLifecycleService: ) # -- End pre-flight ----------------------------------------------- + # -- Invariant Reconciliation (blocks on failure) ----------------- + # Spec: reconciliation runs at the start of Strategize to + # verify plan invariants hold before processing begins. + self._run_invariant_reconciliation(plan) + # -- End reconciliation ------------------------------------------- + plan.processing_state = ProcessingState.PROCESSING plan.timestamps.strategize_started_at = datetime.now() plan.timestamps.updated_at = datetime.now() @@ -1309,6 +1526,9 @@ class PlanLifecycleService: # Layer 4: Consult Error Pattern Database for preventive guidance self._consult_error_patterns(plan) + # Invariant Reconciliation: verify invariants before Execute + self._run_invariant_reconciliation(plan) + # Transition to Execute phase — set processing_state first so that # the phase-state validator sees QUEUED (valid in any phase) when # the phase assignment triggers re-validation. @@ -1484,6 +1704,9 @@ class PlanLifecycleService: plan_id, plan.phase, plan.state or ProcessingState.QUEUED ) + # Invariant Reconciliation: verify invariants before Apply + self._run_invariant_reconciliation(plan) + # Transition to Apply phase — set processing_state first so that # the phase-state validator sees QUEUED (valid in any phase) when # the phase assignment triggers re-validation.