benchmarks/invariant_reconciliation_bench.py

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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