Merge pull request 'feat(plan): wire invariant reconciliation actor auto-invocation' (#1205) from feature/invariant-actor-autowire into master

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

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

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)}"
+    )

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()

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

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

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.