feat(plan): wire invariant reconciliation actor auto-invocation #1205

Merged
freemo merged 1 commits from feature/invariant-actor-autowire into master 2026-04-05 06:08:57 +00:00
7 changed files with 1106 additions and 0 deletions
+7
View File
@@ -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
@@ -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
@@ -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)}"
)
@@ -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()
@@ -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
@@ -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
@@ -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.