cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(actor): implement built-in invariant reconciliation actor
CI / lint (pull_request) Successful in 14s
Details
CI / typecheck (pull_request) Successful in 32s
Details
CI / security (pull_request) Successful in 30s
Details
CI / quality (pull_request) Successful in 18s
Details
CI / benchmark-publish (pull_request) Has been skipped
Details
CI / build (pull_request) Successful in 16s
Details
CI / unit_tests (pull_request) Successful in 1m59s
Details
CI / integration_tests (pull_request) Successful in 3m47s
Details
CI / docker (pull_request) Successful in 47s
Details
CI / coverage (pull_request) Successful in 5m48s
Details
CI / lint (push) Successful in 13s
Details
CI / typecheck (push) Successful in 35s
Details
CI / security (push) Successful in 34s
Details
CI / quality (push) Successful in 19s
Details
CI / build (push) Successful in 17s
Details
CI / unit_tests (push) Successful in 2m27s
Details
CI / integration_tests (push) Successful in 2m59s
Details
CI / benchmark-regression (push) Has been skipped
Details
CI / coverage (push) Successful in 4m9s
Details
CI / benchmark-publish (push) Successful in 16m43s
Details
CI / docker (push) Successful in 39s
Details
CI / benchmark-regression (pull_request) Successful in 30m2s
Details

Browse Source 
Add InvariantReconciliationActor that runs at the start of the Strategize
phase to reconcile invariants from four scopes (global, project, action,
plan). The actor detects conflicts, resolves them using specificity-based
precedence (plan > action > project > global), honours non_overridable
global invariants, records invariant_enforced decisions, and produces a
reconciled InvariantSet.

Changes:
- New: src/cleveragents/actor/reconciliation.py
  - InvariantReconciliationActor class with collect_invariants() and run()
  - reconcile_invariants() pure function
  - ScopeInvariants, ConflictRecord, ReconciliationResult dataclasses
- Modified: src/cleveragents/domain/models/core/invariant.py
  - Added non_overridable: bool field to Invariant model
- New: features/invariant_reconciliation_actor.feature (26 BDD scenarios)
- New: features/steps/invariant_reconciliation_actor_steps.py
- New: robot/invariant_reconciliation_actor.robot
- New: robot/helper_invariant_reconciliation.py
- New: benchmarks/invariant_reconciliation_bench.py

