CHANGELOG.md

@@ -234,6 +234,16 @@ ensuring data is stored with proper parameter values.
   under `.opencode/` are covered by the lint gate.
 
 ### Fixed
+- **InvariantService thread safety** (#7524): Added `threading.RLock` to protect
+  `_invariants` (dict) and `_enforcement_records` (list) from concurrent access
+  during parallel plan execution. Prevents `RuntimeError: dictionary changed size
+  during iteration` and data corruption when multiple subplans simultaneously
+  call `add_invariant()`, `list_invariants()`, `remove_invariant()`,
+  `get_effective_invariants()`, or `enforce_invariants()`. Also adds three
+  thread-safe helper read methods: `get_enforcement_records()`, `get_invariant()`,
+  and `get_invariants_snapshot()`. Includes comprehensive BDD tests for concurrent
+  access patterns (adds, lists, removes, enforcement, and mixed operations).
+
 - **ACMS execute-phase assembler respects project-level hot_max_tokens** (#11035): Fixed
   ``_resolve_hot_max_tokens()`` to read ``hot_max_tokens`` from
   ``context_policy_json["acms_config"]["hot_max_tokens"]`` — the correct sub-key written

CONTRIBUTORS.md

@@ -38,6 +38,7 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed the plan artifacts JSON completeness fix (#9084): ensured `validation_summary` and `apply_summary` are correctly included in `_build_artifacts_dict`, removing stale `@tdd_expected_fail` tags from Behave scenarios to enable full regression test coverage.
 * 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-SUP tracking prefix fix across agent-system-specification.md, automation-tracking.md documentation, replacing the incorrect `AUTO-BUG-POL` prefix with the correct `AUTO-BUG-SUP` prefix used by the bug-hunt-pool-supervisor agent (#7875).
+* HAL 9000 has contributed thread safety to InvariantService (issue #7524): added `threading.RLock` protection for all shared mutable state (_invariants dict, _enforcement_records list) across add, list, remove, effective-set computation, and enforcement operations, preventing RuntimeError: dictionary changed size during iteration in multi-threaded parallel plan execution environments.
 * HAL 9000 has contributed the plugin entry point security hardening fix (#7476): enforced entry point allowlist validation before importing plugin modules to prevent malicious plugin loading.
 * HAL 9000 has contributed the benchmark workflow separation (#9040): moved the benchmark-regression job out of the default PR workflow into a dedicated scheduled workflow, reducing median PR CI turnaround time from 99-132 minutes to under 30 minutes.
 * HAL 9000 has contributed the plan tree JSON/YAML command envelope fix (#9163): wrapped `agents plan tree --format json/yaml` output in the spec-required command envelope structure, added summary statistics, decision_ids mapping, child_plans list, and accurate timing measurement.

features/invariant_service_thread_safety.feature

@@ -0,0 +1,54 @@
+@mock_only
+Feature: InvariantService thread safety (Issue #7524)
+  As a CleverAgents developer
+  I want InvariantService to be safe for concurrent plan execution
+  So that parallel subplans sharing the same singleton instance do not
+  cause RuntimeError: dictionary changed size during iteration or
+  data corruption in _invariants / _enforcement_records
+
+  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 invariant service _lock should be a threading.RLock
+
+  # -----------------------------------------------------------------
+  # Concurrent adds
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent add_invariant from multiple threads does not raise RuntimeError
+    When 15 threads concurrently add 4 invariants each to the invariant service
+    Then no RuntimeError should have been raised during concurrent invariant creation
+    And the invariant service should contain at least 40 active invariants
+
+  # -----------------------------------------------------------------
+  # Concurrent lists
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent list_invariants from multiple threads is data-race free
+    Given 15 invariants pre-stored in the invariant service
+    When 8 threads concurrently call list_invariants on the service
+    Then no exception should have been raised during concurrent listing
+
+  # -----------------------------------------------------------------
+  # Mixed access (add + list)
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent add and list from multiple threads is data-race free
+    When 6 threads concurrently add invariants and 4 threads concurrently list invariants
+    Then no exception should have been raised during mixed concurrent access
+
+  # -----------------------------------------------------------------
+  # Mixed enforcement + listing
+  # -----------------------------------------------------------------
+
+  Scenario: Concurrent enforce_invariant calls do not corrupt enforcement records
+    Given 8 invariants pre-stored in the invariant service with plan "PLAN_THREAD_TEST"
+    When 4 threads concurrently enforce all stored invariants each from plan "PLAN_THREAD_TEST"
+    Then no exception should have been raised during concurrent enforcement
+    And the invariant service should contain at least 28 enforcement records

features/steps/invariant_service_thread_safety_steps.py

@@ -0,0 +1,283 @@
+"""Step definitions for invariant_service_thread_safety.feature (Issue #7524).
+
+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 / _enforcement_records
+- Lost updates or phantom reads under concurrent plan execution
+"""
+
+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(thread_idx: int, idx: int) -> str:
+    """Generate a deterministic invariant text for a given worker."""
+    return f"Thread-{thread_idx}-invariant-{idx} (concurrency guard #{thread_idx * 100 + idx})"
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a thread-safe invariant service")
+def step_given_thread_safe_service(context: Any) -> None:
+    context.invariant_svc = InvariantService()
+    context.errors: list[Exception] = []
+    context.pre_stored_ids: list[str] = []
+
+
+# ---------------------------------------------------------------------------
+# Lock attribute checks
+# ---------------------------------------------------------------------------
+
+
+@then("the invariant service should have a _lock attribute")
+def step_then_has_lock(context: Any) -> None:
+    assert hasattr(context.invariant_svc, "_lock"), (
+        "InvariantService must have a _lock attribute"
+    )
+
+
+@then("the invariant service _lock should be a threading.RLock")
+def step_then_lock_is_rlock(context: Any) -> None:
+    lock = context.invariant_svc._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()
+
+
+# ---------------------------------------------------------------------------
+# Scenario 2: Concurrent adds
+# ---------------------------------------------------------------------------
+
+
+@when("{n:d} threads concurrently add {k:d} invariants each to the invariant service")
+def step_when_concurrent_adds(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.invariant_svc.add_invariant(
+                    text=_make_invariant_text(thread_idx, i),
+                    scope=InvariantScope.GLOBAL,
+                    source_name="concurrent-test",
+                )
+            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 invariant creation")
+def step_then_no_runtime_error(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 adds: {runtime_errors}"
+    )
+
+
+@then("the invariant service should contain at least {n:d} active invariants")
+def step_then_at_least_invariants(context: Any, n: int) -> None:
+    snapshot = context.invariant_svc.get_invariants_snapshot()
+    assert len(snapshot) >= n, (
+        f"Expected at least {n} active invariants, got {len(snapshot)}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Scenario 3: Concurrent lists
+# ---------------------------------------------------------------------------
+
+
+@given("{n:d} invariants pre-stored in the invariant service")
+def step_given_pre_stored_invariants(context: Any, n: int) -> None:
+    context.pre_stored_ids = []
+    for i in range(n):
+        inv = context.invariant_svc.add_invariant(
+            text=f"pre-stored-invariant-{i}",
+            scope=InvariantScope.GLOBAL,
+            source_name="prestore",
+        )
+        context.pre_stored_ids.append(inv.id)
+
+
+@when("{n:d} threads concurrently call list_invariants on the service")
+def step_when_concurrent_lists(context: Any, n: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def worker(thread_idx: int) -> None:
+        for _ in range(10):
+            try:
+                results = context.invariant_svc.list_invariants()
+                _ = len(results)  # ensure we actually iterate the list
+            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 listing")
+def step_then_no_exc_list(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during concurrent list: {context.errors}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Scenario 4: Mixed access (add + list)
+# ---------------------------------------------------------------------------
+
+
+@when(
+    "{a:d} threads concurrently add invariants and "
+    "{num_listers:d} threads concurrently list invariants"
+)
+def step_when_mixed_access(context: Any, a: int, num_listers: int) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def add_worker(thread_idx: int) -> None:
+        for i in range(5):
+            try:
+                context.invariant_svc.add_invariant(
+                    text=f"mixed-add-t{thread_idx}-{i}",
+                    scope=InvariantScope.PROJECT,
+                    source_name="mixed-test",
+                )
+            except Exception as exc:
+                with lock:
+                    errors.append(exc)
+
+    def list_worker(thread_idx: int) -> None:
+        for _ in range(10):
+            try:
+                results = context.invariant_svc.list_invariants()
+                _ = len(results)
+            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(num_listers)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    context.errors = errors
+
+
+@then("no exception should have been raised during mixed concurrent access")
+def step_then_no_exc_mixed(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during mixed concurrent access: {context.errors}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Scenario 5: Enforcement + listing (uses pre-stored invariants)
+# ---------------------------------------------------------------------------
+
+
+@given('{n:d} invariants pre-stored in the invariant service with plan "{plan_id}"')
+def step_given_pre_stored_with_plan(context: Any, n: int, plan_id: str) -> None:
+    context.pre_stored_ids = []
+    for i in range(n):
+        inv = context.invariant_svc.add_invariant(
+            text=f"enforce-test-invariant-{i}",
+            scope=InvariantScope.PLAN,
+            source_name=plan_id,
+        )
+        context.pre_stored_ids.append(inv.id)
+
+
+@when(
+    '{n:d} threads concurrently enforce all stored invariants each from plan "{plan_id}"'
+)
+def step_when_concurrent_enforcement(context: Any, n: int, plan_id: str) -> None:
+    errors: list[Exception] = []
+    lock = threading.Lock()
+
+    def worker(thread_idx: int) -> None:
+        for _ in range(3):
+            try:
+                # Take a snapshot of active invariants at this moment
+                snapshot = context.invariant_svc.get_invariants_snapshot()
+                # Filter to the target plan scope
+                filtered = [
+                    inv
+                    for inv in snapshot
+                    if inv.scope == InvariantScope.PLAN and inv.source_name == plan_id
+                ]
+                if not filtered:
+                    continue
+                context.invariant_svc.enforce_invariants(
+                    plan_id=plan_id,
+                    invariants=filtered,
+                    actor_response=f"concurrent-enforcer-t{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 enforcement")
+def step_then_no_exc_enforce(context: Any) -> None:
+    assert not context.errors, (
+        f"Exception(s) raised during concurrent enforcement: {context.errors}"
+    )
+
+
+@then("the invariant service should contain at least {n:d} enforcement records")
+def step_then_at_least_records(context: Any, n: int) -> None:
+    records = context.invariant_svc.get_enforcement_records()
+    assert len(records) >= n, (
+        f"Expected at least {n} enforcement records, got {len(records)}"
+    )

src/cleveragents/application/services/invariant_service.py

@@ -14,11 +14,20 @@ a dict keyed by invariant ID.
 Effective invariants are computed using plan > action > project > global
 order.  See ``merge_invariants`` for de-duplication semantics.
 
+## Thread Safety
+
+All public methods are protected by a reentrant lock
+(``threading.RLock``). This allows the same thread to call multiple
+public methods without deadlocking, and prevents race conditions when
+multiple threads access shared state (``_invariants`` dict and
+``_enforcement_records`` list) during parallel plan execution.
+
 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
@@ -48,8 +57,23 @@ logger = structlog.get_logger(__name__)
 class InvariantService:
     """Service for managing invariant constraints.
 
-    Provides add, list, remove (soft-delete), effective-set computation,
-    and enforcement record creation.  All storage is in-memory.
+    All storage is in-memory and thread-safe via a reentrant lock
+    (``threading.RLock``).  Multiple threads may call public methods
+    concurrently without risk of ``RuntimeError: dictionary changed size
+    during iteration`` or data corruption.
+
+    Publicly synchronised methods
+    -----------------------------------
+    * ``add_invariant``, ``list_invariants``, ``remove_invariant``
+      (lifecycle operations)
+    * ``get_effective_invariants`` (merged precedence chain)
+    * ``enforce_invariants`` (creates enforcement records and emits events)
+
+    Helper read methods
+    -----------------------------------
+    Additional thread-safe read accessors are provided for scenarios that
+    need to inspect internal state without triggering a lifecycle operation:
+    * ``get_enforcement_records``, ``get_invariant``, ``get_invariants_snapshot``
     """
 
     def __init__(self, event_bus: EventBus | None = None) -> None:
@@ -58,6 +82,7 @@ class InvariantService:
         Args:
             event_bus: Optional EventBus for domain event emission.
         """
+        self._lock = RLock()
         self._invariants: dict[str, Invariant] = {}
         self._enforcement_records: list[InvariantEnforcementRecord] = []
         self._logger = logger.bind(service="invariant")
@@ -72,6 +97,9 @@ class InvariantService:
     ) -> Invariant:
         """Add a new invariant with validation.
 
+        Thread-safe: acquires the internal reentrant lock before accessing
+        shared state.
+
         Args:
             text: The natural-language constraint text.
             scope: The scope at which this invariant applies.
@@ -99,13 +127,14 @@ class InvariantService:
             source_name=source_name.strip(),
         )
 
-        self._invariants[invariant.id] = invariant
-        self._logger.info(
-            "Invariant added",
-            invariant_id=invariant.id,
-            scope=scope.value,
-            source_name=source_name,
-        )
+        with self._lock:
+            self._invariants[invariant.id] = invariant
+            self._logger.info(
+                "Invariant added",
+                invariant_id=invariant.id,
+                scope=scope.value,
+                source_name=source_name,
+            )
         return invariant
 
     def list_invariants(
@@ -116,6 +145,11 @@ class InvariantService:
     ) -> list[Invariant]:
         """Filter and list invariants.
 
+        Thread-safe: acquires the internal reentrant lock before accessing
+        shared state.  When ``effective=True`` calls are delegated through
+        ``get_effective_invariants()`` which also locks -- RLock makes this
+        path deadlock-free (re-entrant acquisition).
+
         Args:
             scope: Filter by scope (None = all scopes).
             source_name: Filter by source name (None = all sources).
@@ -132,19 +166,23 @@ 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]
+            if scope is not None:
+                result = [inv for inv in result if inv.scope == scope]
 
-        if source_name is not None:
-            result = [inv for inv in result if inv.source_name == source_name]
+            if source_name is not None:
+                result = [inv for inv in result if inv.source_name == source_name]
 
         return result
 
     def remove_invariant(self, invariant_id: str) -> Invariant:
         """Soft-delete an invariant by setting active=False.
 
+        Thread-safe: acquires the internal reentrant lock before accessing
+        shared state.
+
         Args:
             invariant_id: The ULID of the invariant to remove.
 
@@ -157,16 +195,19 @@ 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,
+        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
             )
-        # 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
 
     def get_effective_invariants(
@@ -180,6 +221,11 @@ class InvariantService:
         Collects active invariants from each scope tier and merges them
         using plan > action > project > global precedence.
 
+        Thread-safe: acquires the internal reentrant lock before accessing
+        shared state.  If called from another locked public method (e.g.
+        ``list_invariants`` with ``effective=True``), RLock handles the
+        re-entrant acquisition so no deadlock occurs.
+
         Args:
             plan_id: Optional plan identifier to collect plan-scoped
                 invariants.
@@ -194,28 +240,28 @@ class InvariantService:
         Returns:
             Merged, de-duplicated list of effective invariants.
         """
-        active = [inv for inv in self._invariants.values() if inv.active]
-
-        plan_invs = [
-            inv
-            for inv in active
-            if inv.scope == InvariantScope.PLAN
-            and (plan_id is None or inv.source_name == plan_id)
-        ]
-        action_invs = [
-            inv
-            for inv in active
-            if inv.scope == InvariantScope.ACTION
-            and action_name is not None  # Only include when explicitly requested
-            and (inv.source_name == action_name or action_name == "*")
-        ]
-        project_invs = [
-            inv
-            for inv in active
-            if inv.scope == InvariantScope.PROJECT
-            and (project_name is None or inv.source_name == project_name)
-        ]
-        global_invs = [inv for inv in active if inv.scope == InvariantScope.GLOBAL]
+        with self._lock:
+            active = [inv for inv in self._invariants.values() if inv.active]
+            plan_invs = [
+                inv
+                for inv in active
+                if inv.scope == InvariantScope.PLAN
+                and (plan_id is None or inv.source_name == plan_id)
+            ]
+            action_invs = [
+                inv
+                for inv in active
+                if inv.scope == InvariantScope.ACTION
+                and action_name is not None  # Only include when explicitly requested
+                and (inv.source_name == action_name or action_name == "*")
+            ]
+            project_invs = [
+                inv
+                for inv in active
+                if inv.scope == InvariantScope.PROJECT
+                and (project_name is None or inv.source_name == project_name)
+            ]
+            global_invs = [inv for inv in active if inv.scope == InvariantScope.GLOBAL]
 
         return merge_invariants(plan_invs, action_invs, project_invs, global_invs)
 
@@ -416,6 +462,10 @@ class InvariantService:
         Called by the Invariant Reconciliation Actor at the start of
         Strategize to record ``invariant_enforced`` decisions.
 
+        Thread-safe: acquires the internal reentrant lock before accessing
+        shared state (both ``_enforcement_records`` list and ``_invariants``
+        dict for event emission lookups).
+
         Args:
             plan_id: The plan being checked.
             invariants: The invariants to enforce.
@@ -435,44 +485,27 @@ class InvariantService:
             raise ValidationError("Plan ID must not be empty")
 
         violated_ids = set(violated_invariant_ids or [])
-        records: list[InvariantEnforcementRecord] = []
         response = actor_response or ""
-        for inv in invariants:
-            enforced = inv.id not in violated_ids
-            record = InvariantEnforcementRecord(
-                invariant_id=inv.id,
-                enforced=enforced,
-                actor_response=response,
-                decision_id=str(ULID()),
-            )
-            records.append(record)
-            if not enforced and self._event_bus is not None:
-                try:
-                    self._event_bus.emit(
-                        DomainEvent(
-                            event_type=EventType.INVARIANT_VIOLATED,
-                            plan_id=plan_id,
-                            details={
-                                "invariant_id": inv.id,
-                                "invariant_text": inv.text,
-                                "scope": inv.scope.value,
-                            },
-                        )
-                    )
-                except Exception:
-                    self._logger.warning(
-                        "event_bus_emit_failed",
-                        event_type="INVARIANT_VIOLATED",
-                        plan_id=plan_id,
-                        exc_info=True,
-                    )
 
-        self._enforcement_records.extend(records)
-        self._logger.info(
-            "Invariants enforced",
-            plan_id=plan_id,
-            count=len(records),
-        )
+        with self._lock:
+            records: list[InvariantEnforcementRecord] = []
+            for inv in invariants:
+                enforced = inv.id not in violated_ids
+                record = InvariantEnforcementRecord(
+                    invariant_id=inv.id,
+                    enforced=enforced,
+                    actor_response=response,
+                    decision_id=str(ULID()),
+                )
+                records.append(record)
+
+            self._enforcement_records.extend(records)
+            self._logger.info(
+                "Invariants enforced",
+                plan_id=plan_id,
+                count=len(records),
+            )
+
         if self._event_bus is not None:
             for record in records:
                 try:
@@ -494,7 +527,33 @@ class InvariantService:
                         plan_id=plan_id,
                         exc_info=True,
                     )
-            # Emit a single INVARIANT_RECONCILED for the batch
+
+            # Emit INVARIANT_VIOLATED events outside the lock to reduce
+            # hold time on _lock (event bus I/O should not block
+            # concurrent invariants access).
+            for inv in invariants:
+                if inv.id in violated_ids:
+                    try:
+                        self._event_bus.emit(
+                            DomainEvent(
+                                event_type=EventType.INVARIANT_VIOLATED,
+                                plan_id=plan_id,
+                                details={
+                                    "invariant_id": inv.id,
+                                    "invariant_text": inv.text,
+                                    "scope": inv.scope.value,
+                                },
+                            )
+                        )
+                    except Exception:
+                        self._logger.warning(
+                            "event_bus_emit_failed",
+                            event_type="INVARIANT_VIOLATED",
+                            plan_id=plan_id,
+                            exc_info=True,
+                        )
+
+            # Emit a single INVARIANT_RECONCILED for the batch outside lock
             try:
                 self._event_bus.emit(
                     DomainEvent(
@@ -512,4 +571,52 @@ class InvariantService:
                     plan_id=plan_id,
                     exc_info=True,
                 )
+
         return records
+
+    # ------------------------------------------------------------------ #
+    # Thread-safe helper read accessors                                    #
+    # ------------------------------------------------------------------ #
+
+    def get_enforcement_records(self) -> list[InvariantEnforcementRecord]:
+        """Return a snapshot of all enforcement records.
+
+        Thread-safe: acquires the internal reentrant lock to build copies of
+        each record so callers receive a consistent, independent snapshot.
+
+        Returns:
+            A list of ``InvariantEnforcementRecord`` objects copied from
+            the internal registry.
+        """
+        with self._lock:
+            return [record.model_copy() for record in self._enforcement_records]
+
+    def get_invariant(self, invariant_id: str) -> Invariant | None:
+        """Return a single invariant by ID, or ``None`` if absent.
+
+        Thread-safe: acquires the internal reentrant lock before lookup.
+
+        Args:
+            invariant_id: The ULID of the invariant to retrieve.
+
+        Returns:
+            The ``Invariant`` if found, otherwise ``None``.
+        """
+        with self._lock:
+            inv = self._invariants.get(invariant_id)
+            if inv is not None:
+                return inv.model_copy()
+            return None
+
+    def get_invariants_snapshot(self) -> list[Invariant]:
+        """Return a snapshot of all active invariants currently registered.
+
+        Thread-safe: acquires the internal reentrant lock to build a
+        consistent copy of the active set at the moment of invocation.
+
+        Returns:
+            A list of ``Invariant`` objects (copies) that are currently
+            active (``active=True``).
+        """
+        with self._lock:
+            return [inv.model_copy() for inv in self._invariants.values() if inv.active]

tests/application/services/test_invariant_service_thread_safety.py

@@ -0,0 +1,119 @@
+"""Thread-safety regressions for InvariantService."""
+
+from __future__ import annotations
+
+import threading
+from collections.abc import Callable
+
+from cleveragents.application.services.invariant_service import InvariantService
+from cleveragents.domain.models.core.invariant import InvariantScope
+
+
+def _run_threads(*workers: Callable[[], None]) -> list[BaseException]:
+    errors: list[BaseException] = []
+    errors_lock = threading.Lock()
+    start = threading.Barrier(len(workers))
+
+    def guarded(worker: Callable[[], None]) -> None:
+        try:
+            start.wait(timeout=5)
+            worker()
+        except BaseException as exc:
+            with errors_lock:
+                errors.append(exc)
+
+    threads = [threading.Thread(target=guarded, args=(worker,)) for worker in workers]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=10)
+        if thread.is_alive():
+            errors.append(TimeoutError(f"{thread.name} did not finish"))
+
+    return errors
+
+
+def test_add_list_and_remove_are_safe_under_concurrent_access() -> None:
+    service = InvariantService()
+    initial = [
+        service.add_invariant(
+            text=f"initial invariant {index}",
+            scope=InvariantScope.GLOBAL,
+            source_name="thread-test",
+        )
+        for index in range(20)
+    ]
+
+    def add_worker(worker_index: int) -> Callable[[], None]:
+        def run() -> None:
+            for index in range(40):
+                service.add_invariant(
+                    text=f"worker {worker_index} invariant {index}",
+                    scope=InvariantScope.PROJECT,
+                    source_name="thread-test",
+                )
+
+        return run
+
+    def list_worker() -> None:
+        for _ in range(200):
+            assert all(inv.active for inv in service.list_invariants())
+            assert len(service.get_invariants_snapshot()) >= 1
+
+    def remove_worker() -> None:
+        for inv in initial:
+            service.remove_invariant(inv.id)
+
+    errors = _run_threads(
+        add_worker(1),
+        add_worker(2),
+        add_worker(3),
+        list_worker,
+        list_worker,
+        remove_worker,
+    )
+
+    assert errors == []
+    assert len(service.list_invariants(scope=InvariantScope.PROJECT)) == 120
+    assert len(service.list_invariants(scope=InvariantScope.GLOBAL)) == 0
+
+
+def test_enforce_invariants_is_safe_under_concurrent_access() -> None:
+    service = InvariantService()
+    plan_id = "PLAN_THREAD_TEST"
+    invariants = [
+        service.add_invariant(
+            text=f"plan invariant {index}",
+            scope=InvariantScope.PLAN,
+            source_name=plan_id,
+        )
+        for index in range(12)
+    ]
+
+    def enforce_worker(worker_index: int) -> Callable[[], None]:
+        def run() -> None:
+            for _ in range(25):
+                service.enforce_invariants(
+                    plan_id=plan_id,
+                    invariants=invariants,
+                    actor_response=f"worker-{worker_index}",
+                )
+
+        return run
+
+    def records_reader() -> None:
+        for _ in range(200):
+            records = service.get_enforcement_records()
+            assert len(records) % len(invariants) == 0
+
+    errors = _run_threads(
+        enforce_worker(1),
+        enforce_worker(2),
+        enforce_worker(3),
+        enforce_worker(4),
+        records_reader,
+        records_reader,
+    )
+
+    assert errors == []
+    assert len(service.get_enforcement_records()) == 4 * 25 * len(invariants)