CHANGELOG.md

@@ -199,6 +199,15 @@ Changed `wf10_batch.robot` to be less likely to create files, and
   `features/architecture.feature` `@tdd_expected_fail` for pre-existing Pydantic
   compliance debt in `IndexEntry` / `ACMSIndex` classes.
 
+- **InvariantService thread safety (#8209):** Added a ``threading.RLock`` (re-entrant lock) to
+  ``InvariantService`` that guards all shared mutable state
+  (`_invariants` dict and `_enforcement_records` list). Every public method acquires the lock
+  before reading or writing shared data. This prevents
+  ``RuntimeError: dictionary changed size during iteration`` and data corruption under concurrent
+  plan execution when the service is registered as a DI Singleton shared across threads. 7 new
+  Behave scenarios verify lock presence, reentrancy, concurrent add/list/remove/effective-set/enforce,
+  mixed operations, and singleton shared-instance safety across up to 20 threads.
+
 - **Cross-actor subgraph cycle detection reads actor_ref field** (#1431): Fixed
   `_detect_subgraph_cycles()`, `_map_node()`, and the `compile_actor()` main loop
   in `src/cleveragents/actor/compiler.py` to read `actor_ref` from the top-level

CONTRIBUTORS.md

@@ -17,7 +17,7 @@ Below are some of the specific details of various contributions.
 * Jeffrey Phillips Freeman has contributed an implementation for the invariant propagation fix (PR #10881 / issue #9131): added `_propagate_invariant_decisions()` to `SubplanService` to propagate all `invariant_enforced` decisions from parent plans to child plan decision trees during subplan spawn, satisfying the specification requirement for invariant propagation across hierarchical plan execution.
 * Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements.
 * HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool.
-* HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) for parallel plan execution.
+* HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) and thread-safe InvariantService (#8209) with ``threading.RLock`` protection for all shared state mutations under concurrent plan execution.
 * HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption.
 * HAL 9000 has contributed the bug-hunt-pool-supervisor non-blocking tracking fix (#7875 / PR #7957): updated step 5 to be best-effort and added rule 9 to prevent the automation-tracking-manager call from blocking the main supervisor loop.
 * Jeffrey Phillips Freeman has contributed the complete AUTO-BUG-POOL to AUTO-BUG-SUP tracking prefix fix across agent-system-specification.md, automation-tracking.md documentation and agent-system-specification.md spec document, replaced with correct `AUTO-BUG-SUP` prefix used by the bug-hunt-pool-supervisor agent (#7875).

features/invariant_service_thread_safety.feature

@@ -0,0 +1,87 @@
+@mock_only
+Feature: InvariantService thread safety (#8209)
+  As a CleverAgents developer
+  I want InvariantService to be safe for concurrent access from multiple threads
+  So that parallel plan execution with a shared singleton service instance does
+  not cause RuntimeError: dictionary changed size during iteration or
+  data corruption in the invariants dict and enforcement records list
+
+  Background:
+    Given a thread-safe invariant service
+
+  # -----------------------------------------------------------------
+  # Basic lock presence
+  # -----------------------------------------------------------------
+
+  Scenario: InvariantService has a reentrant lock attribute
+    Then the invariant service should have a _lock attribute
+    And the _lock should be a threading.RLock
+
+  # -----------------------------------------------------------------
+  # Concurrent add_invariant
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent add from multiple threads does not raise RuntimeError
+    When 10 threads concurrently add 5 invariants each to the service
+    Then no RuntimeError should have been raised during concurrent add
+    And the service should contain at least 1 invariant
+
+  # -----------------------------------------------------------------
+  # Concurrent list (read) operations
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent list from multiple threads does not raise RuntimeError
+    Given 20 invariants pre-added to the service
+    When 10 threads concurrently list invariants from the service
+    Then no exception should have been raised during concurrent list
+
+  Scenario: Concurrent add and list from multiple threads is data-race free
+    When 5 threads concurrently add invariants and 5 threads concurrently list invariants
+    Then no exception should have been raised during concurrent add and list
+
+  # -----------------------------------------------------------------
+  # Concurrent remove operations
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent remove from multiple threads does not raise RuntimeError
+    Given 20 invariants pre-added to the service with known IDs
+    When 5 threads concurrently remove invariants by ID from the service
+    Then no exception should have been raised during concurrent remove
+    And remaining active invariants should be consistent (no orphans)
+
+  # -----------------------------------------------------------------
+  # Concurrent effective-set computation
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent get_effective_invariants does not raise RuntimeError
+    Given 20 mixed-scope invariants pre-added to the service
+    When 10 threads concurrently call get_effective_invariants from the service
+    Then no exception should have been raised during concurrent effective-set computation
+
+  # -----------------------------------------------------------------
+  # Concurrent enforcement record creation
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent enforce_invariants does not raise RuntimeError
+    Given 10 invariants pre-added to the service from different scopes
+    When 5 threads concurrently enforce invariants on the service for the same plan
+    Then no exception should have been raised during concurrent enforcement
+    And at least 1 enforcement record should exist
+
+  # -----------------------------------------------------------------
+  # Combined mixed operations under contention
+  # -----------------------------------------------------------------
+
+  Scenario: Mixed add, list, remove, enforce from multiple threads is safe
+    When 5 threads concurrently add invariants, list invariants, and get effective set
+    Then no RuntimeError should have been raised during mixed concurrent operations
+    And the invariant count after mixed operations should be non-negative
+
+  # -----------------------------------------------------------------
+  # Singleton safety: same instance shared across threads
+  # -----------------------------------------------------------------
+
+  Scenario: Shared singleton instance survives concurrent access from 20 threads
+    When 20 threads each add 3 invariants and then list them back
+    Then no exception should have been raised during concurrent singleton access
+    And the service invariant count should be non-negative

features/steps/invariant_service_thread_safety_steps.py

@@ -0,0 +1,516 @@
+"""Step definitions for InvariantService thread-safety tests (#8209).
+
+Verifies that all public methods of InvariantService are safe for
+concurrent access from multiple threads, preventing:
+- RuntimeError: dictionary changed size during iteration
+- Data corruption in _invariants dict and _enforcement_records list
+- Lost updates under concurrent plan execution when the service is a
+  DI Singleton shared across threads
+
+Based on the pattern established by context_tier_thread_safety_tests (#7547).
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any
+
+from behave import given, then, when
+
+from cleveragents.application.services.invariant_service import InvariantService
+from cleveragents.domain.models.core.invariant import InvariantScope
+
+__all__: list[str] = []
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_invariant(text: str, scope: InvariantScope, source_name: str) -> Any:
+    """Build a test Invariant via the service."""
+    return None  # we add through the service directly instead
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a thread-safe invariant service")
+def step_given_thread_safe_service(context: Any) -> None:
+    context.inv_service = InvariantService()
+    context.errors: list[Exception] = []
+
+
+# ---------------------------------------------------------------------------
+# Lock attribute checks
+# ---------------------------------------------------------------------------
+
+
+@then("the invariant service should have a _lock attribute")
+def step_then_has_lock(context: Any) -> None:
+    assert hasattr(context.inv_service, "_lock"), (
+        "InvariantService must have a _lock attribute"
+    )
+
+
+@then("the _lock should be a threading.RLock")
+def step_then_lock_is_rlock(context: Any) -> None:
+    # threading.RLock() returns an instance of _RLock (internal type).
+    # We verify it has the acquire/release interface of a reentrant lock.
+    lock = context.inv_service._lock
+    assert hasattr(lock, "acquire"), "_lock must have acquire()"
+    assert hasattr(lock, "release"), "_lock must have release()"
+    # Verify reentrancy: acquiring twice from the same thread must not deadlock.
+    lock.acquire()
+    lock.acquire()
+    lock.release()
+    lock.release()
+
+
+# ---------------------------------------------------------------------------
+# Concurrent add_invariant
+# ---------------------------------------------------------------------------
+
+
+@given("{n:d} invariants pre-added to the service with known IDs")
+def step_given_pre_added_invariants(context: Any, n: int) -> None:
+    context.inv_ids = []
+    for i in range(n):
+        inv = context.inv_service.add_invariant(
+            text=f"Pre-added invariant {i}",
+            scope=InvariantScope.GLOBAL,
+            source_name="test-system",
+        )
+        context.inv_ids.append(inv.id)
+
+
+
+
+@given("{n:d} invariants pre-added to the service")
+def step_given_pre_added_invariants_basic(context: Any, n: int) -> None:
+    """Add N invariants with GLOBAL scope and known IDs."""
+    context.inv_ids = []
+    for i in range(n):
+        inv = context.inv_service.add_invariant(
+            text=f"Pre-added invariant {i}",
+            scope=InvariantScope.GLOBAL,
+            source_name="test-system",
+        )
+        context.inv_ids.append(inv.id)
+
+
+
+@given("{n:d} invariants pre-added to the service from different scopes")
+def step_given_pre_added_different_scopes(context: Any, n: int) -> None:
+    """Add N invariants cycling through GLOBAL/PROJECT/PLAN scopes."""
+    for i in range(n):
+        scope = [
+            InvariantScope.GLOBAL,
+            InvariantScope.PROJECT,
+            InvariantScope.PLAN,
+        ][i % 3]
+        context.inv_service.add_invariant(
+            text=f"Enforcement invariant {i}",
+            scope=scope,
+            source_name="test-plan" if scope == InvariantScope.PLAN else f"source-{i}",
+        )
+
+@given("{n:d} mixed-scope invariants pre-added to the service")
+def step_given_mixed_scope_invariants(context: Any, n: int) -> None:
+    for i in range(n):
+        scope = (i % 3) + 1
+        scope_map = {
+            1: InvariantScope.GLOBAL,
+            2: InvariantScope.PROJECT,
+            3: InvariantScope.PLAN,
+        }
+        context.inv_service.add_invariant(
+            text=f"Mixed invariant {i}",
+            scope=scope_map[scope],
+            source_name=f"source-{i}",
+        )
+
+
+@when("{n:d} threads concurrently add {k:d} invariants each to the service")
+def step_when_concurrent_add(context: Any, n: int, k: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def worker(thread_idx: int) -> None:
+        for i in range(k):
+            try:
+                context.inv_service.add_invariant(
+                    text=f"Thread-{thread_idx}-invariant-{i}",
+                    scope=InvariantScope.GLOBAL,
+                    source_name=f"worker-{thread_idx}",
+                )
+            except Exception as exc:
+                with lock:
+                    errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+@then("no RuntimeError should have been raised during concurrent add")
+def step_then_no_runtime_error_add(context: Any) -> None:
+    runtime_errors = [e for e in context.errors if isinstance(e, RuntimeError)]
+    assert not runtime_errors, (
+        f"RuntimeError(s) raised during concurrent add: {runtime_errors}"
+    )
+
+
+@then("the service should contain at least {n:d} invariant")
+def step_then_contains_at_least(context: Any, n: int) -> None:
+    invariants = context.inv_service.list_invariants()
+    count = len(invariants)
+    assert count >= n, f"Expected at least {n} invariants, got {count}"
+
+
+# ---------------------------------------------------------------------------
+# Concurrent list (read) operations
+# ---------------------------------------------------------------------------
+
+
+@when("{n:d} threads concurrently list invariants from the service")
+def step_when_concurrent_list(context: Any, n: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def worker(thread_idx: int) -> None:
+        for _ in range(5):
+            try:
+                context.inv_service.list_invariants()
+            except Exception as exc:
+                with lock:
+                    errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+@then("no exception should have been raised during concurrent list")
+def step_then_no_exc_list(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during concurrent list: {context.errors}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Concurrent add + list
+# ---------------------------------------------------------------------------
+
+
+@when(
+    "{a:d} threads concurrently add invariants and "
+    "{list_threads:d} threads concurrently list invariants"
+)
+def step_when_concurrent_add_and_list(context: Any, a: int, list_threads: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def add_worker(thread_idx: int) -> None:
+        for i in range(3):
+            try:
+                context.inv_service.add_invariant(
+                    text=f"Add-worker-{thread_idx}-inv-{i}",
+                    scope=InvariantScope.GLOBAL,
+                    source_name=f"add-worker-{thread_idx}",
+                )
+            except Exception as exc:
+                with lock:
+                    errors.append(exc)
+
+    def list_worker(thread_idx: int) -> None:
+        for _ in range(5):
+            try:
+                context.inv_service.list_invariants()
+            except Exception as exc:
+                with lock:
+                    errors.append(exc)
+
+    threads = [threading.Thread(target=add_worker, args=(t,)) for t in range(a)]
+    threads += [threading.Thread(target=list_worker, args=(t,)) for t in range(list_threads)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+@then("no exception should have been raised during concurrent add and list")
+def step_then_no_exc_add_list(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during concurrent add+list: {context.errors}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Concurrent remove operations
+# ---------------------------------------------------------------------------
+
+
+@when(
+    "{n:d} threads concurrently remove invariants by ID from the service"
+)
+def step_when_concurrent_remove(context: Any, n: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+    inv_ids = list(context.inv_ids)[:n]  # each thread removes one unique ID
+
+    def worker(thread_idx: int) -> None:
+        if thread_idx < len(inv_ids):
+            try:
+                context.inv_service.remove_invariant(inv_ids[thread_idx])
+            except Exception as exc:
+                with lock:
+                    errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+@then("no exception should have been raised during concurrent remove")
+def step_then_no_exc_remove(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during concurrent remove: {context.errors}"
+    )
+
+
+@then("remaining active invariants should be consistent (no orphans)")
+def step_then_active_invariants_consistent(context: Any) -> None:
+    # Verify that all active invariants' IDs are still in the dict
+    active_ids = set()
+    for inv_id, inv in context.inv_service._invariants.items():
+        if inv.active:
+            active_ids.add(inv_id)
+
+    listed = context.inv_service.list_invariants()
+    listed_ids = {inv.id for inv in listed}
+
+    assert active_ids == listed_ids, (
+        f"Mismatch between _invariants active set ({active_ids}) and "
+        f"list_invariants output ({listed_ids})"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Concurrent effective-set computation
+# ---------------------------------------------------------------------------
+
+
+@when(
+    "{n:d} threads concurrently call get_effective_invariants from the service"
+)
+def step_when_concurrent_effective(context: Any, n: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def worker(thread_idx: int) -> None:
+        for _ in range(5):
+            try:
+                context.inv_service.get_effective_invariants(plan_id="test-plan")
+            except Exception as exc:
+                with lock:
+                    errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+@then(
+    "no exception should have been raised during concurrent effective-set computation"
+)
+def step_then_no_exc_effective(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during concurrent get_effective_invariants: {context.errors}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Concurrent enforcement record creation
+# ---------------------------------------------------------------------------
+
+
+@when(
+    "{n:d} threads concurrently enforce invariants on the service for the same plan"
+)
+def step_when_concurrent_enforce(context: Any, n: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def worker(thread_idx: int) -> None:
+        # Build a small set of invariants for this thread
+        plan_inv = context.inv_service.add_invariant(
+            text=f"Plan invariant enforced by thread {thread_idx}",
+            scope=InvariantScope.PLAN,
+            source_name="test-plan",
+        )
+        try:
+            context.inv_service.enforce_invariants(
+                plan_id="test-plan",
+                invariants=[plan_inv],
+            )
+        except Exception as exc:
+            with lock:
+                errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+@then("no exception should have been raised during concurrent enforcement")
+def step_then_no_exc_enforce(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during concurrent enforce_invariants: {context.errors}"
+    )
+
+
+@then("at least {n:d} enforcement record should exist")
+def step_then_enforcement_record_exists(context: Any, n: int) -> None:
+    with context.inv_service._lock:
+        count = len(context.inv_service._enforcement_records)
+    assert count >= n, f"Expected at least {n} enforcement records, got {count}"
+
+
+# ---------------------------------------------------------------------------
+# Combined mixed operations under contention
+# ---------------------------------------------------------------------------
+
+
+@when(
+    "{n:d} threads concurrently add invariants, list invariants, and get effective set"
+)
+def step_when_concurrent_mixed(context: Any, n: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def worker(thread_idx: int) -> None:
+        for i in range(3):
+            try:
+                context.inv_service.add_invariant(
+                    text=f"Mixed-thread-{thread_idx}-inv-{i}",
+                    scope=[InvariantScope.GLOBAL, InvariantScope.PROJECT, InvariantScope.PLAN][thread_idx % 3],
+                    source_name=f"mixed-worker-{thread_idx}-{i}",
+                )
+            except Exception:
+                pass
+        try:
+            context.inv_service.list_invariants()
+        except Exception as exc:
+            with lock:
+                errors.append(exc)
+        try:
+            context.inv_service.get_effective_invariants()
+        except Exception as exc:
+            with lock:
+                errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+@then("no RuntimeError should have been raised during mixed concurrent operations")
+def step_then_no_runtime_error_mixed(context: Any) -> None:
+    runtime_errors = [e for e in context.errors if isinstance(e, RuntimeError)]
+    assert not runtime_errors, (
+        f"RuntimeError(s) raised during mixed concurrent ops: {runtime_errors}"
+    )
+
+
+@then("the invariant count after mixed operations should be non-negative")
+def step_then_inv_count_non_negative(context: Any) -> None:
+    invariants = context.inv_service.list_invariants()
+    assert len(invariants) >= 0, "Invariant count should never be negative"
+
+
+# ---------------------------------------------------------------------------
+# Singleton safety: same instance shared across threads
+# ---------------------------------------------------------------------------
+
+
+@when("{n:d} threads each add {k:d} invariants and then list them back")
+def step_when_concurrent_singleton(context: Any, n: int, k: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def worker(thread_idx: int) -> None:
+        for i in range(k):
+            try:
+                context.inv_service.add_invariant(
+                    text=f"Singleton-thread-{thread_idx}-inv-{i}",
+                    scope=InvariantScope.GLOBAL,
+                    source_name=f"singleton-worker-{thread_idx}",
+                )
+                # List all to verify consistency
+                _ = context.inv_service.list_invariants()
+            except Exception as exc:
+                with lock:
+                    errors.append(exc)
+
+    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+
+
+@then("the service invariant count should be non-negative")
+def step_then_inv_count_non_negative_simple(context: Any) -> None:
+    invariants = context.inv_service.list_invariants()
+    assert len(invariants) >= 0, "Invariant count should never be negative"
+
+@then("no exception should have been raised during concurrent singleton access")
+def step_then_no_exc_singleton(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during concurrent singleton access: {context.errors}"
+    )
+
+
+@then(
+    "the service invariant count should be non-negative and the enforcement record count should be non-negative"
+)
+def step_then_both_counts_non_negative(context: Any) -> None:
+    invariants = context.inv_service.list_invariants()
+    assert len(invariants) >= 0, f"Expected non-negative count, got {len(invariants)}"
+
+    with context.inv_service._lock:
+        record_count = len(context.inv_service._enforcement_records)
+    assert record_count >= 0, f"Expected non-negative records, got {record_count}"

src/cleveragents/application/services/invariant_service.py

@@ -9,6 +9,17 @@ lifecycle operations.
 Uses in-memory storage (same pattern as ``PlanLifecycleService``) with
 a dict keyed by invariant ID.
 
+## Thread Safety
+
+A ``threading.RLock`` (re-entrant lock) guards all shared mutable state
+(``_invariants``, ``_enforcement_records``).  Every public method acquires
+the lock before reading or writing shared data; the lock is re-entrant so
+that nested calls from within a locked block do **not** deadlock.
+
+This prevents ``RuntimeError: dictionary changed size during iteration`` and
+data corruption under concurrent plan execution when the service is
+registered as a DI Singleton shared across threads.
+
 ## Merge Precedence
 
 Effective invariants are computed using plan > project > global order.
@@ -19,6 +30,7 @@ Based on ``docs/specification.md`` and implementation plan Stage M3.5.
 
 from __future__ import annotations
 
+from threading import RLock
 from typing import TYPE_CHECKING
 
 import structlog
@@ -46,6 +58,10 @@ class InvariantService:
 
     Provides add, list, remove (soft-delete), effective-set computation,
     and enforcement record creation.  All storage is in-memory.
+
+    Thread safety is provided via a ``threading.RLock`` that protects all
+    shared state mutations so that concurrent readers and writers cannot
+    race or raise ``RuntimeError: dictionary changed size during iteration``.
     """
 
     def __init__(self, event_bus: EventBus | None = None) -> None:
@@ -56,6 +72,7 @@ class InvariantService:
         """
         self._invariants: dict[str, Invariant] = {}
         self._enforcement_records: list[InvariantEnforcementRecord] = []
+        self._lock = RLock()  # guards _invariants and _enforcement_records
         self._logger = logger.bind(service="invariant")
         self._sanitizer = PromptSanitizer()
         self._event_bus = event_bus
@@ -95,7 +112,9 @@ class InvariantService:
             source_name=source_name.strip(),
         )
 
-        self._invariants[invariant.id] = invariant
+        with self._lock:
+            self._invariants[invariant.id] = invariant
+
         self._logger.info(
             "Invariant added",
             invariant_id=invariant.id,
@@ -127,7 +146,8 @@ class InvariantService:
                 project_name=source_name if scope == InvariantScope.PROJECT else None,
             )
 
-        result = [inv for inv in self._invariants.values() if inv.active]
+        with self._lock:
+            result = [inv for inv in self._invariants.values() if inv.active]
 
         if scope is not None:
             result = [inv for inv in result if inv.scope == scope]
@@ -152,15 +172,17 @@ class InvariantService:
         if not invariant_id or not invariant_id.strip():
             raise ValidationError("Invariant ID must not be empty")
 
-        inv = self._invariants.get(invariant_id)
-        if inv is None:
-            raise NotFoundError(
-                resource_type="invariant",
-                resource_id=invariant_id,
-            )
-        # Invariant is frozen (immutable); create a new instance with active=False
-        deactivated = inv.model_copy(update={"active": False})
-        self._invariants[invariant_id] = deactivated
+        with self._lock:
+            inv = self._invariants.get(invariant_id)
+            if inv is None:
+                raise NotFoundError(
+                    resource_type="invariant",
+                    resource_id=invariant_id,
+                )
+            # Invariant is frozen (immutable); create a new instance with active=False
+            deactivated = inv.model_copy(update={"active": False})
+            self._invariants[invariant_id] = deactivated
+
         self._logger.info("Invariant removed (soft-delete)", invariant_id=invariant_id)
         return deactivated
 
@@ -183,7 +205,8 @@ class InvariantService:
         Returns:
             Merged, de-duplicated list of effective invariants.
         """
-        active = [inv for inv in self._invariants.values() if inv.active]
+        with self._lock:
+            active = [inv for inv in self._invariants.values() if inv.active]
 
         plan_invs = [
             inv
@@ -264,7 +287,9 @@ class InvariantService:
                         exc_info=True,
                     )
 
-        self._enforcement_records.extend(records)
+        with self._lock:
+            self._enforcement_records.extend(records)
+
         self._logger.info(
             "Invariants enforced",
             plan_id=plan_id,