fix(concurrency): add thread safety to InvariantService

Added threading.RLock() to protect _invariants dict and _enforcement_records list
All public methods (add_invariant, remove_invariant, list_invariants, get_effective_invariants, enforce_invariants) now use lock protection
Prevents RuntimeError: dictionary changed size during iteration
Prevents data corruption from concurrent access
Follows existing pattern from AutonomyController and AutonomyGuardrailService

ISSUES CLOSED: #7524

features/invariant_service_thread_safety.feature

@@ -0,0 +1,93 @@
+Feature: InvariantService Thread Safety
+  As a developer
+  I want InvariantService to be thread-safe
+  So that concurrent access from parallel plan execution doesn't cause data corruption
+
+  Background:
+    Given I have an InvariantService instance
+
+  # ================================================================
+  # Concurrent Add Operations
+  # ================================================================
+
+  Scenario: Multiple threads can safely add invariants concurrently
+    When 5 threads concurrently add invariants with unique texts
+    Then all invariants should be stored without data loss
+    And the total count should be 5
+
+  Scenario: Concurrent adds don't cause dictionary changed size errors
+    When 10 threads concurrently add invariants
+    Then no RuntimeError should be raised
+    And all invariants should be retrievable
+
+  # ================================================================
+  # Concurrent List Operations
+  # ================================================================
+
+  Scenario: Multiple threads can safely list invariants concurrently
+    Given I have added 5 invariants
+    When 5 threads concurrently list invariants
+    Then all threads should complete successfully
+    And each thread should see consistent data
+
+  Scenario: Listing while adding doesn't cause iteration errors
+    When 3 threads add invariants while 3 threads list invariants concurrently
+    Then no RuntimeError should be raised
+    And the final count should match the number of adds
+
+  # ================================================================
+  # Concurrent Remove Operations
+  # ================================================================
+
+  Scenario: Multiple threads can safely remove invariants concurrently
+    Given I have added 5 invariants
+    When 3 threads concurrently remove different invariants
+    Then all removals should succeed
+    And the remaining count should be 2
+
+  Scenario: Removing while listing doesn't cause errors
+    Given I have added 5 invariants
+    When 2 threads remove invariants while 2 threads list invariants concurrently
+    Then no RuntimeError should be raised
+    And the final state should be consistent
+
+  # ================================================================
+  # Concurrent Enforcement Operations
+  # ================================================================
+
+  Scenario: Multiple threads can safely enforce invariants concurrently
+    Given I have added 3 invariants
+    When 5 threads concurrently enforce the same invariants
+    Then all enforcement records should be stored
+    And the total record count should be 15
+
+  Scenario: Enforcement records are not lost under concurrent access
+    Given I have added 2 invariants
+    When 5 threads concurrently enforce invariants
+    Then all 10 enforcement records should be stored
+    And no records should be duplicated or lost
+
+  # ================================================================
+  # Mixed Concurrent Operations
+  # ================================================================
+
+  Scenario: Mixed operations (add, list, remove, enforce) are thread-safe
+    When 2 threads add, 2 threads list, 2 threads remove, and 2 threads enforce concurrently
+    Then no errors should occur in thread safety test
+    And the final state should be consistent
+
+  # ================================================================
+  # Concurrent Effective Invariants
+  # ================================================================
+
+  Scenario: Getting effective invariants is thread-safe
+    Given I have added invariants at different scopes
+    When 5 threads concurrently get effective invariants
+    Then all threads should complete successfully
+    And all threads should see consistent merged results
+
+  Scenario: Effective invariants computation under concurrent modifications
+    Given I have added invariants at different scopes
+    When 3 threads add invariants while 3 threads get effective invariants concurrently
+    Then no errors should occur in thread safety test
+    And the results should be consistent

features/steps/invariant_service_thread_safety_steps.py

