cleveragents-core/docs/modules/invariant-reconciliation.md

# Invariant Reconciliation Module

**Package:** `cleveragents.actor.reconciliation`, `cleveragents.application.services.invariant_service`
**Introduced:** v3.8.0 (the actor was originally prototyped in M3/v3.2.0; automatic
phase transition invocation shipped in M8/v3.7.0, issue #1941)

The invariant reconciliation module enforces user-defined constraints
(*invariants*) at every plan phase transition. Invariants are declarative rules
that must hold throughout plan execution — for example, "all output files must
be UTF-8 encoded" or "never modify files outside the project directory". The
`InvariantReconciliationActor` merges invariants from every scope, resolves
conflicts deterministically, and blocks plan execution whenever a violation is
detected.

For further background, see:
- [`docs/reference/invariants.md`](../reference/invariants.md) — data model and scope hierarchy
- [ADR-016 — Invariant System](../adr/ADR-016-invariant-system.md) — design rationale
- [Core API Reference — Invariant Service](../api/core.md#invariant-service) — API surface
- [Architecture — Plan Lifecycle](../architecture.md#plan-lifecycle)

---

## Purpose

Invariants provide a safety net that prevents plans from violating
organisation-defined constraints. The reconciliation actor runs at the start of
each Strategize, Execute, and Apply phase to produce the effective invariant
set. If any invariant is violated, the phase transition is **blocked** with a
`ReconciliationBlockedError`, an `INVARIANT_VIOLATED` event is emitted, and the
plan remains in its current phase until the violation is resolved.

---

## Invariant Scopes

Invariants can be defined at four scopes with strict precedence:

```
plan > action > project > global
```

Higher-specificity scopes override lower-specificity ones for the same
normalised text. The sole exception is the `non_overridable` flag on global
invariants.

!!! warning "Security: `non_overridable` global invariants cannot be bypassed"
    A `global`-scoped invariant with `non_overridable=True` **always wins** over
    plan-, action-, or project-scoped invariants. Use this to enforce
    organisation-wide guardrails such as "Never write outside the project
    directory" or "Never access credentials files". Plan-level actors cannot
    override these protections.

| Scope | Description |
|-------|-------------|
| `global` | Applies to all plans across every project |
| `project` | Applies to all plans within a project |
| `action` | Applies to all plans that use a specific action |
| `plan` | Applies only to a single plan instance |

---

## Automatic Invocation

`PlanLifecycleService` automatically invokes the reconciliation actor at the
beginning of these phase transitions:

| Transition | Method |
|------------|--------|
| Strategize start | `start_strategize()` |
| Execute start | `execute_plan()` |
| Apply start | `apply_plan()` |

If reconciliation fails during any transition, the phase change is blocked and
`INVARIANT_VIOLATED` is emitted. After a correction is applied, the actor runs
again via a `CORRECTION_APPLIED` event subscription (best-effort; it does not
delay correction completion).

---

## Core Classes

### `InvariantService`

```python
from cleveragents.application.services.invariant_service import InvariantService
from cleveragents.domain.models.core.invariant import InvariantScope

# Preferred: resolve via the dependency injection container
service = container.invariant_service()

# Testing: construct directly if you supply the required collaborators
service = InvariantService(event_bus=event_bus)
```

**Key methods**

| Method | Returns | Description |
|--------|---------|-------------|
| `add_invariant(text, scope, source_name, *, non_overridable=False)` | `Invariant` | Validates and persists a new invariant |
| `list_invariants(scope, source_name, effective=False)` | `list[Invariant]` | Lists stored invariants; `effective=True` yields the merged precedence chain |
| `remove_invariant(invariant_id)` | `Invariant` | Soft-deletes an invariant (sets `active=False`) |
| `get_effective_invariants(plan_id, project_name)` | `list[Invariant]` | Returns merged invariants using plan > action > project > global |
| `enforce_invariants(plan_id, invariants, actor_response, violated_invariant_ids)` | `list[InvariantEnforcementRecord]` | Creates enforcement records and emits events |

**Events emitted**

| Event | When |
|-------|------|
| `INVARIANT_VIOLATED` | Each violated invariant ID supplied to `enforce_invariants()` |
| `INVARIANT_ENFORCED` | Each record created during enforcement |
| `INVARIANT_RECONCILED` | Once per enforcement batch summarising the reconciliation result |

### `InvariantReconciliationActor`

```python
from cleveragents.actor.reconciliation import InvariantReconciliationActor

actor = InvariantReconciliationActor(
    invariant_service=container.invariant_service(),
    decision_service=container.decision_service(),
)
result = actor.run(
    plan_id="01ARZ3…",
    project_name="local/my-app",
    action_name="builtin/plan-execute",
)
```

The actor is registered as `builtin/invariant-reconciliation` in the actor
registry and is invoked automatically by `PlanLifecycleService`. A coroutine
variant `reconcile()` is also available when the full event/decision pipeline is
needed.

**`run()` return value**

| Field | Type | Description |
|-------|------|-------------|
| `reconciled_set` | `InvariantSet` | Effective invariant set after precedence resolution |
| `conflicts` | `list[ConflictRecord]` | Conflicts detected and their resolutions |
| `violated_ids` | `list[str]` | Identifiers of invariants that were violated |

### `ReconciliationResult`

```python
@dataclass
class ReconciliationResult:
    plan_id: str
    resolved_invariants: list[Invariant]
    conflicts_detected: int
    conflicts_resolved: int
    decisions: list[ReconciliationDecision]
    success: bool
```

### `ReconciliationDecision`

```python
@dataclass
class ReconciliationDecision:
    winner: Invariant
    loser: Invariant
    rationale: str  # e.g. "plan scope > project scope"
    timestamp: datetime
```

---

## Reconciliation Algorithm

The reconciliation algorithm (see **Invariant Reconciliation Algorithm** in the
specification) performs:

1. **Collect** — load invariants from all scopes (global → project → action → plan).
2. **Normalise** — canonicalise text to group duplicates (case and whitespace agnostic).
3. **Detect** — flag conflicts between invariants that constrain the same behaviour.
4. **Resolve** — apply precedence (`plan > action > project > global`), honouring
   `non_overridable` global invariants.
5. **Record** — emit `invariant_enforced` decisions and audit entries.
6. **Emit** — publish `INVARIANT_RECONCILED` and any `INVARIANT_VIOLATED` events.

```python
invariants = service.get_effective_invariants(plan_id, project_name)

records = service.enforce_invariants(
    plan_id=plan_id,
    invariants=invariants,
    actor_response=actor_response_text,
    violated_invariant_ids=["inv-001", "inv-002"],
)
# → emits INVARIANT_ENFORCED / INVARIANT_RECONCILED
```

---

## Failure Behaviour

When reconciliation detects a violation:

1. `ReconciliationBlockedError` is raised and the phase transition aborts.
2. `INVARIANT_VIOLATED` is emitted for each violated invariant.
3. The plan remains in its previous phase until the offending invariants are
   corrected and reconciliation succeeds.

```python
from cleveragents.domain.exceptions import ReconciliationBlockedError

try:
    lifecycle_service.start_execute(plan_id)
except ReconciliationBlockedError as exc:
    logger.error("Phase transition blocked", violated=exc.violated_invariant_ids)
```

Additional operational failures (`InvariantLoadError`, `ReconciliationTimeoutError`) are
propagated to callers so that the lifecycle service can surface actionable
errors to operators.

---

## Managing Invariants via CLI

```bash
# Add a global invariant
agents invariant add --scope global \
    "All output files must be UTF-8 encoded"

# Add a project-scoped invariant
agents invariant add --scope project --project local/my-app \
    "Never modify files outside src/"

# Add a plan-scoped invariant (non-overridable)
agents invariant add --scope plan --plan 01ARZ3... --non-overridable \
    "Do not delete any existing tests"

# List effective invariants for a plan
agents invariant list --plan 01ARZ3... --effective

# Remove an invariant (soft-delete)
agents invariant remove inv-001
```

---

## Domain Model

### `Invariant`

```python
from cleveragents.domain.models.core.invariant import Invariant, InvariantScope

invariant = Invariant(
    text="All output files must be UTF-8 encoded",
    scope=InvariantScope.PROJECT,
    source_name="local/my-app",
    non_overridable=False,
)
```

| Field | Type | Description |
|-------|------|-------------|
| `invariant_id` | `str` (ULID) | Unique identifier |
| `text` | `str` (≤ 2,000 chars) | Constraint description |
| `scope` | `InvariantScope` | `global`, `project`, `action`, or `plan` |
| `source_name` | `str` | Project name, action name, or plan ID (scope dependent) |
| `non_overridable` | `bool` | Prevents higher-specificity overrides when `True` |
| `active` | `bool` | `False` after soft-delete |
| `created_at` | `datetime` (UTC) | Creation timestamp |

### `InvariantScope`

```python
InvariantScope.GLOBAL
InvariantScope.PROJECT
InvariantScope.ACTION
InvariantScope.PLAN
```

### `InvariantSet`

```python
inv_set = InvariantSet(invariants=[inv1, inv2, inv3])
effective = inv_set.get_effective()  # list[Invariant] after precedence resolution
```

### `ConflictRecord`

```python
record = ConflictRecord(
    text="Never modify files outside src/",
    winner_scope=InvariantScope.PLAN,
    loser_scope=InvariantScope.PROJECT,
    resolution="plan-scope invariant takes precedence",
)
```

---

## Merge Precedence Helper

```python
from cleveragents.domain.models.core.invariant import merge_invariants

merged = merge_invariants(
    global_invariants=[...],
    project_invariants=[...],
    action_invariants=[...],
    plan_invariants=[...],
)
```

Rules applied by `merge_invariants()`:

1. De-duplicate invariants using normalised text.
2. Apply precedence `plan > action > project > global`.
3. Honour `non_overridable=True` global invariants regardless of scope.
4. Record conflicts as `ConflictRecord` instances for auditability.

---

## Dependency Injection

`InvariantService` is registered as a singleton in the application container and
the actor is exposed as a factory-bound dependency:

```python
class Container(containers.DeclarativeContainer):
    invariant_service = providers.Singleton(
        InvariantService,
        uow_factory=uow_factory,
        event_bus=event_bus,
    )

    invariant_reconciliation_actor = providers.Factory(
        InvariantReconciliationActor,
        invariant_service=invariant_service,
        decision_service=decision_service,
    )
```

Always obtain services through the container to share caches and maintain
cross-cutting instrumentation.

---

## Usage Examples

### Trigger reconciliation manually

```python
actor = container.invariant_reconciliation_actor()
result = await actor.reconcile(plan_id="plan-01JXYZ")

if not result.success:
    logger.error(
        "Reconciliation failed",
        conflicts=result.conflicts_detected,
    )
else:
    for decision in result.decisions:
        logger.info(
            "%r beat %r: %s",
            decision.winner.text,
            decision.loser.text,
            decision.rationale,
        )
```

### Listen for reconciliation events

```python
@event_bus.subscribe(INVARIANT_RECONCILED)
async def on_reconciled(event):
    metrics.record("invariant.decisions", event.decision_count)

@event_bus.subscribe(INVARIANT_VIOLATED)
async def on_violated(event):
    alerts.raise_("plan-blocked", plan_id=event.plan_id)
```

---

## Testing

The invariant system is covered by:
- Behave scenarios in `features/invariant_enforcement.feature`
- Robot Framework suites in `robot/invariants.robot`

```python
service = InvariantService(event_bus=FakeEventBus())

# Add invariants
service.add_invariant(
    text="All output files must be UTF-8 encoded",
    scope=InvariantScope.PROJECT,
    source_name="local/my-app",
)

# Retrieve effective set
effective = service.get_effective_invariants(
    plan_id="01ARZ3…",
    project_name="local/my-app",
)
assert any(i.text.endswith("UTF-8 encoded") for i in effective)
```

---

## Related Documentation

- [ADR-016 Invariant System](../adr/ADR-016-invariant-system.md)
- [ADR-007 Decision Tree & Correction](../adr/ADR-007-decision-tree-and-correction.md)
- [Architecture — Invariant Reconciliation](../architecture.md#invariant-reconciliation)
- [Architecture — Plan Lifecycle](../architecture.md#plan-lifecycle)
- [Core API Reference — Invariant Service](../api/core.md#invariant-service)
- [Shell Safety Module](shell-safety.md)