diff --git a/CHANGELOG.md b/CHANGELOG.md index 71833a437..7a2a1b9c8 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index ff96cca40..210de5505 100644 --- a/CONTRIBUTORS.md +++ b/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. diff --git a/features/invariant_service_thread_safety.feature b/features/invariant_service_thread_safety.feature new file mode 100644 index 000000000..faa0d4457 --- /dev/null +++ b/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 diff --git a/features/steps/invariant_service_thread_safety_steps.py b/features/steps/invariant_service_thread_safety_steps.py new file mode 100644 index 000000000..96eaf78cb --- /dev/null +++ b/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)}" + ) diff --git a/src/cleveragents/application/services/invariant_service.py b/src/cleveragents/application/services/invariant_service.py index 10e587ce8..8c2d058d1 100644 --- a/src/cleveragents/application/services/invariant_service.py +++ b/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] diff --git a/tests/application/services/test_invariant_service_thread_safety.py b/tests/application/services/test_invariant_service_thread_safety.py new file mode 100644 index 000000000..95b9ea2a1 --- /dev/null +++ b/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)