@@ -0,0 +1,830 @@
+"""Step definitions for invariant_service_thread_safety.feature.
+
+Tests thread safety of InvariantService under concurrent access.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from typing import Any
+
+from behave import given, then, when  # type: ignore[import-untyped]
+
+from cleveragents.application.services.invariant_service import InvariantService
+from cleveragents.domain.models.core.invariant import InvariantScope
+
+# ================================================================
+# Background Steps
+# ================================================================
+
+
+@given("I have an InvariantService instance")
+def step_create_service(context: Any) -> None:
+    """Create a fresh InvariantService instance."""
+    context.service = InvariantService()
+    context.errors: list[Exception] = []
+    context.results: dict[str, Any] = {}
+
+
+# ================================================================
+# Setup Steps
+# ================================================================
+
+
+@given("I have added {count:d} invariants")
+def step_add_invariants(context: Any, count: int) -> None:
+    """Add a specified number of invariants to the service."""
+    context.invariant_ids = []
+    for i in range(count):
+        inv = context.service.add_invariant(
+            text=f"Invariant {i}",
+            scope=InvariantScope.GLOBAL,
+            source_name="test",
+        )
+        context.invariant_ids.append(inv.id)
+
+
+@given("I have added invariants at different scopes")
+def step_add_scoped_invariants(context: Any) -> None:
+    """Add invariants at different scopes for testing."""
+    context.invariant_ids = []
+    scopes = [InvariantScope.GLOBAL, InvariantScope.PROJECT, InvariantScope.PLAN]
+    for i, scope in enumerate(scopes):
+        for j in range(3):
+            inv = context.service.add_invariant(
+                text=f"Invariant {scope.value} {j}",
+                scope=scope,
+                source_name=f"source_{i}",
+            )
+            context.invariant_ids.append(inv.id)
+
+
+# ================================================================
+# Concurrent Add Operations
+# ================================================================
+
+
+@when("{count:d} threads concurrently add invariants with unique texts")
+def step_concurrent_add_unique(context: Any, count: int) -> None:
+    """Add invariants from multiple threads concurrently."""
+    context.errors = []
+    context.added_ids = []
+    lock = threading.Lock()
+
+    def add_invariant(thread_id: int) -> None:
+        try:
+            inv = context.service.add_invariant(
+                text=f"Thread {thread_id} Invariant",
+                scope=InvariantScope.GLOBAL,
+                source_name="test",
+            )
+            with lock:
+                context.added_ids.append(inv.id)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=count) as executor:
+        futures = [executor.submit(add_invariant, i) for i in range(count)]
+        for future in as_completed(futures):
+            future.result()
+
+
+@when("{count:d} threads concurrently add invariants")
+def step_concurrent_add(context: Any, count: int) -> None:
+    """Add invariants from multiple threads concurrently."""
+    context.errors = []
+    context.added_ids = []
+    lock = threading.Lock()
+
+    def add_invariant(thread_id: int) -> None:
+        try:
+            inv = context.service.add_invariant(
+                text=f"Invariant from thread {thread_id}",
+                scope=InvariantScope.GLOBAL,
+                source_name="test",
+            )
+            with lock:
+                context.added_ids.append(inv.id)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=count) as executor:
+        futures = [executor.submit(add_invariant, i) for i in range(count)]
+        for future in as_completed(futures):
+            future.result()
+
+
+# ================================================================
+# Concurrent List Operations
+# ================================================================
+
+
+@when("{count:d} threads concurrently list invariants")
+def step_concurrent_list(context: Any, count: int) -> None:
+    """List invariants from multiple threads concurrently."""
+    context.errors = []
+    context.list_results = []
+    lock = threading.Lock()
+
+    def list_invariants(thread_id: int) -> None:
+        try:
+            result = context.service.list_invariants()
+            with lock:
+                context.list_results.append(result)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=count) as executor:
+        futures = [executor.submit(list_invariants, i) for i in range(count)]
+        for future in as_completed(futures):
+            future.result()
+
+
+@when(
+    "{add_count:d} threads add invariants while {list_count:d} threads list invariants concurrently"
+)
+def step_concurrent_add_and_list(context: Any, add_count: int, list_count: int) -> None:
+    """Add and list invariants concurrently from different threads."""
+    context.errors = []
+    context.added_ids = []
+    context.list_results = []
+    lock = threading.Lock()
+
+    def add_invariant(thread_id: int) -> None:
+        try:
+            inv = context.service.add_invariant(
+                text=f"Invariant from add thread {thread_id}",
+                scope=InvariantScope.GLOBAL,
+                source_name="test",
+            )
+            with lock:
+                context.added_ids.append(inv.id)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    def list_invariants(thread_id: int) -> None:
+        try:
+            result = context.service.list_invariants()
+            with lock:
+                context.list_results.append(result)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=add_count + list_count) as executor:
+        futures = []
+        for i in range(add_count):
+            futures.append(executor.submit(add_invariant, i))
+        for i in range(list_count):
+            futures.append(executor.submit(list_invariants, i))
+        for future in as_completed(futures):
+            future.result()
+
+
+# ================================================================
+# Concurrent Remove Operations
+# ================================================================
+
+
+@when("{count:d} threads concurrently remove different invariants")
+def step_concurrent_remove(context: Any, count: int) -> None:
+    """Remove invariants from multiple threads concurrently."""
+    context.errors = []
+    context.removed_ids = []
+    lock = threading.Lock()
+
+    def remove_invariant(thread_id: int) -> None:
+        try:
+            if thread_id < len(context.invariant_ids):
+                inv_id = context.invariant_ids[thread_id]
+                result = context.service.remove_invariant(inv_id)
+                with lock:
+                    context.removed_ids.append(result.id)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=count) as executor:
+        futures = [executor.submit(remove_invariant, i) for i in range(count)]
+        for future in as_completed(futures):
+            future.result()
+
+
+@when(
+    "{remove_count:d} threads remove invariants while {list_count:d} threads list invariants concurrently"
+)
+def step_concurrent_remove_and_list(
+    context: Any, remove_count: int, list_count: int
+) -> None:
+    """Remove and list invariants concurrently from different threads."""
+    context.errors = []
+    context.removed_ids = []
+    context.list_results = []
+    lock = threading.Lock()
+    remove_index = 0
+
+    def remove_invariant(thread_id: int) -> None:
+        nonlocal remove_index
+        try:
+            with lock:
+                if remove_index < len(context.invariant_ids):
+                    inv_id = context.invariant_ids[remove_index]
+                    remove_index += 1
+                else:
+                    return
+            result = context.service.remove_invariant(inv_id)
+            with lock:
+                context.removed_ids.append(result.id)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    def list_invariants(thread_id: int) -> None:
+        try:
+            result = context.service.list_invariants()
+            with lock:
+                context.list_results.append(result)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=remove_count + list_count) as executor:
+        futures = []
+        for i in range(remove_count):
+            futures.append(executor.submit(remove_invariant, i))
+        for i in range(list_count):
+            futures.append(executor.submit(list_invariants, i))
+        for future in as_completed(futures):
+            future.result()
+
+
+# ================================================================
+# Concurrent Enforcement Operations
+# ================================================================
+
+
+@when("{count:d} threads concurrently enforce the same invariants")
+def step_concurrent_enforce(context: Any, count: int) -> None:
+    """Enforce invariants from multiple threads concurrently."""
+    context.errors = []
+    context.enforcement_records = []
+    lock = threading.Lock()
+
+    # Get the invariants to enforce
+    invariants = context.service.list_invariants()
+
+    def enforce_invariants(thread_id: int) -> None:
+        try:
+            records = context.service.enforce_invariants(
+                plan_id=f"plan_{thread_id}",
+                invariants=invariants,
+            )
+            with lock:
+                context.enforcement_records.extend(records)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=count) as executor:
+        futures = [executor.submit(enforce_invariants, i) for i in range(count)]
+        for future in as_completed(futures):
+            future.result()
+
+
+@when("{count:d} threads concurrently enforce invariants")
+def step_concurrent_enforce_all(context: Any, count: int) -> None:
+    """Enforce invariants from multiple threads concurrently."""
+    context.errors = []
+    context.enforcement_records = []
+    lock = threading.Lock()
+
+    # Get the invariants to enforce
+    invariants = context.service.list_invariants()
+
+    def enforce_invariants(thread_id: int) -> None:
+        try:
+            records = context.service.enforce_invariants(
+                plan_id=f"plan_{thread_id}",
+                invariants=invariants,
+            )
+            with lock:
+                context.enforcement_records.extend(records)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=count) as executor:
+        futures = [executor.submit(enforce_invariants, i) for i in range(count)]
+        for future in as_completed(futures):
+            future.result()
+
+
+# ================================================================
+# Mixed Concurrent Operations
+# ================================================================
+
+
+@when(
+    "{add_count:d} threads add, {list_count:d} threads list, {remove_count:d} threads remove, and {enforce_count:d} threads enforce concurrently"
+)
+def step_concurrent_mixed(
+    context: Any,
+    add_count: int,
+    list_count: int,
+    remove_count: int,
+    enforce_count: int,
+) -> None:
+    """Perform mixed operations concurrently from different threads."""
+    context.errors = []
+    context.added_ids = []
+    context.list_results = []
+    context.removed_ids = []
+    context.enforcement_records = []
+    lock = threading.Lock()
+    remove_index = 0
+
+    def add_invariant(thread_id: int) -> None:
+        try:
+            inv = context.service.add_invariant(
+                text=f"Add thread {thread_id}",
+                scope=InvariantScope.GLOBAL,
+                source_name="test",
+            )
+            with lock:
+                context.added_ids.append(inv.id)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    def list_invariants(thread_id: int) -> None:
+        try:
+            result = context.service.list_invariants()
+            with lock:
+                context.list_results.append(result)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    def remove_invariant(thread_id: int) -> None:
+        nonlocal remove_index
+        try:
+            with lock:
+                if remove_index < len(context.invariant_ids):
+                    inv_id = context.invariant_ids[remove_index]
+                    remove_index += 1
+                else:
+                    return
+            result = context.service.remove_invariant(inv_id)
+            with lock:
+                context.removed_ids.append(result.id)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    def enforce_invariants(thread_id: int) -> None:
+        try:
+            invariants = context.service.list_invariants()
+            records = context.service.enforce_invariants(
+                plan_id=f"plan_{thread_id}",
+                invariants=invariants,
+            )
+            with lock:
+                context.enforcement_records.extend(records)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    total_threads = add_count + list_count + remove_count + enforce_count
+    with ThreadPoolExecutor(max_workers=total_threads) as executor:
+        futures = []
+        for i in range(add_count):
+            futures.append(executor.submit(add_invariant, i))
+        for i in range(list_count):
+            futures.append(executor.submit(list_invariants, i))
+        for i in range(remove_count):
+            futures.append(executor.submit(remove_invariant, i))
+        for i in range(enforce_count):
+            futures.append(executor.submit(enforce_invariants, i))
+        for future in as_completed(futures):
+            future.result()
+
+
+@when(
+    "{thread_count:d} threads perform random operations (add/list/remove/enforce) for {iterations:d} iterations each"
+)
+def step_concurrent_random_operations(
+    context: Any, thread_count: int, iterations: int
+) -> None:
+    """Perform random operations from multiple threads."""
+    import random
+
+    context.errors = []
+    context.added_ids = []
+    context.list_results = []
+    context.removed_ids = []
+    context.enforcement_records = []
+    lock = threading.Lock()
+
+    def random_operations(thread_id: int) -> None:
+        try:
+            for _ in range(iterations):
+                op = random.choice(["add", "list", "remove", "enforce"])
+                if op == "add":
+                    inv = context.service.add_invariant(
+                        text=f"Thread {thread_id} Invariant",
+                        scope=InvariantScope.GLOBAL,
+                        source_name="test",
+                    )
+                    with lock:
+                        context.added_ids.append(inv.id)
+                elif op == "list":
+                    result = context.service.list_invariants()
+                    with lock:
+                        context.list_results.append(result)
+                elif op == "remove" and context.added_ids:
+                    with lock:
+                        if context.added_ids:
+                            inv_id = context.added_ids.pop(0)
+                    try:
+                        context.service.remove_invariant(inv_id)
+                        with lock:
+                            context.removed_ids.append(inv_id)
+                    except Exception:
+                        pass
+                elif op == "enforce":
+                    invariants = context.service.list_invariants()
+                    if invariants:
+                        records = context.service.enforce_invariants(
+                            plan_id=f"plan_{thread_id}",
+                            invariants=invariants,
+                        )
+                        with lock:
+                            context.enforcement_records.extend(records)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=thread_count) as executor:
+        futures = [executor.submit(random_operations, i) for i in range(thread_count)]
+        for future in as_completed(futures):
+            future.result()
+
+
+# ================================================================
+# Concurrent Effective Invariants
+# ================================================================
+
+
+@when("{count:d} threads concurrently get effective invariants")
+def step_concurrent_effective(context: Any, count: int) -> None:
+    """Get effective invariants from multiple threads concurrently."""
+    context.errors = []
+    context.effective_results = []
+    lock = threading.Lock()
+
+    def get_effective(thread_id: int) -> None:
+        try:
+            result = context.service.get_effective_invariants()
+            with lock:
+                context.effective_results.append(result)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=count) as executor:
+        futures = [executor.submit(get_effective, i) for i in range(count)]
+        for future in as_completed(futures):
+            future.result()
+
+
+@when(
+    "{add_count:d} threads add invariants while {get_count:d} threads get effective invariants concurrently"
+)
+def step_concurrent_add_and_effective(
+    context: Any, add_count: int, get_count: int
+) -> None:
+    """Add invariants and get effective invariants concurrently."""
+    context.errors = []
+    context.added_ids = []
+    context.effective_results = []
+    lock = threading.Lock()
+
+    def add_invariant(thread_id: int) -> None:
+        try:
+            inv = context.service.add_invariant(
+                text=f"Invariant {thread_id}",
+                scope=InvariantScope.GLOBAL,
+                source_name="test",
+            )
+            with lock:
+                context.added_ids.append(inv.id)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    def get_effective(thread_id: int) -> None:
+        try:
+            result = context.service.get_effective_invariants()
+            with lock:
+                context.effective_results.append(result)
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    with ThreadPoolExecutor(max_workers=add_count + get_count) as executor:
+        futures = []
+        for i in range(add_count):
+            futures.append(executor.submit(add_invariant, i))
+        for i in range(get_count):
+            futures.append(executor.submit(get_effective, i))
+        for future in as_completed(futures):
+            future.result()
+
+
+# ================================================================
+# Stress Tests
+# ================================================================
+
+
+@when(
+    "{thread_count:d} threads perform operations concurrently for {iterations:d} iterations each"
+)
+def step_stress_test(context: Any, thread_count: int, iterations: int) -> None:
+    """Perform stress test with many threads and iterations."""
+    import random
+
+    context.errors = []
+    context.operation_count = 0
+    lock = threading.Lock()
+
+    def stress_operations(thread_id: int) -> None:
+        try:
+            for i in range(iterations):
+                op = random.choice(["add", "list", "enforce"])
+                if op == "add":
+                    context.service.add_invariant(
+                        text=f"Thread {thread_id} Iter {i}",
+                        scope=InvariantScope.GLOBAL,
+                        source_name="test",
+                    )
+                elif op == "list":
+                    context.service.list_invariants()
+                elif op == "enforce":
+                    invariants = context.service.list_invariants()
+                    if invariants:
+                        context.service.enforce_invariants(
+                            plan_id=f"plan_{thread_id}_{i}",
+                            invariants=invariants,
+                        )
+                with lock:
+                    context.operation_count += 1
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    start_time = time.time()
+    with ThreadPoolExecutor(max_workers=thread_count) as executor:
+        futures = [executor.submit(stress_operations, i) for i in range(thread_count)]
+        for future in as_completed(futures):
+            future.result()
+    context.elapsed_time = time.time() - start_time
+
+
+@when("{count:d} threads perform rapid sequential operations")
+def step_rapid_operations(context: Any, count: int) -> None:
+    """Perform rapid sequential operations from multiple threads."""
+    context.errors = []
+    context.operation_count = 0
+    lock = threading.Lock()
+
+    def rapid_ops(thread_id: int) -> None:
+        try:
+            for i in range(100):
+                context.service.add_invariant(
+                    text=f"Thread {thread_id} Op {i}",
+                    scope=InvariantScope.GLOBAL,
+                    source_name="test",
+                )
+                context.service.list_invariants()
+                with lock:
+                    context.operation_count += 1
+        except Exception as e:
+            with lock:
+                context.errors.append(e)
+
+    start_time = time.time()
+    with ThreadPoolExecutor(max_workers=count) as executor:
+        futures = [executor.submit(rapid_ops, i) for i in range(count)]
+        for future in as_completed(futures):
+            future.result()
+    context.elapsed_time = time.time() - start_time
+
+
+# ================================================================
+# Assertion Steps
+# ================================================================
+
+
+@then("all invariants should be stored without data loss")
+def step_assert_no_data_loss(context: Any) -> None:
+    """Assert that all added invariants are stored."""
+    assert len(context.added_ids) > 0, "No invariants were added"
+    all_invariants = context.service.list_invariants()
+    assert len(all_invariants) >= len(context.added_ids), (
+        f"Expected at least {len(context.added_ids)} invariants, "
+        f"got {len(all_invariants)}"
+    )
+
+
+@then("the total count should be {expected:d}")
+def step_assert_count(context: Any, expected: int) -> None:
+    """Assert the total count of invariants."""
+    all_invariants = context.service.list_invariants()
+    assert len(all_invariants) == expected, (
+        f"Expected {expected} invariants, got {len(all_invariants)}"
+    )
+
+
+@then("no RuntimeError should be raised")
+def step_assert_no_runtime_error(context: Any) -> None:
+    """Assert that no RuntimeError was raised."""
+    runtime_errors = [e for e in context.errors if isinstance(e, RuntimeError)]
+    assert len(runtime_errors) == 0, (
+        f"Expected no RuntimeError, got {len(runtime_errors)}: {runtime_errors}"
+    )
+
+
+@then("all invariants should be retrievable")
+def step_assert_retrievable(context: Any) -> None:
+    """Assert that all invariants can be retrieved."""
+    all_invariants = context.service.list_invariants()
+    assert len(all_invariants) > 0, "No invariants are retrievable"
+
+
+@then("all threads should complete successfully")
+def step_assert_all_threads_complete(context: Any) -> None:
+    """Assert that all threads completed without errors."""
+    assert len(context.errors) == 0, (
+        f"Expected no errors, got {len(context.errors)}: {context.errors}"
+    )
+
+
+@then("each thread should see consistent data")
+def step_assert_consistent_data(context: Any) -> None:
+    """Assert that each thread saw consistent data."""
+    if len(context.list_results) > 1:
+        first_result = context.list_results[0]
+        for result in context.list_results[1:]:
+            assert len(result) == len(first_result), (
+                f"Inconsistent data: expected {len(first_result)} invariants, "
+                f"got {len(result)}"
+            )
+
+
+@then("the final count should match the number of adds")
+def step_assert_final_count_matches_adds(context: Any) -> None:
+    """Assert that the final count matches the number of adds."""
+    all_invariants = context.service.list_invariants()
+    assert len(all_invariants) == len(context.added_ids), (
+        f"Expected {len(context.added_ids)} invariants, got {len(all_invariants)}"
+    )
+
+
+@then("all removals should succeed")
+def step_assert_removals_succeed(context: Any) -> None:
+    """Assert that all removals succeeded."""
+    assert len(context.errors) == 0, (
+        f"Expected no errors, got {len(context.errors)}: {context.errors}"
+    )
+
+
+@then("the remaining count should be {expected:d}")
+def step_assert_remaining_count(context: Any, expected: int) -> None:
+    """Assert the remaining count of invariants."""
+    all_invariants = context.service.list_invariants()
+    assert len(all_invariants) == expected, (
+        f"Expected {expected} remaining invariants, got {len(all_invariants)}"
+    )
+
+
+@then("all enforcement records should be stored")
+def step_assert_records_stored(context: Any) -> None:
+    """Assert that all enforcement records are stored."""
+    assert len(context.enforcement_records) > 0, "No enforcement records were stored"
+
+
+@then("the total record count should be {expected:d}")
+def step_assert_record_count(context: Any, expected: int) -> None:
+    """Assert the total count of enforcement records."""
+    assert len(context.enforcement_records) == expected, (
+        f"Expected {expected} records, got {len(context.enforcement_records)}"
+    )
+
+
+@then("all {expected:d} enforcement records should be stored")
+def step_assert_all_records_stored(context: Any, expected: int) -> None:
+    """Assert that all enforcement records are stored."""
+    assert len(context.enforcement_records) == expected, (
+        f"Expected {expected} records, got {len(context.enforcement_records)}"
+    )
+
+
+@then("no records should be duplicated or lost")
+def step_assert_no_duplicates(context: Any) -> None:
+    """Assert that no records are duplicated or lost."""
+    # Check that we have the expected number of records
+    assert len(context.enforcement_records) > 0, "No records stored"
+
+
+@then("no errors should occur in thread safety test")
+def step_assert_no_errors_thread_safety(context: Any) -> None:
+    """Assert that no errors occurred in thread safety test."""
+    assert len(context.errors) == 0, (
+        f"Expected no errors, got {len(context.errors)}: {context.errors}"
+    )
+
+
+@then("the final state should be consistent")
+def step_assert_final_state_consistent(context: Any) -> None:
+    """Assert that the final state is consistent."""
+    # Verify we can list all invariants without errors
+    all_invariants = context.service.list_invariants()
+    assert isinstance(all_invariants, list), "list_invariants should return a list"
+
+
+@then("all operations should complete successfully")
+def step_assert_all_operations_complete(context: Any) -> None:
+    """Assert that all operations completed successfully."""
+    assert len(context.errors) == 0, (
+        f"Expected no errors, got {len(context.errors)}: {context.errors}"
+    )
+
+
+@then("the final invariant count should match expected value")
+def step_assert_final_invariant_count(context: Any) -> None:
+    """Assert that the final invariant count is valid."""
+    all_invariants = context.service.list_invariants()
+    assert isinstance(all_invariants, list), "list_invariants should return a list"
+    assert len(all_invariants) >= 0, "Invariant count should be non-negative"
+
+
+@then("all enforcement records should be accounted for")
+def step_assert_records_accounted(context: Any) -> None:
+    """Assert that all enforcement records are accounted for."""
+    assert len(context.enforcement_records) >= 0, "Record count should be non-negative"
+
+
+@then("all threads should see consistent merged results")
+def step_assert_consistent_merged_results(context: Any) -> None:
+    """Assert that all threads see consistent merged results."""
+    if len(context.effective_results) > 1:
+        first_result = context.effective_results[0]
+        for result in context.effective_results[1:]:
+            assert len(result) == len(first_result), (
+                f"Inconsistent merged results: expected {len(first_result)}, "
+                f"got {len(result)}"
+            )
+
+
+@then("the results should be consistent")
+def step_assert_results_consistent(context: Any) -> None:
+    """Assert that the results are consistent."""
+    assert len(context.errors) == 0, (
+        f"Expected no errors, got {len(context.errors)}: {context.errors}"
+    )
+
+
+@then("no deadlocks should occur")
+def step_assert_no_deadlocks(context: Any) -> None:
+    """Assert that no deadlocks occurred."""
+    # If we got here, no deadlocks occurred
+    assert True
+
+
+@then("the final state should be valid")
+def step_assert_final_state_valid(context: Any) -> None:
+    """Assert that the final state is valid."""
+    all_invariants = context.service.list_invariants()
+    assert isinstance(all_invariants, list), "list_invariants should return a list"
+
+
+@then("all operations should complete within reasonable time")
+def step_assert_reasonable_time(context: Any) -> None:
+    """Assert that operations completed within reasonable time."""
+    # Reasonable time: less than 30 seconds for the test
+    assert context.elapsed_time < 30, (
+        f"Operations took too long: {context.elapsed_time:.2f}s"
+    )
+
+
+@then("no thread should be starved")
+def step_assert_no_starvation(context: Any) -> None:
+    """Assert that no thread was starved."""
+    # If we got here and all operations completed, no starvation occurred
+    assert context.operation_count > 0, "No operations were performed"

src/cleveragents/application/services/invariant_service.py

@@ -14,11 +14,19 @@ a dict keyed by invariant ID.
 Effective invariants are computed using plan > project > global order.
 See ``merge_invariants`` for de-duplication semantics.
 
+## Thread Safety
+
+All public methods are protected by a reentrant lock (``threading.RLock``)
+to ensure safe concurrent access to ``_invariants`` dict and
+``_enforcement_records`` list. This allows the service to be safely shared
+across threads during parallel plan execution.
+
 Based on ``docs/specification.md`` and implementation plan Stage M3.5.
 """
 
 from __future__ import annotations
 
+import threading
 from typing import TYPE_CHECKING
 
 import structlog
@@ -54,6 +62,7 @@ class InvariantService:
         Args:
             event_bus: Optional EventBus for domain event emission.
         """
+        self._lock: threading.RLock = threading.RLock()
         self._invariants: dict[str, Invariant] = {}
         self._enforcement_records: list[InvariantEnforcementRecord] = []
         self._logger = logger.bind(service="invariant")
@@ -95,7 +104,8 @@ 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,13 +137,14 @@ 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
 
@@ -152,15 +163,16 @@ 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,21 +195,22 @@ 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
-            for inv in active
-            if inv.scope == InvariantScope.PLAN
-            and (plan_id is None or inv.source_name == plan_id)
-        ]
-        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]
+            plan_invs = [
+                inv
+                for inv in active
+                if inv.scope == InvariantScope.PLAN
+                and (plan_id is None or inv.source_name == plan_id)
+            ]
+            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, project_invs, global_invs)
 
@@ -264,7 +277,8 @@ 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,