Closes #549
This commit was merged in pull request #561.
This commit is contained in:
Jeffrey Phillips Freeman
2026-03-04 09:38:49 +00:00
committed by Jeffrey Phillips Freeman
parent 93e3893d69
commit abd4c6de49
8 changed files with 1566 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/invariant_reconciliation_bench.py
+174
View File
@@ -0,0 +1,174 @@
"""ASV benchmarks for Invariant Reconciliation Actor throughput.
Measures the performance of:
- Reconciliation with varying numbers of invariants per scope
- Conflict detection and resolution overhead
- Decision recording throughput
- Full pipeline (collect → reconcile → record)
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
# Ensure the local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.actor.reconciliation import ( # noqa: E402
InvariantReconciliationActor,
ScopeInvariants,
reconcile_invariants,
)
from cleveragents.application.services.decision_service import DecisionService # noqa: E402
from cleveragents.application.services.invariant_service import InvariantService # noqa: E402
from cleveragents.domain.models.core.invariant import ( # noqa: E402
Invariant,
InvariantScope,
)
def _make_invariant(text: str, scope: InvariantScope, source: str) -> Invariant:
"""Create an invariant without going through the service."""
return Invariant(text=text, scope=scope, source_name=source)
def _build_scope_invariants(
n_global: int = 0,
n_project: int = 0,
n_action: int = 0,
n_plan: int = 0,
) -> ScopeInvariants:
"""Build a ScopeInvariants with the given counts (unique texts)."""
return ScopeInvariants(
global_invariants=[
_make_invariant(f"global-rule-{i}", InvariantScope.GLOBAL, "system")
for i in range(n_global)
],
project_invariants=[
_make_invariant(f"project-rule-{i}", InvariantScope.PROJECT, "local/api")
for i in range(n_project)
],
action_invariants=[
_make_invariant(f"action-rule-{i}", InvariantScope.ACTION, "local/deploy")
for i in range(n_action)
],
plan_invariants=[
_make_invariant(f"plan-rule-{i}", InvariantScope.PLAN, "PLAN01")
for i in range(n_plan)
],
)
def _build_conflicting_scope_invariants(n_conflicts: int) -> ScopeInvariants:
"""Build scope invariants where each text appears at global and plan."""
return ScopeInvariants(
global_invariants=[
_make_invariant(f"shared-rule-{i}", InvariantScope.GLOBAL, "system")
for i in range(n_conflicts)
],
project_invariants=[],
action_invariants=[],
plan_invariants=[
_make_invariant(f"shared-rule-{i}", InvariantScope.PLAN, "PLAN01")
for i in range(n_conflicts)
],
)
class ReconciliationAlgorithmSuite:
"""Benchmark the pure reconciliation algorithm (no I/O)."""
params = [1, 10, 50, 100]
param_names = ["invariants_per_scope"]
def setup(self, n: int) -> None:
self._scope_invs = _build_scope_invariants(
n_global=n, n_project=n, n_action=n, n_plan=n
)
def time_reconcile_no_conflicts(self, n: int) -> None:
"""Reconcile N invariants per scope with no conflicts."""
reconcile_invariants(self._scope_invs)
class ConflictResolutionSuite:
"""Benchmark conflict detection and resolution."""
params = [1, 10, 50, 100]
param_names = ["conflict_count"]
def setup(self, n: int) -> None:
self._scope_invs = _build_conflicting_scope_invariants(n)
def time_reconcile_with_conflicts(self, n: int) -> None:
"""Reconcile N conflicting invariant pairs."""
reconcile_invariants(self._scope_invs)
class FullPipelineSuite:
"""Benchmark the full actor pipeline (collect + reconcile + record)."""
params = [1, 10, 50]
param_names = ["invariants_per_scope"]
def setup(self, n: int) -> None:
self._n = n
def time_full_pipeline(self, n: int) -> None:
"""Run full actor pipeline with N invariants per scope."""
inv_svc = InvariantService()
dec_svc = DecisionService()
for i in range(n):
inv_svc.add_invariant(f"global-rule-{i}", InvariantScope.GLOBAL, "system")
inv_svc.add_invariant(
f"project-rule-{i}", InvariantScope.PROJECT, "local/api"
)
inv_svc.add_invariant(
f"action-rule-{i}", InvariantScope.ACTION, "local/deploy"
)
inv_svc.add_invariant(
f"plan-rule-{i}", InvariantScope.PLAN, "01JQAAAAAAAAAAAAAAAAAAAA01"
)
actor = InvariantReconciliationActor(
invariant_service=inv_svc,
decision_service=dec_svc,
)
actor.run(
plan_id="01JQAAAAAAAAAAAAAAAAAAAA01",
project_name="local/api",
action_name="local/deploy",
)
class DecisionRecordingSuite:
"""Benchmark decision recording throughput."""
params = [1, 10, 50, 100]
param_names = ["invariant_count"]
def setup(self, n: int) -> None:
self._n = n
def time_record_decisions(self, n: int) -> None:
"""Record N invariant_enforced decisions."""
inv_svc = InvariantService()
dec_svc = DecisionService()
for i in range(n):
inv_svc.add_invariant(f"global-rule-{i}", InvariantScope.GLOBAL, "system")
actor = InvariantReconciliationActor(
invariant_service=inv_svc,
decision_service=dec_svc,
)
actor.run(plan_id="01JQAAAAAAAAAAAAAAAAAAAA01")
features/invariant_reconciliation_actor.feature
+236
View File
@@ -0,0 +1,236 @@
Feature: Invariant Reconciliation Actor
As a plan strategizer
I want invariants from all scopes to be reconciled automatically
So that the Strategize phase operates under a clear, conflict-free invariant set
Background:
Given a fresh InvariantService for reconciliation
And a fresh DecisionService for reconciliation
# === Single-scope invariants (no conflicts) ===
@single_scope
Scenario: Single global invariant with no conflicts
Given a global invariant "Never delete production data" from source "system"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the reconciled set should contain "Never delete production data"
And 0 conflicts should be detected
And 1 invariant_enforced decision should be recorded
@single_scope
Scenario: Single plan invariant with no conflicts
Given a plan invariant "Mock all network calls" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the reconciled set should contain "Mock all network calls"
And 0 conflicts should be detected
@single_scope
Scenario: Multiple global invariants with no conflicts
Given a global invariant "Never delete production data" from source "system"
And a global invariant "All APIs must maintain backward compatibility" from source "system"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 2 invariants
And 0 conflicts should be detected
And 2 invariant_enforced decisions should be recorded
# === Multi-scope compatible invariants ===
@multi_scope
Scenario: Multi-scope invariants that are all compatible
Given a global invariant "Never delete production data" from source "system"
And a project invariant "Use ORM for all queries" from source "local/api-service" for project "local/api-service"
And a plan invariant "Mock all network calls" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile with-project "local/api-service" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 3 invariants
And the reconciled set should contain "Never delete production data"
And the reconciled set should contain "Use ORM for all queries"
And the reconciled set should contain "Mock all network calls"
And 0 conflicts should be detected
And 3 invariant_enforced decisions should be recorded
@multi_scope
Scenario: Four-scope invariants all compatible
Given a global invariant "Never delete production data" from source "system"
And a project invariant "Use ORM for all queries" from source "local/api-service" for project "local/api-service"
And an action invariant "Separate commits per file" from source "local/code-coverage" for action "local/code-coverage"
And a plan invariant "Mock all network calls" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile with-project-action "local/api-service" and "local/code-coverage" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 4 invariants
And 0 conflicts should be detected
And 4 invariant_enforced decisions should be recorded
# === Conflict detection and specificity resolution ===
@conflict
Scenario: Plan invariant overrides project invariant on same text
Given a project invariant "Use REST for all APIs" from source "local/api-service" for project "local/api-service"
And a plan invariant "Use REST for all APIs" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile with-project "local/api-service" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the winning invariant for "use rest for all apis" should be from "plan" scope
And 1 conflict should be detected
@conflict
Scenario: Plan overrides global on same text
Given a global invariant "All tests must pass" from source "system"
And a plan invariant "All tests must pass" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the winning invariant for "all tests must pass" should be from "plan" scope
And 1 conflict should be detected
@conflict
Scenario: Action invariant overrides project invariant
Given a project invariant "Coverage above 80%" from source "local/api-service" for project "local/api-service"
And an action invariant "Coverage above 80%" from source "local/code-coverage" for action "local/code-coverage"
When I reconcile with-project-action "local/api-service" and "local/code-coverage" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the winning invariant for "coverage above 80%" should be from "action" scope
And 1 conflict should be detected
@conflict
Scenario: Project invariant overrides global invariant
Given a global invariant "Database timeout 10s" from source "system"
And a project invariant "Database timeout 10s" from source "local/api-service" for project "local/api-service"
When I reconcile with-project "local/api-service" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the winning invariant for "database timeout 10s" should be from "project" scope
And 1 conflict should be detected
# === non_overridable global invariant ===
@non_overridable
Scenario: Non-overridable global invariant blocks plan override
Given a non_overridable global invariant "Payment processing must be idempotent" from source "system"
And a plan invariant "Payment processing must be idempotent" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the winning invariant for "payment processing must be idempotent" should be from "global" scope
And 1 conflict should be detected
And the conflict reason should mention "non_overridable"
@non_overridable
Scenario: Non-overridable global invariant blocks project override
Given a non_overridable global invariant "Audit logging is mandatory" from source "system"
And a project invariant "Audit logging is mandatory" from source "local/api-service" for project "local/api-service"
When I reconcile with-project "local/api-service" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the winning invariant for "audit logging is mandatory" should be from "global" scope
And 1 conflict should be detected
And the conflict reason should mention "non_overridable"
@non_overridable
Scenario: Non-overridable global invariant blocks action override
Given a non_overridable global invariant "Security scan required" from source "system"
And an action invariant "Security scan required" from source "local/deploy" for action "local/deploy"
When I reconcile with-action "local/deploy" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the winning invariant for "security scan required" should be from "global" scope
And 1 conflict should be detected
@non_overridable
Scenario: Regular global invariant can be overridden by plan
Given a global invariant "Use verbose logging" from source "system"
And a plan invariant "Use verbose logging" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the winning invariant for "use verbose logging" should be from "plan" scope
# === invariant_enforced decision recording ===
@decisions
Scenario: Each reconciled invariant gets an invariant_enforced decision
Given a global invariant "Never delete production data" from source "system"
And a project invariant "Use ORM for all queries" from source "local/api-service" for project "local/api-service"
And a plan invariant "Mock all network calls" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile with-project "local/api-service" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then 3 invariant_enforced decisions should be recorded
And all recorded decisions should be of type "invariant_enforced"
And each decision should reference the plan "01JQAAAAAAAAAAAAAAAAAAAA01"
@decisions
Scenario: Decisions include rationale about scope
Given a global invariant "Never delete production data" from source "system"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the decision rationale should mention "global"
@decisions
Scenario: Decision question includes invariant text
Given a plan invariant "Mock all network calls" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the decision question should contain "Mock all network calls"
# === Reconciled InvariantSet accessibility ===
@invariant_set
Scenario: Reconciled InvariantSet is returned and accessible
Given a global invariant "Never delete production data" from source "system"
And a plan invariant "Mock all network calls" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the result should contain a reconciled InvariantSet
And the InvariantSet should have 2 invariants
And the InvariantSet invariants should be Invariant model instances
@invariant_set
Scenario: Empty scope produces empty reconciled set
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 0 invariants
And 0 conflicts should be detected
And 0 invariant_enforced decisions should be recorded
# === Validation and edge cases ===
@validation
Scenario: Empty plan_id raises ValueError
Then running reconciliation with empty plan_id should raise ValueError
@validation
Scenario: Inactive invariants are excluded from reconciliation
Given a global invariant "Never delete production data" from source "system"
And an inactive global invariant "Deprecated rule" from source "system"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And the reconciled set should not contain "Deprecated rule"
@validation
Scenario: Case-insensitive conflict detection
Given a global invariant "Use HTTPS only" from source "system"
And a plan invariant "use https only" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I reconcile plan-only for "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the reconciled set should contain 1 invariant
And 1 conflict should be detected
@validation
Scenario: Parent decision ID is passed to recorded decisions
Given a global invariant "Never delete production data" from source "system"
When I reconcile with-parent "01JQBBBBBBBBBBBBBBBBBBBB01" for plan "01JQAAAAAAAAAAAAAAAAAAAA01"
Then the recorded decisions should have parent "01JQBBBBBBBBBBBBBBBBBBBB01"
# === Collect invariants separately ===
@validation
Scenario: Creating actor with None invariant_service raises ValueError
Then creating an actor with None invariant_service should raise ValueError
@validation
Scenario: Creating actor with None decision_service raises ValueError
Then creating an actor with None decision_service should raise ValueError
@validation
Scenario: Reconciliation with no plan_id scope omits plan invariants
Given a global invariant "Global rule" from source "system"
When I reconcile without-plan-id using project "local/api-service"
Then the collected plan invariants should be empty
@collection
Scenario: Collect invariants gathers from all scopes
Given a global invariant "Global rule" from source "system"
And a project invariant "Project rule" from source "local/api-service" for project "local/api-service"
And an action invariant "Action rule" from source "local/deploy" for action "local/deploy"
And a plan invariant "Plan rule" from source "01JQAAAAAAAAAAAAAAAAAAAA01"
When I collect invariants for plan "01JQAAAAAAAAAAAAAAAAAAAA01" with project "local/api-service" and action "local/deploy"
Then the collected scope invariants should have 1 global invariant
And the collected scope invariants should have 1 project invariant
And the collected scope invariants should have 1 action invariant
And the collected scope invariants should have 1 plan invariant
features/steps/invariant_reconciliation_actor_steps.py
+423
View File
@@ -0,0 +1,423 @@
"""Step definitions for invariant_reconciliation_actor.feature.
Tests the InvariantReconciliationActor: multi-scope collection,
conflict detection, specificity-based resolution with non_overridable
support, invariant_enforced decision recording, and reconciled
InvariantSet production.
"""
from __future__ import annotations
from behave import given, then, when # type: ignore[import-untyped]
from cleveragents.actor.reconciliation import (
InvariantReconciliationActor,
ReconciliationResult,
)
from cleveragents.application.services.decision_service import DecisionService
from cleveragents.application.services.invariant_service import InvariantService
from cleveragents.domain.models.core.decision import DecisionType
from cleveragents.domain.models.core.invariant import (
Invariant,
InvariantScope,
InvariantSet,
)
# ================================================================
# Background / Setup
# ================================================================
@given("a fresh InvariantService for reconciliation")
def step_fresh_invariant_service(context):
"""Create a fresh InvariantService instance."""
context.invariant_service = InvariantService()
@given("a fresh DecisionService for reconciliation")
def step_fresh_decision_service(context):
"""Create a fresh DecisionService instance."""
context.decision_service = DecisionService()
# ================================================================
# Adding Invariants
# ================================================================
@given('a global invariant "{text}" from source "{source}"')
def step_add_global_invariant(context, text, source):
"""Add a global-scope invariant."""
context.invariant_service.add_invariant(
text=text, scope=InvariantScope.GLOBAL, source_name=source
)
@given('a non_overridable global invariant "{text}" from source "{source}"')
def step_add_non_overridable_global(context, text, source):
"""Add a non_overridable global-scope invariant."""
inv = context.invariant_service.add_invariant(
text=text, scope=InvariantScope.GLOBAL, source_name=source
)
# Set non_overridable on the stored invariant
inv.non_overridable = True
@given('a project invariant "{text}" from source "{source}" for project "{project}"')
def step_add_project_invariant(context, text, source, project):
"""Add a project-scope invariant."""
context.invariant_service.add_invariant(
text=text, scope=InvariantScope.PROJECT, source_name=source
)
@given('an action invariant "{text}" from source "{source}" for action "{action}"')
def step_add_action_invariant(context, text, source, action):
"""Add an action-scope invariant."""
context.invariant_service.add_invariant(
text=text, scope=InvariantScope.ACTION, source_name=source
)
@given('a plan invariant "{text}" from source "{source}"')
def step_add_plan_invariant(context, text, source):
"""Add a plan-scope invariant."""
context.invariant_service.add_invariant(
text=text, scope=InvariantScope.PLAN, source_name=source
)
@given('an inactive global invariant "{text}" from source "{source}"')
def step_add_inactive_global(context, text, source):
"""Add a global invariant then soft-delete it."""
inv = context.invariant_service.add_invariant(
text=text, scope=InvariantScope.GLOBAL, source_name=source
)
context.invariant_service.remove_invariant(inv.id)
# ================================================================
# Running the actor
# ================================================================
def _build_actor(context):
"""Build the reconciliation actor from context services."""
return InvariantReconciliationActor(
invariant_service=context.invariant_service,
decision_service=context.decision_service,
)
@when('I reconcile plan-only for "{plan_id}"')
def step_run_reconciliation(context, plan_id):
"""Run the reconciliation actor with plan-only scope."""
actor = _build_actor(context)
context.reconciliation_result = actor.run(plan_id=plan_id)
@when('I reconcile with-project "{project}" for plan "{plan_id}"')
def step_run_reconciliation_with_project(context, project, plan_id):
"""Run the reconciliation actor with plan and project scope."""
actor = _build_actor(context)
context.reconciliation_result = actor.run(plan_id=plan_id, project_name=project)
@when('I reconcile with-project-action "{project}" and "{action}" for plan "{plan_id}"')
def step_run_reconciliation_with_project_and_action(context, project, action, plan_id):
"""Run the reconciliation actor with plan, project, and action scope."""
actor = _build_actor(context)
context.reconciliation_result = actor.run(
plan_id=plan_id, project_name=project, action_name=action
)
@when('I reconcile with-action "{action}" for plan "{plan_id}"')
def step_run_reconciliation_with_action(context, action, plan_id):
"""Run the reconciliation actor with plan and action scope."""
actor = _build_actor(context)
context.reconciliation_result = actor.run(plan_id=plan_id, action_name=action)
@when('I reconcile with-parent "{parent_id}" for plan "{plan_id}"')
def step_run_reconciliation_with_parent(context, parent_id, plan_id):
"""Run the reconciliation actor with a parent decision ID."""
actor = _build_actor(context)
context.reconciliation_result = actor.run(
plan_id=plan_id, parent_decision_id=parent_id
)
@when(
'I collect invariants for plan "{plan_id}" '
'with project "{project}" and action "{action}"'
)
def step_collect_invariants(context, plan_id, project, action):
"""Collect invariants without full reconciliation."""
actor = _build_actor(context)
context.collected_scope = actor.collect_invariants(
plan_id=plan_id, project_name=project, action_name=action
)
# ================================================================
# Reconciled set assertions
# ================================================================
@then("the reconciled set should contain {count:d} invariant")
@then("the reconciled set should contain {count:d} invariants")
def step_reconciled_count(context, count):
"""Assert the reconciled set has the expected number of invariants."""
result: ReconciliationResult = context.reconciliation_result
actual = len(result.reconciled_set.invariants)
assert actual == count, f"Expected {count} invariants, got {actual}"
@then('the reconciled set should contain "{text}"')
def step_reconciled_contains_text(context, text):
"""Assert the reconciled set contains an invariant with the given text."""
result: ReconciliationResult = context.reconciliation_result
texts = [inv.text for inv in result.reconciled_set.invariants]
assert text in texts, f"'{text}' not found in {texts}"
@then('the reconciled set should not contain "{text}"')
def step_reconciled_not_contains_text(context, text):
"""Assert the reconciled set does not contain the given text."""
result: ReconciliationResult = context.reconciliation_result
texts = [inv.text for inv in result.reconciled_set.invariants]
assert text not in texts, f"'{text}' unexpectedly found in {texts}"
# ================================================================
# Conflict assertions
# ================================================================
@then("{count:d} conflict should be detected")
@then("{count:d} conflicts should be detected")
def step_conflict_count(context, count):
"""Assert the expected number of conflicts."""
result: ReconciliationResult = context.reconciliation_result
actual = len(result.conflicts)
assert actual == count, f"Expected {count} conflicts, got {actual}"
@then('the winning invariant for "{key}" should be from "{scope}" scope')
def step_winning_scope(context, key, scope):
"""Assert the winner for a given normalised key is from the expected scope."""
result: ReconciliationResult = context.reconciliation_result
for inv in result.reconciled_set.invariants:
if inv.text.strip().lower() == key.strip().lower():
assert inv.scope.value == scope, (
f"Expected winner scope '{scope}', got '{inv.scope.value}'"
)
return
raise AssertionError(f"No invariant found with key '{key}'")
@then('the conflict reason should mention "{text}"')
def step_conflict_reason_mentions(context, text):
"""Assert at least one conflict reason contains the given text."""
result: ReconciliationResult = context.reconciliation_result
reasons = [c.reason for c in result.conflicts]
assert any(text in r for r in reasons), (
f"'{text}' not found in conflict reasons: {reasons}"
)
# ================================================================
# Decision assertions
# ================================================================
@then("{count:d} invariant_enforced decision should be recorded")
@then("{count:d} invariant_enforced decisions should be recorded")
def step_decision_count(context, count):
"""Assert the expected number of enforced decisions."""
result: ReconciliationResult = context.reconciliation_result
actual = len(result.enforced_decision_ids)
assert actual == count, f"Expected {count} decisions, got {actual}"
@then('all recorded decisions should be of type "{dtype}"')
def step_all_decisions_type(context, dtype):
"""Assert all recorded decisions are of the specified type."""
result: ReconciliationResult = context.reconciliation_result
for did in result.enforced_decision_ids:
decision = context.decision_service.get_decision(did)
assert decision.decision_type == DecisionType(dtype), (
f"Decision {did} type is {decision.decision_type}, expected {dtype}"
)
@then('each decision should reference the plan "{plan_id}"')
def step_decisions_reference_plan(context, plan_id):
"""Assert each decision references the given plan."""
result: ReconciliationResult = context.reconciliation_result
for did in result.enforced_decision_ids:
decision = context.decision_service.get_decision(did)
assert decision.plan_id == plan_id, (
f"Decision {did} plan_id is {decision.plan_id}, expected {plan_id}"
)
@then('the decision rationale should mention "{text}"')
def step_decision_rationale_mentions(context, text):
"""Assert at least one decision rationale mentions the text."""
result: ReconciliationResult = context.reconciliation_result
for did in result.enforced_decision_ids:
decision = context.decision_service.get_decision(did)
if text.lower() in decision.rationale.lower():
return
raise AssertionError(f"No decision rationale mentions '{text}'")
@then('the decision question should contain "{text}"')
def step_decision_question_contains(context, text):
"""Assert at least one decision question contains the text."""
result: ReconciliationResult = context.reconciliation_result
for did in result.enforced_decision_ids:
decision = context.decision_service.get_decision(did)
if text in decision.question:
return
raise AssertionError(f"No decision question contains '{text}'")
@then('the recorded decisions should have parent "{parent_id}"')
def step_decisions_have_parent(context, parent_id):
"""Assert all recorded decisions have the given parent decision ID."""
result: ReconciliationResult = context.reconciliation_result
for did in result.enforced_decision_ids:
decision = context.decision_service.get_decision(did)
assert decision.parent_decision_id == parent_id, (
f"Decision {did} parent is {decision.parent_decision_id}, "
f"expected {parent_id}"
)
# ================================================================
# InvariantSet assertions
# ================================================================
@then("the result should contain a reconciled InvariantSet")
def step_result_has_invariant_set(context):
"""Assert the result contains a reconciled InvariantSet."""
result: ReconciliationResult = context.reconciliation_result
assert result.reconciled_set is not None
assert isinstance(result.reconciled_set, InvariantSet)
@then("the InvariantSet should have {count:d} invariants")
def step_invariant_set_count(context, count):
"""Assert the InvariantSet has the expected count."""
result: ReconciliationResult = context.reconciliation_result
actual = len(result.reconciled_set.invariants)
assert actual == count, f"Expected {count} invariants in set, got {actual}"
@then("the InvariantSet invariants should be Invariant model instances")
def step_invariant_set_types(context):
"""Assert each element in the InvariantSet is an Invariant instance."""
result: ReconciliationResult = context.reconciliation_result
for inv in result.reconciled_set.invariants:
assert isinstance(inv, Invariant), (
f"Expected Invariant, got {type(inv).__name__}"
)
# ================================================================
# Validation / error assertions
# ================================================================
@then("running reconciliation with empty plan_id should raise ValueError")
def step_empty_plan_id_raises(context: object) -> None:
"""Assert ValueError is raised for empty plan_id."""
actor = _build_actor(context)
try:
actor.run(plan_id="")
raise AssertionError("Expected ValueError for empty plan_id")
except ValueError:
pass
@then("creating an actor with None invariant_service should raise ValueError")
def step_none_invariant_service_raises(context: object) -> None:
"""Assert ValueError when invariant_service is None."""
try:
InvariantReconciliationActor(
invariant_service=None, # type: ignore[arg-type]
decision_service=context.decision_service,
)
raise AssertionError("Expected ValueError for None invariant_service")
except ValueError:
pass
@then("creating an actor with None decision_service should raise ValueError")
def step_none_decision_service_raises(context: object) -> None:
"""Assert ValueError when decision_service is None."""
try:
InvariantReconciliationActor(
invariant_service=context.invariant_service,
decision_service=None, # type: ignore[arg-type]
)
raise AssertionError("Expected ValueError for None decision_service")
except ValueError:
pass
@when('I reconcile without-plan-id using project "{project}"')
def step_reconcile_without_plan_id(context: object, project: str) -> None:
"""Collect invariants without specifying a plan_id."""
actor = _build_actor(context)
context.collected_scope = actor.collect_invariants(
project_name=project,
)
@then("the collected plan invariants should be empty")
def step_collected_plan_empty(context: object) -> None:
"""Assert the collected plan invariants list is empty."""
actual = len(context.collected_scope.plan_invariants)
assert actual == 0, f"Expected 0 plan invariants, got {actual}"
# ================================================================
# Collection assertions
# ================================================================
@then("the collected scope invariants should have {count:d} global invariant")
@then("the collected scope invariants should have {count:d} global invariants")
def step_collected_global_count(context, count):
"""Assert the collected scope has the expected global count."""
actual = len(context.collected_scope.global_invariants)
assert actual == count, f"Expected {count} global invariants, got {actual}"
@then("the collected scope invariants should have {count:d} project invariant")
@then("the collected scope invariants should have {count:d} project invariants")
def step_collected_project_count(context, count):
"""Assert the collected scope has the expected project count."""
actual = len(context.collected_scope.project_invariants)
assert actual == count, f"Expected {count} project invariants, got {actual}"
@then("the collected scope invariants should have {count:d} action invariant")
@then("the collected scope invariants should have {count:d} action invariants")
def step_collected_action_count(context, count):
"""Assert the collected scope has the expected action count."""
actual = len(context.collected_scope.action_invariants)
assert actual == count, f"Expected {count} action invariants, got {actual}"
@then("the collected scope invariants should have {count:d} plan invariant")
@then("the collected scope invariants should have {count:d} plan invariants")
def step_collected_plan_count(context, count):
"""Assert the collected scope has the expected plan count."""
actual = len(context.collected_scope.plan_invariants)
assert actual == count, f"Expected {count} plan invariants, got {actual}"
robot/helper_invariant_reconciliation.py
+219
View File
@@ -0,0 +1,219 @@
"""Robot Framework helper for invariant reconciliation actor integration tests.
Exercises the full reconciliation pipeline: collect → reconcile → enforce,
then verifies the reconciled set is respected by subsequent operations.
"""
from __future__ import annotations
import json
import sys
from pathlib import Path
# Ensure source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.actor.reconciliation import ( # noqa: E402
InvariantReconciliationActor,
)
from cleveragents.application.services.decision_service import ( # noqa: E402
DecisionService,
)
from cleveragents.application.services.invariant_service import ( # noqa: E402
InvariantService,
)
from cleveragents.domain.models.core.decision import DecisionType # noqa: E402
from cleveragents.domain.models.core.invariant import InvariantScope # noqa: E402
def _run_reconciliation_pipeline() -> dict[str, object]:
"""Run the full reconciliation pipeline and return results as dict."""
inv_svc = InvariantService()
dec_svc = DecisionService()
# Add invariants across scopes
inv_svc.add_invariant(
"Never delete production data",
InvariantScope.GLOBAL,
"system",
)
inv_svc.add_invariant(
"Use ORM for all queries",
InvariantScope.PROJECT,
"local/api-service",
)
inv_svc.add_invariant(
"Separate commits per file",
InvariantScope.ACTION,
"local/code-coverage",
)
inv_svc.add_invariant(
"Mock all network calls",
InvariantScope.PLAN,
"01JQAAAAAAAAAAAAAAAAAAAA01",
)
actor = InvariantReconciliationActor(
invariant_service=inv_svc,
decision_service=dec_svc,
)
result = actor.run(
plan_id="01JQAAAAAAAAAAAAAAAAAAAA01",
project_name="local/api-service",
action_name="local/code-coverage",
)
return {
"invariant_count": len(result.reconciled_set.invariants),
"conflict_count": len(result.conflicts),
"decision_count": len(result.enforced_decision_ids),
"invariant_texts": [inv.text for inv in result.reconciled_set.invariants],
}
def _run_conflict_resolution() -> dict[str, object]:
"""Run reconciliation with conflicts and verify resolution."""
inv_svc = InvariantService()
dec_svc = DecisionService()
inv_svc.add_invariant(
"All tests must pass",
InvariantScope.GLOBAL,
"system",
)
inv_svc.add_invariant(
"All tests must pass",
InvariantScope.PLAN,
"01JQAAAAAAAAAAAAAAAAAAAA01",
)
actor = InvariantReconciliationActor(
invariant_service=inv_svc,
decision_service=dec_svc,
)
result = actor.run(plan_id="01JQAAAAAAAAAAAAAAAAAAAA01")
winner_scope = ""
for inv in result.reconciled_set.invariants:
if inv.text.lower() == "all tests must pass":
winner_scope = inv.scope.value
break
return {
"invariant_count": len(result.reconciled_set.invariants),
"conflict_count": len(result.conflicts),
"winner_scope": winner_scope,
}
def _run_non_overridable_test() -> dict[str, object]:
"""Verify non_overridable global invariants block overrides."""
inv_svc = InvariantService()
dec_svc = DecisionService()
inv = inv_svc.add_invariant(
"Payment processing must be idempotent",
InvariantScope.GLOBAL,
"system",
)
inv.non_overridable = True
inv_svc.add_invariant(
"Payment processing must be idempotent",
InvariantScope.PLAN,
"01JQAAAAAAAAAAAAAAAAAAAA01",
)
actor = InvariantReconciliationActor(
invariant_service=inv_svc,
decision_service=dec_svc,
)
result = actor.run(plan_id="01JQAAAAAAAAAAAAAAAAAAAA01")
winner_scope = ""
for r_inv in result.reconciled_set.invariants:
if "idempotent" in r_inv.text.lower():
winner_scope = r_inv.scope.value
break
return {
"invariant_count": len(result.reconciled_set.invariants),
"conflict_count": len(result.conflicts),
"winner_scope": winner_scope,
"has_non_overridable_reason": any(
"non_overridable" in c.reason for c in result.conflicts
),
}
def _run_decision_verification() -> dict[str, object]:
"""Verify invariant_enforced decisions are properly recorded."""
inv_svc = InvariantService()
dec_svc = DecisionService()
inv_svc.add_invariant(
"Never delete production data",
InvariantScope.GLOBAL,
"system",
)
inv_svc.add_invariant(
"Mock all network calls",
InvariantScope.PLAN,
"01JQAAAAAAAAAAAAAAAAAAAA01",
)
actor = InvariantReconciliationActor(
invariant_service=inv_svc,
decision_service=dec_svc,
)
result = actor.run(plan_id="01JQAAAAAAAAAAAAAAAAAAAA01")
# Verify decisions
all_correct_type = True
all_correct_plan = True
for did in result.enforced_decision_ids:
decision = dec_svc.get_decision(did)
if decision.decision_type != DecisionType.INVARIANT_ENFORCED:
all_correct_type = False
if decision.plan_id != "01JQAAAAAAAAAAAAAAAAAAAA01":
all_correct_plan = False
return {
"decision_count": len(result.enforced_decision_ids),
"all_correct_type": all_correct_type,
"all_correct_plan": all_correct_plan,
}
def main() -> None:
"""Entry point for Robot Framework helper."""
if len(sys.argv) < 2:
print("Usage: helper_invariant_reconciliation.py <command>")
sys.exit(1)
command = sys.argv[1]
handlers = {
"reconcile": _run_reconciliation_pipeline,
"conflict": _run_conflict_resolution,
"non-overridable": _run_non_overridable_test,
"decisions": _run_decision_verification,
}
handler = handlers.get(command)
if handler is None:
print(f"Unknown command: {command}")
sys.exit(1)
result = handler()
print(json.dumps(result))
print("reconciliation-ok")
if __name__ == "__main__":
main()
robot/invariant_reconciliation_actor.robot
+53
View File
@@ -0,0 +1,53 @@
*** Settings ***
Documentation Integration tests for Invariant Reconciliation Actor
... Verifies reconciliation followed by actor execution respecting invariants
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_invariant_reconciliation.py
*** Test Cases ***
Full Reconciliation Pipeline With Four Scopes
[Documentation] Reconcile invariants from global, project, action, and plan scopes
${result}= Run Process ${PYTHON} ${HELPER} reconcile cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} reconciliation-ok
Should Contain ${result.stdout} "invariant_count": 4
Should Contain ${result.stdout} "conflict_count": 0
Should Contain ${result.stdout} "decision_count": 4
Conflict Resolution By Specificity
[Documentation] Plan-scope invariant overrides global on same text
${result}= Run Process ${PYTHON} ${HELPER} conflict cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} reconciliation-ok
Should Contain ${result.stdout} "invariant_count": 1
Should Contain ${result.stdout} "conflict_count": 1
Should Contain ${result.stdout} "winner_scope": "plan"
Non-Overridable Global Invariant Blocks Override
[Documentation] Non-overridable global invariant wins over plan
${result}= Run Process ${PYTHON} ${HELPER} non-overridable cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} reconciliation-ok
Should Contain ${result.stdout} "winner_scope": "global"
Should Contain ${result.stdout} "has_non_overridable_reason": true
Decision Recording Verification
[Documentation] All invariant_enforced decisions correctly recorded
${result}= Run Process ${PYTHON} ${HELPER} decisions cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} reconciliation-ok
Should Contain ${result.stdout} "decision_count": 2
Should Contain ${result.stdout} "all_correct_type": true
Should Contain ${result.stdout} "all_correct_plan": true
src/cleveragents/actor/reconciliation.py
+447
View File
@@ -0,0 +1,447 @@
"""Built-in Invariant Reconciliation Actor.
Runs at the start of the Strategize phase to reconcile invariants from
four scopes: global, project, action, and plan.
**Algorithm**:
1. Collect invariants from all scopes.
2. Group by normalised text (case-insensitive, stripped).
3. Detect conflicts: same text present at different scopes with
contradictory intent, or different invariants that the grouping
cannot resolve automatically.
4. Resolve conflicts using specificity: plan > action > project > global.
Exception: ``non_overridable`` global invariants always win.
5. Record an ``invariant_enforced`` decision for each active invariant.
6. Return a reconciled ``InvariantSet``.
Based on specification §19440-19600 and issue #549.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
import structlog
from cleveragents.domain.models.core.decision import DecisionType
from cleveragents.domain.models.core.invariant import (
Invariant,
InvariantScope,
InvariantSet,
)
if TYPE_CHECKING:
from cleveragents.application.services.decision_service import DecisionService
from cleveragents.application.services.invariant_service import InvariantService
logger = structlog.get_logger(__name__)
# Precedence order (index 0 = highest)
_SCOPE_PRECEDENCE: dict[InvariantScope, int] = {
InvariantScope.PLAN: 0,
InvariantScope.ACTION: 1,
InvariantScope.PROJECT: 2,
InvariantScope.GLOBAL: 3,
}
@dataclass(frozen=True)
class ConflictRecord:
"""A detected conflict between invariants from different scopes.
Attributes:
key: Normalised invariant text used for grouping.
winner: The invariant that prevailed after resolution.
losers: Invariants that were overridden.
reason: Human-readable explanation of the resolution.
"""
key: str
winner: Invariant
losers: list[Invariant]
reason: str
@dataclass(frozen=True)
class ReconciliationResult:
"""Output of the reconciliation process.
Attributes:
reconciled_set: The final effective ``InvariantSet``.
conflicts: All detected conflicts with resolution details.
enforced_decision_ids: ULIDs of ``invariant_enforced`` decisions.
"""
reconciled_set: InvariantSet
conflicts: list[ConflictRecord]
enforced_decision_ids: list[str]
@dataclass(frozen=False)
class ScopeInvariants:
"""Invariants collected from each scope tier.
Convenience container passed to the reconciliation algorithm.
"""
global_invariants: list[Invariant] = field(default_factory=list)
project_invariants: list[Invariant] = field(default_factory=list)
action_invariants: list[Invariant] = field(default_factory=list)
plan_invariants: list[Invariant] = field(default_factory=list)
def all_invariants(self) -> list[Invariant]:
"""Return a flat list of all invariants across scopes."""
return (
self.global_invariants
+ self.project_invariants
+ self.action_invariants
+ self.plan_invariants
)
def _normalise_key(text: str) -> str:
"""Return a case-insensitive, whitespace-stripped key for grouping."""
if not text:
return ""
return text.strip().lower()
def _scope_rank(scope: InvariantScope) -> int:
"""Return the precedence rank for a scope (lower = higher priority)."""
return _SCOPE_PRECEDENCE.get(scope, 99)
def _resolve_group(
invariants: list[Invariant],
) -> tuple[Invariant, list[ConflictRecord]]:
"""Resolve a group of invariants sharing the same normalised text.
Args:
invariants: All invariants with the same normalised text.
Returns:
A tuple of (winner, conflict_records).
Raises:
ValueError: If *invariants* is empty.
"""
if not invariants:
raise ValueError("Cannot resolve an empty invariant group")
if len(invariants) == 1:
return invariants[0], []
key = _normalise_key(invariants[0].text)
# Check for non_overridable global invariants first
non_overridable_globals = [
inv
for inv in invariants
if inv.scope == InvariantScope.GLOBAL and inv.non_overridable
]
if non_overridable_globals:
winner = non_overridable_globals[0]
losers = [inv for inv in invariants if inv is not winner]
conflict = ConflictRecord(
key=key,
winner=winner,
losers=losers,
reason=(
f"Global invariant is non_overridable; "
f"overrides {len(losers)} lower-scope variant(s)"
),
)
return winner, [conflict]
# Standard specificity resolution: highest precedence wins
sorted_invs = sorted(invariants, key=lambda inv: _scope_rank(inv.scope))
winner = sorted_invs[0]
losers = sorted_invs[1:]
conflicts: list[ConflictRecord] = []
if losers:
conflicts.append(
ConflictRecord(
key=key,
winner=winner,
losers=losers,
reason=(
f"{winner.scope.value} scope overrides "
f"{', '.join(inv.scope.value for inv in losers)} scope(s)"
),
)
)
return winner, conflicts
def reconcile_invariants(
scope_invariants: ScopeInvariants,
) -> tuple[list[Invariant], list[ConflictRecord]]:
"""Run the reconciliation algorithm on collected invariants.
Groups invariants by normalised text, resolves each group, and
returns the winners along with any detected conflicts.
Args:
scope_invariants: Invariants from all four scopes.
Returns:
Tuple of (reconciled_invariants, conflict_records).
"""
all_invs = [inv for inv in scope_invariants.all_invariants() if inv.active]
# Group by normalised text
groups: dict[str, list[Invariant]] = {}
for inv in all_invs:
key = _normalise_key(inv.text)
groups.setdefault(key, []).append(inv)
reconciled: list[Invariant] = []
all_conflicts: list[ConflictRecord] = []
for _key, group in groups.items():
winner, conflicts = _resolve_group(group)
reconciled.append(winner)
all_conflicts.extend(conflicts)
return reconciled, all_conflicts
class InvariantReconciliationActor:
"""Built-in actor that reconciles invariants at Strategize start.
Collects invariants from global, project, action, and plan scopes,
detects and resolves conflicts using specificity-based precedence,
records ``invariant_enforced`` decisions, and produces a reconciled
``InvariantSet`` for downstream actors.
Usage::
actor = InvariantReconciliationActor(
invariant_service=inv_svc,
decision_service=dec_svc,
)
result = actor.run(plan_id="...", project_name="...", action_name="...")
"""
#: Actor name used for registration / lookup.
ACTOR_NAME: str = "builtin/invariant-reconciliation"
def __init__(
self,
*,
invariant_service: InvariantService,
decision_service: DecisionService,
) -> None:
if invariant_service is None:
raise ValueError("invariant_service must not be None")
if decision_service is None:
raise ValueError("decision_service must not be None")
self._invariant_service = invariant_service
self._decision_service = decision_service
self._logger = logger.bind(actor=self.ACTOR_NAME)
# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------
def collect_invariants(
self,
*,
plan_id: str | None = None,
project_name: str | None = None,
action_name: str | None = None,
) -> ScopeInvariants:
"""Collect invariants from all four scopes.
Args:
plan_id: Optional plan identifier.
project_name: Optional project name.
action_name: Optional action name.
Returns:
``ScopeInvariants`` with invariants grouped by scope.
"""
svc = self._invariant_service
return ScopeInvariants(
global_invariants=svc.list_invariants(scope=InvariantScope.GLOBAL),
project_invariants=(
svc.list_invariants(
scope=InvariantScope.PROJECT, source_name=project_name
)
if project_name
else []
),
action_invariants=(
svc.list_invariants(
scope=InvariantScope.ACTION, source_name=action_name
)
if action_name
else []
),
plan_invariants=(
svc.list_invariants(scope=InvariantScope.PLAN, source_name=plan_id)
if plan_id
else []
),
)
def run(
self,
*,
plan_id: str,
project_name: str | None = None,
action_name: str | None = None,
parent_decision_id: str | None = None,
) -> ReconciliationResult:
"""Execute the full reconciliation lifecycle.
1. Collect invariants from all scopes.
2. Reconcile conflicts.
3. Record ``invariant_enforced`` decisions.
4. Return the result.
Args:
plan_id: ULID of the plan entering Strategize.
project_name: Optional project name for scoping.
action_name: Optional action name for scoping.
parent_decision_id: Optional parent decision for tree wiring.
Returns:
A ``ReconciliationResult`` with the reconciled set, conflicts,
and decision IDs.
Raises:
ValueError: If *plan_id* is empty or blank.
"""
if not plan_id or not plan_id.strip():
raise ValueError("plan_id must not be empty")
self._logger.info(
"reconciliation.start",
plan_id=plan_id,
project_name=project_name,
action_name=action_name,
)
# Step 1: Collect
scope_invs = self.collect_invariants(
plan_id=plan_id,
project_name=project_name,
action_name=action_name,
)
total = len(scope_invs.all_invariants())
self._logger.info(
"reconciliation.collected",
plan_id=plan_id,
total_invariants=total,
global_count=len(scope_invs.global_invariants),
project_count=len(scope_invs.project_invariants),
action_count=len(scope_invs.action_invariants),
plan_count=len(scope_invs.plan_invariants),
)
# Step 2: Reconcile
reconciled, conflicts = reconcile_invariants(scope_invs)
if conflicts:
self._logger.info(
"reconciliation.conflicts_detected",
plan_id=plan_id,
conflict_count=len(conflicts),
)
for conflict in conflicts:
self._logger.info(
"reconciliation.conflict_resolved",
plan_id=plan_id,
key=conflict.key,
winner_scope=conflict.winner.scope.value,
reason=conflict.reason,
)
# Step 3: Record decisions
decision_ids = self._record_enforcement_decisions(
plan_id=plan_id,
invariants=reconciled,
parent_decision_id=parent_decision_id,
)
# Step 4: Build result
reconciled_set = InvariantSet(invariants=reconciled)
self._logger.info(
"reconciliation.complete",
plan_id=plan_id,
effective_count=len(reconciled),
conflict_count=len(conflicts),
decision_count=len(decision_ids),
)
return ReconciliationResult(
reconciled_set=reconciled_set,
conflicts=conflicts,
enforced_decision_ids=decision_ids,
)
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
def _record_enforcement_decisions(
self,
*,
plan_id: str,
invariants: list[Invariant],
parent_decision_id: str | None,
) -> list[str]:
"""Record an ``invariant_enforced`` decision per invariant.
Args:
plan_id: The plan ULID.
invariants: Reconciled invariants to enforce.
parent_decision_id: Optional parent decision for tree wiring.
Returns:
List of decision ULIDs created.
"""
decision_ids: list[str] = []
for inv in invariants:
decision = self._decision_service.record_decision(
plan_id=plan_id,
decision_type=DecisionType.INVARIANT_ENFORCED,
question=(
f"Should invariant be enforced? [{inv.scope.value}] {inv.text}"
),
chosen_option=f"Enforce: {inv.text}",
parent_decision_id=parent_decision_id,
rationale=(
f"Invariant from {inv.scope.value} scope "
f"(source: {inv.source_name}) enforced during "
f"reconciliation"
),
confidence_score=1.0,
)
decision_ids.append(decision.decision_id)
# Also record enforcement in the invariant service
self._invariant_service.enforce_invariants(
plan_id=plan_id,
invariants=invariants,
actor_response="Reconciliation complete",
)
return decision_ids
__all__ = [
"ConflictRecord",
"InvariantReconciliationActor",
"ReconciliationResult",
"ScopeInvariants",
"reconcile_invariants",
]
src/cleveragents/domain/models/core/invariant.py
+7
View File
@@ -83,6 +83,13 @@ class Invariant(BaseModel):
default=True,
description="Whether this invariant is active (False = soft-deleted)",
)
non_overridable: bool = Field(
default=False,
description=(
"When True and scope is GLOBAL, this invariant cannot be "
"overridden by lower-scope invariants during reconciliation"
),
)
@field_validator("text")
@classmethod
vulture_whitelist.py
+7
View File
@@ -729,3 +729,10 @@ EscalationDecision # noqa: B018, F821
HistoricalOutcome # noqa: B018, F821
OperationContext # noqa: B018, F821
autonomy_controller # noqa: B018, F821
# Invariant Reconciliation Actor — public API (M6, issue #549)
InvariantReconciliationActor # noqa: B018, F821
ReconciliationResult # noqa: B018, F821
ConflictRecord # noqa: B018, F821
ScopeInvariants # noqa: B018, F821
reconcile_invariants # noqa: B018, F821